Testing Code That Uses Task Groups

Migrating from asyncio.gather to a task group changes the type that reaches the caller, and tests written for the old shape stop asserting anything useful. pytest.raises(ValueError) fails with "DID NOT RAISE" even though a ValueError was raised, because what escaped the block was an ExceptionGroup containing it. Testing structured concurrency means asserting on the group's contents, on the cancellation of siblings, and on the guarantee that nothing outlives the block.

Prerequisites

• Python 3.11+ for asyncio.TaskGroup, ExceptionGroup and except*; on 3.10 use anyio.create_task_group() with the exceptiongroup backport.
• anyio >= 4.0 if the code is backend-portable, plus the setup in running one test on asyncio and Trio.
• pytest >= 8.0.

Solution

Assert on the group's shape, then on the side effects its cancellation was supposed to produce.

import anyio
import pytest

pytestmark = pytest.mark.anyio


async def fan_out(urls, fetch, cancelled):
    async def run(url):
        try:
            return await fetch(url)
        finally:
            # Observable cleanup: how the test proves this task was stopped.
            if anyio.get_cancelled_exc_class() and not _completed(url):
                cancelled.append(url)

    async with anyio.create_task_group() as tg:
        for url in urls:
            tg.start_soon(run, url)


async def test_one_failure_cancels_the_siblings(failing_fetch):
    cancelled: list[str] = []

    # The group is what propagates; matching the leaf type would not raise.
    with pytest.raises(ExceptionGroup) as excinfo:
        await fan_out(["/a", "/b", "/c"], failing_fetch, cancelled)

    # Shape: exactly one real failure, the rest cancelled cleanly.
    assert len(excinfo.value.exceptions) == 1
    assert isinstance(excinfo.value.exceptions[0], TimeoutError)

    # Behaviour: the siblings actually stopped, rather than being left running.
    assert sorted(cancelled) == ["/c"]

Why this works

A task group's __aexit__ waits for every child. If a child raises, the group cancels the remaining children, waits for them to finish unwinding, and then raises an ExceptionGroup containing every exception that actually escaped — cancellations that were absorbed cleanly do not appear. That is why counting exceptions distinguishes "one thing failed and the others stopped properly" from "three independent failures".

The cancellation is delivered as an exception at each sibling's next await, so a sibling's finally runs and can record the fact. Observing that record is the only way a test outside the group can verify the cancellation happened, because the group owns the task objects and never exposes them.

Edge cases and failure modes

• A child that swallows cancellation. except BaseException: pass inside a task means the group waits forever for it. The test symptom is a hang, which is why these tests need a deadline — see failing fast with pytest-timeout.
• Matching with except* outside a group. except* always binds an ExceptionGroup, even when one exception matched, so group.exceptions is what to iterate — not group itself.
• Nested groups. An inner group's failure propagates as a group inside the outer group's group. Python flattens some of this; assert on the flattened contents rather than the nesting depth.
• start_soon with a coroutine object. Both asyncio's and AnyIO's task groups want a function and its arguments, not an already-created coroutine. Passing run(url) instead of run, url raises a confusing type error.
• Exceptions raised after the block. Code placed after async with runs only if the group exited cleanly. Cleanup that must happen regardless belongs in a finally around the whole block.

Using start instead of sleeping for readiness

The commonest reason a task-group test is flaky is a sleep standing in for "the server is listening now". start() removes it.

import anyio
import pytest

pytestmark = pytest.mark.anyio


async def serve(port_holder, *, task_status=anyio.TASK_STATUS_IGNORED):
    listener = await anyio.create_tcp_listener(local_port=0)
    port_holder.append(listener.extra(anyio.abc.SocketAttribute.local_address)[1])
    # Signals the caller that initialisation is complete. start() returns here.
    task_status.started()
    async with listener:
        await listener.serve(handle)


async def test_client_connects_to_the_started_server():
    ports: list[int] = []
    async with anyio.create_task_group() as tg:
        # start() does NOT return until task_status.started() is called.
        await tg.start(serve, ports)

        # No sleep anywhere: the port is known and the listener is bound.
        async with await anyio.connect_tcp("127.0.0.1", ports[0]) as stream:
            await stream.send(b"ping")
            assert await stream.receive() == b"pong"

        tg.cancel_scope.cancel()          # stop the server; the block then exits

start() is the difference between a test that waits exactly as long as startup takes and one that waits a fixed guess. It also carries information: whatever the task passes to started() is returned by start(), so a server can hand back its bound port rather than the test fishing it out of a shared list.

The tg.cancel_scope.cancel() at the end is necessary because a serving task never finishes on its own, and the group will not exit until every child does. Forgetting it produces a test that hangs at the closing brace of the async with, which is a confusing place for a hang until the rule is internalised: the block waits for all children, always.

Nesting, and what the group flattens

Real code nests groups: a coordinator opens one, each child opens another. The exception shape that results is the part teams get wrong, because it is not simply "a group of groups".

import anyio
import pytest

pytestmark = pytest.mark.anyio


async def test_nested_group_failures_are_readable(flaky_shard):
    with pytest.raises(BaseExceptionGroup) as excinfo:
        async with anyio.create_task_group() as outer:
            for shard in ("a", "b"):
                outer.start_soon(process_shard, shard, flaky_shard)

    # Flatten before asserting: the nesting depth is an implementation detail
    # of how many groups happened to be open, and it changes under refactoring.
    leaves = list(_flatten(excinfo.value))
    assert [type(e) for e in leaves] == [ValueError]
    assert str(leaves[0]) == "shard b is corrupt"


def _flatten(exc):
    if isinstance(exc, BaseExceptionGroup):
        for inner in exc.exceptions:
            yield from _flatten(inner)
    else:
        yield exc

Asserting on flattened leaves rather than on the nesting is the durable choice. Adding an intermediate group — because a coordinator grew a retry wrapper, say — changes the depth without changing what went wrong, and a test that asserted on depth breaks for no reason a reader can act on.

Python's own except* does some flattening for you — matching a leaf type reaches into nested groups — but the group it binds preserves the original structure, so a hand-written flatten is still the clearest thing to assert on. The helper is six lines, lives in the test package, and pays for itself the first time a coordinator gains a layer.

Proving nothing outlives the block

The headline promise of structured concurrency is that no task escapes its scope. That promise is worth a test, because the ways it breaks are subtle and none of them fail loudly on their own.

import anyio
import pytest

pytestmark = pytest.mark.anyio


async def test_no_task_outlives_the_group(resource_tracker):
    async with anyio.create_task_group() as tg:
        tg.start_soon(worker, resource_tracker)
        tg.start_soon(worker, resource_tracker)

    # Every resource the workers took has been returned by the time the
    # block exits — the block waited for them, so this cannot be a race.
    assert resource_tracker.outstanding == 0
    assert resource_tracker.opened == resource_tracker.closed

The assertion is placed after the async with, and that position is what makes it meaningful. Inside the block it would be racing the workers; outside it, the group has already waited for every child, so any imbalance is a genuine leak rather than a timing artefact.

Three leaks this catches are worth naming. A worker that starts a nested task with asyncio.create_task rather than through the group escapes the scope entirely and keeps running after the block exits. A worker whose cleanup awaits something slow can be cancelled mid-cleanup, releasing half its resources. And a worker holding a lock when it is cancelled releases it only if the acquisition used async with rather than a manual acquire/release pair.

For a suite that uses task groups widely, the same check generalises into an autouse fixture asserting that no tasks remain pending at the end of every test, as described in pytest-asyncio in depth. Under a function-scoped runtime a stray task dies with the loop and the leak is invisible; under any shared scope it survives, and the fixture converts "the suite goes strange after test forty" into "test thirty-nine leaked a task".

Frequently Asked Questions

Why does pytest.raises(ValueError) fail when a task group child raised ValueError? Because what propagated was an ExceptionGroup wrapping it, not the ValueError itself. Match on ExceptionGroup and assert on its exceptions list, or use pytest.raises(ExceptionGroup) with a check on the contents. Python 3.11's except* exists for exactly this.

How do I assert that sibling tasks were cancelled? Give each task an observable cleanup side effect — appending to a list in its finally, releasing a lock, closing a fake connection — and assert on that after the group exits. Asserting on task.cancelled() is not possible from outside, because the group owns the task objects.

What is the difference between start_soon and start?start_soon schedules the task and returns immediately; start runs the task until it calls task_status.started() and only then returns, so the caller knows initialisation finished. Use start whenever the test needs the task to be ready before it proceeds, which removes a whole class of sleep-based waiting.

Related

• Testing with AnyIO & Trio — the cancel-scope semantics these groups are built on.
• Testing Cancellation and Cleanup Paths — asserting that the unwinding did what it should.
• Replacing Sleep-Based Waits with Polling Assertions — what to do when start is unavailable.
• Reading Tracebacks & Exception Chains — how a nested group prints, and which frames matter.

← Back to Testing with AnyIO & Trio