skip to content

How do you test that a coroutine's cancellation and timeout branches actually run?

level: seniorimportance: should knowfreq 42%

answer

  1. The unhappy exit needs a test
  2. Cancel is only a request
  3. Await it before asserting
  4. Assert cancelled() plus the cleanup
  5. Tiny timeout, never-completing await

basics

~10 s

Start the coroutine as a task, await an event so it is running, then cancel and await it, asserting task.cancelled() and the cleanup. Test the timeout branch with a tiny asyncio.timeout budget.

solid answer

~40 s

Cancellation is a code path, so it needs a test that reaches it. Create the task, await an `asyncio.Event` the coroutine sets so you know it is running (cancelling before it starts skips the branch entirely), then `task.cancel()`. Cancel only *requests* cancellation — the test must `await` the task for `CancelledError` to be delivered and the cleanup to run. Then assert two things: `task.cancelled()` is `True`, proving the coroutine re-raised rather than swallowing it, and the observable cleanup happened. For the timeout branch, wrap work that never completes in `async with asyncio.timeout(0.01):` and assert `TimeoutError` was raised — never sleep the real duration and never assert on measured elapsed time. Since 3.11 `asyncio.TimeoutError` is the builtin `TimeoutError`, and `CancelledError` derives from `BaseException`, so `except Exception:` will not catch it.

code

python · 31 lines
python
import asyncio


class PickListBuilder:
    def __init__(self) -> None:
        self.reservation_released = False

    async def build(self, started: asyncio.Event) -> None:
        started.set()
        try:
            await asyncio.Event().wait()      # never completes on its own
        except asyncio.CancelledError:
            self.reservation_released = True  # the cleanup branch under test
            raise


async def main() -> None:
    builder = PickListBuilder()
    started = asyncio.Event()
    task = asyncio.create_task(builder.build(started))
    await started.wait()                      # cancel only once it is running
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        pass
    assert task.cancelled()                   # it did not swallow the cancel
    assert builder.reservation_released


asyncio.run(main())

go deeper

for a junior

Be ready to say that Task.cancel only requests cancellation and that the task must be awaited before you can assert anything, and that asyncio.CancelledError is what the coroutine sees.

for a middle

Explain the mechanics: CancelledError is thrown in at the suspension point, it derives from BaseException so except Exception misses it, and cleanup code must re-raise after releasing resources.

for a senior

Demonstrate the production judgement: cleanup on the cancel path is what prevents leaked resources when clients disconnect, so assert both task.cancelled() and the side effect, and test timeouts with a tiny budget rather than real waiting.

for a principal

Own the strategy: decide where cancellation boundaries and deadlines live in the system, and make 'every long-running coroutine has a tested cancel path' a reviewable standard rather than a per-author habit.

### Cancellation and timeout are code paths, so they need tests Every long-running coroutine has at least two exits: the happy path, and the path where somebody cancels it or a deadline expires. The second exit is where the resource cleanup lives — releasing a warehouse reservation, returning a connection, flushing a partial pick-list — and it is almost never exercised by the happy-path tests. Untested cleanup code is how a service that looks fine in CI leaks reservations in production every time a client disconnects. ### Driving the cancellation branch The recipe has four steps, and skipping any one of them produces a test that passes without testing anything. **1. Start the coroutine as a task.** `asyncio.create_task` schedules it; the test keeps the task object, because that is the handle you cancel and await. **2. Wait until it has actually reached the await you intend to interrupt.** If you cancel immediately, the task may be cancelled before its body has run a single line, and the `except` branch you wanted to test never executes — the test goes green having proved nothing. Use an `asyncio.Event` the coroutine sets at the point of interest, not a sleep. **3. Cancel, then await.** `Task.cancel()` only *requests* cancellation: it arranges for `CancelledError` to be thrown into the coroutine at its current suspension point the next time the loop runs it. Nothing has happened yet when `cancel()` returns. The test must `await` the task (or gather it) for the cancellation to be delivered and the cleanup to run. **4. Assert on both the outcome and the side effect.** `task.cancelled()` must be `True`, and the observable cleanup must have happened. Asserting only the side effect misses a coroutine that swallows the cancellation; asserting only `cancelled()` misses cleanup that silently did not run. ### Why `cancelled()` is the sharp assertion `asyncio.CancelledError` inherits from `BaseException`, not `Exception` — that has been true since 3.8, and it exists precisely so that a blanket `except Exception:` in application code does not accidentally eat a cancellation. A coroutine that catches `CancelledError` and returns normally instead of re-raising is a real and common bug: the task finishes *successfully*, `task.cancelled()` is `False`, awaiting it yields a value, and the surrounding structured-concurrency machinery believes the shutdown succeeded when it did not. That single assertion catches the bug, which is why it belongs in the test even when the cleanup assertion already passes. The correct shape in the code under test is catch, clean up, re-raise: ```python try: await do_work() except asyncio.CancelledError: release_reservation() raise ``` A `finally:` block is the alternative when the cleanup is unconditional; the `except` form is what you want when the cleanup is specific to being cancelled. ### The timeout branch Test it by making the work *never* complete and the budget *tiny*, never by sleeping the real duration: ```python async with asyncio.timeout(0.01): await never_completes() ``` `asyncio.timeout` and `asyncio.timeout_at` arrived in 3.11 and are the modern form; `asyncio.wait_for` still exists and does the same job for a single awaitable. Since 3.11 `asyncio.TimeoutError` is an alias of the builtin `TimeoutError`, so `except TimeoutError:` is the correct catch on 3.14 — older material that carefully distinguishes the two is out of date. The mechanism to explain in an interview: a timeout scope works *by cancelling* the task inside it and converting that cancellation into `TimeoutError` at the boundary. So the coroutine under test sees `CancelledError` internally, and the caller sees `TimeoutError`. Code that swallows `CancelledError` therefore breaks timeouts too — the deadline expires and nothing stops. That is one bug with two symptoms, and one test for each. Two anti-patterns to name explicitly. First, asserting on elapsed wall-clock time (`assert 0.09 < elapsed < 0.12`) reintroduces exactly the machine-speed dependence you were trying to avoid; assert that the branch was taken, not how long it took. Second, using a real multi-second timeout in a test makes the suite slow for no additional coverage: the code path is identical at 10 ms. ### Edges worth a sentence `asyncio.shield` protects an inner awaitable from an outer cancellation, and a test for shielded code must assert that the inner work *continued*. `Task.uncancel` and `Task.cancelling` (3.11) exist so that a nested timeout scope can absorb its own cancellation without breaking an outer one; they matter when you write your own scope-like helper, and asserting on `cancelling()` counts is how you prove such a helper does not swallow an outer cancel.

  • How do you prove the coroutine did not swallow the cancellation?
    Await the cancelled task and assert `task.cancelled()` is `True`. A coroutine that catches `CancelledError` and returns instead of re-raising completes successfully, so `cancelled()` is `False` and awaiting it yields a value. Asserting only the cleanup side effect would still pass, which is why both assertions belong in the test.
  • Why should the test wait for an event before calling Task.cancel()?
    `create_task` only schedules; the coroutine may not have executed a line. Cancelling immediately can kill it before it reaches the await you meant to interrupt, so the `except`/`finally` branch never runs and the test passes without covering anything. Awaiting an event the coroutine sets pins the exact moment it is inside the region under test.
  • Why does code that swallows CancelledError also break timeouts?
    A timeout scope works by cancelling the task inside it and converting that cancellation into `TimeoutError` at the boundary. If the inner coroutine catches `CancelledError` and carries on, the deadline expires but nothing stops — one bug with two symptoms, which is why both branches deserve their own test.

Testing only the happy path is like fire-drilling a warehouse by walking out of the front door: the exit you never rehearse is the one that fails when the alarm is real.

saying these in an interview costs you the question

  • Catches CancelledError and never re-raises it
  • Expects Task.cancel to stop the coroutine immediately
  • Asserts on measured elapsed time to verify a timeout
  • Uses except Exception expecting to catch CancelledError
  • Cancels the task before it has reached any await
  • Sleeps the real timeout duration inside the test

context