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

Porting a pytest-asyncio Suite to AnyIO

Porting an asyncio test suite to AnyIO is a mechanical change with a handful of sharp edges, and doing it module by module rather than all at once is what keeps it reviewable. The order matters: isolate the plugins first, replace primitives second, restructure fixtures third, and only enable the second backend once the module is green on the first. Skipping straight to the backend matrix produces failures whose cause is ambiguous between the port and the runtime.

Prerequisites

  • anyio[trio] >= 4.0 alongside the existing pytest-asyncio, which stays installed until the migration finishes.
  • pytest >= 8.0, Python 3.10+ (3.11+ removes the TimeoutError alias problem described below).
  • The collection-mode rules from configuring asyncio_mode, since both plugins will be active at once.
  • A green suite before you start; porting on top of existing failures makes every diagnosis ambiguous.

Solution

Move one directory, keeping the two plugins apart by invocation rather than by hope.

Bash
# Two invocations, one repository. Neither plugin sees the other's tests.
pytest tests/legacy -q -p no:anyio          # still pytest-asyncio
pytest tests/ported -q -p no:asyncio        # now AnyIO
Python
# tests/ported/conftest.py
import pytest


@pytest.fixture(params=["asyncio"])          # trio added at the very end
def anyio_backend(request):
    return request.param
Python
# BEFORE — tests/legacy/test_worker.py
import asyncio

import pytest


@pytest.mark.asyncio
async def test_worker_drains_the_queue():
    queue = asyncio.Queue()
    done = asyncio.Event()
    await queue.put({"id": 1})

    async def worker():
        item = await queue.get()
        processed.append(item)
        done.set()

    task = asyncio.create_task(worker())
    await asyncio.wait_for(done.wait(), timeout=1.0)
    await task
    assert processed == [{"id": 1}]
Python
# AFTER — tests/ported/test_worker.py
import anyio
import pytest

pytestmark = pytest.mark.anyio


async def test_worker_drains_the_queue():
    send, receive = anyio.create_memory_object_stream[dict](max_buffer_size=1)
    done = anyio.Event()
    await send.send({"id": 1})

    async def worker():
        item = await receive.receive()
        processed.append(item)
        done.set()

    async with anyio.create_task_group() as tg:
        tg.start_soon(worker)                 # the group owns the task
        with anyio.fail_after(1.0):           # a cancel scope, not a wrapper
            await done.wait()

    assert processed == [{"id": 1}]
The four substitutions that make up most of a port Four rows pairing an asyncio construct with its AnyIO replacement: asyncio.Event becomes anyio.Event, asyncio.Queue becomes a memory object stream, asyncio.wait_for becomes fail_after or move_on_after, and create_task plus gather becomes a task group. Each row notes what changes in behaviour rather than only in spelling. Most of the diff is these four lines, repeated asyncio anyio what changes asyncio.Event() anyio.Event() nothing; no clear() in anyio asyncio.Queue() memory object stream closing is observable asyncio.wait_for(c, t) with fail_after(t): scopes a block, not one await create_task + gather create_task_group() raises ExceptionGroup Only the last row changes what a test asserts; the first three are spelling.
Three of the four substitutions are mechanical. The fourth changes the exception type that reaches the test, which is where a careless port silently stops asserting.

Why this works

AnyIO implements its primitives on top of whichever runtime is active, so the replacements are behaviourally equivalent where equivalence is possible and explicit where it is not. anyio.Event has no clear() because Trio's does not, and an event that can be cleared is a source of races anyway; a memory object stream has closing semantics because both runtimes can express them and a queue cannot.

Keeping the two plugins in separate invocations during the migration matters because both implement pytest_pyfunc_call. Within one invocation the registration order decides which claims a coroutine test, and that order is not something a repository should depend on. Two commands cost nothing and remove the ambiguity entirely.

Edge cases and failure modes

  • asyncio.TimeoutError versus TimeoutError. They are the same class from Python 3.11 and different before it, so a test written as pytest.raises(asyncio.TimeoutError) fails against anyio.fail_after on 3.10. Match TimeoutError and require 3.11, or catch both during the transition.
  • gather semantics assumed in assertions. gather returns the first exception and leaves siblings running; a task group cancels siblings and raises a group. Any pytest.raises(ValueError) around ported concurrency code needs updating.
  • Session-scoped async fixtures. They have no single runtime under AnyIO. Split them, as below.
  • loop.call_later and friends. There is no portable equivalent; code relying on them must stay pinned to asyncio or be restructured around a task and a sleep.
  • Tests that assert on asyncio.all_tasks(). No portable equivalent exists, by design — task groups are meant to make the check unnecessary. Replace with an assertion on resource balance.

Restructuring the fixtures

The fixture change is the part that needs thought rather than search-and-replace, because AnyIO gives each test item its own runtime.

Python
# BEFORE — one pool for the whole session, on one loop
import pytest_asyncio


@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def pool():
    pool = await create_pool(DSN)
    try:
        yield pool
    finally:
        await pool.close()
Python
# AFTER — expensive discovery synchronous and shared; the handle per test
import pytest


@pytest.fixture(scope="session")
def dsn():
    # Synchronous: no runtime, so it can genuinely be session-scoped.
    with PostgresContainer("postgres:16-alpine") as container:
        yield container.get_connection_url()


@pytest.fixture
async def pool(dsn):
    # Per item, on that item's runtime. Cheap because the server is already up.
    pool = await create_pool(dsn)
    try:
        yield pool
    finally:
        await pool.close()

The cost of this change is one pool creation per test rather than one per session, which for an in-process pool against an already-running server is single-digit milliseconds. The benefit is that nothing is shared across runtimes, so the whole class of cross-loop failures disappears rather than being managed.

Where per-test pool creation genuinely is too slow — a TLS handshake to a remote service, say — the escape hatch is anyio.from_thread.start_blocking_portal(), which gives a session-scoped portal that synchronous code can use to run coroutines. It works, it is supported, and it is one more moving part; reach for it after measuring rather than in anticipation.

Splitting a session-scoped async fixture for AnyIO Before the port, a single session-scoped async fixture creates a pool on the session loop and every test uses it. After the port, a synchronous session fixture holds the container and yields a connection string, and each test item builds its own pool from that string on its own runtime. One runtime per item changes where the pool lives before async pool fixture, session + session loop every test shares one pool and one loop it must not outlive after sync dsn fixture, session scope [asyncio] pool [trio] pool nothing crosses a runtime boundary
The container start — the genuinely slow part — is still paid once. Only the cheap connection is duplicated, which is what makes the arrangement affordable.

Sequencing the migration

Do it directory by directory, and in this order within each directory.

Move the files. Create the target directory, move one module, and add the pytestmark. Nothing else. Run it; it should pass on asyncio immediately if the module used no asyncio-specific primitives.

Replace primitives. Work through the four substitutions above. The compiler will not help here, so a grep for asyncio\. in the ported directory is the checklist, and it should end empty.

Restructure fixtures. Split the session-scoped async ones. This is where a module most often needs a design decision rather than an edit.

Fix the assertions. Update timeout types and group matching. pytest -q names them.

Add Trio. Only now. Failures at this point are genuine portability findings — a scheduling assumption, a cancellation difference — rather than porting mistakes, and keeping the two categories separate is what makes the second category worth reading.

Bash
# The per-directory checklist, mechanically
grep -rn "asyncio\." tests/ported/ | grep -v "^Binary"     # should be empty
pytest tests/ported -q -p no:asyncio                        # green on asyncio
sed -i 's/params=\["asyncio"\]/params=["asyncio", "trio"]/' tests/ported/conftest.py
pytest tests/ported -q -p no:asyncio                        # now both

Keeping each directory's port as its own change makes the review tractable and the revert cheap. A single change porting forty modules is one nobody reads carefully, and the portability findings in it are indistinguishable from the porting mistakes.

What the port is actually worth

It is worth being honest about the benefit before spending a week on it, because the answer differs sharply by project.

For a library that must support Trio, the port is not optional. The alternative is a second test suite, which drifts, or no coverage of the second runtime at all, which means the support claim is untested.

For an application that will only ever run on asyncio, the case rests entirely on the API. Cancel scopes are genuinely better than wait_for for anything with nested deadlines; task groups are better than gather for anything where a sibling must not outlive its peers; memory object streams are better than queues because closing is observable. If the suite already fights those three things, the port pays. If it does not, the port is churn.

Two costs belong on the other side of that ledger. Every engineer who touches the suite has to learn a second vocabulary, which is small but real, and the anyio dependency joins the test requirements permanently. Neither is a reason to avoid the port, but both are reasons to decide deliberately rather than because a blog post recommended it. Writing the decision down, with the reason, saves the argument being had again in six months when somebody notices two async idioms in one repository.

For a codebase in the middle — an internal service with some concurrency and no Trio requirement — a useful compromise is to adopt AnyIO's primitives in new tests without parametrising the backend. The API improves, the runtime stays pinned to asyncio, and the door to the second backend stays open at the cost of one line in a fixture.

When the port pays for itself Three project types. A library that must support Trio has to port, since the alternative is an untested support claim. An asyncio-only application ports only if it is already fighting timeouts, gather semantics or queue closing. A middle case adopts AnyIO primitives without parametrising the backend, keeping the option open. Three answers, depending on what ships Trio must work the matrix is the proof of the support claim port fully both backends everywhere asyncio only port if nested deadlines or gather semantics hurt otherwise churn measure the pain first undecided use anyio primitives pin the backend option stays open one line to enable trio
The right-hand option is underused. It takes the API improvement immediately and defers the runtime decision to whenever it actually arises.

Frequently Asked Questions

Can both plugins stay installed during the migration? Yes, and they usually must. Keep pytest-asyncio governing the unported directories and AnyIO governing the ported ones, running them as separate invocations so neither claims the other's tests. Remove pytest-asyncio only once the last module has moved.

What replaces a session-scoped async fixture? A synchronous session-scoped fixture that performs the expensive discovery, plus a per-test async fixture that builds a cheap handle from it. AnyIO gives each test item its own runtime, so a session-scoped async fixture has no single runtime to live on.

Do assertions change meaning during the port? Two do. asyncio.wait_for raises asyncio.TimeoutError while anyio.fail_after raises TimeoutError, which are the same class from Python 3.11 but not before. And gather's first-exception behaviour becomes a task group's ExceptionGroup, so pytest.raises on a leaf type stops matching.

← Back to Testing with AnyIO & Trio