Timeouts, Cancellation & Deadlines

A hung test is worse than a failing one. A failure names a file, a line and an assertion; a hang produces a job that runs to the platform's limit and is killed with no output, at which point the only available diagnosis is "something in the suite didn't finish". Async and threaded suites hang for a small number of well-understood reasons, and every one of them is preventable by layering deadlines so that waiting always has an end.

Prerequisites

pytest >= 8.0 and pytest-timeout >= 2.3.
Python 3.11+ for asyncio.timeout and asyncio.TaskGroup; on 3.10 use async_timeout or anyio.fail_after, which behave the same way.
A working understanding of how cancellation is delivered in asyncio — as an exception at the next await, not as a thread kill.
The loop-lifetime rules from pytest-asyncio in depth, since a mismatched loop is a common cause of a hang that looks like a timeout bug.

Core concept: three layers, three jobs

Deadlines in a test suite come in three layers and conflating them is the usual mistake.

The suite ceiling is a wall-clock limit applied to every test by pytest-timeout. Its job is containment: no test may run forever, whatever goes wrong. It is not an assertion and should never be tight enough to fail a merely slow test.

The per-test override raises or lowers the ceiling for a specific test that is legitimately slower or that is deliberately exercising a hang. Its job is to keep the ceiling honest for everyone else.

The per-operation deadline lives inside the test body, wrapping the call under test. Its job is assertion: it states that this operation must complete, or must give up, within a bound, and it is the only one of the three that tests the code's own timeout behaviour.

Only the innermost layer tests the code. The outer two exist so that when the innermost is missing or wrong, the failure is still legible.

Step-by-step implementation

1. Set the suite ceiling

# pyproject.toml
[tool.pytest.ini_options]
timeout = 60
# "thread" dumps every thread's stack before killing the process, which is the
# only useful output when the hang is a deadlock rather than a slow call.
timeout_method = "thread"

signal is the other method: it raises inside the running test at the exact line, producing an ordinary pytest failure with a real traceback. It is better when it works, and it only works on the main thread of a Unix process, so a suite that uses worker threads or runs on Windows needs thread.

2. Override where a test is genuinely slower

import pytest


@pytest.mark.timeout(300)              # a migration test against a real database
def test_full_migration_applies(engine_migrated):
    assert engine_migrated.dialect.has_table(engine_migrated.connect(), "widget")

Raising the global ceiling to accommodate one slow test costs every other test its containment. The marker keeps the exception local and documented.

3. Put the real assertion inside the test

import asyncio

import pytest


async def test_client_gives_up_on_a_stalled_server(stalling_server):
    # This asserts the CLIENT's timeout behaviour. Without it, the test would
    # pass only because pytest-timeout eventually killed the process.
    with pytest.raises(TimeoutError):
        async with asyncio.timeout(0.5):
            await stalling_server.client.fetch("/never-responds")

    # And the cleanup path must have run.
    assert stalling_server.client.open_connections == 0

4. Test the cancellation path explicitly

import asyncio

import pytest


async def test_cancellation_releases_the_lock(worker, lock):
    task = asyncio.create_task(worker.run_forever())
    await worker.started.wait()        # deterministic: wait for the signal, not a sleep

    task.cancel()
    with pytest.raises(asyncio.CancelledError):
        await task                     # awaiting the cancelled task re-raises

    assert not lock.locked(), "the worker's finally block did not release the lock"

The last assertion is the test. task.cancel() raising CancelledError proves only that cancellation was delivered; whether the coroutine's finally ran correctly is a separate fact, and it is the one that matters in production.

5. Shield only what must complete

import asyncio


async def flush_and_close(buffer, connection):
    try:
        await stream_forever(buffer, connection)
    finally:
        # The flush must finish even if we are being cancelled — but it gets its
        # own bound, so a stuck flush cannot outlive the cancellation forever.
        async with asyncio.timeout(2.0):
            await asyncio.shield(buffer.flush(connection))

An unshielded await inside a finally is itself cancelled immediately when the outer cancellation is already in flight, so the flush never happens. A shield without a bound is the opposite failure: cancellation can no longer stop it. Both together is the correct shape, and it is worth a test of its own.

Verification

Prove the ceiling works by writing a test that deliberately hangs, running it once, and then deleting it — or keeping it behind a marker that only runs on demand:

import time

import pytest


@pytest.mark.skip(reason="run manually to verify the timeout configuration")
@pytest.mark.timeout(2)
def test_timeout_configuration_is_live():
    time.sleep(30)

$ pytest -q -m "" --no-skip tests/test_meta.py
+++++++++++++++++++++ Timeout +++++++++++++++++++++
~~~~~~~~ Stack of MainThread (140234...) ~~~~~~~~
  File "tests/test_meta.py", line 9, in test_timeout_configuration_is_live
    time.sleep(30)
+++++++++++++++++++ Timeout ++++++++++++++++++++++

Seeing the stack dump once is worth more than assuming the configuration is correct. Teams routinely discover at this point that timeout_method was defaulting to signal inside a threaded suite, where it silently does nothing.

Troubleshooting

Symptom	Root cause	Fix
Timeout never fires	`signal` method inside a worker thread	Set `timeout_method = "thread"`
Coroutine keeps running after the deadline	No await point reached; blocked in sync code	Move blocking work to `asyncio.to_thread`
`CancelledError` swallowed, task never stops	`except BaseException` or a bare `except:` in the coroutine	Re-raise `CancelledError` after cleanup
Cleanup skipped on timeout	`await` in `finally` cancelled immediately	Wrap in `asyncio.shield` with its own deadline
Timeout fires on a healthy CI runner	Bound set near the average, not the worst case	Raise it an order of magnitude; it is a net, not an assertion
`TimeoutError` vs `asyncio.TimeoutError` confusion	They are the same class from 3.11	Catch `TimeoutError`; drop the alias

Why a coroutine can ignore a deadline

asyncio.timeout does not stop code. It schedules a cancellation on the task, and that cancellation is delivered as a CancelledError raised at the next suspension point. A coroutine with no suspension point available cannot receive it.

That is exactly what happens when a coroutine calls blocking code: time.sleep(30), a synchronous requests.get, a CPU-bound loop, a C extension that does not release the GIL. The loop itself is blocked, so it cannot even run the callback that would deliver the cancellation. The deadline expires, nothing happens, and the suite ceiling eventually kills the process — with a stack that correctly points at the blocking call.

The lower timeline is why the suite ceiling exists. No amount of per-operation deadline discipline helps when the loop itself cannot run.

A related failure is the coroutine that receives the cancellation and refuses it. except Exception does not catch CancelledError in Python 3.8+, because it inherits from BaseException — but except BaseException, a bare except:, or a finally that awaits something slow all give the coroutine an opportunity to keep running. Code that catches CancelledError to do cleanup must re-raise it; anything else silently converts a cancellation into a delay, and the test that would have caught it is the one asserting the task ends up in the cancelled state.

Replacing waits with conditions

Every time.sleep and await asyncio.sleep(0.5) in a test is a deadline in disguise — one with no diagnosis when it is too short and no speed when it is too long. The replacement is always a condition plus a bound.

import asyncio


async def wait_until(predicate, *, timeout=2.0, interval=0.01):
    """Poll until predicate() is true, or fail with a useful message."""
    async with asyncio.timeout(timeout):
        while not predicate():
            await asyncio.sleep(interval)    # yields; never a fixed total wait


async def test_worker_drains_the_queue(worker, queue):
    await queue.put({"id": 1})
    # Returns as soon as the condition holds — typically in one interval.
    await wait_until(lambda: queue.empty())
    assert worker.processed == [{"id": 1}]

This runs in roughly one polling interval when things are healthy and fails in timeout seconds with a TimeoutError when they are not — versus a fixed sleep(2), which always costs two seconds and, when the work takes 2.1 seconds on a loaded runner, produces a failure that looks like a bug in the worker. Better still is an explicit signal from the code under test: an asyncio.Event the worker sets, which removes polling entirely. The trade-off between the two, and the cases where polling is the only option, is the subject of replacing sleep-based waits with polling assertions.

Deadlines for threaded and blocking code

Nothing in asyncio helps a synchronous test that hangs on a lock, a socket read or a subprocess. The standard library primitives all accept a timeout, and the discipline is simply to pass one every time.

import queue
import subprocess
import threading

import pytest


def test_worker_publishes_within_the_budget(worker, results: queue.Queue):
    worker.start()
    try:
        # queue.get() without a timeout is an unbounded wait; with one it is an
        # assertion that the worker produced something in time.
        item = results.get(timeout=5)
    except queue.Empty:
        pytest.fail("worker produced nothing within 5 seconds")
    finally:
        worker.stop()
        worker.join(timeout=5)
        assert not worker.is_alive(), "worker ignored stop()"

    assert item["status"] == "done"


def test_cli_exits_rather_than_waiting_for_input():
    # A subprocess with no timeout is the most common source of a hung suite,
    # because the child inherits stdin and blocks on a prompt nobody sees.
    completed = subprocess.run(
        ["python", "-m", "myapp.cli", "--check"],
        capture_output=True,
        timeout=30,
        stdin=subprocess.DEVNULL,   # a closed stdin turns a prompt into an error
        text=True,
    )
    assert completed.returncode == 0, completed.stderr

Four standard-library calls account for most synchronous hangs: Queue.get, Lock.acquire, Thread.join and subprocess.run/communicate. All four default to waiting forever, and all four take a timeout argument. A lint rule or a review habit that rejects any of them without one removes an entire category of unreproducible CI failures.

subprocess deserves the extra note above. A child process that reads from stdin will block indefinitely when stdin is an interactive terminal locally and an inherited pipe in CI — the classic "works on my machine, hangs in the pipeline" case. Passing stdin=subprocess.DEVNULL converts the wait into an immediate EOFError inside the child, which surfaces as a normal non-zero exit with a traceback in stderr.

None of these produce a diagnosis when they hang; all four produce a precise one when bounded. The default is the bug.

Making a hang reportable

When a hang does reach CI despite the layers above, the run should still yield enough information to diagnose it. Two configuration lines cover most of the gap.

faulthandler is enabled by pytest automatically, and faulthandler_timeout tells it to dump every thread's stack after a given number of seconds without finishing a test:

[tool.pytest.ini_options]
faulthandler_timeout = 45     # below the suite ceiling, so the dump lands first

Setting it below timeout is the detail that matters: the dump has to happen while the process is still alive. A faulthandler_timeout of 45 with a timeout of 60 gives fifteen seconds in which the stacks are written and flushed before pytest-timeout kills the process.

The second line is -p no:cacheprovider plus -v in the CI invocation, so the log shows which test started even when none finished. A hung run whose last log line is the name of the offending test is a two-minute diagnosis; one that buffered its output and printed nothing is an hour of bisecting. The broader set of habits for making CI failures self-describing — artifact capture, container logs, structured output — is collected in capturing artifacts from a failed CI test run.

Choosing the numbers

Timeout values that cause flakiness are almost always set near the observed duration rather than an order of magnitude above it. The reasoning that produces a bad number is "this takes 40 ms, so 200 ms is generous" — which ignores that CI runners are shared, that a cold import can add a second, and that the same suite will one day run on a machine with a tenth of the cores.

A workable rule: the suite ceiling sits at ten to sixty seconds, high enough that no healthy test approaches it; per-test overrides are used sparingly and documented with the reason; and per-operation deadlines inside tests are chosen to be well above the healthy case but well below the suite ceiling, so that when they fire the failure is attributed to the operation rather than to the runner. Where a test genuinely needs to assert on a tight bound — "the cache must answer in under 5 ms" — that is a benchmark, not a test, and belongs with pytest-benchmark, which measures repeatedly and reports distributions rather than failing on a single sample.

One more consideration applies to suites that run under pytest-xdist. With eight workers on a four-core runner, every test is competing for CPU, and a bound calibrated on an idle machine will fire spuriously. The fix is not to raise every timeout but to measure under the configuration that actually runs in CI: run the suite locally with the same worker count, take the slowest observed duration from --durations=0, and set the ceiling well clear of that. A timeout tuned against a serial run and deployed to a parallel one is the single most common source of a suite that is green locally and intermittently red in the pipeline. Record the chosen numbers and the measurement they came from in a comment next to the configuration, because the next person to see a spurious timeout will otherwise double it, and doubling an already-generous bound simply doubles how long a genuine hang takes to surface.

Frequently Asked Questions

What is the difference between the signal and thread timeout methods? The signal method uses SIGALRM, works only on the main thread of a Unix process, and interrupts the running code with a traceback at the exact line. The thread method runs a watchdog thread that dumps every thread's stack and then kills the process; it works on Windows and inside threads but cannot produce a normal test failure. Use signal where it is available and thread where it is not.

Why does my coroutine keep running after asyncio.timeout expires? Cancellation is delivered as an exception at the next await point. A coroutine that is blocked in synchronous code — a CPU loop, a blocking socket read, time.sleep — never reaches an await and therefore never sees the cancellation. Move blocking work to asyncio.to_thread so the loop keeps a suspension point available.

Should cleanup code be shielded from cancellation? Only the part that must complete, and only with a bound. asyncio.shield around a short release or rollback is legitimate; shielding a whole cleanup coroutine means a timeout cannot stop it, which reintroduces the hang the timeout existed to prevent. Shield the smallest possible region and give it its own deadline.

Is a global pytest timeout enough on its own? No. A suite-wide ceiling stops a hang from consuming the job, but it tells you nothing about whether the code's own timeout logic works. Per-operation deadlines inside tests assert the behaviour; the global ceiling is a safety net for the cases the assertions miss.

How do I choose a timeout value that is not flaky? Set it an order of magnitude above the observed worst case, not just above the average. A test that normally takes 50 ms gets a 5-second ceiling: high enough that a slow runner never trips it, low enough that a genuine hang is caught in seconds rather than at the job limit.

Configure the suite ceiling properly in failing fast with pytest-timeout.
Cover the cleanup paths with testing cancellation and cleanup paths.
Remove fixed waits using replacing sleep-based waits with polling assertions.
Get a stack dump from a deadlocked threaded test with dumping stacks on deadlock with faulthandler.
When the hang is an await that never resolves, trace it with tracking down a hung await with task stacks.

← Back to Testing Async & Concurrent Python