Database Migration Testing Best Practices: Catch Schema Bugs Early

07 Sep 2026 — 6 min read

A bad database migration is one of the worst production incidents you can have. Unlike a bad code deploy, you can't just roll back — the data is already mutated. A migration that drops a column, truncates a table, or corrupts data during a lock-heavy ALTER can take hours to recover from.

Most teams test their application code thoroughly. Most teams barely test their migrations at all. This guide covers best practices for making database migrations a first-class testing concern.

Why Migration Testing Fails in Practice

The standard approach is to run migrations against a dev database and see if the application still works. This fails to catch:

Migrations that work on empty dev data but break on production-sized tables — an ALTER TABLE ADD COLUMN with a default value backfills every row; on a 500M row table, that's a 45-minute table lock.
Rollback failures — most teams don't test rollbacks until they need one in production.
Migration ordering bugs — migration 48 assumes data inserted by migration 47, but on a fresh database they run in the wrong order.
Idempotency failures — running a migration twice causes errors on some tools but is required for some deployment patterns.
Data integrity issues — a migration correctly changes the schema but corrupts existing data in edge cases.

Core Best Practice 1: Test Migrations Against a Copy of Production Data

The only reliable way to know if a migration is safe is to run it against real data at real scale.

#!/bin/bash
# scripts/test-migration-against-prod-copy.sh

# 1. Restore latest production backup to test database
pg_restore \
  --host=test-db.internal \
  --dbname=testdb \
  --clean \
  --if-exists \
  /backups/prod-latest.dump

# 2. Run pending migrations
flyway -url=jdbc:postgresql://test-db.internal/testdb migrate

# 3. Verify row counts haven't changed unexpectedly
psql -h test-db.internal -d testdb -c "
  SELECT 
    schemaname,
    relname AS table_name,
    n_live_tup AS row_count
  FROM pg_stat_user_tables
  ORDER BY n_live_tup DESC
  LIMIT 20;
"

# 4. Run smoke tests
python scripts/post-migration-smoke-test.py

This script should run in CI for every migration PR, not just before deployments.

Core Best Practice 2: Test Rollbacks Explicitly

Every migration needs a corresponding rollback, and every rollback needs a test:

Flyway Undo Migrations

-- V47__add_user_preferences.sql (forward migration)
ALTER TABLE users ADD COLUMN preferences JSONB DEFAULT '{}';
CREATE INDEX idx_users_preferences ON users USING gin(preferences);

-- U47__add_user_preferences.sql (undo migration)
DROP INDEX IF EXISTS idx_users_preferences;
ALTER TABLE users DROP COLUMN IF EXISTS preferences;

Test the rollback:

def test_migration_v47_rollback(test_db):
    """Verify migration V47 can be rolled back without data loss."""
    # Apply migration
    flyway.migrate(test_db, target="47")
    
    # Insert data using new schema
    test_db.execute(
        "INSERT INTO users (id, name, preferences) VALUES (1, 'Alice', '{\"theme\": \"dark\"}'::jsonb)"
    )
    
    # Roll back
    flyway.undo(test_db, target="46")
    
    # Verify user still exists (no data loss except the preferences column)
    row = test_db.execute("SELECT id, name FROM users WHERE id = 1").fetchone()
    assert row is not None
    assert row["name"] == "Alice"
    
    # Verify column is gone
    columns = test_db.execute(
        "SELECT column_name FROM information_schema.columns WHERE table_name = 'users'"
    ).fetchall()
    assert "preferences" not in [c["column_name"] for c in columns]

Alembic Downgrade Testing

# tests/test_migrations.py
import pytest
from alembic import command
from alembic.config import Config
from sqlalchemy import create_engine, text

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

def test_upgrade_downgrade_roundtrip(alembic_cfg, test_db_engine):
    """Each migration can be upgraded and downgraded cleanly."""
    # Get list of migration revisions
    from alembic.script import ScriptDirectory
    scripts = ScriptDirectory.from_config(alembic_cfg)
    revisions = list(scripts.walk_revisions())
    
    for revision in reversed(revisions):
        # Upgrade to this revision
        command.upgrade(alembic_cfg, revision.revision)
        
        # Downgrade back
        if revision.down_revision:
            command.downgrade(alembic_cfg, revision.down_revision)
            
            # Re-upgrade (idempotency check)
            command.upgrade(alembic_cfg, revision.revision)
    
    # Final state should match head
    command.upgrade(alembic_cfg, "head")

Core Best Practice 3: Test Migration Idempotency

A migration that fails halfway through and must be re-run should not leave the database in an inconsistent state:

def test_migration_idempotent(alembic_cfg):
    """Running the same migration twice doesn't fail or corrupt data."""
    # First run
    command.upgrade(alembic_cfg, "head")
    
    # Insert test data
    with engine.connect() as conn:
        conn.execute(text("INSERT INTO users (name) VALUES ('Alice')"))
        conn.commit()
    
    # Second run (simulate re-run after partial failure)
    # Should not raise, should not duplicate data
    command.upgrade(alembic_cfg, "head")
    
    with engine.connect() as conn:
        count = conn.execute(text("SELECT COUNT(*) FROM users")).scalar()
    
    assert count == 1, f"Idempotent re-run created duplicate data: {count} rows"

Core Best Practice 4: Test Data Integrity After Migration

Schema changes often transform data. Test that the transformation is correct:

def test_migration_v52_splits_fullname_column(test_db):
    """Migration V52 correctly splits full_name into first_name/last_name."""
    # Setup: create users with the old schema
    test_db.execute("""
        INSERT INTO users (id, full_name) VALUES
        (1, 'Alice Smith'),
        (2, 'Bob Jones'),
        (3, 'Carol'),          -- no last name
        (4, 'Dave van de Berg') -- compound last name
    """)
    
    # Run migration
    flyway.migrate(test_db, target="52")
    
    # Verify data transformation
    rows = test_db.execute(
        "SELECT id, first_name, last_name FROM users ORDER BY id"
    ).fetchall()
    
    assert rows[0] == (1, "Alice", "Smith")
    assert rows[1] == (2, "Bob", "Jones")
    assert rows[2] == (3, "Carol", None)      # NULL last name for single-name users
    assert rows[3] == (4, "Dave", "van de Berg")  # compound last name preserved
    
    # Verify row count preserved (no rows lost)
    count = test_db.execute("SELECT COUNT(*) FROM users").scalar()
    assert count == 4

Core Best Practice 5: Lock and Performance Testing

Migrations on large tables can lock the entire table. Test for this before going to production:

def test_migration_does_not_lock_table(test_db_with_large_dataset):
    """Migration V55 uses ADD COLUMN ... NOT NULL DEFAULT to avoid full table lock."""
    # Start a long-running read in background
    import threading
    read_results = []
    read_errors = []
    
    def background_reader():
        try:
            with test_db_with_large_dataset.connect() as conn:
                # This should not be blocked by the migration
                result = conn.execute(text(
                    "SELECT COUNT(*) FROM orders"
                )).scalar()
                read_results.append(result)
        except Exception as e:
            read_errors.append(str(e))
    
    reader_thread = threading.Thread(target=background_reader)
    reader_thread.start()
    
    # Run migration
    start_time = time.monotonic()
    flyway.migrate(test_db_with_large_dataset, target="55")
    migration_time = time.monotonic() - start_time
    
    reader_thread.join(timeout=5.0)
    
    assert not read_errors, f"Concurrent reads failed: {read_errors}"
    assert read_results, "No read results received — possible lock"
    assert migration_time < 10.0, \
        f"Migration took {migration_time:.1f}s — may cause production timeout"

Core Best Practice 6: Test Migration Ordering

Migrations must work both on fresh databases and on existing databases with data:

@pytest.mark.parametrize("starting_version", ["empty", "v1", "v25", "v48"])
def test_migration_works_from_any_baseline(alembic_cfg, test_db_factory, starting_version):
    """Migration chain works regardless of starting schema version."""
    db = test_db_factory(starting_version)
    
    # Apply all pending migrations
    command.upgrade(alembic_cfg, "head")
    
    # Application schema validation
    inspector = inspect(db)
    required_tables = ["users", "orders", "products", "sessions"]
    existing_tables = inspector.get_table_names()
    
    for table in required_tables:
        assert table in existing_tables, \
            f"Table '{table}' missing after migration from {starting_version}"

CI/CD Integration

Migrations should be tested in CI before any code that depends on them is merged:

# .github/workflows/migration-tests.yml
name: Migration Tests

on:
  pull_request:
    paths:
      - 'migrations/**'
      - 'alembic/versions/**'
      - 'flyway/sql/**'

jobs:
  migration-tests:
    runs-on: ubuntu-latest

    services:
      postgres:
        image: postgres:16-alpine
        env:
          POSTGRES_DB: testdb
          POSTGRES_USER: test
          POSTGRES_PASSWORD: test
        ports:
          - 5432:5432
        options: --health-cmd pg_isready --health-interval 10s

    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 pytest alembic sqlalchemy psycopg2-binary

      - name: Test upgrade from empty
        run: |
          alembic upgrade head
        env:
          DATABASE_URL: postgresql://test:test@localhost/testdb

      - name: Test downgrade roundtrip
        run: pytest tests/test_migrations.py -v
        env:
          DATABASE_URL: postgresql://test:test@localhost/testdb

      - name: Test data integrity
        run: pytest tests/test_migration_data.py -v
        env:
          DATABASE_URL: postgresql://test:test@localhost/testdb

Migration Testing Checklist

For every migration PR:

Forward migration works on empty database
Forward migration works on database with existing data
Rollback/downgrade works without data loss
Idempotent run doesn't fail or duplicate data
Data transformation is tested if migration transforms existing data
No table-level locks on large tables (verify with pg_locks monitoring)
Performance tested on production-sized data (or representative sample)
Row counts validated before and after
Application smoke tests pass after migration

Common Migration Bugs to Catch

Dropping a column still referenced in code: The migration works, but the application crashes. Run application tests after migration tests.

NOT NULL column without default: ALTER TABLE ADD COLUMN foo INTEGER NOT NULL fails if any existing rows can't get a value. Add a DEFAULT or backfill first.

Missing index on foreign key: Adding an FK constraint without a supporting index can slow queries by orders of magnitude. Validate index creation is part of the migration.

Migration version conflicts: Two branches create V47_* migrations. Most tools fail noisily here, but test for it explicitly in CI.

Assumes empty table: A migration that does UPDATE users SET role = 'user' assumes all existing users should get 'user' role. Test with data that has edge cases.

Database migrations are permanent changes to your most critical data. The testing patterns here — rollback testing, data integrity verification, idempotency checks, and production-scale validation — turn migrations from high-risk deployments into just another pull request. The investment in migration tests pays for itself the first time they prevent a production incident that would have required a multi-hour restore from backup.