Python Testing & Debugging Beyond unittestAdvanced Python testing playbooks
Async & Concurrency

Running One Test on asyncio and Trio

A library that claims asyncio and Trio support has to prove it, and the cheapest proof is running the same test body on both. AnyIO makes the backend a fixture, so parametrising that fixture turns every marked test into two items whose failures are attributed by runtime. The setup is four lines; the value is finding, on the first run, the two or three places where the code assumed asyncio's scheduler.

Prerequisites

  • anyio >= 4.0 installed as anyio[trio], so the Trio backend actually imports.
  • pytest >= 8.0, with pytest-asyncio either absent or kept away from these tests — see configuring asyncio_mode.
  • Code under test that uses AnyIO primitives rather than asyncio ones; the survey is in testing with AnyIO and Trio.

Solution

Override the plugin's anyio_backend fixture with a parametrised one, and mark the module.

Python
# tests/anyio/conftest.py
import pytest


@pytest.fixture(params=["asyncio", "trio"])
def anyio_backend(request):
    # One fixture, two parameters: every marked test becomes two items.
    return request.param
Python
# tests/anyio/test_pipeline.py
import anyio
import pytest

pytestmark = pytest.mark.anyio          # AnyIO collects every test in this file


async def test_worker_signals_completion():
    done = anyio.Event()                # portable: no asyncio.Event here
    seen: list[int] = []

    async def worker():
        seen.append(1)
        done.set()

    async with anyio.create_task_group() as tg:
        tg.start_soon(worker)
        with anyio.fail_after(1.0):     # portable deadline, both runtimes
            await done.wait()

    assert seen == [1]
Bash
pytest tests/anyio -q
Plain text
tests/anyio/test_pipeline.py::test_worker_signals_completion[asyncio] PASSED
tests/anyio/test_pipeline.py::test_worker_signals_completion[trio] PASSED
2 passed in 0.08s
One test body expanded across two runtimes A single marked test resolves the anyio_backend fixture, which is parametrised with asyncio and trio. Two items are produced. The asyncio item runs on an event loop with a deterministic FIFO ready queue; the trio item runs under a nursery with deliberately randomised scheduling, so ordering assumptions fail there. Same body, two schedulers test_worker_signals @pytest.mark.anyio anyio_backend params: 2 [asyncio] FIFO ready queue · deterministic order [trio] randomised order · finds ordering assumptions
Trio's randomised scheduling is the feature that makes the second item worth running: it turns a latent ordering assumption into a reproducible failure.

Why this works

AnyIO's pytest plugin runs a marked coroutine test by calling anyio.run with the backend named by the anyio_backend fixture. Because that fixture is an ordinary pytest fixture, parametrising it multiplies the test items exactly as parametrising any other fixture would, and the backend name appears in the item id.

Everything pytest can do with parametrisation therefore applies: -k trio selects one runtime, pytest.param("trio", marks=pytest.mark.xfail) records a known gap, and a failure reports which runtime it occurred on without any extra instrumentation. The plugin does not special-case any of this; it simply reads the fixture.

Edge cases and failure modes

  • trio not installed. The trio parameter raises ImportError at test time rather than skipping, which is correct — a silently missing matrix entry is worse than a loud one. Depend on anyio[trio].
  • pytest-asyncio in auto mode nearby. It will claim these tests before AnyIO sees them, and the anyio_backend fixture is then unused. Keep them in a directory with its own invocation.
  • An asyncio import inside the test body. asyncio.get_running_loop() raises on the Trio item. Move backend-specific assertions into a pinned module.
  • Fixtures that are async. They run on the same backend as the requesting test, which is correct, but it means an async fixture cannot be session-scoped across differing backends. Keep expensive setup synchronous.
  • Time-based assertions. Trio and asyncio schedule timers differently enough that a tight assert elapsed < 0.05 will eventually fail on one of them. Assert on ordering and events instead.

Passing options to a backend

The fixture may return a tuple of the backend name and an options dictionary, which is how uvloop and Trio's own knobs are configured.

Python
import pytest

BACKENDS = [
    pytest.param(("asyncio", {"use_uvloop": False}), id="asyncio"),
    pytest.param(("asyncio", {"use_uvloop": True}), id="uvloop"),
    pytest.param(("trio", {}), id="trio"),
]


@pytest.fixture(params=BACKENDS)
def anyio_backend(request):
    return request.param
Plain text
tests/anyio/test_pipeline.py::test_worker_signals_completion[asyncio] PASSED
tests/anyio/test_pipeline.py::test_worker_signals_completion[uvloop]  PASSED
tests/anyio/test_pipeline.py::test_worker_signals_completion[trio]    PASSED

Explicit id values matter here. Without them the tuple's repr becomes the parameter id, producing identifiers such as test_worker[anyio_backend0] that are useless in a report and unstable across refactors — the readable-id problem covered in generating readable test IDs.

Three backends is usually one too many for every test. A practical arrangement runs asyncio and Trio everywhere, and adds the uvloop entry only for the handful of tests that exercise socket handling, where uvloop's different implementation genuinely could diverge.

Reading a backend-specific failure

When one item fails and its sibling passes, the diagnosis is nearly always one of three things, and they are distinguishable from the traceback alone.

An AttributeError or RuntimeError naming an asyncio API means backend-specific code leaked into a portable test. The fix is to replace the call with its AnyIO equivalent or to pin the test.

A timeout or a hang on Trio only usually means the code depends on a task being scheduled before another. Trio randomises that order deliberately, so the failure is the bug surfacing rather than Trio misbehaving. The repair is a real synchronisation primitive — an anyio.Event, a memory object stream — in place of the implicit ordering.

A cancellation-shaped failure, where cleanup did not run or an exception changed type, points at a difference in how the two runtimes deliver cancellation. Trio's cancellation is level-triggered within a cancel scope; asyncio's is edge-triggered per task. Code that catches BaseException broadly behaves differently under the two, which is worth knowing about regardless of which runtime ships.

Three shapes of backend-specific failure Three cards. An attribute or runtime error naming an asyncio API means backend-specific code leaked into a portable test. A hang or timeout on Trio only means the code relied on scheduling order. A cancellation-shaped failure points at the different cancellation semantics of the two runtimes. One item red, its sibling green asyncio API named get_running_loop, call_soon, all_tasks leaked into the body fix: use the anyio equivalent, or pin hangs on trio only task A assumed to run before task B a real ordering bug fix: an explicit event or a stream rendezvous cleanup differs finally skipped, or the exception type changed cancellation semantics fix: stop catching BaseException broadly
The middle card is the most valuable outcome of running two backends: an ordering assumption that would have survived years of asyncio-only testing.

Fixtures across two backends

Async fixtures run on whichever backend the requesting test resolved to, which is convenient and imposes one real constraint: a fixture cannot be shared across items that resolved differently.

Python
import anyio
import pytest


@pytest.fixture
async def broker():
    # Runs twice per test function — once per backend item — because each
    # item gets its own runtime and therefore its own fixture instance.
    send, receive = anyio.create_memory_object_stream[dict](max_buffer_size=4)
    async with send, receive:
        yield send, receive

This is correct but it means any expensive async setup is paid once per backend, doubling the cost of the very fixtures that were already the slow ones. The remedy is the same split recommended throughout this section: keep the expensive part synchronous and session-scoped, and let only a thin async handle be per-test.

Python
import pytest
from testcontainers.postgres import PostgresContainer


@pytest.fixture(scope="session")
def dsn():
    # Synchronous and session-scoped: no loop, no backend, no duplication.
    with PostgresContainer("postgres:16-alpine") as container:
        yield container.get_connection_url()


@pytest.fixture
async def connection(dsn):
    # Async and per-item: cheap to build twice, once per backend.
    async with await connect(dsn) as conn:
        yield conn
Splitting setup so only the cheap half runs twice A session-scoped synchronous fixture starts a container once and yields a connection string. Both the asyncio item and the trio item then build their own async connection from that string, so the expensive container start is paid once while only the cheap connection is duplicated. Expensive once, cheap twice container (session, sync) 3 s, once per run [asyncio] connection fixture a few milliseconds [trio] connection fixture a few milliseconds
Without the split, the container fixture would have to be async and per-item, and the backend matrix would double the slowest part of the suite rather than the fastest.

The same reasoning applies to anything with a handshake — a broker connection, a warmed cache, a compiled schema. Discover it once synchronously, connect to it cheaply per item, and the second backend costs milliseconds rather than seconds.

One consequence is worth anticipating: a synchronous session fixture cannot await anything, so any setup that genuinely requires the network has to run through a blocking client rather than an async one. For a container start, a schema migration or a health poll that is not a hardship — the synchronous library exists and is usually simpler. Where no synchronous path exists, anyio.from_thread.start_blocking_portal() gives a session-scoped portal that can run coroutines from synchronous code, at the cost of one more moving part; reach for it only when the blocking alternative is genuinely missing.

Keeping the matrix affordable

Every parametrised test runs twice, so the matrix has a real cost on a large suite, and two adjustments keep it proportionate.

Run both backends where the runtime could plausibly matter — concurrency primitives, cancellation, streams, timeouts — and pin everything else to one. A test that posts JSON and asserts on a status code learns nothing from a second runtime, and pinning it with a local anyio_backend fixture returning "asyncio" costs one line per module.

Then move the full matrix off the pull-request path. Deriving the parameter list from an environment variable keeps one configuration and one set of tests, with the breadth decided per job:

Python
import os

import pytest

_ALL = os.environ.get("TEST_ALL_BACKENDS") == "1"


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

Reading the environment at import time is acceptable here specifically because parametrisation is decided during collection; there is no later point at which the choice could be made. The result is a fast default run and a nightly job that exercises both runtimes, which is the same fast-and-thorough split used for integration tests.

Frequently Asked Questions

Why does a test pass on asyncio and fail on Trio? Usually a scheduling assumption. asyncio's ready queue is FIFO and deterministic; Trio deliberately randomises the order in which equally-ready tasks are scheduled, so code that relied on task A always running before task B fails there. That is a real bug being exposed, not a Trio incompatibility.

How do I run only one backend while debugging? Select by parameter id: pytest -k trio runs only the Trio items, and -k 'not trio' runs the rest. The backend is an ordinary parametrisation, so every selection mechanism pytest offers works on it.

Can I pass options such as uvloop to a backend? Yes. Return a tuple of the backend name and an options dictionary from the anyio_backend fixture — for example ("asyncio", {"use_uvloop": True}) — and AnyIO passes the options through to the runner.

← Back to Testing with AnyIO & Trio