How to Write Effective Integration Tests with Testcontainers

Testcontainers lets you spin up real databases, message queues, and services as Docker containers directly inside your test suite. Your integration tests run against the same PostgreSQL, Redis, or Kafka that your application uses in production instead of flaky mocks or in-memory substitutes. In Python, testcontainers-python (currently at v4.14.2) integrates with pytest fixtures that start a container before tests and tear it down after. You get isolated, reproducible, and parallelizable integration tests that catch bugs that unit tests and mocks cannot.

Below: setup with pytest, testing services beyond databases, performance patterns, and CI/CD configuration.

Why Mocks and In-Memory Databases Are Not Enough

Mocking db.execute() verifies your code calls execute with the right SQL string. It does not verify that the SQL is syntactically valid, that the schema exists, or that the query returns expected results with real data types and constraints. You could have a test suite where every mock-based test passes while your actual queries fail against a real PostgreSQL instance.

SQLite as a stand-in test database is a common compromise, but it misses PostgreSQL-specific features entirely. JSONB operators, ON CONFLICT DO UPDATE (upsert), ARRAY types, full-text search with tsvector, LISTEN/NOTIFY for pub/sub, and advisory locks all behave differently or do not exist in SQLite. If your application depends on any of these - and most non-trivial PostgreSQL applications do - your SQLite-based tests are lying to you about whether the code works.

A shared test database introduces a different class of problems. State accumulates between test runs, causing order-dependent failures. The classic symptom: “works on my machine but fails in CI.” Manual cleanup scripts are brittle and slow. Someone forgets to run them, or the cleanup misses a table, and suddenly tests start failing for reasons unrelated to the code under test.

Testcontainers solves all three problems. It creates a fresh, isolated container for each test session (or each test, depending on how you scope your fixtures). Every run starts from a clean state, and containers are destroyed after tests complete.

The performance cost is modest. Starting a PostgreSQL container takes 2-4 seconds. A Redis container starts in under 1 second. Amortized across a full test suite using session-scoped fixtures, this overhead is negligible. You pay the startup cost once, then run hundreds of tests against the same container.

Testcontainers exists across many languages - Java/Kotlin (the original), Go, Node.js, Rust, .NET, and Python. The Python library integrates with pytest natively and covers the most common use cases out of the box.

Setting Up Testcontainers with pytest

Installation

Install testcontainers with the extras for the services you need:

pip install testcontainers[postgresql,redis,kafka]

Or with uv :

uv add testcontainers[postgresql]

You need Docker running locally, or a Docker-compatible runtime. Podman works if you enable the Docker compatibility socket. Colima works on macOS as a lightweight Docker Desktop alternative.

Basic PostgreSQL Fixture

The core pattern lives in your conftest.py. A session-scoped fixture starts the container once, and all tests in the session share it:

import pytest
from testcontainers.postgres import PostgresContainer


@pytest.fixture(scope="session")
def pg_container():
    with PostgresContainer("postgres:16") as pg:
        yield pg

The with block handles the full lifecycle - it pulls the image if needed, starts the container, waits for PostgreSQL to accept connections, and stops the container when the session ends.

Getting the Connection URL

The container exposes a get_connection_url() method that returns a SQLAlchemy -compatible URL:

url = pg_container.get_connection_url()
# postgresql+psycopg2://test:test@localhost:32789/test

The port is randomly assigned each time, so there are no conflicts if you run tests in parallel or have other PostgreSQL instances running.

SQLAlchemy Integration

Wire the container into SQLAlchemy with layered fixtures:

from sqlalchemy import create_engine
from sqlalchemy.orm import Session


@pytest.fixture(scope="session")
def engine(pg_container):
    return create_engine(pg_container.get_connection_url())


@pytest.fixture
def db_session(engine):
    with Session(engine) as session:
        yield session
        session.rollback()

The engine fixture is session-scoped (created once), while db_session is function-scoped (created fresh for each test). The rollback() call ensures each test gets a clean slate without restarting the container.

Running Migrations

If your project uses Alembic for schema migrations, apply them in the fixture before yielding:

from alembic.config import Config
from alembic import command


@pytest.fixture(scope="session")
def pg_container():
    with PostgresContainer("postgres:16") as pg:
        alembic_cfg = Config("alembic.ini")
        alembic_cfg.set_main_option(
            "sqlalchemy.url",
            pg.get_connection_url()
        )
        command.upgrade(alembic_cfg, "head")
        yield pg

This ensures your tests run against the exact same schema that production uses - migrations and all.

Transaction Rollback for Per-Test Isolation

The most effective isolation pattern wraps each test in a transaction that rolls back after the test completes:

from sqlalchemy import event


@pytest.fixture
def db_session(engine):
    connection = engine.connect()
    transaction = connection.begin()
    session = Session(bind=connection)

    # Start a nested savepoint
    nested = connection.begin_nested()

    @event.listens_for(session, "after_transaction_end")
    def restart_savepoint(session, trans):
        nonlocal nested
        if trans.nested and not trans._parent.nested:
            nested = connection.begin_nested()

    yield session

    session.close()
    transaction.rollback()
    connection.close()

No data is actually committed, so no cleanup queries are needed. This is the fastest isolation method available.

Testing Real Services: Redis, Kafka, and Custom Containers

Testcontainers works with more than relational databases. Anything with a Docker image can be spun up as a test dependency.

Redis

Test your caching logic, pub/sub, and Redis Streams against a real Redis instance:

from testcontainers.redis import RedisContainer
import redis


@pytest.fixture(scope="session")
def redis_client():
    with RedisContainer("redis:7.4") as container:
        client = redis.Redis.from_url(
            container.get_connection_url()
        )
        yield client

Kafka

Test producer/consumer logic including serialization, consumer groups, and delivery guarantees:

from testcontainers.kafka import KafkaContainer
from kafka import KafkaProducer, KafkaConsumer


@pytest.fixture(scope="session")
def kafka_container():
    with KafkaContainer("confluentinc/cp-kafka:7.7") as kafka:
        yield kafka


def test_produce_and_consume(kafka_container):
    bootstrap = kafka_container.get_bootstrap_server()
    producer = KafkaProducer(bootstrap_servers=bootstrap)
    producer.send("test-topic", b"hello")
    producer.flush()

    consumer = KafkaConsumer(
        "test-topic",
        bootstrap_servers=bootstrap,
        auto_offset_reset="earliest",
        consumer_timeout_ms=5000,
    )
    messages = [msg.value for msg in consumer]
    assert b"hello" in messages

Generic Containers

For services without a dedicated module, use GenericContainer. Here is MinIO for S3-compatible storage testing:

from testcontainers.generic import GenericContainer


@pytest.fixture(scope="session")
def minio_container():
    container = (
        GenericContainer("minio/minio:latest")
        .with_exposed_ports(9000)
        .with_env("MINIO_ROOT_USER", "admin")
        .with_env("MINIO_ROOT_PASSWORD", "password")
        .with_command("server /data")
    )
    with container:
        yield container

Elasticsearch

Test index creation, mappings, and queries against a real Elasticsearch node:

from testcontainers.elasticsearch import ElasticsearchContainer
from elasticsearch import Elasticsearch


@pytest.fixture(scope="session")
def es_client():
    with ElasticsearchContainer("elasticsearch:8.15.0") as es:
        client = Elasticsearch(es.get_url())
        yield client

Custom Wait Strategies

Testcontainers checks port availability by default, but you can add custom wait strategies for more reliable startup detection:

from testcontainers.core.waiting_utils import wait_for_logs

with PostgresContainer("postgres:16") as pg:
    wait_for_logs(
        pg,
        "database system is ready to accept connections",
        timeout=30,
    )

Multi-Container Networks

Connect multiple containers on a shared network so they can communicate by container name:

from testcontainers.core.network import Network

with Network() as network:
    pg = PostgresContainer("postgres:16")
    pg.with_network(network)
    pg.with_network_aliases("db")

    with pg:
        # Other containers on the same network
        # can reach postgres at "db:5432"
        yield pg

This simulates multi-service architectures where your app container talks to a database container by hostname.

Patterns for Fast, Reliable Integration Tests

Naively starting a container per test is slow. These patterns make Testcontainers-based tests fast enough for continuous development.

Session-Scoped Containers

Start the container once per test session with scope="session". All tests share the same container. Isolation comes from transactions or database-level cleanup, not container restarts. This turns a 2-4 second PostgreSQL startup into a one-time cost instead of per-test overhead.

Transaction Rollback Pattern

Wrap each test in a BEGIN / ROLLBACK using a nested SAVEPOINT. No data is committed, no cleanup queries run. This is the fastest isolation method because the database does almost no work between tests.

Database Template Pattern

For tests that need a fully independent database (perhaps because they test DDL operations or connection pooling), PostgreSQL’s template databases are fast:

CREATE DATABASE test_run_001 TEMPLATE template_db;

PostgreSQL copies templates as a file-level operation, making it near-instant for small schemas. Create the schema once in a template database, then stamp out copies for each test or test group that needs full isolation.

Parallel Test Execution

With pytest-xdist , each worker gets its own session-scoped container on a different random port. No coordination needed - Testcontainers handles port assignment automatically. Alternatively, use a single shared container with per-worker databases or schemas for lower resource usage.

Test Data Factories

Use factory_boy or polyfactory to generate test data that gets inserted into the Testcontainers database. Factories are more maintainable than SQL fixture files and adapt automatically when your schema changes:

import factory
from myapp.models import User

class UserFactory(factory.alchemy.SQLAlchemyModelFactory):
    class Meta:
        model = User
        sqlalchemy_session_persistence = "commit"

    username = factory.Faker("user_name")
    email = factory.Faker("email")

Selective Test Running

Mark integration tests so you can skip them during quick local iterations:

@pytest.mark.integration
def test_user_creation(db_session):
    ...

Run only unit tests locally with pytest -m "not integration". Run the full suite in CI where container startup time is acceptable.

CI/CD Integration and Troubleshooting

GitHub Actions

Docker is pre-installed on ubuntu-latest runners, so Testcontainers works without extra configuration. No services: block needed - Testcontainers manages everything through the Docker socket at /var/run/docker.sock.

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - run: pip install -e ".[test]"
      - run: pytest

GitLab CI

Use docker:dind as a service and set the Docker host:

test:
  image: python:3.12
  services:
    - docker:dind
  variables:
    DOCKER_HOST: tcp://docker:2375
  script:
    - pip install -e ".[test]"
    - pytest

If the socket path differs from the default, set TESTCONTAINERS_DOCKER_SOCKET_OVERRIDE=/var/run/docker.sock.

The Ryuk Container

Testcontainers starts a helper container called Ryuk (testcontainers/ryuk) that monitors test containers and removes them if your test process crashes without cleanup. This prevents leaked containers from accumulating on CI runners.

If your CI environment does not allow privileged containers, disable Ryuk with TESTCONTAINERS_RYUK_DISABLED=true. Be aware that crashed test runs may leave containers behind. Set TESTCONTAINERS_RYUK_CONTAINER_TIMEOUT=60 to clean up containers idle for more than 60 seconds.

Docker Compose Module

For multi-service architectures already defined in a docker-compose.yml, testcontainers-python provides a DockerCompose class:

from testcontainers.compose import DockerCompose

with DockerCompose("/path/to/project") as compose:
    host = compose.get_service_host("webapp", 8080)
    port = compose.get_service_port("webapp", 8080)
    # Test against http://{host}:{port}

This is useful when your application depends on multiple services that need to talk to each other - rather than wiring up individual containers and networks, you reuse your existing Compose file.

Podman Compatibility

Testcontainers works with Podman if you enable Docker compatibility mode. The team does not actively test against Podman, so edge cases exist, but the core functionality works for most use cases. Key setup steps:

• Enable the Podman Docker compatibility socket
• Set DOCKER_HOST to point to the Podman socket (e.g., unix:///run/user/1000/podman/podman.sock)
• For rootless Podman, you may need to disable Ryuk (TESTCONTAINERS_RYUK_DISABLED=true)

Common Errors and Debugging

For verbose debugging output, set TESTCONTAINERS_LOG_LEVEL=DEBUG. This shows container lifecycle events including the container ID, which you can use with docker logs <container-id> to see the service’s stdout/stderr when a test fails.

Putting It All Together

Here is a complete conftest.py that sets up PostgreSQL and Redis for a test suite:

import pytest
from testcontainers.postgres import PostgresContainer
from testcontainers.redis import RedisContainer
from sqlalchemy import create_engine, event
from sqlalchemy.orm import Session
import redis


@pytest.fixture(scope="session")
def pg_container():
    with PostgresContainer("postgres:16") as pg:
        yield pg


@pytest.fixture(scope="session")
def redis_client():
    with RedisContainer("redis:7.4") as container:
        client = redis.Redis.from_url(
            container.get_connection_url()
        )
        yield client


@pytest.fixture(scope="session")
def engine(pg_container):
    return create_engine(pg_container.get_connection_url())


@pytest.fixture
def db_session(engine):
    connection = engine.connect()
    transaction = connection.begin()
    session = Session(bind=connection)

    nested = connection.begin_nested()

    @event.listens_for(session, "after_transaction_end")
    def restart_savepoint(session, trans):
        nonlocal nested
        if trans.nested and not trans._parent.nested:
            nested = connection.begin_nested()

    yield session

    session.close()
    transaction.rollback()
    connection.close()

And a test file using these fixtures:

from myapp.models import User


@pytest.mark.integration
def test_create_user(db_session):
    user = User(username="testuser", email="test@example.com")
    db_session.add(user)
    db_session.flush()

    result = db_session.query(User).filter_by(
        username="testuser"
    ).first()
    assert result is not None
    assert result.email == "test@example.com"


@pytest.mark.integration
def test_cache_user(db_session, redis_client):
    user = User(username="cached", email="cached@example.com")
    db_session.add(user)
    db_session.flush()

    redis_client.set(f"user:{user.id}", user.email)
    cached = redis_client.get(f"user:{user.id}")
    assert cached == b"cached@example.com"

Each test runs against real PostgreSQL and Redis instances. The transaction rollback ensures test_create_user does not leave data behind that would affect test_cache_user. The containers start once and are shared across the entire session, keeping the total overhead to a few seconds regardless of how many tests you write.