Mocking Async Dependencies with AsyncMock

The checkout service's retry logic is tested against a mock payments client, and every test is green. In production, the first retry crashes with TypeError: charge() missing 1 required positional argument: 'idempotency_key' — the real client's signature changed three weeks ago, and the mock accepted anything. Elsewhere in the suite, a MagicMock standing in for an async client fails with object MagicMock can't be used in 'await' expression, so someone "fixed" it by making the production code check inspect.isawaitable(). And the timeout test does not really test a timeout, because the mock returns instantly. Mocks for async dependencies need to be awaitable, keep the real signatures, fail in scripted ways, and be slow on command. unittest.mock provides all of that in the standard library once you know which constructor to reach for.

Prerequisites

• Python 3.11+, standard library unittest.mock; the test runner examples use pytest with pytest-asyncio, set up as in testing asyncio code with pytest-asyncio.
• The testing principles from Testing Async Code: mock at the network boundary, test logic with real objects.
• Familiarity with the retry and timeout patterns being tested, such as exponential backoff with jitter.

1. Replace the dependency with an autospecced mock

create_autospec(Class, instance=True) builds a mock from the real class: each async def method becomes an AsyncMock, each regular method a MagicMock, and every call is checked against the real signature. It is the default choice for any client object.

import asyncio
from unittest.mock import AsyncMock, create_autospec

import pytest


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

    def region(self) -> str:
        return "eu"


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")


@pytest.mark.asyncio
async def test_autospec_enforces_the_contract() -> None:
    client = create_autospec(PaymentsClient, instance=True)
    assert isinstance(client.charge, AsyncMock)          # async method -> awaitable mock
    assert not isinstance(client.region, AsyncMock)       # sync method stays synchronous
    with pytest.raises(TypeError):
        await client.charge("acct")                       # missing argument: caught in the test

A bare AsyncMock() is awaitable but accepts any arguments and invents any attribute on access, so a renamed method or an extra parameter never breaks the test. A bare MagicMock() is not awaitable at all. Autospec avoids both failure modes with one line.

Verify: add a required parameter to PaymentsClient.charge and run the suite; every test that calls charge with the old arguments fails with TypeError, pointing at the call sites that production would have broken.

2. Script responses and failures with side_effect

Retry, fallback and circuit-breaker logic is exercised by sequences of outcomes. Assign an iterable to side_effect and each await consumes the next item: exception instances are raised, anything else is returned.

@pytest.mark.asyncio
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


@pytest.mark.asyncio
async def test_gives_up_after_three_failures() -> None:
    client = create_autospec(PaymentsClient, instance=True)
    client.charge.side_effect = ConnectionError("refused")     # raised on every await

    with pytest.raises(RuntimeError, match="payments unavailable"):
        await checkout(client, "acct", 500)
    assert client.charge.await_count == 3


@pytest.mark.asyncio
async def test_behaviour_depends_on_arguments() -> None:
    client = create_autospec(PaymentsClient, instance=True)

    async def fake_charge(account: str, cents: int) -> dict:
        if cents > 10_000:
            raise PermissionError("limit exceeded")
        return {"id": f"ch_{account}_{cents}"}

    client.charge.side_effect = fake_charge                     # an async function is awaited
    assert await checkout(client, "a1", 500) == "ch_a1_500"

A list runs out: an extra, unexpected await raises StopAsyncIteration, which is useful — it fails a test whose code retries more often than intended. When the response depends on arguments, use an async def side effect rather than a growing list.

Verify: change the retry limit in checkout from 3 to 4; test_retries_then_succeeds still passes, but test_gives_up_after_three_failures fails on await_count, which is exactly the regression the test exists to catch.

3. Assert on awaits, not only on calls

AsyncMock tracks calls and awaits separately. A coroutine function that is called but never awaited does nothing — the classic missing-await bug — and only the await assertions detect it.

@pytest.mark.asyncio
async def test_called_but_never_awaited() -> None:
    notify = AsyncMock(return_value=None)

    coro = notify("order-1")                     # bug in code under test: no await
    notify.assert_called_once_with("order-1")    # passes — the call happened
    with pytest.raises(AssertionError):
        notify.assert_awaited_once_with("order-1")   # fails — nothing ran
    coro.close()                                 # silence the never-awaited warning in this demo


@pytest.mark.asyncio
async def test_await_history() -> None:
    client = create_autospec(PaymentsClient, instance=True)
    client.charge.return_value = {"id": "ch_1"}
    await checkout(client, "acct", 500)
    client.charge.assert_awaited_once_with("acct", 500)
    assert client.charge.await_args_list[0].args == ("acct", 500)

Prefer assert_awaited_once_with, assert_awaited_with and await_args_list throughout async tests. await_count increases when the await starts, not when it finishes — a detail that matters in step 5, where a slow mock can be asserted as awaited while it is still pending.

Verify: remove an await from a call site in the code under test; the corresponding assert_awaited_* assertion fails even though assert_called_* would pass. Combine with the warning filters from the pytest-asyncio guide so the stray coroutine also fails the build.

4. Mock async context managers and iterators

Async clients are often used through async with and async for: sessions, transactions, cursors, streaming responses. MagicMock supports the async dunder methods directly — __aenter__ and __aexit__ are configured as AsyncMock automatically, and __aiter__ accepts an iterable of values.

from unittest.mock import AsyncMock, MagicMock


async def load_rows(pool) -> list[dict]:
    async with pool.acquire() as conn:
        return await conn.fetch("SELECT id FROM orders")


async def stream_chunks(response) -> bytes:
    return b"".join([chunk async for chunk in response.aiter_bytes()])


@pytest.mark.asyncio
async def test_async_context_manager() -> None:
    pool = MagicMock()
    conn = pool.acquire.return_value.__aenter__.return_value
    conn.fetch = AsyncMock(return_value=[{"id": 1}, {"id": 2}])

    assert await load_rows(pool) == [{"id": 1}, {"id": 2}]
    pool.acquire.return_value.__aexit__.assert_awaited_once()     # connection released


@pytest.mark.asyncio
async def test_async_iterator() -> None:
    response = MagicMock()
    response.aiter_bytes.return_value.__aiter__.return_value = [b"he", b"llo"]
    assert await stream_chunks(response) == b"hello"

Asserting on __aexit__ is the interesting part: it verifies that the code under test released the connection, including on error paths. Make conn.fetch raise and the same assertion checks that the context manager still exits — the property that keeps connection pools from exhausting in production.

Verify: change load_rows to call pool.acquire().__aenter__() manually without exiting; the __aexit__ assertion fails.

5. Make dependencies slow or stuck on command

Timeouts, cancellation and backpressure only happen when a dependency is slow, so the mock must be able to be slow. Two techniques cover almost everything: an asyncio.Event that holds the await until the test releases it, and a side effect that sleeps.

import asyncio
from unittest.mock import patch

import pytest


async def fetch_rate(currency: str) -> float:          # rates.py, the real dependency
    await asyncio.sleep(1)
    return 1.1


async def convert(amount: float, currency: str) -> float:
    return round(amount * await fetch_rate(currency), 2)


@pytest.mark.asyncio
async def test_holds_until_released() -> None:
    release = asyncio.Event()

    async def held(currency: str) -> float:
        await release.wait()
        return 3.0

    with patch(f"{__name__}.fetch_rate", side_effect=held) as fake:  # patch() makes an AsyncMock
        task = asyncio.create_task(convert(1, "USD"))
        await asyncio.sleep(0)
        assert not task.done()                      # in flight: assert intermediate state here
        fake.assert_awaited_once_with("USD")        # counted when the await starts
        release.set()
        assert await task == 3.0


@pytest.mark.asyncio
async def test_timeout_fires_against_a_hanging_dependency() -> None:
    client = create_autospec(PaymentsClient, instance=True)

    async def hang(*args, **kwargs):
        await asyncio.Event().wait()                # never returns

    client.charge.side_effect = hang
    with pytest.raises(TimeoutError):
        async with asyncio.timeout(0.05):
            await client.charge("acct", 500)

patch() detects that the target is an async def function and substitutes an AsyncMock automatically, so patched coroutine functions stay awaitable. The event-held pattern is what lets a test assert on the in-between state — requests queued, a semaphore full, a breaker half-open — which is where most concurrency bugs live. For timeouts of seconds or minutes, replace the real sleep with virtual time as shown in controlling time in asyncio tests.

Verify: both tests pass in milliseconds; removing release.set() makes the first one hang, which your per-test timeout converts into a named failure.

Verification

Async mocks are doing their job when:

• Signatures are enforced: client mocks come from create_autospec, and a changed method signature breaks the tests that call it.
• Failure sequences are explicit: retry and fallback tests script outcomes with side_effect lists or async functions, and assert the exact number of awaits.
• Awaits are asserted: tests use assert_awaited_* rather than assert_called_* for coroutine methods.
• Resource release is verified: __aexit__ assertions cover success and error paths for pools, sessions and transactions.
• Slowness is controllable: timeout and cancellation tests use held or hanging side effects, never real network latency.

Pitfalls & edge cases

• Mocking what you own. Mocking internal async helpers couples tests to implementation details. Mock the outermost boundary — the HTTP client, the driver — and let internal coroutines run for real.
• return_value of a coroutine function. Setting return_value to a coroutine object makes the mock return a coroutine that must itself be awaited, producing double-await bugs in tests. Set return_value to the final value; AsyncMock wraps it.
• Autospec and instance attributes. Attributes created in __init__ are not visible to create_autospec. Set them on the mock explicitly, or declare them as class-level annotations so autospec can see them.
• Sharing a mock across tests. A module-level mock accumulates await_count and side_effect state across tests. Create mocks inside each test or in a function-scoped fixture.
• Synchronous mocks inside async code. A MagicMock passed where the code awaits raises a TypeError in the test; do not "fix" production code to tolerate non-awaitables. Fix the mock.

Frequently Asked Questions

How do I mock an async function in Python?

Use unittest.mock.AsyncMock, or patch the function with unittest.mock.patch, which automatically substitutes an AsyncMock when the target is an async def function. Awaiting the mock returns its return_value or applies its side_effect. For client objects, create_autospec(Class, instance=True) builds AsyncMocks for async methods and checks call signatures.

Why does MagicMock fail with can't be used in await expression?

Calling a MagicMock returns another MagicMock, which is not awaitable. Async methods need AsyncMock, whose calls return awaitables. Replace the MagicMock with AsyncMock for the specific method, or build the whole client with create_autospec so async methods are detected automatically from the real class.

What is the difference between assert_called and assert_awaited on AsyncMock?

assert_called checks that the mock was called, which only creates a coroutine. assert_awaited checks that the resulting coroutine was actually awaited, meaning the code ran. A missing await in the code under test passes assert_called but fails assert_awaited, so async tests should assert on awaits.

How do I mock an async context manager like async with session?

Use MagicMock, which supports aenter and aexit as AsyncMocks. Configure the object returned inside the block through mock.aenter.return_value, and assert mock.aexit.assert_awaited_once() to verify that the resource was released. Async iterators work similarly by setting aiter.return_value to a list of items.

Related

• Testing Async Code — up to the topic overview for loops, time control and race reproduction.
• Controlling time in asyncio tests — pairing slow mocks with a virtual clock for long timeouts.
• Resilience, Cancellation & Error Handling — the section overview for retry, timeout and breaker behaviour.