Skip to content

Detecting Leaked Tasks in Tests

A function that starts a background task and returns — fire-and-forget email, a cache refresh, a metrics flush — leaves work running after the test that called it has finished. In production that task may outlive the request, swallow its own errors, or keep the process from shutting down cleanly; in tests it disappears without a trace. Tested with pytest 9.1 and pytest-asyncio 1.4: a handler that launched asyncio.create_task(send_email(...)) and returned passed its test with no warning at all — the plugin tore the loop down and the pending task went with it. An autouse fixture that compares asyncio.all_tasks() before and after each test failed the same test at teardown with "1 task(s) still running after the test: send_email", while a version that tracked and drained its background task passed. The whole suite of three tests took 0.14 s, so the check costs nothing. This guide installs the fixture, handles legitimate background tasks, and fixes what it finds.

Prerequisites

1. See how a leak passes silently

The leak needs only an untracked create_task:

async def send_email(addr: str) -> None:
    await asyncio.sleep(10)                      # stands in for SMTP


async def handle_signup(addr: str) -> str:
    asyncio.create_task(send_email(addr))        # fire-and-forget
    return "ok"


async def test_signup():
    assert await handle_signup("a@example.com") == "ok"     # tested: passes, no warning

Tested: the test passed in 0.08 s and printed nothing about the task that was still sleeping. The bug it hides is real: nothing keeps a reference to the task, so it can be garbage-collected mid-flight; its exceptions are never retrieved; and on shutdown it is cancelled without anyone knowing an email was lost. Tests are the cheapest place to catch it.

Verify: add a print at the end of the background coroutine and confirm it never runs during the test.

2. Fail tests that leave tasks running

An autouse fixture snapshots the running tasks before the test and checks after it:

# conftest.py
import asyncio

import pytest
import pytest_asyncio


@pytest_asyncio.fixture(autouse=True)
async def no_leaked_tasks(request):
    before = set(asyncio.all_tasks())
    yield
    await asyncio.sleep(0)                       # let just-finished tasks complete
    leaked = [t for t in asyncio.all_tasks() - before
              if t is not asyncio.current_task() and not t.done()]
    if leaked and "allow_leaks" not in request.keywords:
        for task in leaked:
            task.cancel()
        pytest.fail(f"{len(leaked)} task(s) still running after the test: "
                    + ", ".join(t.get_coro().__qualname__ for t in leaked))

Tested: the fire-and-forget test now failed at teardown, naming send_email, and a version that tracked its task and drained it passed. Naming the coroutine in the failure message makes the culprit obvious; task.get_name() helps too if your code names its tasks. The fixture cancels what it finds so one leaky test does not affect the next. Register an allow_leaks marker in pytest.ini for the rare test that deliberately leaves something running.

Verify: the suite fails when you add a fire-and-forget create_task anywhere in tested code.

The same leaky handler with and without the fixture A grid of 3 rows by 3 columns. The same leaky handler with and without the fixture test without leak fixture with leak fixture fire-and-forget create_task passed, silent failed: send_email still running tracked task, drained in test passed passed suite run time 0.08 s 0.14 s (3 tests) pytest 9.1, pytest-asyncio 1.4.0.

3. Give background tasks an owner

The fix in application code is to make every background task owned by something with a lifecycle — a registry drained at shutdown, or a TaskGroup:

class Background:
    def __init__(self) -> None:
        self.tasks: set[asyncio.Task] = set()

    def spawn(self, coro, *, name: str) -> asyncio.Task:
        task = asyncio.create_task(coro, name=name)
        self.tasks.add(task)
        task.add_done_callback(self._done)
        return task

    def _done(self, task: asyncio.Task) -> None:
        self.tasks.discard(task)
        if not task.cancelled() and task.exception() is not None:
            log.error("background task %s failed", task.get_name(), exc_info=task.exception())

    async def drain(self, timeout: float) -> None:
        if self.tasks:
            await asyncio.wait(self.tasks, timeout=timeout)
        for task in self.tasks:
            task.cancel()

The registry keeps strong references (so tasks are not garbage-collected), logs failures (so they are not lost), and drains on shutdown. In tests, a fixture that creates the registry and drains it at teardown makes background work part of the test's scope — and the leak fixture then stays green. Often the better fix is not a background task at all: an email belongs in a job queue, as in Background Jobs & Task Queues.

Verify: every create_task in the codebase goes through an owner — a registry, a TaskGroup, or a task the caller awaits.

How the leak check works around each test A flow of 5 stages. How the leak check works around each test snapshot all_tasks() before run the test may spawn tasks sleep(0) let finished tasks settle new, unfinished tasks? compare cancel + fail name each coroutine A dozen lines of fixture turn a silent leak into a named failure.

4. Check for unretrieved exceptions too

A related leak: a task that finished with an exception nobody retrieved. asyncio logs "Task exception was never retrieved" when the task is garbage-collected — often after the test that caused it, attributed to the wrong place. Make it fail the right test:

@pytest.fixture(autouse=True)
def fail_on_unretrieved(caplog):
    yield
    gc.collect()                                  # finalize dropped tasks now, inside this test
    bad = [r for r in caplog.records if "exception was never retrieved" in r.getMessage()]
    if bad:
        pytest.fail(f"{len(bad)} task exception(s) were never retrieved")

Forcing a collection at teardown makes asyncio report dropped failed tasks while the offending test is still the current one. Combined with the running-task check, this covers both ways background work escapes a test: still running, or failed and forgotten. Python's -X dev mode and asyncio debug mode add more warnings of this kind, as described in finding blocking calls with asyncio debug mode.

Verify: a test that spawns a task raising ValueError and drops it fails with the unretrieved-exception message.

5. Handle shared loops and long-lived fixtures

With module- or session-scoped event loops, tasks started by fixtures (a server, a consumer, a relay) legitimately run across many tests. Exclude them by recording the baseline after those fixtures start:

@pytest_asyncio.fixture(autouse=True, loop_scope="module")
async def no_leaked_tasks(request, module_services):      # depends on the long-lived fixtures
    before = set(asyncio.all_tasks())                     # baseline includes their tasks
    yield
    ...

Depending on the long-lived fixtures guarantees they have started before the snapshot, so their tasks are in the baseline and not reported. Name long-lived tasks (create_task(..., name="relay")) so a genuine leak and a service task are easy to tell apart in failure messages. Loop scopes and their trade-offs are covered in choosing event loop scopes in pytest-asyncio.

Verify: the leak fixture is green on a module that uses a shared server fixture, and still catches a fire-and-forget task added to one test.

What should this leaked task become? A decision on What is the task for with 4 outcomes. What should this leaked task become? What is the task for? the caller needs its result await it no background at all parallel parts of one request TaskGroup scoped to the request service-owned background work registry + drain logged failures must survive restarts (email) job queue durable A leak is a task without an owner; give it one.

Verification

Task leaks are caught when:

  • An autouse fixture fails tests that leave new tasks running.
  • Unretrieved task exceptions fail the test that caused them.
  • Long-lived fixture tasks are in the baseline, not reported as leaks.
  • Application code gives every task an owner.

Diagnostic Hook: when the fixture reports a leak, the coroutine's qualified name points at the spawn site; task.get_stack() on the leaked task (before cancelling it) shows where it is waiting, which usually explains why it did not finish.

Pitfalls & edge cases

  • Relying on warnings. Tested: the leak passed silently.
  • Fire-and-forget create_task. No reference, no error handling, no shutdown.
  • Snapshotting before long-lived fixtures start. Their tasks look like leaks.
  • Checking only running tasks. Failed, unretrieved tasks are leaks too.

Frequently Asked Questions

How do I detect tasks left running after a pytest-asyncio test?

Use an autouse async fixture that records asyncio.all_tasks() before the test and fails if new unfinished tasks remain afterwards, cancelling them. In testing it caught a fire-and-forget send_email task.

Does pytest-asyncio warn about leaked tasks?

Not reliably: in testing with pytest-asyncio 1.4, a test that left a task running passed with no warning.

How should background tasks be managed so tests do not leak?

Keep them in an owned registry with a done callback that logs failures and a drain step at shutdown, or use a TaskGroup when the work belongs to one request.

How do I find 'Task exception was never retrieved' in the right test?

Call gc.collect() at teardown so dropped tasks are finalized during the test that created them, and fail if the message was logged.