← back to the archiveCover illustration for “Test asyncio timeouts through cleanup”
TILday 129·today·Published ·by Andy Padia

Test asyncio timeouts through cleanup

In short: An asyncio timeout requests cancellation; test cleanup order and the final outcome before treating a timed-out agent tool as stopped.

python ▸ import asyncio

async def main():
    events = []

    async def operation():
        try:
            events.append("started")
            await asyncio.Event().wait()
        finally:
            events.append("cleanup_started")
            await asyncio.sleep(0.02)
            events.append("cleanup_finished")

    task = asyncio.create_task(operation())
    try:
        await asyncio.wait_for(task, timeout=0.01)
    except TimeoutError:
        events.append("timeout_reported")

    assert events == ["started", "cleanup_started",
                      "cleanup_finished", "timeout_reported"]
    assert task.done() and task.cancelled()
    print(events)

asyncio.run(main())

A timeout expired, cleanup ran, and the caller received a successful-looking fallback. That was one of two synthetic Python 3.14.6 cases checked on 7 October 2026. The difference was a single choice inside the coroutine: propagate cancellation or swallow it.

Test asyncio timeouts through cleanup, not just until the timer fires. My editorial rule for an agent-tool wrapper is to check the final outcome and the order of shutdown events together. A configured deadline tells you when to request cancellation; it does not establish that the work has stopped or that the result will be reported as a timeout.

Why asyncio timeouts can outlast the timer

Python's wait_for documentation says it waits for the awaited operation to finish cancelling. That wait can exceed the timeout. This is useful: releasing a local resource may require work after cancellation arrives. It also means that a timeout value is not a hard upper bound on the caller's wait.

The related CancelledError reference says the exception should normally be raised again after handling it. Catching cancellation to clean up is different from catching it and returning a result. In the second synthetic case, returning "fallback" made wait_for return that value instead of raising TimeoutError.

The synthetic asyncio cleanup test

Save this as timeout_check.py and run python3 timeout_check.py on Python 3.11 or later. It uses no network, database or API credentials. The event that the operation waits on is deliberately never set; cancellation is its exit path.

import asyncio

async def main():
    events = []

    async def operation():
        try:
            events.append("started")
            await asyncio.Event().wait()
        finally:
            events.append("cleanup_started")
            await asyncio.sleep(0.02)
            events.append("cleanup_finished")

    task = asyncio.create_task(operation())
    try:
        await asyncio.wait_for(task, timeout=0.01)
    except TimeoutError:
        events.append("timeout_reported")

    assert events == ["started", "cleanup_started",
                      "cleanup_finished", "timeout_reported"]
    assert task.done() and task.cancelled()
    print(events)

asyncio.run(main())

The useful assertion is the sequence, not a stopwatch threshold. Cleanup must finish before the timeout is reported, and the task must actually be cancelled. The requested sleeps are test inputs, not measured service latency. Don't turn this into a claim that cancellation takes precisely thirty milliseconds on another machine.

The companion negative test explicitly caught CancelledError and returned a fallback. Cleanup still finished, but the final event was returned:fallback; the task was done without being cancelled. That is the failure worth making visible in a wrapper test. A dashboard that counts only returned values could otherwise classify this deliberately abandoned operation as a success.

My judgement for an agent-tool wrapper

For a hypothetical tool that stages a local file before requesting an external action, I want separate evidence for local cleanup and the external action's outcome. This is an engineering recommendation, not a client deployment story. Replace the synthetic cleanup marker with an assertion about the actual resource: the temporary file is absent, or the connection was released through the relevant library's supported mechanism.

Then inspect the caller's response. If a fallback is an intentional product choice, label and test it as a degraded result. Don't let a broad cancellation handler quietly decide what counts as success. Keep the cancellation path out of the normal retry path until the operation's side effects are understood.

Where the asyncio check stops

This example checks one local coroutine receiving one cancellation. It does not test repeated cancellation during cleanup, blocking native code, worker threads or remote services. A local task reaching its terminal state is not a receipt proving that an external action was cancelled. Test those boundaries with the service's own status or cancellation mechanism before promising an end-to-end stop.

There is also no hard-stop guarantee if cleanup itself hangs. The point is to expose that dependency in the test and design a bounded recovery policy for the real resource, not to hide it behind another reassuring timeout number.

What's in it for you

  • Catch a cancellation handler that converts an abandoned operation into apparent success.
  • Verify cleanup and the caller's outcome together before adding a timeout to an agent tool.

A timeout test is finished when cleanup and the final outcome are checked—not when the timer fires.

Sources

#python#testing#reliability
← older drop
EmbeddingGemma 2 needs an indexing contract

related drops

explore all 378 drops →
← back to the archiveday 129