Patching Async Code & Coroutines

An async test can pass while proving nothing at all. Replace a coroutine function with a MagicMock, and calling it returns a Mock rather than a coroutine; the code under test awaits it, gets a TypeError that a broad except swallows, or — more often — never awaits it because the mock stands in for the whole call. The suite is green, RuntimeWarning: coroutine ... was never awaited scrolls past unread, and the behaviour is untested. Patching async code correctly is mostly about making that failure mode impossible.

Prerequisites

Python 3.8+ for unittest.mock.AsyncMock; 3.10+ for the patch detection improvements this guide assumes.
pytest >= 8.0 and a runner for the tests themselves — see pytest-asyncio in depth.
Working knowledge of patch target resolution, since nothing changes there for async: where to patch.
The distinction between Mock, MagicMock and AsyncMock covered in Mock vs MagicMock vs AsyncMock.

Core concept: awaitability is a property of the replacement

AsyncMock differs from MagicMock in exactly one significant way: calling it returns a coroutine, so await mock(...) works and resolves to return_value. Everything else — side_effect, assert_called_with, attribute auto-creation — behaves the same, with an extra family of await-specific assertions layered on.

patch() tries to pick the right class for you by inspecting the target with asyncio.iscoroutinefunction. When that inspection succeeds, everything works. When it fails — and it fails for callables implemented in C, for objects whose __call__ is async, for functions wrapped in a decorator that is not a coroutine function itself — you silently get a MagicMock, and the test stops testing.

The right-hand column is the dangerous one precisely because it is green. assert_called_once_with is satisfied by the call alone, whether or not anything awaited the result.

Step-by-step implementation

1. Make the warning fatal

# pyproject.toml
[tool.pytest.ini_options]
filterwarnings = [
    "error",
    # A never-awaited coroutine in a test is a broken test, not a nuisance.
    "error::RuntimeWarning",
]

This is the single highest-value line in the file. Every mis-patched async call becomes a failure at the point it happens, rather than a warning nobody reads. The general technique is covered in turning warnings into errors with filterwarnings.

2. Patch with autospec

from unittest.mock import patch

import pytest


async def test_order_service_fetches_the_invoice():
    # autospec inspects the real attribute: coroutine functions become
    # AsyncMock, ordinary ones MagicMock, and both get the real signature.
    with patch("myapp.billing.BillingClient.fetch_invoice", autospec=True) as fetch:
        fetch.return_value = Invoice(id="inv_1", total_minor=1234)

        result = await OrderService(BillingClient()).total_for("inv_1")

    assert result == 1234
    # Note: awaited, not merely called.
    fetch.assert_awaited_once()

autospec=True makes the signature real, so a call with a renamed keyword fails here instead of in production. The argument for using it everywhere is made in create_autospec vs patch(autospec=True).

3. Force the class when detection fails

from unittest.mock import AsyncMock, patch


async def test_client_wrapped_in_a_decorator():
    # `retry` returns a plain function wrapping a coroutine, so iscoroutinefunction
    # says False and patch would hand back a MagicMock.
    with patch("myapp.clients.fetch_with_retry", new_callable=AsyncMock) as fetch:
        fetch.return_value = {"status": "ok"}
        assert await consume() == "ok"

Any time a patched async call produces a TypeError: object Mock can't be used in 'await' expression, this is the fix. It is also worth fixing upstream: functools.wraps on an async def wrapper preserves coroutine-ness, and a decorator that does not is a hazard beyond the tests.

4. Handle the async protocols

from unittest.mock import AsyncMock, MagicMock


def make_async_cm(value):
    """An async context manager double, built by hand."""
    cm = MagicMock()
    cm.__aenter__ = AsyncMock(return_value=value)   # what `async with` binds
    cm.__aexit__ = AsyncMock(return_value=False)    # False → exceptions propagate
    return cm


async def test_connection_is_released_on_error(pool):
    connection = AsyncMock()
    pool.acquire = MagicMock(return_value=make_async_cm(connection))

    with pytest.raises(ValueError):
        await run_failing_query(pool)

    # __aexit__ ran, so the connection went back to the pool.
    pool.acquire.return_value.__aexit__.assert_awaited_once()

__aexit__ returning False rather than a Mock matters: a truthy return value from __aexit__ suppresses the exception, so a carelessly configured mock silently swallows the error the test was written to observe.

5. Assert on awaits, and on order

from unittest.mock import AsyncMock, MagicMock


async def test_writes_happen_after_the_read():
    parent = MagicMock()
    parent.attach_mock(AsyncMock(), "read")
    parent.attach_mock(AsyncMock(), "write")

    await pipeline(parent.read, parent.write)

    # mock_calls on the parent interleaves both children in real order.
    assert [name for name, _, _ in parent.mock_calls] == ["read", "write"]

await_args_list gives per-mock ordering; only a shared parent gives the global sequence. When a bug is "the write happened before the read committed", this is the assertion that catches it.

Verification

Confirm the configuration actually catches a mis-patch by writing one deliberately:

from unittest.mock import MagicMock, patch


async def test_deliberately_wrong_patch_now_fails():
    with patch("myapp.clients.fetch", new=MagicMock(return_value={"ok": True})):
        with pytest.raises(TypeError):
            await consume()

E   TypeError: object dict can't be used in 'await' expression

With filterwarnings = ["error"] in place, the softer variant — where the coroutine is created and dropped — also fails, with RuntimeWarning: coroutine 'fetch' was never awaited raised as an error. Seeing both once is what justifies trusting the rest of the suite.

Troubleshooting

Symptom	Root cause	Fix
`object MagicMock can't be used in 'await' expression`	`patch` detection failed	`new_callable=AsyncMock`, or `autospec=True`
`RuntimeWarning: coroutine was never awaited`	A mock replaced the call, so nothing awaited	Make it an error; patch with `AsyncMock`
`assert_called_once` passes but nothing ran	Assertion is about the call, not the await	Use `assert_awaited_once_with`
An exception vanished inside `async with`	`__aexit__` returned a truthy `Mock`	Set `__aexit__ = AsyncMock(return_value=False)`
`async for` over a mock raises `TypeError`	`__aiter__`/`__anext__` not configured	Use `autospec` on an async iterable, or set both
Await order assertions are unreliable	Per-mock lists compared across mocks	Attach both to one parent and read `mock_calls`

Async iterators and streaming responses

Streaming APIs are the other place hand-configured mocks go wrong, because async for needs two dunder methods and a sentinel exception.

from unittest.mock import MagicMock


def async_iter(items):
    """A double for anything consumed with `async for`."""
    iterator = iter(items)

    async def anext_():
        try:
            return next(iterator)
        except StopIteration:
            raise StopAsyncIteration          # the sentinel `async for` expects

    mock = MagicMock()
    mock.__aiter__.return_value = mock
    mock.__anext__ = anext_
    return mock


async def test_processes_every_chunk(handler):
    response = MagicMock()
    response.aiter_bytes.return_value = async_iter([b"one", b"two", b"three"])

    await handler.consume(response)

    assert handler.chunks == [b"one", b"two", b"three"]

StopAsyncIteration rather than StopIteration is the detail that catches everyone: raising the synchronous sentinel inside __anext__ produces a RuntimeError about a coroutine raising StopIteration, which is a confusing report of a simple mistake.

In practice, most streaming tests are better served by the transport-level fakes described in mocking httpx clients with respx, which produce real response objects with real async iteration. Hand-rolled async iterators are for the cases where the thing being iterated is your own abstraction rather than a library's.

Patching where the coroutine is defined versus used

Target resolution works identically for async code, but one async-specific wrinkle trips people up: a module that binds a coroutine function at import time and then schedules it as a task.

# myapp/worker.py
import asyncio

from myapp.clients import fetch_invoice          # bound at import time


async def refresh_all(ids):
    # The name `fetch_invoice` here is worker's own global, not clients'.
    tasks = [asyncio.create_task(fetch_invoice(i)) for i in ids]
    return await asyncio.gather(*tasks)

from unittest.mock import AsyncMock, patch


async def test_refresh_all_fetches_each_id():
    # Patch where it is USED — myapp.worker — not where it is defined.
    with patch("myapp.worker.fetch_invoice", new_callable=AsyncMock) as fetch:
        fetch.return_value = Invoice(id="x", total_minor=0)
        await refresh_all(["a", "b", "c"])

    assert fetch.await_count == 3

Patching myapp.clients.fetch_invoice here would have no effect, because worker already holds its own reference — the ordinary rule from where to patch. What makes the async version harder to spot is create_task: with a MagicMock the task creation itself raises, but inside a gather the error arrives as one entry among several and is easy to misread as a problem with the concurrency rather than with the patch.

The rule is unchanged from synchronous code; only the way the mistake presents itself is different.

Async fixtures that provide the double

Most of the friction in async patching disappears when the double is supplied by a fixture rather than constructed inside each test. The fixture owns the configuration, the tests own the expectations.

from unittest.mock import AsyncMock, create_autospec

import pytest

from myapp.billing import BillingClient


@pytest.fixture
def billing_client():
    """A strict double: real signatures, coroutine methods awaitable."""
    client = create_autospec(BillingClient, instance=True, spec_set=True)
    # Sensible defaults so most tests configure nothing at all.
    client.fetch_invoice.return_value = Invoice(id="inv_1", total_minor=1000)
    client.list_invoices.return_value = []
    return client


async def test_total_uses_the_fetched_invoice(billing_client):
    service = OrderService(billing=billing_client)
    assert await service.total_for("inv_1") == 1000
    billing_client.fetch_invoice.assert_awaited_once_with("inv_1")

create_autospec(..., instance=True, spec_set=True) gives three guarantees at once: coroutine methods become AsyncMock, signatures are enforced, and assigning an attribute the real class does not have raises immediately. That last one catches the commonest drift — a test configuring client.get_invoice after the method was renamed to fetch_invoice, which on a bare MagicMock silently creates a new attribute and passes.

Defaults in the fixture are what keep individual tests short. A test that cares only about the total should not have to configure list_invoices, and a fixture that supplies plausible defaults means the only lines in a test are the ones that matter to it — the same argument made for test data factories, applied to collaborators instead of rows.

One caution: a fixture-provided mock is shared for the duration of one test only, but a session-scoped one accumulates calls across tests and will eventually make assert_awaited_once fail for reasons unrelated to the test that fails. Keep mock fixtures function-scoped, or call reset_mock() in an autouse fixture if a wider scope is unavoidable.

Reading an async mock's recorded state

When an await assertion fails, AsyncMock carries more information than the error message shows, and printing it is usually faster than re-running with a debugger.

def describe(mock) -> str:
    return (
        f"called {mock.call_count}×, awaited {mock.await_count}×\n"
        f"  calls:  {mock.call_args_list}\n"
        f"  awaits: {mock.await_args_list}\n"
    )

called 2×, awaited 1×
  calls:  [call('inv_1'), call('inv_2')]
  awaits: [call('inv_1')]

A gap between call_count and await_count is the diagnosis by itself: something created a coroutine and dropped it. That is a real bug in the code under test roughly half the time — a create_task that was removed, a branch that forgot its await — and a bug in the test the other half. Either way it is a fact the assertion error alone does not mention.

The mirror-image case is await_count higher than expected with identical arguments, which usually means a retry loop is running where the test assumed one attempt. await_args_list shows the repetition directly, and mock.await_args holds only the last one, which is why asserting on it alone hides the retries.

Where patching async code is the wrong tool

Three situations look like patching problems and are not.

Testing concurrency. Patching a coroutine with an AsyncMock that returns instantly removes every suspension point, so the interleaving the test was meant to explore no longer exists. Tests about ordering, cancellation or races need real awaits — see testing threads and race conditions and the cooperative equivalents in the async section.

Testing timeouts. A mock that returns immediately can never time out. Simulating slowness needs a side_effect that actually awaits (async def slow(*a, **kw): await asyncio.sleep(10)), and even then the assertion is usually better expressed against a fake server that stalls.

Replacing an entire client. When five methods of a client need configuring, the mock has become an unnamed second implementation. A fake client with real behaviour is shorter and survives refactoring, as argued in spies, fakes and hand-rolled test doubles.

The common thread is that AsyncMock is excellent at standing in for one call at a boundary and poor at standing in for a subsystem. Used at the boundary it makes tests fast and precise; used deeper it removes the behaviour the test exists to check.

Each step inward trades fidelity for control. The bottom row is where async tests most often stop testing anything about being async.

Making async doubles behave like the real thing

A double that returns instantly is convenient and, for some tests, wrong. Three behaviours are worth reproducing when the code under test is supposed to cope with them.

Latency, when the code has a timeout or a concurrency limit to exercise:

import asyncio
from unittest.mock import AsyncMock


async def slow(*args, **kwargs):
    await asyncio.sleep(0.2)          # a real suspension point, not a busy wait
    return {"status": "ok"}


client.fetch = AsyncMock(side_effect=slow)

Failure then success, for retry logic — side_effect accepts a sequence, and exceptions in it are raised rather than returned:

client.fetch = AsyncMock(side_effect=[TimeoutError, TimeoutError, {"status": "ok"}])

Backpressure, for anything consuming a stream faster than it is produced, which needs a real asyncio.Queue rather than a mock at all.

The first two are cheap and turn "the retry code is covered" into "the retry code demonstrably retries twice and then succeeds". The third is the point at which a mock stops being the right tool, because the behaviour under test is the interaction between producer and consumer and a double for either one removes it.

A useful heuristic: if configuring the double takes more lines than the assertion, the test is probably about the collaborator rather than about the code, and either a fake or a real dependency will be shorter and more honest.

There is one more behaviour worth reproducing deliberately, because forgetting it produces a class of bug that only appears under load: cancellation. A real client being cancelled mid-request raises asyncio.CancelledError at the await, and the calling code is expected to let it propagate after cleaning up. An AsyncMock never does that unless told to, so the cleanup path goes untested until a timeout fires in production. Setting side_effect=asyncio.CancelledError for one test, and asserting the connection was returned to the pool afterwards, closes that gap for the cost of three lines. The same trick covers ConnectionResetError and the transport-specific exceptions a client can raise mid-stream, all of which reach the caller through exactly the same path.

Frequently Asked Questions

Why does patch() give me a MagicMock instead of an AsyncMock?patch() inspects the target and returns an AsyncMock only when it detects a coroutine function. A plain function that returns a coroutine, a callable class instance, or a C-implemented method may not be detected, so you get a MagicMock whose return value is not awaitable. Pass new_callable=AsyncMock explicitly in those cases.

What does 'coroutine was never awaited' mean in a test? Something produced a coroutine object that nothing consumed — usually a MagicMock standing in for an async function, so the call returned a Mock rather than something awaitable, or an await that was accidentally removed. Treat the warning as an error in the test configuration; it almost always means the assertion under it proved nothing.

How do I patch an async context manager? Give the mock's __aenter__ an AsyncMock returning the object the body should receive, and __aexit__ an AsyncMock returning False. MagicMock supplies these automatically when the spec is an async context manager, which is the argument for using autospec rather than configuring by hand.

Can I assert on the order of awaits across several mocks? Yes, by attaching the mocks to a common parent with attach_mock and asserting on the parent's mock_calls, which interleaves the children in call order. AsyncMock records awaits in await_args_list per mock, which gives per-mock ordering but not a global sequence.

Does autospec work on async functions? Yes, and it is the recommended approach. create_autospec inspects the target and produces AsyncMock for coroutine functions and MagicMock for the rest, with signatures enforced on both, so a call with the wrong arguments fails in the test rather than in production.

Work through the context-manager case in patching an async context manager.
Assert sequencing precisely with asserting await order with AsyncMock.
Compare the three mock classes in Mock vs MagicMock vs AsyncMock.
Get the test runner's loop right first with pytest-asyncio in depth.
Replace deep mocking with behaviour using spies, fakes and hand-rolled test doubles.

← Back to Advanced Mocking & Test Doubles in Python