Skip to content

Testing Cancellation and Cleanup Paths

Cancellation paths are the least-tested code in most async services and the most likely to be exercised during an incident — every timeout, every TaskGroup failure, every shutdown runs them. They are also genuinely awkward to test, because a test must get a task to a specific await, cancel it there, and then assert about code that ran while an exception was propagating. The awkwardness is what produces the two common outcomes: no tests at all, or tests built on asyncio.sleep(0.1) that pass locally and flake on CI. Both are avoidable. This guide builds a small suite of cancellation tests — all of which run in 0.23 seconds together — using events rather than sleeps for synchronisation.

Prerequisites

  • Python 3.11+ with pytest, pytest-asyncio and pytest-timeout (pip install pytest pytest-asyncio pytest-timeout).
  • Cancellation semantics from Cancellation Patterns — especially that CancelledError must be re-raised.
  • Async test setup from Testing Async Code.
The shape of a cancellation test 5 ordered steps. The shape of a cancellation test create_task(job) the task under test await started.wait() deterministic, not a sleep task.cancel() at a known point pytest.raises(CancelledError) it must propagate assert the cleanup ran closed once, flushed once The event is what makes the test reproducible on a loaded CI machine.

1. Synchronise with an event, never a sleep

The test needs the task to be at its await before cancelling. asyncio.sleep(0.1) is a guess about how long that takes; an asyncio.Event set by the code under test is a fact:

async def job(resource, started: asyncio.Event, flushed: list):
    async with resource:
        started.set()                                  # "I am now at the await"
        try:
            await asyncio.sleep(10)
        except asyncio.CancelledError:
            flushed.append("flush")
            raise


async def test_cancel_runs_cleanup():
    resource, started, flushed = Resource(), asyncio.Event(), []
    task = asyncio.create_task(job(resource, started, flushed))
    await started.wait()                               # deterministic
    task.cancel()
    with pytest.raises(asyncio.CancelledError):
        await task
    assert task.cancelled()
    assert resource.closed == 1 and not resource.open
    assert flushed == ["flush"]

Where the production code has no natural place for an event, a test double provides one: a fake client whose request() sets an event and then waits forever is both the slow dependency and the synchronisation point. await asyncio.sleep(0) has one legitimate use here — letting a freshly created task begin executing — but it cannot get you to a particular await.

Add timeout = 20 in your pytest configuration via pytest-timeout. A cancellation test that hangs should fail the suite in seconds, not block CI for its full job timeout.

Verify: the test passes with the machine under load, and fails immediately if the task never reaches its await.

How the test waits for the task to reach its await 2 columns contrasting await asyncio.sleep(0.1), await started.wait(). How the test waits for the task to reach its await await asyncio.sleep(0.1) a guess passes on a fast machine flakes on a loaded runner slows the suite down hides real ordering bugs await started.wait() a fact the task signals its own progress no timing assumption runs in microseconds fails loudly if the task never starts Use asyncio.sleep(0) only to let a task begin, never to wait for a specific point in it.

2. Assert that cancellation propagated

The single most valuable assertion is task.cancelled(). It distinguishes a task that was cancelled from one that caught the cancellation and returned normally — the most common cancellation bug there is:

async def test_swallowing_cancellation_is_detected():
    async def bad():
        try:
            await asyncio.sleep(10)
        except asyncio.CancelledError:
            return "swallowed"                         # the bug

    task = asyncio.create_task(bad())
    await asyncio.sleep(0)
    task.cancel()
    result = await task                                # no exception at all!
    assert not task.cancelled()
    assert result == "swallowed"

That test documents the broken behaviour; in a real suite the assertion is inverted — assert task.cancelled() — and it fails the day someone adds an except Exception that happens to catch too much, or a finally that returns.

Pair it with pytest.raises(asyncio.CancelledError) around the await task. Both are needed: the raises clause proves the exception reached the awaiter, and cancelled() proves the task ended in the cancelled state rather than raising a CancelledError it constructed itself.

Verify: introducing a return in an except asyncio.CancelledError block breaks the test.

Did the code handle cancellation correctly? A decision on What state did the task end in with 3 outcomes. Did the code handle cancellation correctly? What state did the task end in? task.cancelled() is True correct it re-raised it returned a value a bug cancellation was swallowed it raised another error suspicious cleanup masked the cancel assert task.cancelled() is the single most valuable line in these tests.

3. Cover the cases that are different code paths

Four scenarios exercise genuinely different machinery, and passing one says nothing about the others.

Cancel mid-await is the case above: cleanup runs, resources close exactly once.

Cancel before the task starts takes a different path entirely — the coroutine never runs, so no cleanup is needed and none should happen:

async def test_cancel_before_start():
    task = asyncio.create_task(job(resource, started, flushed))
    task.cancel()                                      # no await in between
    with pytest.raises(asyncio.CancelledError):
        await task
    assert resource.closed == 0 and flushed == []      # nothing was acquired

Cleanup that hangs verifies that the cleanup path itself is bounded — a finally that awaits a dead connection otherwise blocks shutdown indefinitely:

async def test_cleanup_timeout_is_bounded():
    ...
    started = loop.time()
    task.cancel()
    with pytest.raises(asyncio.CancelledError):
        await task
    assert loop.time() - started < 0.5                 # bounded by the cleanup's own timeout

A sibling failing in a TaskGroup checks the cancellation your code will actually receive most often in production:

async def test_taskgroup_cancels_siblings():
    with pytest.raises(BaseExceptionGroup):
        async with asyncio.TaskGroup() as tg:
            tg.create_task(child(1))
            tg.create_task(child(2))
            tg.create_task(failing())
    assert sorted(cancelled) == [1, 2]

Verify: all four exist for every component that holds a resource across an await.

The cancellation cases worth a test each A grid of 4 rows by 2 columns. The cancellation cases worth a test each scenario what it proves assertion cancel mid-await cleanup runs resource closed exactly once cancel before first run no cleanup needed nothing was acquired cleanup that hangs cleanup is bounded teardown under a deadline a sibling fails the group cancels every child recorded cancellation Each is a different code path; passing one says nothing about the others.

4. Test that shielded work completes

Code that protects a critical section with asyncio.shield deserves a test proving the protection works, because the failure mode — a commit that silently does not happen — is invisible otherwise:

async def test_shield_protects_a_critical_section():
    committed = []
    task = asyncio.create_task(job_with_shielded_commit(committed))
    await asyncio.sleep(0)
    task.cancel()
    with contextlib.suppress(asyncio.CancelledError):
        await task
    await asyncio.sleep(0.1)                           # let the shielded work finish
    assert committed == [True]

The await after the cancellation is the part people leave out. The shielded task outlives the cancelled one by design, so the assertion must be made after it has had a chance to complete — in production that is the shutdown sequence waiting for it, and in the test it is an explicit wait. Better still, have the shielded work set an event and await that, keeping the test deterministic.

Verify: removing the shield makes the test fail.

5. Run cancellation tests as part of shutdown tests

Component-level cancellation tests catch the mechanics; a service-level test catches the ordering. The useful integration test starts the service, sends work, triggers shutdown, and asserts on what finished:

async def test_shutdown_drains_in_flight_work():
    service = await start_service()
    handles = [await service.submit(job) for job in jobs]
    await service.shutdown(grace=2.0)
    assert all(h.done() for h in handles)
    assert service.rejected_after_shutdown == 0
    assert not asyncio.all_tasks() - {asyncio.current_task()}

That last assertion — no leftover tasks — catches the leaks that component tests miss, because asyncio.all_tasks() sees everything the service created, including the background task someone started in a constructor. Graceful shutdown and signals covers the sequence being tested.

Verify: the test fails when a background task is created without being tracked.

Verification

Cancellation testing is adequate when:

  • No test uses a sleep to wait for a state — events or test doubles provide the synchronisation.
  • task.cancelled() is asserted, alongside pytest.raises(asyncio.CancelledError).
  • All four scenarios are covered for each component that holds resources.
  • Cleanup is bounded, proven by a deadline assertion.
  • Shielded work is verified to complete after the cancellation.
  • A service-level test asserts no tasks survive shutdown.

Pitfalls & edge cases

  • asyncio.sleep(0.1) as synchronisation. It passes locally and flakes on a loaded CI runner; worse, it hides ordering bugs.
  • Catching Exception in the test. CancelledError derives from BaseException, so a pytest.raises(Exception) will not match it.
  • Forgetting pytest-timeout. A hanging cancellation test blocks the whole suite.
  • Asserting only that no exception escaped. A swallowed cancellation also produces no exception; check the task's state.
  • Event loop reuse between tests. Leftover tasks from one test can fail the next; assert on asyncio.all_tasks() in a fixture teardown.
  • Testing only the happy cancellation. Cancellation during cleanup, and cancellation of an already-finishing task, are separate paths.

Frequently Asked Questions

How do I test that a coroutine handles cancellation correctly?

Start it with create_task, wait for an asyncio.Event the coroutine sets when it reaches its await, then cancel it. Assert that awaiting the task raises CancelledError, that task.cancelled() is True, and that the cleanup side effects — resources closed, buffers flushed — happened exactly once.

Why are my cancellation tests flaky?

Almost always because they use asyncio.sleep to wait for the task to reach a particular point. That is a timing assumption that fails on a loaded CI machine. Have the code under test — or a test double standing in for a slow dependency — set an event, and await that instead.

How do I detect that code swallowed a CancelledError?

Assert task.cancelled() after awaiting the task. A task that caught the cancellation and returned normally has cancelled() False and produces no exception, so a test that only checks "nothing raised" passes against the bug.

How do I test asyncio.shield in a cancellation test?

Cancel the outer task, suppress the CancelledError, then wait for the shielded work to finish before asserting on its effect — ideally by awaiting an event the shielded coroutine sets. Removing the shield should make that assertion fail, which is how you know the test is meaningful.

Should cancellation tests use pytest-timeout?

Yes. Cancellation bugs manifest as hangs, and without a timeout a single broken test blocks the entire suite until the CI job's own limit. A global timeout of ten to twenty seconds turns a hang into a fast, clear failure.