async with pool.acquire() as conn: is one of the most common lines in async Python and one of the easiest to mock incorrectly. The manager has two async protocol methods, the variable after as is bound to whatever __aenter__ returns rather than to the manager itself, and __aexit__'s return value decides whether an exception in the body propagates or vanishes. Get the last one wrong and the test passes while silently swallowing the very error it was written to observe.
These mistakes are especially costly because async context managers are almost always guarding a scarce resource — a pooled connection, a lock, a transaction, an open stream — and the whole point of the manager is that the resource is released on every exit path. A mock that suppresses exceptions or hands the body the wrong object makes a test pass precisely in the situation where the real resource would leak. The fix is a few explicit lines rather than reliance on auto-configured mocks, plus an assertion that cleanup actually ran. Once written as a small helper, the pattern is reusable for every pool, session, lock and client in an async codebase, and the helper itself becomes the place where the rule about __aexit__ is encoded once rather than remembered everywhere.
Prerequisites
- Python 3.8+ for
AsyncMock; 3.10+ is recommended for the improved auto-detection inpatch. pytest >= 8.0with an async runner, as in pytest-asyncio in depth.- The async mocking basics from patching async code and coroutines.
Solution
A helper builds the manager correctly once; tests use it everywhere.
from unittest.mock import AsyncMock, MagicMock
import pytest
def async_cm(value):
"""An async context manager double that behaves like a real one."""
cm = MagicMock()
cm.__aenter__ = AsyncMock(return_value=value) # what `as` binds to
cm.__aexit__ = AsyncMock(return_value=False) # False: exceptions propagate
return cm
async def test_failed_query_still_releases_the_connection(repo):
connection = AsyncMock()
connection.fetch.side_effect = RuntimeError("query failed")
repo.pool.acquire = MagicMock(return_value=async_cm(connection))
with pytest.raises(RuntimeError, match="query failed"):
await repo.load_orders("cus_1")
manager = repo.pool.acquire.return_value
manager.__aexit__.assert_awaited_once() # cleanup ran on the failure path
exc_type, exc, _tb = manager.__aexit__.await_args.args
assert exc_type is RuntimeError # and saw the real exception
__aexit__ set explicitly.Why this works
async with expands to an await on __aenter__, the body, and an await on __aexit__ with the exception triple — three None values on success. Python then checks the truthiness of what __aexit__ returned: truthy means "I handled it, suppress the exception", falsy means "let it propagate". A MagicMock's auto-created __aexit__ returns another mock object, which is truthy, so every exception raised in the body disappears.
Setting __aexit__ to an AsyncMock(return_value=False) restores the real manager's behaviour for any manager that does not deliberately suppress exceptions — which is nearly all of them. Recording the call also gives the test something to assert on: that cleanup was awaited, and with which exception, which is the evidence that the resource would have been released in production.
The binding of as deserves the same care. Python evaluates the expression after async with, awaits its __aenter__, and binds the result — so async with pool.acquire() as conn gives conn whatever __aenter__ returned, not the manager. A double that returns itself from __aenter__ works for managers like locks, where the manager and the resource are the same object, and silently misleads for pools and sessions, where the body expects a different object entirely. Configuring __aenter__'s return value explicitly, every time, removes the ambiguity.
Edge cases and failure modes
- Forgetting
__aenter__'s return value. The body receives a generic mock instead of the connection it expects, and assertions on it test nothing. Always set it explicitly. - Managers that should suppress. A few managers —
contextlib.suppress, some retry helpers — legitimately returnTrue. Mirror that deliberately and name it in the test. acquire()being awaited itself. Some APIs areasync with await pool.acquire()rather thanasync with pool.acquire(). The first needsacquireto be anAsyncMockreturning the manager; the second needs a plainMagicMock. Match the real API.- Nested managers. A transaction inside a connection inside a pool needs each level configured. The helper composes:
async_cm(async_cm(transaction))is not right, but returningasync_cm(conn)fromacquireandasync_cm(tx)fromconn.transactionis. - Using a real manager when available. For your own managers, a fake implementation with real
__aenter__/__aexit__methods is often clearer than a mock and cannot get the protocol wrong.
Testing both exit paths
A context manager exists to guarantee cleanup, and the guarantee has two halves. The success path — body completes, __aexit__ receives three Nones, resources are released — is what every test exercises by accident. The failure path — body raises, __aexit__ receives the exception, resources are still released, the exception still reaches the caller — is what production exercises during an incident and tests exercise almost never.
Writing both as explicit tests turns the guarantee into something the suite checks rather than something the code claims. The failure-path test is the more valuable of the two, and it should assert three things: the exception propagated to the caller, __aexit__ was awaited exactly once, and it received the real exception type. That last check catches a subtle bug where intermediate code catches the original exception and raises a different one, leaving the manager to clean up with the wrong information — harmless for a connection, but significant for a transaction manager that decides between commit and rollback based on whether an exception occurred.
A third test is worth adding for any manager wrapping a transaction: that on the success path the commit happened and on the failure path the rollback did. With a hand-configured double this is a matter of giving the transaction mock commit and rollback as AsyncMocks and asserting which one was awaited. It is the test that proves the manager does its job rather than merely that Python called its methods, and it is the one that fails when someone refactors the manager and gets the condition backwards.
A fake manager instead of a mock
For managers you own, a small fake class is often clearer than any mock configuration, because it implements the protocol in ordinary Python and cannot get the suppression rule wrong by accident.
class FakeConnection:
def __init__(self):
self.executed: list[str] = []
self.released = False
self.exit_exception: type | None = None
async def execute(self, sql: str) -> None:
self.executed.append(sql)
class FakePool:
def __init__(self):
self.connection = FakeConnection()
def acquire(self):
return self # the pool is its own manager here
async def __aenter__(self):
return self.connection
async def __aexit__(self, exc_type, exc, tb):
self.connection.released = True
self.connection.exit_exception = exc_type
return False # never suppress
Tests then read naturally — assert pool.connection.released, assert pool.connection.exit_exception is RuntimeError — with no knowledge of how mocks record awaits. The fake is also reusable across every test that needs a pool, which is where it pays back over repeated mock configuration, and it can be checked against the real pool with a small contract suite in the same way as any other fake.
The trade-off is a few dozen lines of test support code. For a manager used in one or two tests, the helper function above is enough. For the connection pool, the database session or the HTTP client that half the suite depends on, the fake is the better long-term choice — it is written once, reviewed once, and every test that uses it inherits a correct protocol implementation without having to know what one looks like, for the same reasons argued in writing an in-memory fake repository.
Frequently Asked Questions
Why does my test pass even though the body raised inside async with?
Because the mocked __aexit__ returned a truthy value. An unconfigured MagicMock or AsyncMock returns a Mock object, which is truthy, and a truthy return from __aexit__ tells Python to suppress the exception. Set __aexit__ to an AsyncMock with return_value=False so exceptions propagate as they would with the real manager.
What does async with bind the target variable to?
The value returned by awaiting __aenter__, not the manager object itself. If a test configures the manager but forgets __aenter__'s return value, the body receives a generic Mock rather than the connection or session it expects, and assertions on that object test nothing meaningful.
Can autospec handle async context managers?
Yes. create_autospec on a class that defines __aenter__ and __aexit__ produces a mock with AsyncMock versions of both, with the real signatures. You still need to set __aenter__'s return value, but __aexit__'s signature is enforced and the protocol is correctly async.
Related
- Patching Async Code & Coroutines — the wider rules for async doubles.
- Asserting Await Order with AsyncMock — checking the sequence of enter, body and exit calls.
- Testing Async Generators and Context Managers — testing a real manager rather than a double.
- Rolling Back Every Test with Nested Transactions — when the real transaction manager is the better choice.
← Back to Patching Async Code & Coroutines