Chapter 17 · Testing Asynchronous Code

Chapter goals: run async tests with pytest-asyncio and anyio, understand how event-loop scoping affects fixtures, and write async integration tests on top of httpx.AsyncClient.

17.1 Why Vanilla pytest Can't Run async def

pytest's collector only calls test functions. Calling async def test_xxx() returns a coroutine object — it is never awaited, so pytest can only warn "coroutine was never awaited" and mark the test as passed (a false green!). Async tests therefore need a plugin to take over execution and drive coroutines through an event loop.

Two mainstream options:

Plugin	Event loop	Notes
pytest-asyncio	asyncio only	Full-featured, deep fixture integration; first choice for FastAPI/SQLAlchemy projects
anyio (ships its own pytest plugin)	asyncio + trio	Comes free with anyio; ideal for libraries that need a trio backend

17.2 pytest-asyncio: Two Modes and Basic Usage

pytest-asyncio has strict (default) and auto discovery modes:

• strict: every async test must be explicitly marked with @pytest.mark.asyncio;
• auto: all async def test_* are recognized as asyncio tests automatically.

# pyproject.toml
[tool.pytest.ini_options]
asyncio_mode = "auto"          # or "strict" — pick one per team

# test_async_basic.py
import asyncio
import pytest


@pytest.mark.asyncio                 # Required in strict mode; optional in auto mode
async def test_runs_in_a_loop():
    assert asyncio.get_running_loop() is not None


async def fetch_user(client, uid):   # Sample async function under test
    return {"id": uid}


@pytest.mark.asyncio
async def test_fetch_user():
    user = await fetch_user(None, 42)
    assert user["id"] == 42

The false-green trap

If you forget the plugin and the marker, pytest emits PytestUnhandledCoroutineWarning yet the result may still be PASSED. In CI add -W error::pytest.PytestUnhandledCoroutineWarning to make it fail loudly.

17.3 Event Loop Scope: loop_scope

By default pytest-asyncio gives each test its own new event loop (function level), which is the most isolated. But some resources bind to a specific loop (connection pools, for example) and error out when reused across loops:

RuntimeError: Task ... got Future ... attached to a different loop

The fix is pinning adjacent tests to the same loop:

# test_loop_scope.py
import asyncio
import pytest


@pytest.mark.asyncio(loop_scope="module")   # All tests in this module share one module-level loop
async def test_remember_loop():
    global _loop
    _loop = asyncio.get_running_loop()


@pytest.mark.asyncio(loop_scope="module")
async def test_same_loop():
    global _loop
    assert asyncio.get_running_loop() is _loop

Official advice: adjacent tests should share the same loop scope (unifying on module within a module is common). Async fixtures follow suit — align them with loop_scope:

# conftest.py
import pytest


@pytest.fixture(loop_scope="module", scope="module")
async def engine():
    """Module-level async engine: built once, shared by the whole module on one loop."""
    eng = await create_test_engine()
    yield eng
    await eng.dispose()

The rule of thumb: fixture scope ≤ its bound loop_scope. Otherwise a function-level loop gets destroyed while a module-level fixture still lives on it — hello, "different loop" errors.

17.4 The anyio Plugin: One Codebase, Two Backends

anyio ships a pytest plugin. The anyio_backend fixture picks the backend and can be parametrized so one test runs under both asyncio and trio:

# test_anyio_demo.py
import anyio
import pytest


@pytest.mark.anyio                       # Requires an anyio_backend fixture to exist
async def test_sleep():
    await anyio.sleep(0.01)


@pytest.fixture
def anyio_backend():
    return "asyncio"                     # Switch to "trio" to run on trio


# Parametrizing across backends:
@pytest.fixture(params=["asyncio", "trio"])
def anyio_backend(request):
    return request.param                 # The test above now runs twice per backend pair

Selection advice: applications (FastAPI etc.) almost always live on asyncio — use pytest-asyncio; when writing generic libraries that want to claim trio compatibility, use anyio. Don't enable both discovery modes simultaneously or tests will run twice.

17.5 In Practice: httpx.AsyncClient Integration Tests

Modern ASGI apps (FastAPI/Starlette/Litestar) don't need a real server — httpx's ASGITransport pushes requests straight into the app object:

# test_api_async.py
import httpx
import pytest
from fastapi import FastAPI

app = FastAPI()


@app.get("/hello/{name}")
async def hello(name: str):
    return {"msg": f"hello {name}"}


@pytest.fixture
async def client():
    # ASGITransport: in-process connection to the ASGI app, zero network overhead
    transport = httpx.ASGITransport(app=app)
    async with httpx.AsyncClient(
        transport=transport, base_url="http://test"
    ) as c:
        yield c


@pytest.mark.anyio
@pytest.fixture(anyio_backend="asyncio")
def backend():
    ...


@pytest.mark.asyncio
async def test_hello(client):
    resp = await client.get("/hello/agno")
    assert resp.status_code == 200
    assert resp.json() == {"msg": "hello agno"}


# If you prefer the anyio plugin, the client fixture declares a backend:
# import anyio
#
# @pytest.fixture
# def anyio_backend():
#     return "asyncio"
#
# @pytest.fixture
# async def client(anyio_backend):
#     transport = httpx.ASGITransport(app=app)
#     async with httpx.AsyncClient(transport=transport,
#                                  base_url="http://test") as c:
#         yield c

Three essentials:

1. Inject ASGITransport(app) via transport= instead of the deprecated AsyncClient(app=app) shortcut;
2. base_url is just a placeholder — no request leaves the machine;
3. Wrap the client in async with inside a fixture so pools close properly; combine with 17.3's loop_scope rules to avoid cross-loop issues.

Chapter Summary

• Vanilla pytest never executes coroutines; hand async tests to pytest-asyncio or anyio;
• strict mode demands explicit @pytest.mark.asyncio; auto mode detects automatically;
• One loop per test by default; share resources via loop_scope="module" pinning tests and async fixtures to one loop;
• The anyio plugin validates asyncio/trio simultaneously through parametrized anyio_backend;
• For in-process integration tests use httpx.ASGITransport(app) + AsyncClient(transport=...).

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. Without any async plugin installed, what happens to a plain async def test_xxx()?

2. What is pytest-asyncio’s default event-loop strategy when loop_scope is not configured?

3. You hit "Future attached to a different loop". What is the most common root cause?

4. For in-process async integration testing of a FastAPI app with httpx, which approach is recommended?

🛠️ Hands-on Practice

1. Write an async rate_limiter class (internally using asyncio.sleep) and at least three pytest-asyncio tests covering "allow / throttle / window reset".
2. Deliberately mix different loop_scopes to produce a failure, read the error, fix it, and write down your conclusion.
3. Rewrite exercise 1 using anyio's parametrized backend to confirm the tests pass on both asyncio and trio.

With deterministic-input async testing mastered, the next chapter explores testing unknown input spaces — enter Hypothesis.