Testing with AnyIO & Trio

A library that advertises asyncio and Trio support has, in practice, two implementations and one test suite — or two test suites that drift. AnyIO removes that choice by defining async primitives once and implementing them on both runtimes, so a single test body can be executed against each. For application code that will only ever run on asyncio the value is smaller but not zero: AnyIO's cancel scopes and task groups have cleaner semantics than the asyncio equivalents, and a test written against them is explicit about scope in a way wait_for never is.

Prerequisites

• anyio >= 4.0 (the release that made task groups and cancel scopes the primary API) plus trio as a test-only dependency.
• pytest >= 8.0. AnyIO ships its own pytest plugin; no separate pytest-anyio package exists.
• Python 3.9+, though 3.11+ is worth having so ExceptionGroup and except* are available natively rather than through the exceptiongroup backport.
• An understanding of what pytest-asyncio does to collection, since the two plugins compete — see pytest-asyncio in depth.

Core concept: the backend is a fixture

AnyIO's pytest integration is built from one idea. Tests marked with @pytest.mark.anyio are run by the plugin, and how they are run is decided by a fixture called anyio_backend. Because it is an ordinary fixture, parametrising it parametrises the runtime: one test function, two test items, one per backend.

The second idea follows from the first: anything the test touches must exist on both runtimes. asyncio.Event does not exist in Trio, so portable tests use anyio.Event. asyncio.wait_for has no Trio equivalent, so portable code uses anyio.fail_after. This is not overhead imposed by the test framework — it is the same constraint the library under test is already living with, made visible.

Step-by-step implementation

1. Install both backends

# pyproject.toml
[project.optional-dependencies]
test = ["pytest>=8.0", "anyio[trio]>=4.0"]

anyio[trio] pulls Trio in as an extra. Without it the trio parameter raises ImportError at test time rather than skipping, which is the right behaviour — a matrix entry that silently disappears is worse than one that fails loudly.

2. Mark the module

import pytest

# Every coroutine test in this module is run by AnyIO rather than pytest-asyncio.
pytestmark = pytest.mark.anyio

3. Parametrise the backend

# conftest.py
import pytest


@pytest.fixture(params=["asyncio", "trio"])
def anyio_backend(request):
    # Overriding the plugin's default fixture turns one test into two items.
    return request.param

The plugin's default implementation returns "asyncio" as a plain string. Returning a tuple lets you pass backend options — ("asyncio", {"use_uvloop": True}) is the usual reason — and the same tuple form works inside params.

4. Use portable primitives

import anyio


async def test_worker_signals_completion():
    done = anyio.Event()                      # not asyncio.Event
    results: list[int] = []

    async def worker():
        results.append(42)
        done.set()

    async with anyio.create_task_group() as tg:
        tg.start_soon(worker)
        # fail_after is a cancel scope: it bounds everything in the block.
        with anyio.fail_after(1.0):
            await done.wait()

    assert results == [42]

Note what is absent: no create_task, no explicit await task, no gather. The task group owns its children, and the async with block does not exit until every child has finished or been cancelled. That is the property worth testing against, because it is the one that stops tasks leaking.

5. Pin the tests that cannot be portable

import pytest


@pytest.fixture
def anyio_backend():
    # This module reaches for asyncio internals; run it on asyncio only.
    return "asyncio"


async def test_uses_loop_internals():
    import asyncio

    loop = asyncio.get_running_loop()
    assert loop.get_debug() is True

A local anyio_backend override shadows the parametrised one for that module. This is the clean way to keep three or four genuinely backend-specific tests without abandoning portability for the rest.

Verification

Backend coverage is visible in the test identifiers, so --collect-only is the check:

pytest tests/test_streams.py --collect-only -q

tests/test_streams.py::test_worker_signals_completion[asyncio]
tests/test_streams.py::test_worker_signals_completion[trio]
tests/test_streams.py::test_uses_loop_internals[asyncio]
3 tests collected

Two items for the portable test and one for the pinned test is exactly the intended shape. If every test shows a single [asyncio] item, the conftest.py override is not being picked up — usually because it sits in a directory above a nearer conftest.py that defines its own. If the tests show no backend suffix at all, pytest-asyncio claimed them first, which is the failure covered next.

Troubleshooting

Cancel scopes are the real portability story

asyncio.wait_for(coro, timeout) applies a deadline to one awaitable. AnyIO applies it to a region of code:

import anyio


async def test_partial_work_is_kept_when_the_deadline_passes(feed):
    received: list[bytes] = []

    # move_on_after does not raise; it cancels the block and records the fact.
    with anyio.move_on_after(0.25) as scope:
        async for chunk in feed.stream():
            received.append(chunk)

    assert scope.cancelled_caught is True     # the deadline actually fired
    assert received, "the stream should have yielded something before the cutoff"

The distinction matters for tests specifically. fail_after raises TimeoutError, which is right when the deadline is a failure; move_on_after returns and sets cancelled_caught, which is right when the deadline is part of the behaviour being tested — a poller that gives up, a batch that flushes on a timer, a stream that returns what it has. Expressing "give up after 250 ms, then assert on what arrived" with asyncio.wait_for requires wrapping the call in a try/except and losing the partial results; the cancel scope keeps them because the cancellation unwinds the loop body rather than the whole coroutine.

Cancel scopes also nest correctly, which wait_for does not do in any obvious way. An outer scope with a 5-second budget containing an inner scope with a 200 ms budget behaves the way the code reads: the inner deadline fires first and only cancels its own block. Testing that behaviour — that an inner timeout does not consume the outer budget, and that an outer timeout does interrupt the inner block — is the kind of assertion that catches real retry-loop bugs, and it is nearly unwriteable against wait_for.

What portability costs

Two things, and both are worth stating plainly before a team commits.

The first is runtime: every parametrised test runs twice. On a suite of 800 async tests that is not a rounding error, and the mitigation is selectivity — parametrise the tests that exercise concurrency primitives, scheduling, cancellation or stream semantics, and pin everything else to one backend. A test that posts JSON and asserts on a status code learns nothing from running on Trio.

The second is expressiveness. Inside a portable test you cannot reach for loop.call_later, loop.run_in_executor, asyncio.all_tasks() or any of the introspection that makes certain assertions easy. anyio.to_thread.run_sync covers the executor case, and anyio.get_current_task() covers part of the introspection case, but the leak-detection fixture that asserts asyncio.all_tasks() is empty has no portable equivalent — task groups are supposed to make it unnecessary, which is true right up until the code under test calls asyncio.create_task directly.

The honest summary: if the library has to support Trio, AnyIO is not a choice but a requirement, and the constraints are the ones the library already has. If it does not, adopt AnyIO for the cancel-scope semantics and pin the backend, which gets the better API without doubling the suite. Migrating an existing asyncio suite is a mechanical but non-trivial exercise, covered step by step in porting a pytest-asyncio suite to AnyIO.

Memory object streams replace queues

asyncio.Queue has no Trio counterpart, and the substitute AnyIO offers is better suited to testing anyway. A memory object stream is a typed, bounded channel split into a send half and a receive half, and closing either half is observable on the other — which is exactly the signal a test needs to assert that a producer finished rather than stalled.

import anyio
import pytest


async def test_producer_closes_the_stream_when_done():
    # Buffer of 0 means every send waits for a receive: the tightest coupling,
    # and the one that surfaces ordering bugs immediately.
    send, receive = anyio.create_memory_object_stream[int](max_buffer_size=0)

    async def produce():
        async with send:                     # closing the send half is the "done" signal
            for value in range(3):
                await send.send(value)

    received: list[int] = []
    async with anyio.create_task_group() as tg:
        tg.start_soon(produce)
        async with receive:
            async for value in receive:      # loop ends when the sender closes
                received.append(value)

    assert received == [0, 1, 2]


async def test_consumer_sees_closure_not_a_hang():
    send, receive = anyio.create_memory_object_stream[int](max_buffer_size=1)
    await send.aclose()

    # A closed stream raises rather than blocking forever — assertable, unlike a hang.
    with pytest.raises(anyio.EndOfStream):
        await receive.receive()

The second test is the one worth copying into any codebase using queues. With asyncio.Queue there is no closure concept at all: a consumer waiting on an empty queue whose producer has died waits forever, and the test that should catch it instead hangs until the suite's timeout kills it. Memory object streams turn that into EndOfStream, which a test can assert on in a millisecond.

max_buffer_size deserves a deliberate choice rather than a default. Zero forces a rendezvous, which makes interleaving deterministic and is the right setting for tests about ordering. A finite buffer lets the producer run ahead by a known amount, which is the right setting for tests about backpressure — fill the buffer, assert the next send blocks, drain one item, assert it unblocks. An unbounded buffer, which AnyIO deliberately makes awkward to request, removes backpressure entirely and with it the ability to test for it.

Running the backend matrix without doubling the bill

Two backends means up to twice the runtime, and the way to avoid paying it on every push is to treat the second backend as a distinct job rather than as part of the default run.

# conftest.py — one backend by default, both when explicitly asked
import os

import pytest

_BACKENDS = ["asyncio", "trio"] if os.environ.get("TEST_ALL_BACKENDS") else ["asyncio"]


@pytest.fixture(params=_BACKENDS)
def anyio_backend(request):
    return request.param

pytest -q                          # pull requests: asyncio only, full speed
TEST_ALL_BACKENDS=1 pytest -q      # merge queue and nightly: both runtimes

Deriving the parameter list from the environment keeps one conftest.py and one set of tests, with the matrix as a runtime decision. Reading an environment variable at import time is normally a smell, but parametrization is decided at collection, so there is no later point at which the choice could be made.

The alternative — marking individual tests for the second backend — is worth resisting. It requires a judgement per test about whether the runtime could matter, that judgement is made when the test is written and never revisited, and the tests that turn out to be backend-sensitive are precisely the ones nobody predicted. Running everything on both backends nightly costs one job and needs no judgement at all.

When a failure does appear on one backend only, the test identifier already names it: test_stream[trio] failing while test_stream[asyncio] passes is a complete bug report for anyone who knows the two runtimes. Reproducing it locally is pytest -k "trio and test_stream", and the diagnosis is almost always one of three things — a scheduling assumption (Trio's scheduler is deliberately randomised, asyncio's is FIFO), a cancellation assumption, or an asyncio API that leaked into portable code. Trio's randomised scheduling is a feature here: it finds ordering assumptions that asyncio's deterministic queue would hide for years.

Async fixtures under AnyIO

AnyIO runs async fixtures the same way it runs tests: through the backend the requesting test resolved to. That gives one guarantee pytest-asyncio needs configuration to achieve — a fixture is always on the same runtime as the test using it, because there is no separate loop lifetime to get out of step.

import anyio
import pytest


@pytest.fixture
async def broker():
    # Runs on whichever backend the requesting test resolved to.
    send, receive = anyio.create_memory_object_stream[dict](max_buffer_size=8)
    async with send, receive:
        yield send, receive
    # Both halves closed here, on the same runtime, before the test item ends.

The constraint that replaces loop scoping is sharper: a session-scoped async fixture is not possible in the general case, because each test item gets a fresh runtime. AnyIO documents this directly — async fixtures are effectively function-scoped unless the backend fixture is widened to match, and widening anyio_backend to session scope means every test in the session shares one runtime and one set of parameters.

In practice that pushes expensive setup into synchronous fixtures wherever it can live there. A Testcontainers instance, a temporary directory, a loaded configuration file: none of these need to await anything, so making them synchronous and session-scoped sidesteps the whole question. Only the thin async wrapper around them — the connection, the client, the stream — needs to be per-test, and per-test is cheap once the expensive part is already running. That split is the same one recommended for database fixtures: a durable resource created synchronously, a disposable handle acquired per test.

Frequently Asked Questions

Do I need Trio installed to use AnyIO in tests? No. AnyIO runs on asyncio by default and only imports Trio when a test is parametrised onto the trio backend. Install trio as a test-only dependency when you want the second backend in the matrix; without it, the anyio_backend fixture simply yields asyncio and the suite runs as before.

Why does my AnyIO test say 'fixture anyio_backend not found'? The AnyIO plugin supplies that fixture only when it is active. Either the plugin is not installed, or pytest-asyncio claimed the test first in auto mode. Check with pytest --fixtures | grep anyio, and isolate backend-parametrised tests in a directory whose conftest.py sets asyncio_mode = "strict".

Can I call asyncio APIs inside an AnyIO test? Only in tests pinned to the asyncio backend. Calling asyncio.get_running_loop() or loop.call_soon in a test that also runs on Trio raises at runtime, because no asyncio loop exists there. Move backend-specific assertions into their own test marked for that backend alone.

What replaces asyncio.wait_for in AnyIO?anyio.fail_after(seconds) for a deadline that raises TimeoutError, and anyio.move_on_after(seconds) for one that returns quietly with cancelled_caught set. Both are cancel scopes, so they apply to everything inside the block rather than to a single awaitable.

Is AnyIO slower than running on asyncio directly? The abstraction costs a thin layer of indirection per call, which is immaterial next to any real I/O. What costs measurable time is running the whole suite twice, once per backend — parametrise only the tests whose behaviour could differ between runtimes, and pin the rest to one backend.

Related guides

• Put the backend matrix in place with running one test on asyncio and Trio.
• Assert on structured-concurrency semantics in testing code that uses task groups.
• Move an existing suite across with porting a pytest-asyncio suite to AnyIO.
• Compare fixture lifetimes between the two plugins in pytest-asyncio vs anyio scoping trade-offs.
• Bound the whole suite regardless of backend with timeouts, cancellation and deadlines.

← Back to Testing Async & Concurrent Python