29

如何定义异步设备并在异步测试中使用它们？

以下代码都在同一个文件中，失败得很惨。测试运行程序是否明确调用了夹具并且没有等待？

@pytest.fixture
async def create_x(api_client):
    x_id = await add_x(api_client)
    return api_client, x_id

async def test_app(create_x, auth):
    api_client, x_id = create_x
    resp = await api_client.get(f'my_res/{x_id}', headers=auth)
    assert resp.status == web.HTTPOk.status_code

生产

==================================== ERRORS ====================================
_____________ ERROR at setup of test_app[pyloop] ______________

api_client = <aiohttp.test_utils.TestClient object at 0x7f27ec954f60>

    @pytest.fixture
    async def create_x(api_client):
>       x_id = await add_x(api_client)
...
... cannot show the full trace and pathnames sorry
...    

in __await__
    ret = yield from self._coro /home/mbb/.pyenv/versions/3.6.3/envs/mr/lib/python3.6/site-packages/aiohttp/test_utils.py:245: in request
    method, self.make_url(path), *args, **kwargs /home/mbb/.pyenv/versions/mr/lib/python3.6/site-packages/aiohttp/helpers.py:104: in __iter__
    ret = yield from self._coro /home/mbb/.pyenv/versions/mr/lib/python3.6/site-packages/aiohttp/client.py:221: in _request
    with timer:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <aiohttp.helpers.TimerContext object at 0x7f27ec9875c0>

    def __enter__(self):
        task = current_task(loop=self._loop)

        if task is None:
>           raise RuntimeError('Timeout context manager should be used '
                               'inside a task') E           RuntimeError: Timeout context manager should be used inside a task

/home/mbb/.pyenv/versions/mr/lib/python3.6/site-packages/aiohttp/helpers.py:717: RuntimeError
=========================== 1 error in 1.74 seconds ============================ Process finished with exit code 0

我知道我可能会做

@pytest.fixture
def create_x(loop, api_client):
    x_id = loop.run_until_complete(add_x(api_client))
    return api_client, x_id

但我想知道是否存在更简单/最优雅的方式。我在 pytest、pytest-asyncio、pytest-aiohttp 的项目页面中都找不到清晰简单的示例/解释。

我使用 Python 3.6.3、pytest 3.4.2、pytest-asyncio 0.8.0 和 pytest-aiohttp 0.3.0

非常感谢您的帮助

4

4 回答 4

52

您只需要将测试标记为异步

@pytest.mark.asyncio
async def test_app(create_x, auth):
    api_client, x_id = create_x
    resp = await api_client.get(f'my_res/{x_id}', headers=auth)
    assert resp.status == web.HTTPOk.status_code

这告诉 pytest 在事件循环中运行测试，而不是直接调用它。

灯具可以标记为正常

@pytest.fixture
async def create_x(api_client):
    x_id = await add_x(api_client)
    return api_client, x_id
于 2018-04-20T11:21:10.660 回答
13

PyTest 本身不支持协程函数，因此您需要为其安装额外的框架

  • pytest-aiohttp
  • pytest-asyncio
  • pytest-三重奏
  • pytest-tornasync

如果你使用pytest-aiohttp，你的问题就这样解决了

import asyncio
import pytest

from app import db


url = 'postgresql://postgres:postgres@localhost:5432'


@pytest.fixture(scope='session')
def loop():
    return asyncio.get_event_loop()


@pytest.fixture(scope='session', autouse=True)
async def prepare_db(loop):
    async with db.with_bind(f'{url}/postgres') as engine:
        await engine.status(db.text('CREATE DATABASE test_db'))

    await db.set_bind(f'{url}/test_db')
    await db.gino.create_all()

    yield
    await db.bind.close()

    async with db.with_bind(f'{url}/postgres') as engine:
        await engine.status(db.text('DROP DATABASE test_db'))

主要思想是使用同步loop-fixture，异步夹具将使用它

于 2019-08-06T20:25:09.640 回答
1

使用 pytest-asyncio 如果将以下内容添加到 pytest.ini 配置文件中，则不需要标记：

# pytest.ini
[pytest]
...
asyncio_mode=auto
...

添加后，您可以创建一个异步夹具和测试，只需要异步语法

# test_file.py
import pytest
from some-aio-library import AsyncClient

@pytest.fixture
async def test_async_client():
    async with AsyncClient() as client:
        yield client


async def test_something(test_async_client: AsyncClient):
    result = await test_async_client.get_some_data()
    
    assert result.data == "some data"
于 2022-02-02T00:35:44.527 回答
1

将其放入conftest.py以标记所有异步函数：

import inspect
import pytest

def pytest_collection_modifyitems(config, items):
    for item in items:
        if inspect.iscoroutinefunction(item.function):
            item.add_marker(pytest.mark.asyncio)
于 2021-11-14T22:06:36.973 回答