Async fixtures with pytest

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? producing I know I could probably do

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

3

3 Answers

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
4

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

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

ncG1vNJzZmirpJawrLvVnqmfpJ%2Bse6S7zGiorp2jqbawutJoa3JxY2uEc4COmqqyppNis6rE066pnqtdrLa1tIypsK2do6k%3D

 Share!