在async异步函数中使用Flask test_client()时报RuntimeError怎么解决？

• 测试函数添加了@pytest.mark.asyncio装饰器，pytest运行该用例时会提前在当前线程启动一个asyncio事件循环
• Flask同步test_client()底层依赖asgiref库的AsyncToSync工具适配异步视图逻辑，AsyncToSync运行时要求当前线程没有正在运行的事件循环（它需要自行创建独立的事件循环运行异步代码），二者直接冲突就会抛出该报错
• 代码中直接同步调用client.post()没有加await，也是触发冲突的直接原因

方案1：拆分测试用例（最稳妥） #

把异步逻辑测试和Flask接口测试拆成两个独立用例，接口测试用例不加@pytest.mark.asyncio，避免提前启动事件循环：

# 第一个用例：单独测试异步save_token逻辑
@pytest.mark.asyncio
async def test_save_token_logic(monkeypatch):
    mock_save_token = mock.AsyncMock(name = "mock_save_token")
    mock_save_token.return_value = await save_token_(None, payload, 'domain.org', True, 89)
    monkeypatch.setattr(views, 'save_token', mock_save_token)
    await views.save_token(None, payload, 'domain.org', True, 9)
    # 注意原代码此处参数匹配有误，调用传的domain_id是9，原期望写的是89，已修正
    assert mock_save_token.call_args_list == [mock.call(None, payload, 'domain.org', True, 9)]

# 第二个用例：同步测试接口
def test_post_sugar_token(monkeypatch):
    mock_save_token = mock.AsyncMock(name = "mock_save_token")
    # 同步用例里异步mock返回值直接赋值即可，无需await
    mock_save_token.return_value = {'valid':True}
    monkeypatch.setattr(views, 'save_token', mock_save_token)

    # 原代码请求头拼写错误，Autherization修正为Authorization
    headers = {'Authorization': 'ehrdmek2492.fkeompvmw.04294002'}
    data = {
        'password': '12345',
        'key':'Hi',
        'instance': 'my.domain',
        'domain_id': 1
    }

    client = app.test_client()
    res = client.post('/token/sugar/', data = data, headers = headers)

    assert res.status_code == 200
    assert res.content_type == 'application/json'
    assert res.json == {'valid':True}

方案2：使用Flask异步测试客户端 #

如果需要在同一个异步用例里完成测试，改用Flask的异步调用方式，通过await执行接口请求：

@pytest.mark.asyncio
async def test_post_sugar_token(monkeypatch):
    mock_save_token = mock.AsyncMock(name = "mock_save_token")
    mock_save_token.return_value = await save_token_(None, payload, 'domain.org', True, 89)
    monkeypatch.setattr(views, 'save_token', mock_save_token)

    await views.save_token(None, payload, 'domain.org', True, 9)
    assert mock_save_token.call_args_list == [mock.call(None, payload, 'domain.org', True, 9)]

    headers = {'Authorization': 'ehrdmek2492.fkeompvmw.04294002'}
    data = {
        'password': '12345',
        'key':'Hi',
        'instance': 'my.domain',
        'domain_id': 1
    }

    # 异步模式下用async with创建测试客户端，await调用请求方法
    async with app.test_client() as client:
        res = await client.post('/token/sugar/', data = data, headers = headers)

    assert res.status_code == 200
    assert res.content_type == 'application/json'
    assert res.json == {'valid':True}

内容的提问来源于stack exchange，提问作者Anuruddha Bandara