Testing with AnyIO and Trio

Switch the test suite first (Phase 1 of the migration playbook) and keep it green after every commit. This reference covers the anyio pytest plugin, running every test on both backends, moving off pytest-asyncio, pytest-trio, deterministic time with MockClock, deterministic ordering, and how to test cancellation, timeouts, exception groups and FastAPI/httpx code. The examples were run with anyio 4.15, trio 0.34, pytest 9, pytest-trio 0.8, httpx 0.28 and FastAPI 0.157 on Python 3.13. asyncio-only test tricks (event-loop fixtures, loop.slow_callback_duration) are in the asyncio advisor.

#Contents

• The anyio pytest plugin
• Choosing backends: the anyio_backend fixture
• Async fixtures
• Moving from pytest-asyncio
• pytest-trio
• Deterministic time: MockClock and autojump
• Deterministic ordering
• Testing cancellation and cleanup
• Testing timeouts
• Testing exception groups
• Fakes instead of asyncio mocks
• FastAPI, Starlette and httpx under anyio
• Threads and blocking portals in tests
• Warnings, leaks and the CI matrix
• Checklist

#The anyio pytest plugin

The plugin ships inside anyio and registers itself through a pytest entry point. There is nothing extra to install, and pytest-asyncio is not needed.

# pyproject.toml
[tool.pytest.ini_options]
anyio_mode = "auto"            # recent anyio 4.x; omit and use @pytest.mark.anyio on older releases
filterwarnings = [
    "error::RuntimeWarning",   # "coroutine ... was never awaited" fails the test
    "error::ResourceWarning",  # unclosed streams/sockets fail the test
]

# Strict mode: mark per module
import pytest

pytestmark = pytest.mark.anyio

If anyio auto mode and pytest-asyncio auto mode are both on, anyio emits a PytestConfigWarning ("AnyIO auto mode has been enabled together with pytest-asyncio auto mode"). Pick one owner per test: during a gradual move, keep pytest-asyncio in strict mode and let anyio take only the tests marked anyio.

How it works, in one paragraph: a marked test (or auto mode) gets the anyio_backend fixture added. The plugin's pytest_pyfunc_call hook runs the coroutine on a test runner for that backend. Async fixtures are wrapped only when the requesting test uses anyio_backend, and the test and its async fixtures share one runner (one event loop) while any of them is alive.

#Choosing backends: the anyio_backend fixture

The built-in fixture is module-scoped and parametrised over anyio.get_available_backends(), so in recent anyio 4.x every test runs once per installed backend: test_x[asyncio] and, if trio is importable, test_x[trio]. Older releases ran only asyncio. Override it in conftest.py so the matrix does not depend on which anyio is installed:

# conftest.py
import importlib.util

import pytest

HAS_TRIO = importlib.util.find_spec("trio") is not None


@pytest.fixture(
    scope="module",                          # keep the built-in scope: see the ScopeMismatch note below
    params=[
        pytest.param("asyncio", id="asyncio"),
        pytest.param("trio", id="trio",
                     marks=pytest.mark.skipif(not HAS_TRIO, reason="trio not installed")),
    ],
)
def anyio_backend(request):
    return request.param

The value is either a backend name or a (name, options) tuple. The options go to the backend runner:

Other knobs:

ScopeMismatch. A module- or session-scoped async fixture depends on anyio_backend. If you override anyio_backend with the default function scope, pytest fails with ScopeMismatch: You tried to access the function scoped fixture anyio_backend with a module scoped request object. Give the override a scope at least as wide as the widest async fixture (scope="session" if you have session-scoped async fixtures).

#Async fixtures

Async fixtures are plain @pytest.fixture functions (async def with return or yield). There is no special decorator such as pytest_asyncio.fixture.

import anyio
import pytest


class FakeVectorStore:
    def __init__(self) -> None:
        self.closed = False
        self.docs: dict[str, str] = {}

    async def upsert(self, doc_id: str, text: str) -> None:
        await anyio.lowlevel.checkpoint()
        self.docs[doc_id] = text

    async def aclose(self) -> None:
        await anyio.lowlevel.checkpoint()
        self.closed = True


@pytest.fixture
async def store():
    s = FakeVectorStore()
    yield s
    await s.aclose()                            # teardown runs on the same backend as the test


@pytest.mark.anyio
async def test_upsert(store):
    await store.upsert("a", "alpha")
    assert store.docs == {"a": "alpha"}

Rules:

• The async fixture runs under anyio only when the test that requests it uses anyio_backend (marked test or auto mode). A sync test that requests an async fixture fails on pytest 9 with "requested an async fixture 'x', with no plugin or hook that handled it"; older pytest handed the test an un-awaited coroutine.
• Calling request.getfixturevalue("some_async_fixture") from inside a running async test raises RuntimeError: Cannot schedule a coroutine in the test runner while another is already running. Request the fixture as a parameter instead.
• A task group opened in a yield fixture must be entered and exited in that fixture (the plugin runs setup and teardown of an async generator fixture in one task). Yield the task group itself, or a client that uses it. Do not yield from inside a cancel scope that the test is expected to exit.
• Fixtures that start background services use tg.start(...) so the test begins only after the service is ready:

import anyio
import pytest


async def fake_embedding_server(port_box: list, *, task_status=anyio.TASK_STATUS_IGNORED) -> None:
    port_box.append(8123)                       # bind, then report readiness
    task_status.started(8123)
    await anyio.sleep_forever()


@pytest.fixture
async def embedding_port():
    async with anyio.create_task_group() as tg:
        port = await tg.start(fake_embedding_server, [])
        yield port
        tg.cancel_scope.cancel()                # stop the server; the group then exits cleanly


@pytest.mark.anyio
async def test_server_ready(embedding_port):
    assert embedding_port == 8123

#Moving from pytest-asyncio

Order of work: switch the plugin with anyio_backend pinned to "asyncio", get green, migrate code, then add "trio" to the fixture params module by module, starting with modules that have no asyncio-only dependencies.

#pytest-trio

Use pytest-trio only for a codebase that targets trio directly and has no anyio layer. A library that must run on both backends should use the anyio plugin, because pytest-trio runs every test on trio only.

import trio


async def test_retry_backoff_is_fast(autojump_clock):     # pytest-trio, trio_mode = true
    start = trio.current_time()
    await trio.sleep(300)                                  # five minutes of backoff, in ~0 real seconds
    assert trio.current_time() - start == 300

Avoid running the same test under both plugins. With both installed, an anyio-marked test belongs to anyio, and an unmarked async def test with no plugin claiming it fails with "async def functions are not natively supported". Turning on trio_mode makes pytest-trio claim unmarked async tests, so do not enable it together with anyio_mode = "auto".

#Deterministic time: MockClock and autojump

trio.testing.MockClock replaces the clock of a trio run. With autojump_threshold=0, whenever every task is blocked the clock jumps to the next deadline. A test that waits on a 30-second timeout then finishes in milliseconds, and the elapsed virtual time is exact.

import anyio
import pytest
from trio.testing import MockClock


@pytest.fixture(params=[pytest.param(("trio", {"clock": MockClock(autojump_threshold=0)}), id="trio-mock")])
def anyio_backend(request):
    return request.param


@pytest.mark.anyio
async def test_agent_budget_gives_up_after_120s():
    start = anyio.current_time()
    with anyio.move_on_after(120) as budget:
        await anyio.sleep(3600)                 # a tool that never answers
    assert budget.cancelled_caught
    assert anyio.current_time() - start == 120  # exact under MockClock, ~0 s of wall time

Caveats:

• MockClock only exists on trio. asyncio has no supported way to swap the loop clock, so keep timeout tests on a trio-only param (above) and test the logic, not the wall-clock duration, on asyncio.
• Each pytest param creates one MockClock at import time. It is reused across tests with that param, which is fine because each test only reads differences of current_time(). Build the clock inside a fixture if a test asserts absolute times.
• Autojump does not wait for real I/O or threads. A test that calls to_thread.run_sync or a real socket can hit its deadline before the thread finishes. Use a small positive threshold or keep threads out of MockClock tests.
• Compare anyio.current_time() differences, never absolute values or time.monotonic(). trio's normal clock adds a large random offset on purpose.

#Deterministic ordering

Never assert that start_soon children run in submission order. trio randomly reverses each batch of runnable tasks, deliberately, so hidden ordering assumptions show up. asyncio happens to be FIFO, which is why a test can pass on asyncio and flake on trio.

import anyio
import pytest


@pytest.mark.anyio
async def test_worker_waits_for_permit():
    limiter = anyio.CapacityLimiter(1)
    log: list[str] = []

    async def worker(name: str, release: anyio.Event) -> None:
        async with limiter:
            log.append(f"{name} start")
            await release.wait()

    first_done, second_done = anyio.Event(), anyio.Event()
    async with anyio.create_task_group() as tg:
        tg.start_soon(worker, "a", first_done)
        tg.start_soon(worker, "b", second_done)
        await anyio.wait_all_tasks_blocked()
        assert len(log) == 1                    # exactly one holds the token; which one is not asserted
        assert limiter.borrowed_tokens == 1
        first_done.set()
        second_done.set()
    assert sorted(log) == ["a start", "b start"]

#Testing cancellation and cleanup

Three properties to check for every function that owns resources: it stops when cancelled, its cleanup actually ran, and it re-raised the cancel exception (the caller's scope reports cancelled_caught).

import anyio
import pytest


class Lease:
    def __init__(self) -> None:
        self.released = False

    async def release(self) -> None:
        await anyio.sleep(0.01)                 # cleanup that needs an await
        self.released = True


async def hold_lease(lease: Lease, started: anyio.Event) -> None:
    try:
        started.set()
        await anyio.sleep_forever()
    finally:
        with anyio.CancelScope(shield=True), anyio.move_on_after(1):
            await lease.release()


@pytest.mark.anyio
async def test_cancel_releases_lease():
    lease, started = Lease(), anyio.Event()
    async with anyio.create_task_group() as tg:
        tg.start_soon(hold_lease, lease, started)
        await started.wait()
        tg.cancel_scope.cancel()
    assert lease.released                       # fails without the shield: the await in finally is skipped
    assert tg.cancel_scope.cancelled_caught

To catch handlers that swallow cancellation (MIG031), run the function in a scope that is already cancelled and assert it did not get past its first checkpoint:

import anyio
import pytest


async def bad_retry(log: list) -> None:
    for attempt in range(3):
        try:
            await anyio.sleep(1)
        except anyio.get_cancelled_exc_class():
            log.append(f"swallowed {attempt}")  # bug: no raise
    log.append("finished")


@pytest.mark.anyio
async def test_bad_retry_ignores_cancellation():
    log: list = []
    with anyio.CancelScope() as scope:
        scope.cancel()
        await bad_retry(log)
    assert log == ["swallowed 0", "swallowed 1", "swallowed 2", "finished"]   # the bug, made visible

A correct function leaves log == [] there. The same test shape also checks the cancellation rules: replace the body under test and assert on what ran.

For thread offload, assert that cancellation returns promptly when you rely on abandon_on_cancel=True or from_thread.check_cancelled(). Measure with time.monotonic() here because the worker thread runs on real time, and keep these tests off the MockClock param.

#Testing timeouts

Exception classes to expect: anyio's fail_after raises the builtin TimeoutError on both backends; trio's own trio.fail_after raises trio.TooSlowError, which is not a TimeoutError subclass.

#Testing exception groups

Task groups raise ExceptionGroup even for a single failure, so pytest.raises(ValueError) no longer matches. Use one of:

import anyio
import pytest


async def boom(msg: str) -> None:
    await anyio.lowlevel.checkpoint()
    raise ValueError(msg)


@pytest.mark.anyio
async def test_raises_group():                   # pytest 8.4+
    with pytest.RaisesGroup(pytest.RaisesExc(ValueError, match="bad shard")):
        async with anyio.create_task_group() as tg:
            tg.start_soon(boom, "bad shard 3")


@pytest.mark.anyio
async def test_group_contains():                 # pytest 8.0+
    with pytest.raises(ExceptionGroup) as excinfo:
        async with anyio.create_task_group() as tg:
            tg.start_soon(boom, "bad shard 3")
    assert excinfo.group_contains(ValueError, match="bad shard")

If the public API unwraps single errors at its boundary (see keeping a single-exception API), test the boundary with plain pytest.raises(ProviderError), and test the internals with RaisesGroup.

#Fakes instead of asyncio mocks

Fakes used in tests must not import asyncio, or the trio param fails with errors such as RuntimeError: no running event loop (see troubleshooting-migration.md). Build them from anyio primitives:

import anyio
import pytest


class FakeLLM:
    """Completes calls only when the test says so; records peak concurrency."""

    def __init__(self) -> None:
        self.gate = anyio.Event()
        self.in_flight = 0
        self.peak = 0

    async def complete(self, prompt: str) -> str:
        self.in_flight += 1
        self.peak = max(self.peak, self.in_flight)
        try:
            await self.gate.wait()
            return prompt.upper()
        finally:
            self.in_flight -= 1


async def fan_out(llm: FakeLLM, prompts: list[str], limit: int) -> list[str]:
    limiter = anyio.CapacityLimiter(limit)
    results: list = [None] * len(prompts)

    async def one(i: int, p: str) -> None:
        async with limiter:
            results[i] = await llm.complete(p)

    async with anyio.create_task_group() as tg:
        for i, p in enumerate(prompts):
            tg.start_soon(one, i, p)
    return results


@pytest.mark.anyio
async def test_fan_out_respects_limit_and_order():
    llm = FakeLLM()
    out: list = []

    async def run() -> None:
        out.extend(await fan_out(llm, ["a", "b", "c", "d"], limit=2))

    async with anyio.create_task_group() as tg:
        tg.start_soon(run)
        await anyio.wait_all_tasks_blocked()
        assert llm.in_flight == 2               # the limiter holds the other two back
        llm.gate.set()
    assert out == ["A", "B", "C", "D"] and llm.peak == 2

Notes:

• unittest.mock.AsyncMock is backend-neutral: awaiting it does not touch any event loop. It is fine for "was it called with X" checks. It has no checkpoint, though, so a fake that must be cancellable or must yield needs a real await anyio.sleep(...) / Event.wait() in a side_effect.
• A streaming fake: give the code under test the receive end of anyio.create_memory_object_stream[str](0) and push tokens from the test with await send.send(token). Closing send ends the consumer's async for.
• Outbound HTTP: use httpx.MockTransport(handler) with httpx.AsyncClient(transport=...). It runs on both backends and needs no network.

#FastAPI, Starlette and httpx under anyio

Two ways to drive an ASGI app in tests:

Because ASGITransport skips lifespan, anything set up there (app.state.http, a task group, a limiter) is missing and the endpoint fails with an AttributeError. Enter the lifespan yourself:

from contextlib import asynccontextmanager

import anyio
import httpx
import pytest
from fastapi import FastAPI, Request


@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.limiter = anyio.CapacityLimiter(4)
    yield


app = FastAPI(lifespan=lifespan)


@app.get("/answer")
async def answer(request: Request) -> dict:
    async with request.app.state.limiter:
        await anyio.sleep(0)
    return {"answer": 42}


@pytest.fixture
async def client():
    async with lifespan(app):                   # or app.router.lifespan_context(app)
        transport = httpx.ASGITransport(app=app)
        async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
            yield c


@pytest.mark.anyio
async def test_answer(client):
    r = await client.get("/answer")
    assert r.status_code == 200 and r.json() == {"answer": 42}

This runs on both backends. If the app depends on asyncpg, aiohttp or another asyncio-only client, pin that test module to "asyncio". ASGITransport re-raises app exceptions by default (raise_app_exceptions=True), so a failing endpoint shows its real traceback instead of a 500. Pass raise_app_exceptions=False to assert on the error response. More lifespan patterns: ai-engineering-anyio.md.

#Threads and blocking portals in tests

• A sync test that must call async code: anyio.run(fn, *args, backend="trio"), or a blocking portal when several calls must share one loop:

import anyio
from anyio.from_thread import start_blocking_portal


async def embed(text: str) -> list[float]:
    await anyio.sleep(0)
    return [float(len(text))]


def test_sync_caller_through_portal():
    with start_blocking_portal(backend="asyncio") as portal:
        assert portal.call(embed, "abc") == [3.0]
        assert portal.call(embed, "hello") == [5.0]

• Never call anyio.run from inside an async test: the loop is already running in that thread, and you get RuntimeError: Already running asyncio in this thread (or trio). A blocking portal started inside an async test does work (it runs its own loop in a new thread), but every portal.call blocks the test's loop, so the test's other tasks freeze. Inside async code, just await.
• Code that calls anyio.from_thread.run must run in a worker thread started by to_thread.run_sync (or be given a token). Called from the main test thread it raises NoEventLoopError: Not running inside an AnyIO worker thread, and no event loop token was provided.
• trio.testing.wait_all_threads_completed() (trio only) waits for trio.to_thread workers, useful after abandon_on_cancel=True.

#Warnings, leaks and the CI matrix

CI steps after a migration:

python scripts/migrate_async.py src/ --check         # no asyncio left outside "# noqa: MIG0xx" adapters
pytest -q -W error::RuntimeWarning                   # every anyio test on asyncio and trio
pytest -q -k asyncio                                 # optional: isolate a backend when triaging a failure

Run the trio column even for an app that ships on asyncio: it is the cheapest detector of leftover asyncio.* calls, hidden ordering assumptions and swallowed cancellation.

#Checklist

• @pytest.mark.asyncio, pytest_asyncio.fixture and event_loop are gone; anyio owns async tests.
• anyio_backend is overridden in conftest.py with explicit params and a scope wide enough for the async fixtures.
• Modules with asyncio-only dependencies pin "asyncio"; everything else runs on both backends.
• Timeout and backoff tests use the trio + MockClock(autojump_threshold=0) param and assert exact virtual elapsed time.
• No test relies on start_soon order or sleep(0) scheduling; it uses events, wait_all_tasks_blocked() or indexed results.
• Cancellation tests assert that cleanup ran and that the cancel exception was not swallowed.
• Group errors are asserted with pytest.RaisesGroup or group_contains.
• Fakes use anyio primitives only; outbound HTTP uses httpx.MockTransport.
• ASGI tests either use TestClient as a context manager or enter the lifespan around ASGITransport.
• RuntimeWarning and ResourceWarning are errors in CI.