Integration, Database & Service Testing

Integration tests are where test suites go to become slow and unreliable, and the reasons are nearly always structural rather than intrinsic. A suite that starts a Postgres container per test, deletes rows in teardown, seeds data by calling the application's own API, and asserts against whatever the staging environment happens to be running today will be slow, order-dependent, and red for reasons unrelated to the change under review. The same coverage, structured differently — one container per session, one transaction per test, data built by explicit factories, contracts verified against a schema — runs in a fraction of the time and fails only when something is genuinely broken.

This section covers that restructuring: transactional database fixtures that give per-test isolation without per-test setup, Testcontainers lifecycles that start real services once and wait for them properly, data factories that make a test's preconditions readable, and contract testing for HTTP APIs that moves the breakage detection to the side that caused it. Readers are assumed comfortable with pytest fixtures, Docker, and at least one SQL toolkit.

The layers, and what each one is allowed to be slow about

The useful mental model is not a pyramid but a set of concentric boundaries. Each layer is allowed to cross exactly one more boundary than the layer inside it, and the cost of a test is dominated by the outermost boundary it crosses.

The corollary is a triage rule for an existing slow suite: find the tests crossing the outer boundaries and ask what each one uniquely catches. A test that spins up Postgres to check that an empty basket totals zero is a unit test wearing a container. Moving it inward is usually a one-line change to accept a repository argument, which is the argument made at length in dependency injection for testability.

Transactional isolation beats cleanup

The most consequential decision in a database suite is how a test leaves the database clean. There are three options and they are not equivalent.

Delete in teardown is the intuitive one and the worst. It is O(tables) per test, it has to respect foreign-key order, it misses anything a test created through a path you forgot about, and an exception mid-test skips the cleanup entirely, poisoning every test after it.

Recreate the schema per test is correct and roughly a hundred times too slow — hundreds of milliseconds of DDL for a test that does two inserts.

Wrap each test in a transaction and roll it back is correct, cheap, and immune to mid-test failures, because the rollback happens in the fixture's finally whether the test passed, failed or raised. The subtlety is that application code often calls commit() itself, which would end your outer transaction — the answer is a savepoint, and both SQLAlchemy and Django provide the machinery.

import pytest
from sqlalchemy import create_engine, event
from sqlalchemy.orm import Session


@pytest.fixture(scope="session")
def engine():
    # One engine, one pool, for the whole run. The DSN points at a container
    # started once per session, not at a developer's local install.
    return create_engine("postgresql+psycopg://test:test@localhost:5432/test")


@pytest.fixture
def db_session(engine):
    connection = engine.connect()
    outer = connection.begin()                 # the transaction we will roll back
    session = Session(bind=connection, join_transaction_mode="create_savepoint")
    try:
        yield session                          # application code may commit freely
    finally:
        session.close()
        outer.rollback()                       # undoes everything, always
        connection.close()

join_transaction_mode="create_savepoint" (SQLAlchemy 2.0) is the important argument: it tells the session to open a SAVEPOINT inside the connection's existing transaction, so a session.commit() inside the application releases the savepoint rather than committing the outer transaction. On 1.4 the same effect needed a manual after_transaction_end event listener that restarted the nested transaction — code that appears in a lot of older conftest.py files and can now be deleted.

What this buys in CI is measurable: a suite of 400 database tests goes from roughly four minutes of setup-dominated runtime to under thirty seconds, and the tests stop caring what order they run in. The detailed walkthrough, including the cases where savepoints are not enough, is in rolling back every test with nested transactions.

Real services, started once, waited for properly

Testcontainers starts a real Docker container from the test process and hands back a connection URL. The value is that the service is the actual one — the same Postgres version, the same Redis eviction behaviour, the same Kafka protocol — rather than an approximation whose differences surface in production.

import pytest
from testcontainers.postgres import PostgresContainer


@pytest.fixture(scope="session")
def postgres():
    # Pinned tag: an unpinned "latest" makes the suite's behaviour depend on
    # when the CI cache was last warmed.
    with PostgresContainer("postgres:16-alpine") as container:
        yield container


@pytest.fixture(scope="session")
def dsn(postgres):
    # The container maps 5432 to an ephemeral host port; never hardcode it.
    return postgres.get_connection_url()

Two mistakes account for most Testcontainers pain. The first is scope: a function-scoped container adds two to five seconds per test, which is how a fifty-test suite becomes a four-minute one. The second is readiness — a container that is running is not a container that is accepting connections, and the gap is where time.sleep(5) gets added and then doubled every time CI is busy. Every Testcontainers module ships a wait strategy that polls the real readiness signal (a log line, a port, a health query); using it removes both the sleep and the flake, as shown in waiting for container readiness without sleep.

Data that says what the test means

The second-largest source of unreadable integration tests is setup. Twenty lines constructing an order, a customer, three line items and a payment before one line of assertion leaves the reader unable to see which of those twenty-one values the test is actually about.

Factories fix this by giving every field a sensible default and letting the test override only what matters:

import factory
from myapp.models import Customer, Order


class CustomerFactory(factory.Factory):
    class Meta:
        model = Customer

    name = factory.Sequence(lambda n: f"customer-{n}")   # unique, deterministic
    country = "GB"
    vat_exempt = False


def test_vat_is_zero_for_exempt_customers(db_session):
    # One overridden field: the reader sees immediately what this test is about.
    order = Order(customer=CustomerFactory(vat_exempt=True), net=100_00)
    assert order.vat() == 0

factory.Sequence rather than faker.name() for anything that must be unique is the detail that keeps factories deterministic. Random data in a factory means a test that fails one run in a thousand when the generator happens to produce a duplicate or an apostrophe — and reproducing it requires the seed, which nobody recorded. Where realistic data genuinely helps, seed the generator explicitly; generating reproducible fake data with Faker covers the seeding hooks, and property-based testing is the right tool when the goal is genuinely to explore the input space rather than to fill in a field.

Contracts, not staging environments

Testing against a shared staging environment produces a suite whose result depends on what somebody else deployed twenty minutes ago. The alternative is to test the agreement rather than the deployment: record the request and response shapes your code depends on, verify your side against them locally, and hand the provider a machine-checkable description they can verify in their own pipeline.

Two mechanisms cover most cases. Schema validation asserts that a real response conforms to the provider's published OpenAPI document, which catches a field that changed type without anyone's tests needing to coordinate. Consumer-driven contracts go further: the consumer's expectations are published as a pact that the provider's pipeline replays against its real implementation, so a breaking change fails the provider's build. Neither replaces the occasional end-to-end smoke test, but both move the failure to the moment and the repository where it can be fixed cheaply — the details are in validating responses against an OpenAPI schema and consumer-driven contract tests with Pact.

Brokers, queues and the assertions worth making

Message-driven systems attract a particular kind of useless test: publish a message, sleep, assert the handler ran. It is slow, it is flaky, and it verifies the broker rather than the code. The assertions that earn their runtime are about the guarantees the broker exists to provide.

Acknowledgement. Does the consumer acknowledge only after the work is durable? The test is to make the handler raise, then assert the message is redelivered rather than lost. With RabbitMQ that means asserting the message reappears after a channel close; with Kafka it means asserting the offset was not committed.

Idempotency. Every at-least-once broker will eventually deliver the same message twice. The test is to call the handler twice with the same payload and assert the second call is a no-op — no duplicate row, no second charge, no second email.

Ordering. Where ordering is guaranteed per key, the test publishes an interleaved sequence across two keys and asserts per-key order is preserved while cross-key order is not assumed.

import pytest


def test_handler_is_idempotent(db_session, order_created_event):
    # At-least-once delivery is a given; the handler must absorb the repeat.
    handle_order_created(order_created_event, db_session)
    handle_order_created(order_created_event, db_session)     # same message id

    orders = db_session.query(Order).filter_by(external_id=order_created_event.id).all()
    assert len(orders) == 1          # the dedupe key did its job


def test_failed_handler_does_not_acknowledge(consumer, broken_handler):
    consumer.register(broken_handler)
    with pytest.raises(RuntimeError):
        consumer.process_next()

    # The message must still be pending for redelivery, not silently dropped.
    assert consumer.unacknowledged_count == 1

Note that neither test needs a running broker. Acknowledgement and idempotency are properties of your handler, testable with an in-memory consumer double, and that is where they belong. Reserve the containerised broker for the small number of tests that genuinely exercise the client library's protocol handling — connection recovery, prefetch limits, consumer-group rebalancing — because those behaviours live in the broker and the client, not in your handler.

The awkward middle case is the handler that does both: consumes a message, writes to the database, publishes a follow-up. That pair of side effects is the classic dual-write problem, and the test worth writing is the one that kills the process between the two. In practice that means asserting the outbox row and the database row land in the same transaction, which is an assertion the transactional fixture above makes trivial: roll back, and neither should exist.

Running integration tests in CI without a flaky pipeline

The operational half of this section is arranging for these tests to run somewhere other than a developer's laptop without becoming the reason nobody trusts the build.

Separate them by marker, not by directory alone. A marker lets the fast suite run on every push and the full suite run on merge, without maintaining two invocations that drift apart:

# pyproject.toml
[tool.pytest.ini_options]
markers = [
    "integration: needs a real database or containerised service",
]
addopts = "--strict-markers"

pytest -m "not integration"      # pull-request feedback in under a minute
pytest -m integration            # merge queue, nightly, or on demand

--strict-markers matters more than it looks: without it, a typo in @pytest.mark.integraton creates a new marker silently, and those tests then run in the fast suite forever. The wider case for strict configuration is made in pytest configuration best practices.

Give every worker its own schema. Under pytest-xdist, derive the database name from the worker id so parallel workers never share mutable state:

import os

import pytest


@pytest.fixture(scope="session")
def database_name():
    # "gw0", "gw1", … under xdist; "master" when running serially.
    worker = os.environ.get("PYTEST_XDIST_WORKER", "master")
    return f"test_{worker}"

Cache the images, not the data. Docker layer caching on the runner removes the image pull, which is usually the largest fixed cost. Caching a populated database volume between runs, by contrast, reintroduces exactly the order dependence transactions were removed to avoid.

Make failures self-describing. An integration failure in CI is expensive to reproduce, so the run should hand you everything needed: the container logs, the last SQL statements, and the test's own captured output. Wiring that up once is the difference between a ten-minute diagnosis and a day of re-running the job with extra prints — the mechanics are in capturing artifacts from a failed CI test run.

Schema changes are the riskiest thing to leave untested

A migration is the one piece of code that runs exactly once against data you cannot see, in an environment you cannot roll back cheaply, at the moment of a deploy. Test suites routinely skip them, because the application's tests create their schema from the model definitions rather than by running the migrations — which means the migrations are never executed until production.

That gap produces two specific failures. The first is drift: the models and the migration history disagree, so a fresh database built from migrations differs from the one the tests ran against. Alembic detects this directly — autogenerate a revision against a migrated database and assert that it is empty:

from alembic.autogenerate import compare_metadata
from alembic.migration import MigrationContext

from myapp.models import Base


def test_models_match_migrations(engine_migrated):
    # engine_migrated points at a database built by running every migration.
    with engine_migrated.connect() as connection:
        context = MigrationContext.configure(connection)
        diff = compare_metadata(context, Base.metadata)

    assert diff == [], f"models and migrations disagree: {diff}"

The second failure is data: a migration that works on an empty schema and fails on real rows, because a new NOT NULL column has no default, or a backfill loop times out on a table with fifty million rows. The test is to apply the migration to a database that already contains representative rows, which the session-scoped container makes cheap. Both checks, plus the downgrade path that most teams discover is broken at the worst possible moment, are covered in testing Alembic migrations in CI.

Building the test database from migrations rather than from metadata.create_all() has one more benefit worth the switch on its own: it makes every test run an implicit smoke test of the migration history. If a revision is unapplyable, the suite cannot start, and the failure arrives at the pull request rather than at the deploy.

Common pitfalls and antipatterns

1. Seeding through the application's own API. Creating a user by calling POST /users couples every test to the endpoint's current behaviour and makes an unrelated validation change break two hundred tests. Root cause: setup as a test. Fix: insert through the model or factory layer directly.
2. Sharing a database between parallel workers. Under pytest-xdist, two workers truncating the same tables produce failures that look like race conditions in your code. Root cause: shared mutable state across processes. Fix: one schema or one database per worker, keyed on PYTEST_XDIST_WORKER.
3. Asserting on auto-increment identifiers. assert order.id == 1 passes exactly once. Root cause: an assertion about the database's internal counter. Fix: assert on values the test supplied, or on relationships.
4. Unpinned container images. postgres:latest makes the suite's behaviour a function of the CI cache. Root cause: an implicit dependency. Fix: pin the tag, and upgrade it deliberately in its own change.
5. Cleaning up with TRUNCATE ... CASCADE in a session fixture. It silently removes reference data the suite seeded once, so tests pass in isolation and fail in bulk. Root cause: cleanup scoped wider than the data it owns. Fix: roll back per test; reserve truncation for a deliberate reset between test groups.
6. Integration tests with no timeout. A container that never becomes ready turns into a job that runs until the CI platform kills it, with no traceback. Root cause: unbounded waiting. Fix: a wait strategy with an explicit timeout, plus a suite-wide ceiling via pytest-timeout.

Frequently Asked Questions

Should integration tests share one database or get one each? One database, many transactions. Starting a database per test costs seconds each and exhausts container resources; starting one per session and wrapping every test in a transaction that rolls back gives the same isolation for a few milliseconds. Use separate databases only when a test needs DDL that cannot run inside a transaction, or when parallel workers must not see each other's schema changes.

How do I stop integration tests from being order-dependent? Roll back rather than delete, seed reference data in a session fixture that never changes, and never let a test rely on an identifier produced by an earlier test. Then prove it by running with pytest-randomly; an order-dependent suite fails within a handful of seeds, and the bisection workflow narrows it to the offending pair.

Are Testcontainers too slow for CI? Only if started per test. A session-scoped container costs one image pull, cached by the runner, plus two to five seconds of startup amortised across the whole suite. What makes them slow is polling readiness with sleeps instead of a real wait strategy, and pulling images without a registry cache.

What belongs in an integration test rather than a unit test? Anything whose behaviour lives outside your code: SQL semantics, constraint violations, transaction isolation, migration ordering, serialization formats on the wire, and authentication handshakes. Business rules above those boundaries belong in fast tests with in-memory fakes.

How do contract tests differ from integration tests against a real service? An integration test proves your code works against the version of the service running right now. A contract test proves the agreement itself — request shape, response shape, status codes — and can be verified by the provider in their own pipeline, so a breaking change is caught by whoever made it rather than by whoever consumes it.

Related guides

• Build the fixture layer first with database fixtures and transactional tests.
• Get real services into the suite with spinning up services with Testcontainers.
• Make preconditions readable using test data factories and builders.
• Move breakage detection upstream with contract testing for HTTP APIs.
• Keep the fast layers fast by faking at the transport in mocking network and HTTP calls, and keep async integration fixtures on one loop with testing async and concurrent Python.

← Back to all guides