0

我正在尝试使用test_client()使用 pytest测试 Flask REST-API 端点。但我得到一个错误说

> RuntimeError: You cannot use AsyncToSync in the same thread as an async event loop - just await the async function directly.

谁能解释我为什么会发生这种情况以及避免此错误的解决方案是什么。

测试功能：

import pytest
from  unittest import mock

from flask import request
from app import create_app
from app.base.views import views

app = create_app()


async def save_token_(sugar, payload, instance, bool, domain_id):
    return {'valid':True}

payload = {
    "password": 'password', 
    "username": 'crm_admin',
    "grant_type": "password"
}

@pytest.mark.asyncio
async def test_post_sugar_token(monkeypatch, aiohttp_client, loop):
    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, 89)]

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

    ##-# using test_client()
    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}
# # ----------------------------------------------------------------------

错误信息

4

1 回答 1

0

我遇到了同样的问题并在烧瓶的 github repo 上打开了一张票： https ://github.com/pallets/flask/issues/4375 。他们友好地解释了这个问题并提供了解决方法。

简而言之，flask 可以处理异步视图，flask 的 test_client 可以在异步上下文中使用，但不能同时使用两者。

引用github问题：

这种情况下的问题是 Flask 代码库与 asyncio 不兼容，因此您无法在 asyncio 循环内运行测试客户端。将单元测试编写为异步测试实际上并没有什么好处，因为 Flask 本身不是异步的。

这是建议的解决方法，针对您的示例稍作调整：

@pytest.mark.asyncio
async def test_post_sugar_token():
    # ... same code as before

    ##-# using test_client()
    def sync_test():
        with app.test_client() as 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}

    loop = asyncio.get_running_loop()
    await loop.run_in_executor(None, sync_test)
于 2021-12-10T18:12:38.240 回答