Python Testing & Debugging Beyond unittestAdvanced Python testing playbooks
Integration & Data

Testing Alembic Migrations in CI

A migration is the one piece of code that runs exactly once, against data nobody on the team can see, at the moment of a deploy. Most test suites never execute it: they build the schema from metadata.create_all() for speed, so the migration chain is tested for the first time in production. Four small tests close that gap, and together they run in a few seconds against the same container the rest of the suite already uses.

Prerequisites

  • alembic >= 1.13 and SQLAlchemy >= 2.0.
  • A real database engine matching production, started once per session — see starting Postgres with testcontainers-python.
  • pytest >= 8.0, and an alembic.ini whose URL can be overridden from a fixture.
  • An empty database per migration test, separate from the one the transactional fixture uses, since these tests commit DDL.

Solution

Python
import pytest
from alembic import command
from alembic.autogenerate import compare_metadata
from alembic.config import Config
from alembic.migration import MigrationContext
from alembic.script import ScriptDirectory
from sqlalchemy import create_engine

from myapp.models import Base


@pytest.fixture
def alembic_cfg(empty_database_dsn):
    cfg = Config("alembic.ini")
    cfg.set_main_option("sqlalchemy.url", empty_database_dsn)
    return cfg


def test_single_head(alembic_cfg):
    # Two concurrent branches each adding a revision produce two heads;
    # `alembic upgrade head` then refuses to run at deploy time.
    heads = ScriptDirectory.from_config(alembic_cfg).get_heads()
    assert len(heads) == 1, f"multiple heads: {heads} — add a merge revision"


def test_models_match_migrations(alembic_cfg, empty_database_dsn):
    command.upgrade(alembic_cfg, "head")
    engine = create_engine(empty_database_dsn)
    with engine.connect() as connection:
        diff = compare_metadata(MigrationContext.configure(connection), Base.metadata)
    # Any entry here is a model change with no migration, or the reverse.
    assert diff == [], f"models and migrations disagree:\n{diff}"


def test_every_revision_upgrades_and_downgrades(alembic_cfg):
    script = ScriptDirectory.from_config(alembic_cfg)
    revisions = list(reversed(list(script.walk_revisions())))   # base → head
    for revision in revisions:
        command.upgrade(alembic_cfg, revision.revision)
    for revision in reversed(revisions):
        command.downgrade(alembic_cfg, revision.down_revision or "base")
Four migration checks and what each one catches Four cards. The single-head check catches concurrent branches that cannot be applied linearly. The drift check compares migrated schema against model metadata. The stairway walks every revision up and back down. The data check applies a migration to rows written in the old schema and asserts the transformed result. Four tests, four different failures caught before deploy single head two branches both added a revision deploy refuses: "multiple heads" milliseconds drift a model changed with no migration, or back production schema differs from tests one upgrade stairway each revision up, then all the way down broken downgrade found in an incident seconds data rows in the old shape then upgrade NOT NULL with no default, bad backfill per data migration
The first three are generic and can be added to any project in an afternoon. The fourth is written per migration, for the ones that touch existing rows.

Why this works

Alembic keeps its history as a directed graph of revisions, and ScriptDirectory exposes that graph without touching a database — which is why the head check costs nothing. compare_metadata is the same machinery alembic revision --autogenerate uses: it reflects the live schema and diffs it against model metadata, so an empty diff after upgrade head means the migrations produce exactly the schema the models describe.

The stairway test walks the graph one revision at a time rather than jumping to head. That matters because a revision can depend on state an earlier revision created, and jumping skips the intermediate points where a later revision's downgrade must leave the schema in a shape its predecessor recognises.

Edge cases and failure modes

  • Autogenerate false positives. Server defaults, type variants and some constraint names are compared imperfectly on certain dialects. Filter known-harmless entries explicitly with include_object rather than loosening the assertion.
  • Irreversible migrations. A downgrade that raises NotImplementedError is honest; the stairway test should skip past it by stopping the downgrade walk at that revision, not by deleting the test.
  • Tests sharing the transactional fixture's database. These tests run DDL and commit it. Give them a separate empty database per test, or they will destroy the schema the rest of the suite is using.
  • SQLite in CI, Postgres in production. Many migrations work on one and fail on the other — ALTER COLUMN is the usual casualty. Run the real engine.
  • Branch labels and dependencies. Projects that use multiple branches deliberately need the head check relaxed to "one head per branch label", which get_heads combined with get_revision handles.

Testing a migration that transforms data

The failures that reach production are rarely DDL — they are migrations that work on an empty table and fail on real rows. Testing those needs rows in the old shape, which means raw SQL, because the model classes already describe the new one.

Python
from alembic import command
from sqlalchemy import create_engine, text


def test_backfill_splits_full_name(alembic_cfg, empty_database_dsn):
    engine = create_engine(empty_database_dsn)

    # 1. Bring the schema to the revision BEFORE the one under test.
    command.upgrade(alembic_cfg, "a1b2c3d4")         # has customer.full_name

    # 2. Insert rows in that old shape. Models cannot be used here.
    with engine.begin() as conn:
        conn.execute(text(
            "INSERT INTO customer (id, full_name) VALUES "
            "(1, 'Ada Lovelace'), (2, 'Plato'), (3, NULL)"
        ))

    # 3. Apply the migration under test.
    command.upgrade(alembic_cfg, "e5f6a7b8")         # splits into first/last

    # 4. Assert on the transformed data, including the awkward rows.
    with engine.connect() as conn:
        rows = conn.execute(text(
            "SELECT id, first_name, last_name FROM customer ORDER BY id"
        )).all()
    assert rows == [(1, "Ada", "Lovelace"), (2, "Plato", None), (3, None, None)]

The single-word name and the NULL are the rows that matter. A backfill written against "first space last" handles the common case and raises on the others, and those are precisely the rows that exist in production and not in a developer's imagination. Choosing representative awkward rows is most of the skill here, and pulling a few anonymised examples of each real shape from a production sample is the most reliable way to choose them.

Sequence for testing a data-bearing migration Four steps left to right. Upgrade to the revision before the one under test. Insert rows in that old schema using raw SQL, including awkward cases such as single-word names and nulls. Upgrade to the revision under test. Assert on the transformed rows. Old shape in, new shape out 1 · upgrade to N−1 schema has full_name only 2 · raw INSERTs 'Ada Lovelace' 'Plato', NULL 3 · upgrade to N the migration under test 4 · assert every row, awkward ones too Step 2 cannot use the ORM: the model classes describe the schema after step 3, not before it.
The awkward rows in step two are the test. A migration that only ever sees the happy-path row has been tested against the one case that was never going to fail.

Where these tests sit in the pipeline

The four checks differ enough in cost that they belong in different stages, and placing them deliberately keeps the fast suite fast without letting migration bugs through.

The single-head check needs no database and runs in milliseconds, so it belongs in the pull-request suite and, ideally, in a pre-commit hook: two developers merging revisions on the same day is the common case, and catching it before the second merge is far cheaper than after.

The drift check needs one upgrade against an empty database — a second or two with a session container — and also belongs on every pull request, because a model change without a migration is the single most frequent migration defect.

The stairway walks every revision twice. On a mature project with two hundred revisions that is a minute or more, which argues for the merge queue rather than every push. A useful compromise is to walk only the revisions added since the main branch on pull requests, and the full history nightly.

The data tests are per migration and usually fast, so they run wherever the migration's own change runs. The volume job above is nightly by nature.

Placing migration checks by cost Three pipeline stages. Every push runs the single-head check, the drift check and the data tests for changed migrations. The merge queue runs the full stairway of upgrades and downgrades. The nightly job restores a production-sized snapshot and times each revision. Cheap checks early, expensive ones on a schedule every push • single head • model/migration drift • data tests for changed migrations seconds merge queue • full stairway up • full stairway down • every irreversible revision marked about a minute nightly • restore a snapshot • time each revision • flag long locks tens of minutes
Each stage catches something the earlier ones cannot. The nightly job is the only place lock duration and backfill time become visible before a deploy.

Marking the slower checks keeps the split explicit in the code rather than in pipeline YAML alone — @pytest.mark.migrations_full on the stairway, selected in the merge-queue job and excluded elsewhere, means a developer running the suite locally gets the fast checks by default and can opt into the full walk when working on a migration.

Running migrations against realistic volume

Correctness on a handful of rows is necessary and not sufficient. A migration that rewrites a column on a fifty-million-row table can hold a lock for an hour, and the only safe time to learn that is before the deploy.

A nightly job that restores an anonymised production snapshot into a container and times alembic upgrade head against it catches the two failure modes that matter: a migration that takes far longer than the deploy window allows, and one that takes an exclusive lock on a hot table. Neither shows up in the per-commit suite, and neither needs to — the job runs on a schedule, reports the duration per revision, and flags anything above a threshold.

Bash
# Nightly: restore the snapshot, then time each revision individually.
pg_restore --no-owner -d "$DSN" snapshot.dump
for rev in $(alembic history -r current:head | awk '{print $3}' | tac); do
  /usr/bin/time -f "%e s  $rev" alembic upgrade "$rev"
done

The per-revision timing is the useful output. A total of twelve minutes tells you little; one revision taking eleven of them names the migration to rewrite — usually by splitting a single ALTER into an additive change, a batched backfill and a later constraint, the same expand–migrate–contract sequence used for API evolution.

Frequently Asked Questions

Why test migrations if the application tests already pass? Because application tests usually build the schema from model metadata, so the migrations never run until a deploy. Drift between models and migration history, an unapplyable revision or a broken downgrade are all invisible to a suite that never executes the migration chain.

Should every migration have a working downgrade? Every migration should either have a downgrade that has been tested or be explicitly marked irreversible with a downgrade that raises. An untested downgrade is worse than none, because it will be trusted during an incident and fail then.

How do I test a migration that transforms existing data? Stamp the database to the revision before it, insert representative rows through raw SQL that matches that older schema, run the upgrade, and assert on the transformed data. Model classes cannot be used for the setup because they describe the new schema, not the old one.

← Back to Database Fixtures & Transactional Tests