Async fixtures with pytest

Question

How do I define async fixtures and use them in async tests?

The following code, all in the same file, fails miserably. Is the fixture called plainly by the test runner and not awaited?

@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

producing

==================================== 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

I know I could probably do

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

but I'd like to know if an easier/most elegant way exists. I cannot find a clear and simple example/explanation in the project pages of either pytest, pytest-asyncio, pytest-aiohttp.

I use Python 3.6.3, pytest 3.4.2, pytest-asyncio 0.8.0 and pytest-aiohttp 0.3.0

Many thanks for your kind help

Answer 1

You only need to mark your tests as async

@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

This tells pytest to run the test inside an event loop rather than calling it directly.

The fixtures can be marked as normal

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

Answer 2

Coroutine functions are not natively supported by PyTest, so you need to install additional framework for it

• pytest-aiohttp
• pytest-asyncio
• pytest-trio
• pytest-tornasync

If you use pytest-aiohttp, your problem solves in this way

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'))

Main idea is using synchronous loop-fixture which will be used by async fixtures

Answer 3

with pytest-asyncio you do not need to mark if you add the following to your pytest.ini config file:

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

with this added you can create an async fixture and tests with only needing async syntax

# 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"

Answer 4

Drop this in conftest.py to mark all async functions:

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)