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

Maryammaryann

What does the api_client fixture look like? Is it also an async fixture? – Pecoraro

@Pecoraro it's an object that offers http functionality, as can be spotted in the test await api_client.get( – Maryammaryann

See this answer from @ivan-medina below for more recent answer. – Toulouse

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

Aluminothermy

Seems odd that this is necessary. The function is async, why would pytest NOT run it in an event loop by default? – Wealthy

@mblakesley Markers can be made optional, github.com/pytest-dev/pytest-asyncio#auto-mode – Siobhansion

If you do not enable auto-mode, you may need to replace @pytest.fixture with @pytest_asyncio.fixture per stackoverflow.com/questions/72996818 – Greensickness

@JoelSullivan this was the fix for me, thank you! – Swearword

Answer 2

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"

Selfseeker

I installed pytest-asyncio, but still get this warning: PytestConfigWarning: Unknown config option: asyncio_mode – Jingo

Answer 3

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

Cacia

Answer 4

For me:

pytest --asyncio-mode=auto solved the issue

You can configure this parameter in your .coveragerc or your pytest.ini

Plotinus