Testing Migrations — Complete Guide
In this tutorial, you will learn about Testing Migrations. We cover key concepts, practical examples, and best practices to help you master this topic.
Learn testing strategies for database migrations: validate up and down migrations, test against production-sized data, verify data integrity, automate migration tests in CI/CD, and ensure rollbacks work correctly before production deployment.
What You Learn
You will learn how to test database migrations effectively: writing unit tests for migration functions, testing against a fresh database in CI, verifying data integrity before and after migrations, testing rollback scenarios, Performance Testing migrations on production-scale data, and automating migration tests in deployment pipelines.
Why It Matters
Untested migrations cause production incidents. A migration that drops a column used by the application crashes the site. A migration that locks a large table causes downtime. A migration with a bug in the down function makes rollback impossible. Testing catches these issues before they reach production.
Real-World Use
DodaTech's migration test suite caught a migration that would have dropped a foreign key constraint without removing dependent data. The test detected the constraint violation during CI. The migration was fixed before deployment, preventing a production incident that would have affected 50,000 orders.
Unit Testing Migrations
# tests/test_migrations.py
"""Unit tests for migration functions."""
import pytest
from sqlalchemy import text
from migrations.versions.abc123_add_phone import upgrade, downgrade
def test_up_migration_creates_column(test_database):
"""Verify up migration adds the phone column."""
upgrade()
result = test_database.execute(
text("SELECT column_name FROM information_schema.columns "
"WHERE table_name = 'users' AND column_name = 'phone'")
).fetchone()
assert result is not None, "phone column should exist after upgrade"
def test_down_migration_drops_column(test_database):
"""Verify down migration removes the phone column."""
upgrade()
downgrade()
result = test_database.execute(
text("SELECT column_name FROM information_schema.columns "
"WHERE table_name = 'users' AND column_name = 'phone'")
).fetchone()
assert result is None, "phone column should not exist after downgrade"
def test_up_down_cycle_preserves_data(test_database):
"""Verify data survives an up-down-up cycle."""
test_database.execute(
text("INSERT INTO users (email, name) VALUES ('test@test.com', 'Test')")
)
original_count = test_database.execute(
text("SELECT COUNT(*) FROM users")
).scalar()
upgrade()
downgrade()
upgrade()
final_count = test_database.execute(
text("SELECT COUNT(*) FROM users")
).scalar()
assert original_count == final_count, "Data should survive migration cycle"
Expected output: Unit tests verify the migration creates the expected schema changes, reverts them correctly, and preserves data through migration cycles. Tests run against a fresh test database in CI.
Integration Testing with Fresh Database
# .github/workflows/migration-tests.yml
name: Migration Tests
on:
pull_request:
paths:
- 'migrations/**'
- '**/models.py'
- '**/schema.prisma'
jobs:
test-migrations:
runs-on: ubuntu-latest
services:
postgres:
image: postgres:16
env:
POSTGRES_PASSWORD: postgres
options: >-
--health-cmd pg_isready
--health-interval 10s
--health-timeout 5s
--health-retries 5
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: Create test database
run: createdb -h localhost -U postgres test_migrations
- name: Apply all migrations
run: alembic upgrade head
env:
DATABASE_URL: postgresql://postgres:postgres@localhost/test_migrations
- name: Verify schema matches models
run: python -c "
from app import db
db.engine.connect()
print('Models loaded successfully against migrated schema')
"
- name: Test application startup
run: |
python -m app.health_check
echo 'Application starts with migrated schema'
- name: Test rollback
run: |
alembic downgrade -1
alembic upgrade head
echo 'Rollback and re-apply works'
- name: Run application test suite
run: pytest tests/
env:
DATABASE_URL: postgresql://postgres:postgres@localhost/test_migrations
Expected output: CI pipeline creates a fresh database, applies all migrations, verifies schema matches models, tests application startup, tests rollback, and runs the full test suite against the migrated database.
Data Integrity Testing
# tests/test_data_integrity.py
"""Verify data integrity before and after migrations."""
import pytest
def test_no_data_loss_after_migration(test_database):
"""Verify row counts match before and after migration cycle."""
tables = ['users', 'orders', 'products']
for table in tables:
before = test_database.execute(
text(f"SELECT COUNT(*) FROM {table}")
).scalar()
upgrade()
downgrade()
after = test_database.execute(
text(f"SELECT COUNT(*) FROM {table}")
).scalar()
assert before == after, (
f"Data loss in {table}: {before} rows before, {after} after"
)
def test_foreign_key_integrity(test_database):
"""Verify foreign key constraints are valid after migration."""
result = test_database.execute(text("""
SELECT
COUNT(*) AS total_orders,
COUNT(DISTINCT o.user_id) AS distinct_users
FROM orders o
LEFT JOIN users u ON o.user_id = u.id
WHERE u.id IS NULL
""")).fetchone()
assert result.orphan_count == 0, (
f"Found {result.orphan_count} orders with missing users"
)
def test_not_null_constraints(test_database):
"""Verify NOT NULL columns have no NULL values."""
result = test_database.execute(text("""
SELECT
SUM(CASE WHEN email IS NULL THEN 1 ELSE 0 END) AS null_emails
FROM users
""")).fetchone()
assert result.null_emails == 0, (
f"Found {result.null_emails} users with NULL email"
)
Expected output: Data integrity tests verify row counts are preserved, foreign key relationships remain valid, and NOT NULL constraints are enforced. These tests catch data corruption caused by migrations.
Performance Testing Migrations
# tests/test_migration_performance.py
"""Benchmark migration execution time."""
import time
import pytest
def test_migration_execution_time(test_database, benchmark_data):
"""Verify migration completes within acceptable time."""
# benchmark_data contains production-scale data
start = time.time()
upgrade()
duration = time.time() - start
max_duration = 30 # seconds
assert duration < max_duration, (
f"Migration took {duration:.2f}s, exceeds {max_duration}s limit"
)
def test_migration_lock_contention(test_database, benchmark_data):
"""Verify migration does not lock tables excessively."""
# Simulate concurrent queries during migration
import threading
lock_detected = threading.Event()
def concurrent_query():
try:
test_database.execute(text("SELECT COUNT(*) FROM users"))
except Exception:
lock_detected.set()
thread = threading.Thread(target=concurrent_query)
thread.start()
upgrade()
thread.join()
assert not lock_detected.is_set(), (
"Migration caused lock contention"
)
def test_rollback_performance(test_database, benchmark_data):
"""Verify rollback completes quickly."""
upgrade()
start = time.time()
downgrade()
duration = time.time() - start
max_rollback_duration = 10 # seconds
assert duration < max_rollback_duration, (
f"Rollback took {duration:.2f}s, exceeds {max_rollback_duration}s limit"
)
Expected output: Performance tests measure migration and rollback execution time against production-scale data. They verify migrations complete within the deployment window and do not cause excessive lock contention.
Testing Data Migrations
# tests/test_data_migration.py
"""Test data transformation correctness."""
import pytest
def test_backfill_correctness(test_database):
"""Verify backfill produces correct values."""
test_database.execute(text("""
INSERT INTO users (email, first_name, last_name, full_name)
VALUES
('alice@test.com', 'Alice', 'Smith', NULL),
('bob@test.com', 'Bob', 'Jones', NULL),
('carol@test.com', 'Carol', 'Williams', NULL)
"""))
run_data_migration()
results = test_database.execute(text("""
SELECT email, full_name FROM users ORDER BY email
""")).fetchall()
assert results[0].full_name == 'Alice Smith'
assert results[1].full_name == 'Bob Jones'
assert results[2].full_name == 'Carol Williams'
def test_dry_run_does_not_modify_data(test_database):
"""Verify dry run mode does not change data."""
original = test_database.execute(
text("SELECT full_name FROM users WHERE email = 'alice@test.com'")
).scalar()
run_data_migration_dry_run()
after_dry_run = test_database.execute(
text("SELECT full_name FROM users WHERE email = 'alice@test.com'")
).scalar()
assert original == after_dry_run, "Dry run should not modify data"
def test_data_migration_idempotency(test_database):
"""Verify running data migration twice is safe."""
run_data_migration()
first_run = test_database.execute(
text("SELECT full_name FROM users ORDER BY email")
).fetchall()
run_data_migration()
second_run = test_database.execute(
text("SELECT full_name FROM users ORDER BY email")
).fetchall()
assert first_run == second_run, "Data migration should be idempotent"
Expected output: Data migration tests verify correct transformation, dry run does not modify data, and running the migration multiple times is safe (idempotent). These tests prevent data corruption.
Test Fixtures Setup
# conftest.py
"""Test fixtures for migration testing."""
import pytest
from sqlalchemy import create_engine, text
@pytest.fixture
def test_database():
"""Create a fresh test database for each test."""
db_url = "postgresql://postgres:postgres@localhost/test_migrations"
engine = create_engine(db_url)
# Apply base schema (pre-migration state)
with engine.begin() as conn:
conn.execute(text("""
CREATE TABLE users (
id SERIAL PRIMARY KEY,
email VARCHAR(255) NOT NULL UNIQUE,
name VARCHAR(200),
created_at TIMESTAMP DEFAULT NOW()
)
"""))
yield engine
# Cleanup
with engine.begin() as conn:
conn.execute(text("DROP TABLE IF EXISTS users CASCADE"))
conn.execute(text("DROP TABLE IF EXISTS orders CASCADE"))
@pytest.fixture
def benchmark_data(test_database):
"""Insert production-scale test data."""
with test_database.begin() as conn:
for i in range(100000):
conn.execute(text(
"INSERT INTO users (email, name) VALUES (:email, :name)"
), {'email': f'user{i}@test.com', 'name': f'User {i}'})
return test_database
Expected output: Test fixtures create a fresh database with pre-migration schema for each test. The benchmark fixture inserts 100,000 rows to simulate production-scale data. Each test runs in isolation.
Common Mistakes
1. Testing Only Up Migrations
Testing only the up migration misses rollback issues. Always test both up and down. Verify the down migration restores the exact previous schema. Test the up-down-up cycle.
2. Testing Against Empty Databases
Migrations that work on empty databases may fail with real data. Test against databases with production-scale data. Include edge cases: large tables, NULL values, duplicate records, and constraint violations.
3. Not Testing Migration Ordering
Migrations applied in the wrong order cause failures. Test that the full migration chain applies cleanly from scratch. Verify that migration dependencies are correct.
4. Skipping Data Migration Tests
Data migrations transform existing records. Test that transformations produce correct values. Test edge cases: NULL fields, empty strings, special characters, and very long values.
5. Not Testing Rollback Performance
A rollback that works on small data may time out on production-scale data. Performance test rollbacks. Verify they complete within the deployment window.
6. Ignoring Migration Interactions
Migrations can interact: one migration adds a column, another uses it. Test the full migration chain, not individual migrations in isolation. Verify schema consistency after all migrations.
Practice Questions
1. What should you test in a migration?
Up migration creates correct schema changes. Down migration reverts them exactly. Data integrity is preserved (row counts, foreign keys, constraints). Performance meets time limits. Rollback completes within acceptable duration.
2. Why test against production-scale data?
Real databases have millions of rows, large indexes, and complex relationships. Migrations that work on small test databases may time out, lock tables, or exhaust memory on production-scale data.
3. What is the up-down-up cycle test?
Apply migration (up), revert it (down), apply it again (up). The schema should be the same after the cycle. Row counts should match. This verifies the migration is fully reversible and idempotent.
4. How do you automate migration tests in CI/CD?
Create a fresh database in CI, apply all migrations, verify schema against models, test application startup, run application test suite, test rollback, and report results. Fail the pipeline if any migration test fails.
Challenge
Build a comprehensive migration test suite that: unit tests individual migrations (up, down, up-down-up), integration tests the full migration chain from scratch, data integrity tests (row counts, constraints, relationships), performance tests against production-scale data, rollback tests with timing verification, and CI/CD integration with fresh database per run.
FAQ
Mini Project: Migration Test Framework
Build a migration test framework that: creates a fresh test database per test run, applies base schema for pre-migration state, runs up and down migrations, verifies schema using information_schema queries, validates data integrity (row counts, constraints, relationships), measures execution time with performance assertions, and integrates with CI/CD pipeline for automated testing on every PR.
What's Next
Now that you understand testing migrations, learn about CI/CD for Migrations for automating migration execution and verification in deployment pipelines.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro