Testing Async & Concurrent Python

Concurrency turns a test suite's assumptions into scheduling bets. A synchronous test either passes or fails on the code; an async test also depends on which loop is running, which task got to the await first, and whether the fixture that built the connection pool outlived the loop it was bound to. The result is the failure mode every team with an async service eventually meets: tests that are green alone, green locally, and red on a loaded CI runner about one run in forty.

This section is about removing those bets. It covers the mechanics of how a coroutine test is collected and executed, how pytest-asyncio's modes and loop scopes decide which loop your fixtures see, how AnyIO runs one test body on both asyncio and Trio, how to make a race condition reproduce on demand instead of once a fortnight, and how timeouts and cancellation keep a hung await from consuming a CI job's entire budget. Readers are assumed fluent in async/await, tasks, and the difference between concurrency and parallelism; nothing here re-explains the event loop from scratch.

How an async test actually runs

A coroutine function is not a test. When pytest collects async def test_x(), calling it produces a coroutine object and returns it — the body never executes. Without a plugin, pytest sees a test that returned a non-None value, emits PytestUnhandledCoroutineWarning, and skips it. Since pytest 8.4 that warning is an error by default, which is an improvement: the older behaviour silently reported a passing suite in which none of the async tests had run.

A plugin fixes this by hooking pytest_pyfunc_call, intercepting the coroutine before pytest treats it as a return value, and driving it to completion on a loop the plugin owns. Everything that makes async testing subtle follows from that one substitution: the loop is created by the plugin, on a schedule you configure rather than one you call, and every async fixture is bound to whichever loop was current when it ran.

The practical consequence is a rule worth memorising: an object's loop affinity is decided at construction, not at use. asyncio.Lock() records no loop in Python 3.10+, but the moment it is first awaited it attaches to the running loop and stays there. A session-scoped fixture that builds an asyncpg pool under a function-scoped loop leaves a pool holding sockets registered with a selector that gets closed after the first test. The second test then fails with RuntimeError: Event loop is closed or, worse, hangs — a failure explored in detail in debugging the event loop is closed RuntimeError.

Choosing a runner

Three options are in real use, and the choice is about the library under test rather than taste.

pytest-asyncio is the default for asyncio-only code. Since 0.23 it separates collection (asyncio_mode) from loop lifetime (loop_scope), and since 0.24 the loop_scope argument is available on both @pytest.mark.asyncio and @pytest_asyncio.fixture. That separation is the feature: it lets one module share a loop across its tests while the rest of the suite stays function-scoped.

# pyproject.toml equivalent config lives in [tool.pytest.ini_options]
import asyncio

import pytest

pytestmark = pytest.mark.asyncio(loop_scope="module")  # one loop for this module


@pytest.fixture(scope="module")
def module_state():
    return {"seen": []}


async def test_first(module_state):
    module_state["seen"].append(id(asyncio.get_running_loop()))
    assert True


async def test_second(module_state):
    # Same loop id as test_first, because loop_scope="module" kept it alive.
    assert module_state["seen"][0] == id(asyncio.get_running_loop())

anyio takes the opposite approach: it defines the test in terms of an abstract async runtime and parametrises over backends, so one test body runs on asyncio and on Trio. For a library author that is not a convenience, it is the test matrix. The cost is that backend-specific APIs (asyncio.get_running_loop, loop.call_soon) are off-limits inside those tests.

Plain asyncio.run() inside a synchronous test remains a legitimate third option for a handful of tests in an otherwise synchronous suite. It brings no plugin, no fixture integration, and no shared loop — which is exactly why it never surprises anyone.

Fixtures that live on the right loop

The single most productive habit in an async suite is to make the fixture that creates a resource and the loop that owns it share a lifetime. pytest-asyncio expresses that directly: @pytest_asyncio.fixture(loop_scope="session", scope="session") states both halves. Get them out of step and you have written a ScopeMismatch waiting to happen — the synchronous analogue of which is covered in fixing ScopeMismatch errors in pytest.

import asyncpg
import pytest_asyncio


@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def pool():
    # Both scopes say "session": the pool and the loop it registers sockets
    # with are created and destroyed together.
    pool = await asyncpg.create_pool(dsn="postgresql://localhost/test", min_size=1)
    try:
        yield pool
    finally:
        await pool.close()          # runs while the session loop is still alive


@pytest_asyncio.fixture(loop_scope="session")
async def conn(pool):
    # Function-scoped fixture, session-scoped loop: the connection is per test,
    # the loop is shared, and nothing crosses loops.
    async with pool.acquire() as connection:
        transaction = connection.transaction()
        await transaction.start()
        try:
            yield connection
        finally:
            await transaction.rollback()   # every test leaves the database clean

In CI this matters twice over. Creating a connection pool per test costs tens of milliseconds of handshake each time, which on a thousand-test suite is minutes of wall clock; and a per-test pool multiplies the server's connection count by the worker count under pytest-xdist, which is how a test job exhausts max_connections on a shared database. The transactional pattern above — one durable pool, one per-test transaction rolled back at teardown — is developed further in database fixtures and transactional tests.

Threads, tasks, and the shape of a race

Async code is single-threaded, which removes data races between instructions but not between await points. Any read-modify-write that spans an await is interruptible, and a second task entering the same function will observe the intermediate state. That is a genuine race with a deterministic reproduction, because the scheduler is cooperative: if you control where the awaits are, you control the interleaving.

import asyncio


class Counter:
    def __init__(self):
        self.value = 0

    async def increment(self, store):
        current = await store.read()      # suspension point: another task can run here
        await asyncio.sleep(0)            # make the window explicit for the test
        await store.write(current + 1)    # writes a value that may now be stale


async def test_lost_update_is_reproducible(store):
    counter = Counter()
    # Two tasks, one shared store: the second read happens before the first write.
    await asyncio.gather(counter.increment(store), counter.increment(store))
    assert await store.read() == 2       # fails: the increment is lost

asyncio.sleep(0) is the one sleep that belongs in a test. It yields to the loop without waiting for wall-clock time, so the interleaving is forced rather than hoped for, and the test fails every run instead of one in a hundred. The same discipline applied to real threads needs threading.Barrier and threading.Event instead, because preemption there is genuinely non-deterministic — the techniques are in testing threads and race conditions.

Deadlines are not optional

An async test that hangs does not fail — it consumes the job's entire time budget and then gets killed, usually with no traceback and no indication of which test was running. A suite without deadlines therefore has a failure mode strictly worse than a red test: a red pipeline with no diagnosis.

Two layers are worth having. pytest-timeout sets a per-test wall-clock ceiling for the whole suite, with --timeout-method=thread producing a stack dump rather than a bare kill. Inside individual tests, asyncio.timeout() (3.11+) or async_timeout scopes a deadline to the operation actually being exercised, which is what turns "something hung" into "the retry loop never exits when the server sends a partial response".

import asyncio

import pytest


async def test_partial_response_does_not_hang(client):
    # A per-operation deadline: this asserts the timeout behaviour of the code,
    # not merely the health of the test runner.
    with pytest.raises(TimeoutError):
        async with asyncio.timeout(0.5):
            await client.fetch_until_complete("/slow-stream")

    # Cleanup must still have run — the deadline is not an excuse to leak.
    assert client.open_connections == 0

The second assertion is the important one. Cancellation in Python is delivered as an exception at the next await, so every finally on the stack runs — but only if the code actually has one. Tests that assert on post-cancellation state are the only reliable way to keep connection and lock cleanup honest, and they are covered in testing cancellation and cleanup paths.

Structured concurrency changes what a test asserts

asyncio.TaskGroup (3.11+) and anyio.create_task_group() change the failure contract of concurrent code, and tests written for the old contract quietly stop checking anything. Under asyncio.gather(), one task raising leaves its siblings running and returns a single exception; under a task group, the first failure cancels every sibling and the block exits with an ExceptionGroup containing every exception that actually escaped. A test that still writes pytest.raises(ValueError) around a task group will fail with "DID NOT RAISE" even though a ValueError was raised, because what propagated was an ExceptionGroup wrapping it.

import asyncio

import pytest


async def fan_out(urls, fetch):
    results = []
    async with asyncio.TaskGroup() as tg:          # 3.11+
        tasks = [tg.create_task(fetch(url)) for url in urls]
    results.extend(task.result() for task in tasks)
    return results


async def test_one_failure_cancels_the_group(fetch_that_fails_on_second):
    # except* semantics: match inside the group, not against it.
    with pytest.raises(ExceptionGroup) as excinfo:
        await fan_out(["/a", "/b", "/c"], fetch_that_fails_on_second)

    # Assert on the contents, which is where the useful information lives.
    assert len(excinfo.value.exceptions) == 1
    assert isinstance(excinfo.value.exceptions[0], TimeoutError)
    # And assert the siblings were cancelled rather than left running.
    assert fetch_that_fails_on_second.cancelled == ["/c"]

Two assertions matter here and neither is obvious. The first is the shape of the exception group: len(excinfo.value.exceptions) distinguishes "one thing went wrong and the rest were cancelled cleanly" from "three things went wrong independently", which are very different bugs with identical log lines. The second is that the siblings were genuinely cancelled — the whole point of structured concurrency is that no task outlives its block, and that guarantee is worth a test, because a sibling that swallows CancelledError breaks it silently.

For codebases still on 3.10, anyio.create_task_group() provides the same semantics and the exceptiongroup backport supplies ExceptionGroup and except*, which makes the migration a version bump rather than a rewrite. The task group guide works through the assertions in full, including the nesting rules that decide whether a group flattens or wraps.

Keeping an async suite fast without making it flaky

Async suites are usually slow for one of two reasons, and the two want opposite remedies. The first is genuine I/O: hundreds of tests each opening a connection, each paying a TLS handshake. The second is artificial waiting: sleeps inserted to "let things settle", which cost the same time on every run whether or not the thing they wait for has happened.

Fix the second first, because it also fixes flakiness. Every await asyncio.sleep(0.2) in a suite is simultaneously a 200 ms tax and a bet that 200 ms is enough on the slowest runner that will ever execute it. Replacing it with an asyncio.Event that the code under test sets, or with a bounded polling loop that asserts a condition, usually reduces the wait to a few milliseconds and removes the failure mode. The mechanics — including how to keep the polling loop from becoming a busy wait — are in replacing sleep-based waits with polling assertions.

Only then attack the I/O, and do it by widening scope rather than by adding concurrency. A session-scoped loop with a session-scoped pool amortises the expensive setup across the whole run; per-test isolation comes from a transaction that rolls back, not from a fresh connection. Where real parallelism is needed, pytest-xdist process workers each get their own loop and their own pool, which is safe precisely because nothing is shared across processes — but it multiplies the resource footprint by the worker count, so the database's connection limit becomes the real ceiling.

One measurement is worth taking before any of this: run with --durations=25 and check whether the slowest tests are slow in setup or in the call phase. Setup-heavy suites are fixed by scope; call-heavy suites are fixed by removing sleeps or by faking the remote service entirely with the techniques in mocking network and HTTP calls. Guessing which of the two you have is how teams spend a week widening fixture scopes on a suite that was sleeping the whole time.

Deciding what to fake at the async boundary

The hardest judgement in an async suite is where to stop being real. Faking too little turns unit tests into integration tests that need a network; faking too much produces a suite that passes while the service cannot talk to anything. The boundary that works is the protocol boundary — the last place where your code's own types cross into somebody else's.

For HTTP that means faking at the transport, not at the client object. respx for httpx and aioresponses for aiohttp intercept below the client's public API, so the code under test still builds real requests, still runs its own retry and header logic, and still parses real response objects. A hand-rolled AsyncMock client, by contrast, skips all of that: it asserts that you called a method, which is a claim about your own code rather than about the exchange. The full argument is in mocking httpx clients with respx.

For databases the calculation flips. A fake repository is cheap and fast for testing business rules, but any test that exercises SQL, migrations, constraints or isolation semantics has to run against the real engine, because those behaviours exist only in the engine. That is what Testcontainers is for: a real Postgres, started once per session, reachable over a real socket from the session loop.

The last row deserves emphasis. Every few years someone proposes a deterministic fake event loop that fires timers instantly. It makes tests fast and makes them test a different program, because the interleavings that produce real bugs are exactly the ones the fake removes. Control time with an injected clock, control I/O with a transport-level fake, and let the loop be the loop — the reasoning behind injecting rather than patching is in injecting a clock instead of patching datetime.

Common pitfalls and antipatterns

1. Creating a shared primitive at import time. LOCK = asyncio.Lock() at module scope binds to whichever loop first awaits it, then fails in every later test. Root cause: module-level state outliving the loop. Fix: build it inside a fixture whose loop_scope matches its use.
2. Mixing pytest-asyncio and anyio in one module. Both claim pytest_pyfunc_call; whichever is registered first wins, and the other's markers are silently ignored. Root cause: two plugins, one hook. Fix: one runner per module, enforced by keeping backend-parametrised tests in their own directory with a local conftest.py.
3. time.sleep() inside an async test. It blocks the loop, so every other task stops, including the one the test is waiting on. Root cause: a synchronous call on the loop thread. Fix: await asyncio.sleep(...), or push blocking work to asyncio.to_thread, and enable asyncio debug mode so the loop reports the block.
4. Fire-and-forget tasks with no reference. asyncio.create_task(coro) without storing the result lets the garbage collector reclaim the task mid-flight, producing the Task was destroyed but it is pending! warning and a test that passes because the work never happened. Root cause: no strong reference. Fix: keep the task in a set, or use a task group that awaits it.
5. Asserting on wall-clock durations. assert elapsed < 0.1 is a measurement of the CI runner, not the code. Root cause: a timing assertion standing in for a behavioural one. Fix: assert on ordering and on call counts, and reserve durations for benchmarks run with pytest-benchmark.
6. Swallowing CancelledError. A bare except Exception around an await does not catch it in 3.8+, but except BaseException does, and a coroutine that absorbs cancellation makes a timeout unable to stop it. Root cause: over-broad exception handling. Fix: re-raise CancelledError after cleanup, and assert in tests that it propagates.

Frequently Asked Questions

Why do async tests pass alone but fail when the whole suite runs? Almost always because an object was created on one event loop and used on another. Connection pools, asyncio.Lock, asyncio.Queue and anything that captured get_event_loop() at construction bind to the loop that was running at the time. Run the suite with -p no:randomly and pytest --setup-show to see which fixture scope created the object, then match the fixture's loop scope to the object's lifetime.

Should I use pytest-asyncio or AnyIO for a new project? Use AnyIO when the library under test targets both asyncio and Trio, or when you want structured concurrency primitives in the tests themselves. Use pytest-asyncio when the code is asyncio-only and you need fine-grained control over loop scope. Both can coexist in one repository, but never in one test module — two plugins collecting the same coroutine function causes double execution or a silent skip.

How do I test that code handles cancellation correctly? Start the coroutine as a task, let it reach the await you care about, then call task.cancel() and await the task inside pytest.raises(asyncio.CancelledError). Assert on the side effects the cleanup path was supposed to produce — a released lock, a closed connection, a flushed buffer — rather than on the exception alone.

Is asyncio.sleep() acceptable in a test? Only as a yield point of zero duration. asyncio.sleep(0) hands control back to the loop so another task can run and is deterministic. Any positive duration is a bet on scheduling that will eventually lose on a loaded CI runner; replace it with an asyncio.Event, a queue await, or a polling assertion with a deadline.

Do I need locks in tests if the code under test is single-threaded async? No, but you still need to reason about interleaving. A single event loop gives you atomicity only between await points, so any read-modify-write that spans an await can be interleaved by another task. Tests that drive two tasks through the same object are how that class of bug is found.

Related guides

• Start with pytest-asyncio in depth for modes, loop scopes and the fixture rules that follow from them.
• Cover two runtimes with one test body using AnyIO and Trio, which also explains what you give up in exchange.
• Make intermittent concurrency bugs reproducible with testing threads and race conditions.
• Keep a hung await from eating the pipeline with timeouts, cancellation and deadlines.
• When an async test fails for reasons the assertion does not explain, move to debugging async code and event loops and the scoping rules in mastering pytest fixtures.

← Back to all guides