Python

Testing Async Python Code with pytest-asyncio and anyio

HelpMeTest

HelpMeTest

— 4 min read
Testing Async Python Code with pytest-asyncio and anyio

Testing async Python code requires an event loop to run coroutines. pytest-asyncio and anyio provide that infrastructure and make async tests look nearly identical to synchronous ones.

The Problem with Async Tests

You can't call await outside a coroutine, so this doesn't work:

# Won't work — pytest runs tests synchronously
def test_fetch():
    result = await fetch_data()  # SyntaxError
    assert result == "data"

You need a test runner that understands coroutines and provides an event loop for each test.

pytest-asyncio

Install:

pip install pytest pytest-asyncio

Mark tests with @pytest.mark.asyncio:

import pytest
import asyncio

async def fetch_data():
    await asyncio.sleep(0.01)
    return "data"

@pytest.mark.asyncio
async def test_fetch():
    result = await fetch_data()
    assert result == "data"

Auto Mode

Instead of marking every test, enable auto mode in pytest.ini or pyproject.toml:

# pytest.ini
[pytest]
asyncio_mode = auto
# pyproject.toml
[tool.pytest.ini_options]
asyncio_mode = "auto"

With auto mode, all async def test_* functions run as async tests automatically:

async def test_fetch():
    result = await fetch_data()
    assert result == "data"

Async Fixtures

Use @pytest_asyncio.fixture for async setup and teardown:

import pytest_asyncio

@pytest_asyncio.fixture
async def db_connection():
    conn = await create_connection("postgresql://localhost/testdb")
    yield conn
    await conn.close()

async def test_insert(db_connection):
    await db_connection.execute("INSERT INTO users VALUES ('alice')")
    row = await db_connection.fetchone("SELECT * FROM users WHERE name='alice'")
    assert row["name"] == "alice"

With asyncio_mode = auto, you can use @pytest.fixture for async fixtures too:

@pytest.fixture
async def db_connection():
    conn = await create_connection("postgresql://localhost/testdb")
    yield conn
    await conn.close()

Testing with anyio

anyio is a compatibility layer over asyncio and trio. Use it when your code should work with multiple backends:

pip install anyio[trio] pytest-anyio
import pytest
import anyio

@pytest.mark.anyio
async def test_concurrent_tasks():
    results = []

    async def task(name):
        await anyio.sleep(0.01)
        results.append(name)

    async with anyio.create_task_group() as tg:
        tg.start_soon(task, "a")
        tg.start_soon(task, "b")
        tg.start_soon(task, "c")

    assert sorted(results) == ["a", "b", "c"]

Run tests against both backends:

# conftest.py
import pytest

@pytest.fixture(params=["asyncio", "trio"])
def anyio_backend(request):
    return request.param

This runs every @pytest.mark.anyio test twice — once with asyncio, once with trio.

Mocking Coroutines

unittest.mock.AsyncMock handles coroutines:

from unittest.mock import AsyncMock, patch

async def get_user(user_id: int) -> dict:
    return await http_client.get(f"/users/{user_id}")

async def test_get_user():
    mock_response = {"id": 1, "name": "alice"}

    with patch("mymodule.http_client.get", new=AsyncMock(return_value=mock_response)):
        result = await get_user(1)

    assert result["name"] == "alice"

For more complex scenarios:

from unittest.mock import AsyncMock

async def test_retry_on_failure():
    mock_client = AsyncMock()
    # First call raises, second call succeeds
    mock_client.get.side_effect = [
        ConnectionError("timeout"),
        {"status": "ok"},
    ]

    result = await fetch_with_retry(mock_client, "/endpoint")
    assert result == {"status": "ok"}
    assert mock_client.get.call_count == 2

Testing Timeouts

Test that your code fails fast when operations take too long:

import asyncio
import pytest

async def slow_operation():
    await asyncio.sleep(10)
    return "result"

@pytest.mark.asyncio
async def test_timeout_raises():
    with pytest.raises(asyncio.TimeoutError):
        await asyncio.wait_for(slow_operation(), timeout=0.1)

With anyio:

import anyio
import pytest

@pytest.mark.anyio
async def test_timeout_with_anyio():
    with pytest.raises(TimeoutError):
        with anyio.fail_after(0.1):
            await slow_operation()

Testing Concurrent Behavior

Use asyncio.gather to run operations concurrently and check results:

@pytest.mark.asyncio
async def test_concurrent_fetches():
    urls = ["/api/user/1", "/api/user/2", "/api/user/3"]
    results = await asyncio.gather(*[fetch(url) for url in urls])

    assert len(results) == 3
    assert all(r["status"] == "ok" for r in results)

Test that your code handles cancellation correctly:

@pytest.mark.asyncio
async def test_cancellation():
    task = asyncio.create_task(long_running_operation())
    await asyncio.sleep(0.01)  # let it start
    task.cancel()

    with pytest.raises(asyncio.CancelledError):
        await task

Event Loop Scope

By default, each test gets a fresh event loop. Control the scope with fixtures:

@pytest_asyncio.fixture(scope="session")
async def shared_client():
    """One client for all tests in the session."""
    async with httpx.AsyncClient(base_url="http://localhost:8000") as client:
        yield client

This avoids creating and destroying expensive resources (database connections, HTTP clients) for each test.

Testing Async Generators

async def stream_events(count: int):
    for i in range(count):
        await asyncio.sleep(0.001)
        yield {"event": i}

@pytest.mark.asyncio
async def test_stream():
    events = []
    async for event in stream_events(5):
        events.append(event)

    assert len(events) == 5
    assert events[0] == {"event": 0}
    assert events[4] == {"event": 4}

Testing FastAPI and HTTPX

FastAPI is async — test it with HTTPX's AsyncClient:

import pytest
from httpx import AsyncClient
from myapp import app

@pytest.mark.asyncio
async def test_create_user():
    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.post("/users", json={"name": "alice"})

    assert response.status_code == 201
    assert response.json()["name"] == "alice"

Parameterizing Async Tests

@pytest.mark.parametrize("input,expected", [
    ("hello", "HELLO"),
    ("world", "WORLD"),
    ("", ""),
])
@pytest.mark.asyncio
async def test_async_upper(input, expected):
    result = await async_upper(input)
    assert result == expected

Common Mistakes

Forgetting to await coroutines:

async def test_bad():
    result = fetch_data()  # forgot await — result is a coroutine, not the value
    assert result == "data"  # always fails

Mixing sync and async code without a bridge:

# Can't call async from sync without run_until_complete or asyncio.run
def sync_function():
    loop = asyncio.get_event_loop()
    return loop.run_until_complete(async_function())  # works but creates issues in tests

Use asyncio.run() for one-off calls, or restructure so tests are always async.

End-to-End Testing Async Systems

Unit tests with mocked coroutines don't catch integration issues — network timeouts, database connection limits, or event ordering bugs under load. HelpMeTest lets you test async HTTP APIs and workflows end-to-end:

Scenario: async webhook delivery
  Given a user submits a form
  When the form is processed asynchronously
  Then a webhook is delivered within 5 seconds
  And the payload contains the form data

HelpMeTest handles timing and retry logic automatically, so you don't need to write asyncio.sleep and polling loops in your tests.

Key Takeaways

  • Use pytest-asyncio with asyncio_mode = auto to avoid marking every async test
  • AsyncMock handles coroutines in unittest.mock — use side_effect for sequences of return values
  • anyio lets you test against both asyncio and trio backends with parametrized fixtures
  • Scope async fixtures carefully — session scope avoids expensive setup/teardown per test
  • Always test cancellation and timeout behavior — these are the hardest bugs to find in production

Read more