Skip to content

Testing Async Code in Python

The retry logic has a unit test, and the test passes. It also takes 31 seconds, because the backoff really sleeps. The circuit breaker test fails one run in forty on CI, because it depends on two tasks finishing in a particular order. The integration suite hangs forever when a mock is not awaited, and the fixture that creates an HTTP client works for the first test in a module and then raises RuntimeError: ... attached to a different loop for the second. None of these are bugs in the code under test; they are the predictable result of testing concurrent, time-dependent code with tools and habits designed for synchronous functions.

Everything this site describes — timeouts, cancellation, retries, backpressure, graceful shutdown — is only as trustworthy as its tests, and those behaviours are precisely the ones that are hardest to test: they involve time, ordering and failure. This section covers how to run coroutines under a test runner without event loop surprises, how to replace async dependencies without losing signature checks, how to control time so that a 30-second breaker test runs in milliseconds, and how to make concurrency bugs reproducible instead of flaky. The parent section, Resilience, Cancellation & Error Handling, describes the behaviours; this one describes how to prove them.

Scope of this section:

  • Running async tests with pytest-asyncio and unittest.IsolatedAsyncioTestCase, and choosing event loop scopes.
  • Replacing async dependencies with AsyncMock and autospecced mocks.
  • Deterministic time: injected clocks and a virtual-time event loop for timeouts, backoff and breakers.
  • Detecting leaked tasks, unawaited coroutines and resources left open by tests.
  • Reproducing races deliberately with controlled interleavings rather than retries.

Architectural principles

  • A test owns its event loop. Every async test runs in a loop whose lifetime is explicit — per test, per module or per session — and every async fixture it uses must live on that same loop. Mixing scopes is the root of "attached to a different loop" errors.
  • Never wait on the wall clock. Real sleeps make suites slow, and slow suites get fewer tests. Time-dependent behaviour is tested by controlling time — injecting a clock or running on a virtual-time loop — not by sleeping and hoping.
  • Mocks must keep the contract. An async dependency replaced by a plain MagicMock is not awaitable, and one replaced by an unspecced AsyncMock accepts any arguments. Use create_autospec so a changed signature breaks the test, not production.
  • Ordering is an input, not an accident. If a behaviour depends on which task runs first, the test must force that order with events or barriers. A test that passes "most of the time" is testing the scheduler, not the code.
  • A test leaves nothing behind. Tasks, connections and threads started by a test are stopped by that test. A leak detector that fails the test turns silent pollution between tests into a clear error.
What an async test has to control 5 stacked layers from Event loop to Cleanup. What an async test has to control Event loop who runs the coroutine which loop scope Dependencies AsyncMock at the boundary autospec signatures Time injected clock virtual-time loop Ordering events as test hooks no reliance on luck Cleanup no leaked tasks pools and threads closed Leave any layer uncontrolled and the test becomes slow, flaky or both.

Execution model: how a test runner drives the loop

A synchronous test runner cannot await, so an async test is always a coroutine that something else runs to completion on an event loop. unittest.IsolatedAsyncioTestCase does this with an asyncio.Runner per test case: asyncSetUp, the test method and asyncTearDown all run on one fresh loop, which is closed afterwards together with its default executor and async generators. pytest-asyncio does the equivalent for pytest, with one addition that matters in practice — the loop's lifetime is configurable. A test marked with loop_scope="module" runs on a loop shared by every test in the module, which lets expensive async fixtures such as a database pool or an HTTP client be created once.

That flexibility is also the main source of test-suite breakage. Many asyncio primitives and virtually every async client bind themselves to the loop that is running when they are first used: locks, queues, futures, transports and connection pools. A fixture that created a client on a module-scoped loop and then yielded it to a test running on a function-scoped loop hands the test an object whose internals belong to a loop that is not running. The rule is simple and absolute: a fixture's loop scope must match the loop scope of every test that uses it.

Time is the other half of the model. The loop's scheduler keeps timers in a heap ordered by loop.time(), and when no callback is ready it blocks in the selector until the earliest timer is due. Nothing in that design requires the clock to be the wall clock. A loop whose time() returns a controllable value, and whose selector advances that value instead of sleeping when nothing is ready, runs every asyncio.sleep(), asyncio.timeout() and call_later() in exact virtual order — instantly. That is the basis for the time-control pattern below and the full walkthrough in controlling time in asyncio tests.

Which loop scope for this fixture? A decision on What does the fixture create with 3 outcomes. Which loop scope for this fixture? What does the fixture create? stateful: queues, locks function loop scope fresh per test expensive client, read-only use module loop scope tests match the scope process-wide resource session loop scope every user matches it The fixture and every test using it must share one loop scope.

Pattern catalogue

pytest-asyncio with explicit loop scopes

The baseline for pytest users. Configure the mode once, and state loop scopes wherever a fixture outlives a single test.

# pip install pytest pytest-asyncio
# pyproject.toml:
#   [tool.pytest.ini_options]
#   asyncio_mode = "strict"
#   asyncio_default_fixture_loop_scope = "function"
import asyncio
import pytest
import pytest_asyncio


@pytest_asyncio.fixture(scope="module", loop_scope="module")
async def client():
    c = await make_client()            # e.g. httpx.AsyncClient()
    yield c
    await c.aclose()


@pytest.mark.asyncio(loop_scope="module")
async def test_health(client):
    response = await client.get("/health")
    assert response.status_code == 200

Use strict mode in libraries and mixed codebases so only explicitly marked tests are treated as async; use asyncio_mode = "auto" in pure-asyncio services to drop the markers. The trade-off of a module-scoped loop is isolation: state left on that loop by one test is visible to the next. Setting it up correctly, including what happens when the scopes disagree, is covered in testing asyncio code with pytest-asyncio.

Standard library: IsolatedAsyncioTestCase

When a project cannot add test dependencies, unittest has first-class async support with a fresh loop per test.

import asyncio
import unittest


class RetryTests(unittest.IsolatedAsyncioTestCase):
    async def asyncSetUp(self) -> None:
        self.queue: asyncio.Queue[int] = asyncio.Queue(maxsize=2)

    async def test_put_blocks_when_full(self) -> None:
        await self.queue.put(1)
        await self.queue.put(2)
        with self.assertRaises(TimeoutError):
            async with asyncio.timeout(0.01):
                await self.queue.put(3)

    async def asyncTearDown(self) -> None:
        while not self.queue.empty():
            self.queue.get_nowait()


if __name__ == "__main__":
    unittest.main()

Per-test loops give the strongest isolation at the cost of re-creating expensive resources for every test. It also runs under pytest unchanged, which makes it a reasonable choice for shared library code.

Autospecced async mocks

Replace an async dependency with a mock that is awaitable and enforces the real signature.

import asyncio
from unittest.mock import create_autospec


class PaymentsClient:
    async def charge(self, account: str, cents: int) -> dict: ...


async def checkout(client: PaymentsClient, account: str, cents: int) -> str:
    for _ in range(3):
        try:
            return (await client.charge(account, cents))["id"]
        except ConnectionError:
            await asyncio.sleep(0)
    raise RuntimeError("payments unavailable")


async def test_retries_then_succeeds() -> None:
    client = create_autospec(PaymentsClient, instance=True)
    client.charge.side_effect = [ConnectionError(), ConnectionError(), {"id": "ch_1"}]
    assert await checkout(client, "acct", 500) == "ch_1"
    assert client.charge.await_count == 3
    client.charge.assert_awaited_with("acct", 500)

create_autospec inspects the class: async def methods become AsyncMock, regular methods become MagicMock, and calling with the wrong arguments raises TypeError. The trade-off is that mocks verify interactions, not behaviour; keep them at the boundary to the network and test real logic with real objects. More, including async context managers and iterators, is in mocking async dependencies with AsyncMock.

Virtual time for timeouts and backoff

Run time-dependent code on a loop whose clock jumps straight to the next timer.

import asyncio
import selectors


class _JumpingSelector(selectors.DefaultSelector):
    def __init__(self) -> None:
        super().__init__()
        self.loop = None

    def select(self, timeout=None):
        if timeout is None or timeout <= 0:
            return super().select(timeout)
        ready = super().select(0)               # real I/O is still served
        if not ready:
            self.loop.advance(timeout)          # nothing ready: jump to the next timer
        return ready


class VirtualClockLoop(asyncio.SelectorEventLoop):
    def __init__(self) -> None:
        selector = _JumpingSelector()
        super().__init__(selector)
        selector.loop = self
        self._now = 0.0

    def time(self) -> float:
        return self._now

    def advance(self, seconds: float) -> None:
        self._now += seconds


def run_virtual(coro):
    with asyncio.Runner(loop_factory=VirtualClockLoop) as runner:
        return runner.run(coro)

A retry loop with exponential backoff of 1, 2, 4 and 8 seconds runs in well under a millisecond on this loop, and loop.time() inside the code reads exactly 0, 1, 3, 7 and 15. The trade-off is scope: code that waits on real threads or sockets sees timers fire as soon as it blocks, so use virtual time for pure-async logic and real time for integration tests.

Forced interleavings for race conditions

Reproduce a race by making the scheduler take the bad path every time.

import asyncio


class Inventory:
    def __init__(self, stock: int, checkpoint: asyncio.Event | None = None) -> None:
        self.stock = stock
        self._checkpoint = checkpoint           # test-only hook between check and act

    async def reserve(self) -> bool:
        if self.stock <= 0:
            return False
        if self._checkpoint is not None:
            await self._checkpoint.wait()        # the gap a real await would create
        self.stock -= 1
        return True


async def test_double_reservation_race() -> None:
    gate = asyncio.Event()
    inventory = Inventory(stock=1, checkpoint=gate)
    first = asyncio.create_task(inventory.reserve())
    second = asyncio.create_task(inventory.reserve())
    await asyncio.sleep(0)                       # both tasks pass the stock check
    gate.set()
    assert await first and await second
    assert inventory.stock == -1                 # the bug, reproduced deterministically

The hook is a small, deliberate seam; the alternative — running the test a thousand times and hoping the scheduler interleaves badly — produces a flaky test that proves nothing when it passes. Once the race is fixed with a lock (see choosing asyncio Lock vs Semaphore vs Event), the same test asserts stock == 0 and one failed reservation.

Forcing the double-reservation race 3 lanes over time. Forcing the double-reservation race task 1 check stock wait on gate stock -= 1 task 2 check stock wait on gate stock -= 1 test gate.set() scheduler steps → The gate makes the bad interleaving happen on every run, not one in forty.

Resource boundaries

Async test suites fail in resource-shaped ways: too slow, too shared, or leaking between tests.

Resource Default risk Boundary to set
Event loops One per test: slow setup for expensive fixtures Module or session loop scope for read-only clients; function scope for anything stateful
Wall-clock time Real sleeps in retries, timeouts, breakers Inject a clock or run on a virtual-time loop; no test sleeps longer than tens of milliseconds
Tasks Background tasks outlive the test and fail later tests Autouse fixture that fails on tasks left running
Connections and pools Fixtures reopen pools per test, exhausting a shared test database One pool per module or session, sized below the database's connection limit
Threads to_thread work still running at loop close delays teardown Await offloaded work in the test; keep executor jobs short
Test timeouts A missing await or a deadlock hangs the whole CI job Per-test deadline with asyncio.timeout() or pytest-timeout

The task-leak row deserves a fixture in every async codebase. It turns a failure that appears three tests later, in an unrelated file, into an error on the test that caused it.

Integrated production example

This conftest.py plus test module combine the patterns: a shared client on a module loop, autospecced dependencies, a virtual-time run for backoff, a per-test deadline, and a leak detector that fails any test leaving tasks behind.

# payments.py — the code under test
import asyncio


class PaymentsClient:
    async def charge(self, account: str, cents: int) -> dict: ...


async def checkout_with_backoff(client: PaymentsClient, account: str, cents: int,
                                attempts: int = 3, base_delay: float = 1.0) -> str:
    for attempt in range(attempts):
        try:
            return (await client.charge(account, cents))["id"]
        except ConnectionError:
            if attempt < attempts - 1:
                await asyncio.sleep(base_delay * 2 ** attempt)
    raise RuntimeError("payments unavailable")


# conftest.py
import asyncio
import functools

import pytest
import pytest_asyncio


@pytest_asyncio.fixture(autouse=True)
async def no_leaked_tasks():
    before = asyncio.all_tasks()
    yield
    await asyncio.sleep(0)                               # let just-finished tasks settle
    leaked = {t for t in asyncio.all_tasks() - before
              if t is not asyncio.current_task() and not t.done()}
    for t in leaked:
        t.cancel()
    if leaked:
        await asyncio.gather(*leaked, return_exceptions=True)
        pytest.fail(f"test leaked {len(leaked)} task(s): {sorted(t.get_name() for t in leaked)}")


@pytest.fixture
def deadline():
    # a factory: asyncio.timeout() must be created inside the running test
    return functools.partial(asyncio.timeout, 2.0)       # no test may hang the suite


# test_checkout.py
import asyncio
from unittest.mock import create_autospec

import pytest

from payments import PaymentsClient, checkout_with_backoff   # code under test
from virtual_clock import VirtualClockLoop           # the class from the pattern above


def run_virtual(coro):
    with asyncio.Runner(loop_factory=VirtualClockLoop) as runner:
        return runner.run(coro)


@pytest.mark.asyncio
async def test_gives_up_after_three_attempts(deadline):
    client = create_autospec(PaymentsClient, instance=True)
    client.charge.side_effect = ConnectionError()
    async with deadline():
        with pytest.raises(RuntimeError, match="payments unavailable"):
            await checkout_with_backoff(client, "acct", 500, base_delay=0)
    assert client.charge.await_count == 3


def test_backoff_schedule_in_virtual_time():
    attempts: list[float] = []

    async def scenario():
        client = create_autospec(PaymentsClient, instance=True)

        async def flaky(account, cents):
            attempts.append(asyncio.get_running_loop().time())
            if len(attempts) < 3:
                raise ConnectionError()
            return {"id": "ch_1"}

        client.charge.side_effect = flaky
        return await checkout_with_backoff(client, "acct", 500, base_delay=1.0)

    assert run_virtual(scenario()) == "ch_1"
    assert attempts == [0.0, 1.0, 3.0]                   # exact, and instant


@pytest.mark.asyncio
async def test_background_refresh_is_stopped(deadline):
    async with deadline():
        refresher = asyncio.create_task(asyncio.sleep(3600), name="token-refresher")
        refresher.cancel()                                # the leak fixture fails without this
        await asyncio.gather(refresher, return_exceptions=True)

The virtual-time test is a plain synchronous test that builds its own runner, so it does not interact with pytest-asyncio's loop at all — a useful property when a suite mixes both styles. The leak detector is autouse, so a new test that forgets to stop a background task fails with the task's name.

Diagnostic Hook — is the async test suite healthy?

Track three numbers in CI. Suite duration per test (total time ÷ test count): a rising value almost always means real sleeps have crept in; anything above a few tens of milliseconds for a unit test deserves a look. Flake rate (tests that fail then pass on retry, per week): async flakiness is nearly always an uncontrolled interleaving or a wall-clock timeout, so each flake is a bug report against the test. Leak-detector failures and "coroutine was never awaited" warnings: run the suite with -W error::RuntimeWarning so an unawaited coroutine fails the build instead of scrolling past. Alert when duration per test doubles or when any flake recurs.

Async suite health signals by urgency 4 bars comparing duration per test rising with the others. Async suite health signals by urgency duration per test rising real sleeps crept in a flake recurs uncontrolled ordering or time never-awaited warning test proves nothing leak detector fails pollutes later tests Bar length encodes urgency, not a measured quantity. Treat every flake as a bug report against the test.

Failure modes

Failure mode Root cause Detection Fix
RuntimeError: ... attached to a different loop Fixture created on one loop, used by a test on another Error on the second test that uses a module or session fixture Match loop_scope on the fixture and every test using it
Async test passes without running Test runner lacks async support, or the marker is missing in strict mode Test finishes instantly; coverage shows no lines executed Install pytest-asyncio and mark tests, or use auto mode; fail on warnings
TypeError: object MagicMock can't be used in 'await' expression Async dependency replaced with a plain MagicMock Error at the first await on the mock create_autospec(Class, instance=True) or AsyncMock
Suite takes minutes Real sleeps in retries, backoff and timeouts Per-test duration report shows seconds-long unit tests Inject delays or run on a virtual-time loop
Intermittent CI failures Assertion depends on task completion order Failure rate that changes with machine load Force ordering with events; assert on outcomes, not order
Later tests fail mysteriously Background tasks leaked from an earlier test Failures move when test order changes Autouse leak-detection fixture
CI job hangs until killed Deadlock or an awaited future that never resolves Job timeout with no failing test named Per-test asyncio.timeout() or pytest-timeout

Frequently Asked Questions

How do I run async test functions with pytest?

Install pytest-asyncio and either mark each coroutine test with @pytest.mark.asyncio or set asyncio_mode = "auto" in the pytest configuration. Without the plugin, recent pytest versions fail async test functions as unsupported rather than running them. Async fixtures use @pytest_asyncio.fixture in strict mode.

Why do I get attached to a different loop errors in pytest?

An async fixture created its object, such as a client, lock or connection pool, on one event loop and a test used it on another. This happens when a module- or session-scoped fixture runs on a different loop scope than the test. Give the fixture and every test that uses it the same loop_scope.

How can I test asyncio timeouts and retries without waiting?

Control time instead of sleeping. Either inject the sleep function or clock into the code under test, or run the test on an event loop whose time method returns a virtual clock that jumps to the next scheduled timer when nothing is ready. Backoffs of minutes then run instantly with exact timestamps.

Should I use unittest.IsolatedAsyncioTestCase or pytest-asyncio?

IsolatedAsyncioTestCase needs no dependencies and gives every test a fresh event loop, which suits libraries and strict isolation. pytest-asyncio integrates with pytest fixtures and allows module- or session-scoped loops for expensive shared resources. Both can coexist in one pytest run.

How do I make a flaky async race-condition test deterministic?

Stop relying on the scheduler to interleave tasks badly. Add a test hook, such as an asyncio.Event awaited between the check and the update, start the competing tasks, let them reach the hook, then release it. The race then happens on every run, and the fixed code can be asserted just as deterministically.