Database Testing Pipeline with Testcontainers and pytest: Catch Schema and Migration Bugs Before They Hit Production

Database tutorial - IT technology blog
Database tutorial - IT technology blog

Quick Start: Up and Running in 5 Minutes

Six months ago, a botched migration took our production database offline for 45 minutes. Every unit test had passed — because they ran against mocks. Real PostgreSQL had different ideas about constraint ordering. That incident is what pushed me to build a proper integration test pipeline with Testcontainers.

Install the dependencies first:

pip install testcontainers pytest sqlalchemy psycopg2-binary alembic

Then a minimal test that spins up a real PostgreSQL container:

import pytest
from testcontainers.postgres import PostgresContainer
from sqlalchemy import create_engine, text

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

def test_basic_connection(postgres_container):
    engine = create_engine(postgres_container.get_connection_url())
    with engine.connect() as conn:
        result = conn.execute(text("SELECT version()"))
        version = result.scalar()
    assert "PostgreSQL 16" in version

Run it:

pytest test_db.py -v

First run downloads the Docker image — about 170MB. After that, containers spin up in 2–3 seconds. This is real PostgreSQL, not a mock or SQLite wearing a Postgres costume. That distinction is easy to dismiss until a mock-passing migration brings your production database down.

Deep Dive: What You Can Actually Test

Testing Migrations End-to-End

Running Alembic migrations against a genuinely clean database — that’s where this setup earns its keep. Unit tests miss a lot: a migration referencing a non-existent column, PostgreSQL-specific syntax that SQLite silently swallows, constraint ordering that only matters on a real engine.

import pytest
from testcontainers.postgres import PostgresContainer
from sqlalchemy import create_engine, text
from alembic.config import Config
from alembic import command

@pytest.fixture(scope="session")
def migrated_db():
    with PostgresContainer("postgres:16") as pg:
        engine = create_engine(pg.get_connection_url())

        alembic_cfg = Config("alembic.ini")
        alembic_cfg.set_main_option("sqlalchemy.url", pg.get_connection_url())
        command.upgrade(alembic_cfg, "head")

        yield engine

def test_users_table_schema(migrated_db):
    with migrated_db.connect() as conn:
        result = conn.execute(text(
            "SELECT column_name, data_type "
            "FROM information_schema.columns "
            "WHERE table_name = 'users' ORDER BY ordinal_position"
        ))
        columns = {row[0]: row[1] for row in result}

    assert "id" in columns
    assert "email" in columns
    assert columns["created_at"] == "timestamp with time zone"

I’ve caught three migration bugs this way that would have required manual intervention in production: a foreign key constraint pointing the wrong direction, a missing index on a high-traffic lookup column, and a VARCHAR(255) where we needed TEXT. None of those would have surfaced in mocked tests.

Testing Query Logic Against Real Data

Mocked databases let you assert that a query was called. Real databases let you assert that the query returns the right data. There’s a big difference.

from sqlalchemy.orm import Session
from your_app.models import User, Order

@pytest.fixture
def db_session(migrated_db):
    with Session(migrated_db) as session:
        yield session
        session.rollback()  # Clean up after each test

def test_user_total_completed_orders(db_session):
    # Arrange: insert real test data
    user = User(email="[email protected]", name="Test User")
    db_session.add(user)

    for i in range(3):
        order = Order(user=user, amount=10.00 * (i + 1), status="completed")
        db_session.add(order)

    db_session.add(Order(user=user, amount=99.00, status="pending"))
    db_session.flush()

    # Act: run the actual query from your application layer
    result = db_session.execute(text(
        "SELECT SUM(amount) FROM orders "
        "WHERE user_id = :uid AND status = 'completed'"
    ), {"uid": user.id}).scalar()

    # Assert: pending order must NOT be included
    assert result == 60.00

That session.rollback() in fixture teardown keeps each test isolated without dropping and recreating the schema. Fast and predictable.

Advanced Usage: Building the Full CI/CD Pipeline

Session-Scoped Container with Per-Test Isolation

Fifty database tests running sequentially will grind your CI to a halt. One container per session — migrations run once — with transaction rollbacks per test for data isolation. Here’s the pattern:

# conftest.py
import pytest
from testcontainers.postgres import PostgresContainer
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from alembic.config import Config
from alembic import command

@pytest.fixture(scope="session")
def db_engine():
    with PostgresContainer("postgres:16") as pg:
        engine = create_engine(
            pg.get_connection_url(),
            pool_size=10,
            max_overflow=20
        )
        alembic_cfg = Config("alembic.ini")
        alembic_cfg.set_main_option("sqlalchemy.url", pg.get_connection_url())
        command.upgrade(alembic_cfg, "head")
        yield engine

@pytest.fixture
def db_session(db_engine):
    connection = db_engine.connect()
    transaction = connection.begin()
    session = Session(bind=connection)
    yield session
    session.close()
    transaction.rollback()
    connection.close()

Layer pytest-xdist on top for parallel execution:

pip install pytest-xdist
pytest -n 4 tests/integration/

This pattern cut our test suite from 8 minutes to 90 seconds.

GitHub Actions Integration

Testcontainers requires Docker. GitHub Actions ships with Docker already available, so the CI config stays minimal:

# .github/workflows/db-tests.yml
name: Database Integration Tests

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest

    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"

      - name: Install dependencies
        run: pip install -r requirements.txt

      - name: Run database integration tests
        run: pytest tests/integration/ -v --tb=short
        env:
          TESTCONTAINERS_RYUK_DISABLED: "false"

Testing Migration Rollbacks

Emergency rollbacks are stressful enough. Discovering your downgrade script throws an error mid-incident makes them worse. Test every migration in both directions:

def test_migration_upgrade_and_downgrade(postgres_container):
    engine = create_engine(postgres_container.get_connection_url())
    alembic_cfg = Config("alembic.ini")
    alembic_cfg.set_main_option("sqlalchemy.url", postgres_container.get_connection_url())

    # Upgrade to latest
    command.upgrade(alembic_cfg, "head")

    # Downgrade one step — this should not throw
    command.downgrade(alembic_cfg, "-1")

    # Upgrade again to verify idempotency
    command.upgrade(alembic_cfg, "head")

    with engine.connect() as conn:
        result = conn.execute(text("SELECT COUNT(*) FROM alembic_version"))
        assert result.scalar() == 1

Practical Tips After 6 Months in Production

Separate fixture scopes deliberately. Use scope="session" for the container and migrations (expensive to repeat), and scope="function" with transaction rollbacks for data isolation (cheap). This combination is what cut our test suite from 8 minutes to 90 seconds.

Pin your database version in tests. Use postgres:16 instead of postgres:latest. Tests won’t randomly break when a new major version drops. Match it exactly to whatever runs in production.

Write a schema assertion test alongside every migration. Every migration that lands on main should have a corresponding test verifying the expected schema state. It becomes a living record of how your database evolved — and catches regressions before they leave CI.

# Test added alongside migration 005_add_user_preferences.py
def test_migration_005_schema(migrated_db):
    with migrated_db.connect() as conn:
        result = conn.execute(text(
            "SELECT data_type FROM information_schema.columns "
            "WHERE table_name = 'user_preferences' AND column_name = 'settings'"
        ))
        data_type = result.scalar()

    # Explicitly assert jsonb, not just json — we need GIN index support
    assert data_type == "jsonb"

Test your seed data scripts. If your application ships with reference data or default configs, include a fixture that runs those scripts and verifies the row counts. We caught a broken seed script this way before it reached staging.

Prepare test fixtures carefully. Sometimes I need edge-case datasets for integration tests — reference data from CSV exports or sanitized production snapshots. When I need to quickly convert CSV to JSON for data imports, I use toolcraft.app/en/tools/data/csv-to-json — runs entirely in the browser, so no data leaves your machine. That matters when the CSV contains pseudonymized test records you can’t upload anywhere.

Add a retry for container startup on CI. On GitHub Actions, Docker occasionally takes a few extra seconds to be ready. Wrap your container fixture in a simple retry if you see intermittent startup failures on fresh runners.

Six months of this pipeline running on every pull request. We’ve caught schema bugs, wrong constraint directions, missing indexes, and two migrations that passed on local SQLite but blew up on real PostgreSQL in staging. The pipeline adds about 90 seconds to CI. It has saved us from at least four production incidents. Worth it.

Share: