pytest-timeout leaves asyncio tasks running after test timeout
0 reputation · 28 Mar 2021, 16:01 UTC
0 reputation · 28 Mar 2021, 16:01 UTC
When a test is aborted by the pytest‑timeout plugin after exceeding its limit, the plugin raises a TimeoutError that marks the test as failed, but it does not automatically cancel any asyncio tasks or threads that the test may have started.
The plugin’s implementation differs between Unix, where it uses SIGALRM, and Windows, where it relies on a watchdog thread, which can leave resources such as event loops, open files, or network connections uncleaned if the test does not cooperate with cancellation.
Given this behavior, the goal is to determine whether pytest‑timeout should propagate cancellation to the asyncio event loop (or equivalent concurrency primitives) so that timed‑out tests clean up their resources without requiring manual constructs like asyncio.wait_for inside each test.
Should the plugin automatically cancel running asyncio tasks when a timeout occurs?
What cross‑platform mechanism could be used to propagate cancellation reliably?
Are there configuration options that could enable this behavior while preserving backward compatibility?
33275 reputation · 28 Mar 2021, 23:56 UTC
No — pytest-timeout does not, and by design cannot safely, cancel asyncio tasks when it fires. It interrupts the test by raising TimeoutError in the executing frame (via SIGALRM on POSIX or a watchdog thread on Windows). Any tasks the test created with asyncio.create_task() keep being scheduled on the event loop after the test is reported as failed. Automatic cancellation would need to be an opt-in feature implemented cooperatively with pytest-asyncio, not a default.
The signal handler (or watchdog) raises an exception at whatever bytecode the main thread happens to be executing. It has no knowledge of the event loop, so it never calls task.cancel(). Two consequences follow:
SIGALRM only reaches the main thread. If the test runs its loop (or a blocking call) in a worker thread, the timeout may not even interrupt the right frame.Not by default. Cancelling every task on timeout injects CancelledError into arbitrary finally blocks and fixture teardown, which would break existing suites that rely on normal cleanup running to completion. It can also mask the original timeout if a cancellation error propagates first. That is why the conservative behavior — fail the test, leave cleanup to the test/fixtures — is the current default.
A reliable implementation needs cooperation from the async plugin, because only it knows which loop belongs to the test:
asyncio.get_event_loop(), which is unreliable across threads).asyncio.all_tasks(loop), cancel each, and await them with a short grace period — essentially what loop.shutdown_asyncgens() plus a manual cancel-and-gather does.TimeoutError, so the failure reason stays accurate.This works on Windows too, because cancellation is requested on the loop object rather than by interrupting bytecode — the watchdog thread just needs a thread-safe handle to the loop (loop.call_soon_threadsafe).
An opt-in ini option would preserve existing behavior, e.g.:
# pytest.ini
[pytest]
timeout = 30
timeout_cancel_asyncio = true # proposed, default false
timeout_cancel_grace = 2 # seconds to wait for cancelled tasksDefaulting to off keeps every current suite unchanged; users with leaky async tests opt in explicitly.
Make the tests cancellation-aware so the plugin's blunt instrument is never the cleanup path:
import asyncio, pytest
@pytest.mark.asyncio
async def test_worker():
async with asyncio.timeout(5): # Python 3.11+; else asyncio.wait_for
await run_worker()Also ensure fixtures close loops and connections in finally blocks, and consider anyio-style cancel scopes if you need structured cleanup.
import asyncio, pytest
@pytest.mark.asyncio
async def test_leak():
t = asyncio.create_task(asyncio.sleep(60))
await asyncio.sleep(30) # killed by --timeout=1 firstRun pytest --timeout=1 -s and add a finalizer that prints t.done() — you will see the task still pending. Also confirm with pip show pytest-timeout pytest-asyncio which versions and which asyncio_mode you are on, since loop ownership differs between strict and auto modes. If your traceback shows asyncio.TimeoutError rather than the plugin's TimeoutError, the timeout is coming from your own wait_for, not the plugin — that distinction changes the fix.
Use comments to ask for clarification. Post a solution as an answer.
No question comments on this page.