Skip to content

Integration Testing Async Services with Real Dependencies

A mocked database agrees with every assumption in your code, which is exactly why it cannot find the bugs that matter: the constraint you forgot, the type the driver coerces differently, the query that deadlocks under concurrency, the pool that runs out. Running the tests against a real PostgreSQL and a real Redis finds those, and the usual objection — that it is slow — is largely untrue when the containers are session-scoped and isolation is a transaction rollback. The suite built in this guide runs six tests against live Postgres and Redis in 0.51 seconds, including fixtures. This covers the fixture stack, the isolation strategies and the details that make such a suite safe to run in parallel on CI.

Prerequisites

  • Python 3.11+ with pytest, pytest-asyncio, asyncpg and redis.
  • Containers for the dependencies — Docker Compose, testcontainers, or services your CI provides.
  • Async test basics from Testing Async Code, including asyncio_mode and loop scopes.
What a mock cannot tell you 2 columns contrasting mocked dependency, real dependency. What a mock cannot tell you mocked dependency fast, and agrees with you tests your call sequence encodes your assumptions never rejects bad SQL never times out or deadlocks real dependency slower, and argues constraint violations surface driver types and coercions pool limits and timeouts transaction and locking behaviour Measured: a six-test suite against real Postgres and Redis ran in 0.51 s.

1. Start the dependency once, per session

Container startup is the expensive part, so it belongs at session scope, with the schema created once:

@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def pool():
    conn = await wait_for(lambda: asyncpg.connect(DSN))
    await conn.execute("DROP TABLE IF EXISTS orders")
    await conn.execute("CREATE TABLE orders (id bigserial primary key, sku text, qty int)")
    await conn.close()
    pool = await asyncpg.create_pool(DSN, min_size=1, max_size=4)
    yield pool
    await pool.close()

Note both scope arguments. scope="session" keeps the fixture alive across tests; loop_scope="session" keeps it on one event loop, without which pytest-asyncio gives each test a fresh loop and the pooled connections — bound to the loop that created them — fail with confusing errors. Tests that use such a fixture need the matching marker:

pytestmark = pytest.mark.asyncio(loop_scope="session")

This mismatch is the most common reason a first attempt at async integration testing fails, and the error messages point everywhere except at the loop scope.

Verify: the fixture's setup output appears once for the whole suite, not once per test.

2. Wait for readiness, do not sleep

A container that has started is not a database that accepts connections. Polling with a deadline is both faster and more reliable than a fixed sleep:

async def wait_for(factory, timeout: float = 30.0):
    deadline = time.monotonic() + timeout
    last = None
    while time.monotonic() < deadline:
        try:
            return await factory()
        except Exception as exc:                       # not ready yet
            last = exc
            await asyncio.sleep(0.1)
    raise TimeoutError(f"dependency not ready: {last!r}")

Against an already-running container this returned in 0.10 s; against a cold start it takes as long as it takes, up to the deadline. Including the last exception in the timeout message turns "the fixture timed out" into "connection refused" or "password authentication failed", which is the difference between a five-minute and a fifty-minute debugging session.

Health checks in docker-compose.yml and testcontainers' wait strategies do the same job at the container level; the in-Python version works everywhere, including against dependencies your CI provides as services.

Verify: stopping the container makes the fixture fail in timeout seconds with the underlying connection error.

The fixture stack for a real dependency 5 ordered steps. The fixture stack for a real dependency reach the container session scope, started once poll until ready connect in a loop, with a deadline create the schema once per session open a transaction one per test roll back teardown, always Session scope for the expensive part, function scope for the isolation.

3. Isolate with a transaction rollback

The fastest isolation is a transaction the test never commits:

@pytest_asyncio.fixture(loop_scope="session")
async def db(pool):
    async with pool.acquire() as conn:
        transaction = conn.transaction()
        await transaction.start()
        try:
            yield conn                                 # the test runs inside it
        finally:
            await transaction.rollback()               # nothing survives

Verified: a test inserting a row saw count = 1, and the next test saw count = 0. Cost per test is microseconds, and there is no cleanup code to forget.

The limitation is real, though: code under test cannot manage its own transactions, because a COMMIT inside the fixture's transaction ends the isolation. Three alternatives, in ascending cost:

  • TRUNCATE the tables between tests — milliseconds, and it lets the code commit. Use TRUNCATE ... RESTART IDENTITY CASCADE to reset sequences too.
  • A schema per parallel worker — one setup per worker, with search_path set per connection, which is what makes pytest-xdist safe.
  • A container per test — seconds, reserved for tests that genuinely destroy the instance.

One measured detail that surprises people: sequences are not rolled back. A test that inserted after two earlier tests got id = 3, not id = 1, because nextval is non-transactional by design. Never assert on specific primary-key values; capture what the insert returns.

Verify: a test that inserts data is followed by one that asserts the table is empty, and both pass.

Four ways to isolate one test from the next A grid of 4 rows by 2 columns. Four ways to isolate one test from the next strategy cost per test limitation transaction rollback microseconds the code cannot commit TRUNCATE the tables milliseconds sequences and triggers persist a schema per worker once per worker needs search_path plumbing a container per test seconds only for destructive tests Rollback is the default; use truncation for code that manages its own transactions.

4. Test the things only a real dependency shows

With the fixtures in place, write the tests a mock could never fail:

async def test_constraint_violation_is_real(db):
    new_id = await db.fetchval(
        "INSERT INTO orders (sku, qty) VALUES ($1, $2) RETURNING id", "abc", 3)
    with pytest.raises(asyncpg.UniqueViolationError):
        await db.execute("INSERT INTO orders (id, sku, qty) VALUES ($1, $2, $3)",
                         new_id, "dup", 1)

That asserts on the driver's actual exception type, which is what your error handling will have to match in production — and which a mock returning a generic Exception would let you get wrong indefinitely. The same applies to Redis semantics:

async def test_cache_roundtrip(cache):
    await cache.set("k", "v", ex=5)
    assert await cache.get("k") == b"v"                # bytes, not str
    assert 0 < await cache.ttl("k") <= 5

b"v" rather than "v" is exactly the kind of detail a mock hides until production. Concurrency behaviour is another: four concurrent 200 ms queries on a four-connection pool completed in 0.32 s, proving the pool is genuinely parallel — the same test with max_size=1 would take four times as long, which is how you catch an accidentally serialised pool.

Verify: each integration test asserts something that a mocked version could not distinguish.

Where the time goes in an integration suite 3 bars comparing readiness check with the others. Where the time goes in an integration suite readiness check 0.10 s four concurrent 200 ms queries 0.32 s six tests, including fixtures 0.51 s Containers already running; a cold container start adds a few seconds, once per session. Per-test cost is dominated by the work, not by the isolation.

5. Keep the suite fast and parallel-safe

Integration suites decay when they get slow, so guard the properties that keep them quick:

  • Session-scoped containers and schemas, function-scoped isolation. Never restart a container per test.
  • Unique namespaces per worker. PYTEST_XDIST_WORKER in the schema name, the Redis database number or a key prefix makes parallel runs safe. The fixture above uses Redis database 1 and flushes it at setup — fine serially, wrong under xdist unless each worker gets its own number.
  • A marker and a default. -m "not integration" for the fast inner loop, the full suite in CI. Make the slow path opt-in for developers and mandatory for merges.
  • pytest-timeout. A test that hangs against a container should fail in seconds; deadlocks are one of the things you are testing for.
  • The same images as production, pinned by tag. Testing against postgres:18 and deploying on 15 finds problems in the wrong direction.

Verify: the suite passes under pytest -n 4 and takes no longer than serially per test.

Verification

An integration suite is healthy when:

  • Containers start once per session, and readiness is polled rather than slept for.
  • Loop scope matches fixture scope, with the matching asyncio marker on tests.
  • Each test starts from a known state, by rollback or truncation.
  • Tests assert driver-specific behaviour — exception types, returned types, TTLs.
  • Parallel runs are namespaced per worker.
  • Per-test overhead is milliseconds, not seconds.

Pitfalls & edge cases

  • Mismatched loop scopes. A session-scoped pool on function-scoped loops fails with errors that mention futures, not loops.
  • Asserting on generated ids. Sequences are not transactional; rollback does not reuse them.
  • Sharing a Redis database across workers. One worker's flushdb wipes another's data mid-test.
  • Committing inside a rollback fixture. The code under test escapes the isolation, and later tests see its rows.
  • Leaking connections between tests. A test that acquires without releasing exhausts a small pool; assert the pool's free count in teardown.
  • Depending on container ordering. Compose starts containers in parallel; poll each dependency independently rather than assuming.

Frequently Asked Questions

How do I run async integration tests against a real database?

Use a session-scoped pytest-asyncio fixture with loop_scope="session" that connects to a containerised database, creates the schema once, and yields a connection pool. Give each test a transaction that the fixture rolls back in teardown, and mark the tests with pytest.mark.asyncio(loop_scope="session").

Why do my async fixtures fail with 'attached to a different loop'?

Because the fixture is session-scoped but the loop is function-scoped, so the pool was created on a loop that no longer exists. Set loop_scope="session" on the fixture and use the matching marker on every test that consumes it.

How do I isolate integration tests from each other?

Wrap each test in a transaction and roll it back — microseconds per test and no cleanup code. If the code under test commits its own transactions, truncate the tables between tests instead, with RESTART IDENTITY to reset sequences, or give each parallel worker its own schema.

How do I wait for a container to be ready in an async test?

Poll: try to connect in a loop with a short sleep and an overall deadline, and include the last exception in the timeout message. Against an already-running container this took 0.10 s. A fixed sleep is both slower on a warm machine and unreliable on a cold one.

Are integration tests too slow to run on every commit?

Usually not, if the containers are session-scoped and isolation is a rollback: six tests against real Postgres and Redis ran in 0.51 s here. Keep container startup out of the per-test path, namespace parallel workers, and mark the suite so developers can skip it locally while CI always runs it.