pyguides

Testing Async Code with pytest

Async code needs async tests. A regular def test_* function can’t await a coroutine, and async def test_* without the right pytest plugin just fails. pytest-asyncio solves this — it handles the event loop and gives you familiar pytest patterns for async code.

Installing pytest-asyncio

pip install pytest-asyncio

That’s it. Once installed, pytest automatically recognises async def tests and handles them correctly.

Your First Async Test

import pytest

async def fetch_user(user_id: int) -> dict:
    await asyncio.sleep(0.1)  # simulates a database call
    return {"id": user_id, "name": "Alice"}

@pytest.mark.asyncio
async def test_fetch_user():
    result = await fetch_user(42)
    assert result == {"id": 42, "name": "Alice"}

The @pytest.mark.asyncio decorator tells pytest to run this test in an async context. Without it, pytest ignores async tests silently in older versions, or raises an error in newer ones.

Running the Test

pytest test_example.py -v

Async Fixtures

Just like regular fixtures, async fixtures use @pytest.fixture. Declare the return type as an async generator if you need teardown:

@pytest.fixture
async def db_connection():
    conn = await connect_to_db("localhost", 5432)
    yield conn
    await conn.close()

Use it in async tests:

@pytest.mark.asyncio
async def test_query(db_connection):
    result = await db_connection.query("SELECT * FROM users")
    assert len(result) > 0

Async fixtures can depend on other fixtures, async or otherwise:

@pytest.fixture
def mock_api():
    return MagicMock()

@pytest.fixture
async def async_db(mock_api):
    return AsyncDBClient(mock_api)

@pytest.mark.asyncio
async def test_integration(async_db, mock_api):
    result = await async_db.fetch_all()
    mock_api.query.assert_called_once()

Fixture Scope with async

By default, async fixtures are function-scoped — created fresh for each test. For module or session scope, specify it explicitly:

@pytest.fixture(scope="module")
async def shared_db():
    conn = await async_connect()
    yield conn
    await conn.cleanup()

This is useful when connecting to a database or service is expensive and you’d like to reuse the connection across tests in the same module.

Patching in Async Tests

unittest.mock.patch works in async tests — just make sure your assertions are also async:

@pytest.mark.asyncio
async def test_fetch_user_with_fallback(mock_get):
    mock_get.return_value.json.return_value = {"id": 1, "name": "Bob"}

    result = await fetch_user_with_api(1, fallback_name="Unknown")
    assert result == "Bob"
    mock_get.assert_called_once()

assert_called_once() is not async — it’s a mock method. The await goes on the coroutine you’re testing.

Event Loop Scope

pytest-asyncio creates an event loop for each async test. By default, the loop is function-scoped — new loop per test. This prevents state bleeding between tests.

If you need a module-scoped event loop (rare):

@pytest.fixture(scope="module")
def event_loop():
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

This is discouraged unless you have a specific reason — function-scoped loops are safer.

Testing Async Context Managers

Use async with in tests:

@pytest.mark.asyncio
async def test_async_queue():
    async with AsyncQueue() as q:
        await q.put("item")
        result = await q.get()
        assert result == "item"

async with requires an async context, which @pytest.mark.asyncio provides.

Common Failure Modes

“Event loop fixture not found.” This means pytest-asyncio isn’t installed, or you’re using an old version. Upgrade: pip install pytest-asyncio --upgrade.

“coroutine was never awaited.” Your test is not marked as async, or you’re calling an async function without await in a regular test function. Add @pytest.mark.asyncio to the test.

Fixture returns coroutine instead of result. If a fixture is async def but you don’t use it in an async test, pytest may not await it. Always pair async fixtures with @pytest.mark.asyncio tests.

Sharing mutable state across async tests. Async tests run concurrently within the same event loop in some configurations. Avoid shared mutable state, or use fixtures with proper cleanup.

Using asyncio.run for Non-async Test Code

Sometimes you call async code from a sync context — for example, in a setup_method:

@pytest.fixture
async def client():
    return await AsyncClient()

@pytest.fixture
def sync_client(client):
    # Run the async client in a sync context for non-async tests
    return asyncio.run(client)

asyncio.run() creates a fresh event loop and runs the coroutine to completion. Don’t call this inside an already-running event loop.

See Also