Testing Alembic Migrations: Verify Before Deploying
A broken migration deployed to production can bring down your entire service. Testing migrations before deployment is non-negotiable. Your tests should verify that (1) migrations apply without errors, (2) the schema changes are correct, (3) data is preserved or transformed properly, and (4) rollbacks work. Alembic integrates cleanly with pytest, allowing you to test on an isolated database.
Setting Up a Test Database
Use an in-memory SQLite database for fast, isolated tests:
# tests/conftest.py
import pytest
from sqlalchemy import create_engine
from alembic.config import Config
from alembic.script import ScriptDirectory
from alembic.runtime.migration import MigrationContext
from alembic.operations import Operations
from app.models import Base
@pytest.fixture(scope='function')
def test_db():
"""Create a fresh in-memory database for each test."""
engine = create_engine('sqlite:///:memory:')
# Create all tables from models (no migrations)
Base.metadata.create_all(engine)
yield engine
# Cleanup
Base.metadata.drop_all(engine)
For more isolation, use a fresh file-based SQLite database:
@pytest.fixture(scope='function')
def test_db(tmp_path):
"""Create a fresh test database in a temp file."""
db_path = tmp_path / 'test.db'
engine = create_engine(f'sqlite:///{db_path}')
Base.metadata.create_all(engine)
yield engine
Base.metadata.drop_all(engine)
Testing a Single Migration
Test that a specific migration applies and produces the expected schema:
# tests/test_migrations.py
import pytest
from sqlalchemy import Column, Integer, String, inspect, text
from alembic.config import Config
from alembic.runtime.migration import MigrationContext
from alembic.operations import Operations
def test_add_phone_column_migration(test_db):
"""Verify the 'add phone column' migration works."""
# Get the current schema before migration
inspector = inspect(test_db)
columns_before = {col['name'] for col in inspector.get_columns('users')}
assert 'phone' not in columns_before, "phone column already exists (migration may be applied)"
# Apply the migration
config = Config('alembic.ini')
config.set_main_option('sqlalchemy.url', str(test_db.url))
ctx = MigrationContext.configure(test_db.connect())
migration = Operations(ctx)
# Run the upgrade function from your migration file
from migrations.versions import _004_add_phone_column
_004_add_phone_column.upgrade()
# Verify the column was added
inspector = inspect(test_db)
columns_after = {col['name'] for col in inspector.get_columns('users')}
assert 'phone' in columns_after, "phone column was not added"
# Verify column type
phone_col = next(col for col in inspector.get_columns('users') if col['name'] == 'phone')
assert phone_col['type'].__class__.__name__ == 'String', "phone column has wrong type"
assert phone_col['nullable'] == True, "phone column should be nullable"
This approach is manual but gives you full control. For larger suites, use Alembic's upgrade command:
def test_add_phone_column_with_upgrade(test_db):
"""Test migration using alembic upgrade."""
config = Config('alembic.ini')
config.set_main_option('sqlalchemy.url', str(test_db.url))
from alembic.runtime.migration import MigrationContext
from alembic.operations import Operations
# Start from base (no migrations applied)
ctx = MigrationContext.configure(test_db.connect())
ops = Operations(ctx)
# Apply a specific migration by calling its upgrade() function
import importlib
migration_module = importlib.import_module('migrations.versions._004_add_phone_column')
migration_module.upgrade()
# Verify result
inspector = inspect(test_db)
columns = {col['name'] for col in inspector.get_columns('users')}
assert 'phone' in columns
Testing Upgrade and Downgrade Cycle
Verify that your migration can be applied and rolled back:
def test_migration_roundtrip(test_db):
"""Apply migration, verify, rollback, and verify again."""
inspector = inspect(test_db)
# Before: verify phone doesn't exist
columns_before = {col['name'] for col in inspector.get_columns('users')}
assert 'phone' not in columns_before
# Apply
from migrations.versions import _004_add_phone_column
_004_add_phone_column.upgrade()
# After upgrade: verify phone exists
inspector = inspect(test_db)
columns_upgraded = {col['name'] for col in inspector.get_columns('users')}
assert 'phone' in columns_upgraded
# Downgrade
_004_add_phone_column.downgrade()
# After downgrade: verify phone is gone
inspector = inspect(test_db)
columns_downgraded = {col['name'] for col in inspector.get_columns('users')}
assert 'phone' not in columns_downgraded
Testing Data Migrations
For migrations that transform data, verify both the schema change and the data integrity:
def test_split_name_migration(test_db):
"""Verify name column is split into first_name and last_name with data preserved."""
from sqlalchemy.orm import Session
from app.models import User
# Insert test data before migration
with Session(test_db) as session:
user = User(username='alice', email='[email protected]', name='Alice Johnson')
session.add(user)
session.commit()
# Apply migration
from migrations.versions import _005_split_name_column
_005_split_name_column.upgrade()
# Verify schema: name is gone, first_name and last_name exist
inspector = inspect(test_db)
columns = {col['name'] for col in inspector.get_columns('users')}
assert 'name' not in columns
assert 'first_name' in columns
assert 'last_name' in columns
# Verify data: name was split correctly
with test_db.connect() as conn:
result = conn.execute(text(
"SELECT first_name, last_name FROM users WHERE username = 'alice'"
)).fetchone()
assert result[0] == 'Alice', f"Expected first_name='Alice', got '{result[0]}'"
assert result[1] == 'Johnson', f"Expected last_name='Johnson', got '{result[1]}'"
Parametrized Tests for Multiple Migrations
Test all migrations in sequence:
import pytest
from pathlib import Path
from alembic.script import ScriptDirectory
def get_all_migrations():
"""Collect all migration files."""
script_dir = ScriptDirectory('alembic')
migrations = []
for revision in script_dir.walk_revisions('base', 'head'):
migrations.append(revision)
return migrations
@pytest.mark.parametrize('migration', get_all_migrations())
def test_all_migrations_apply(test_db, migration):
"""Test that every migration can be applied."""
config = Config('alembic.ini')
config.set_main_option('sqlalchemy.url', str(test_db.url))
from alembic.command import upgrade as alembic_upgrade
# Apply up to this migration
alembic_upgrade(config, migration.revision)
# Verify alembic_version table has the right revision
inspector = inspect(test_db)
with test_db.connect() as conn:
result = conn.execute(text(
"SELECT version_num FROM alembic_version"
)).scalar()
assert result == migration.revision, f"Migration {migration} not applied"
Testing in CI/CD Pipelines
In your CI configuration (GitHub Actions, GitLab CI, etc.), test migrations on multiple database types:
# .github/workflows/test-migrations.yml
name: Test Migrations
on: [push, pull_request]
jobs:
test:
runs-on: ubuntu-latest
services:
postgres:
image: postgres:15
env:
POSTGRES_USER: testuser
POSTGRES_PASSWORD: testpass
POSTGRES_DB: testdb
options: >-
--health-cmd pg_isready
--health-interval 10s
--health-timeout 5s
--health-retries 5
ports:
- 5432:5432
steps:
- uses: actions/checkout@v3
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.11'
- name: Install dependencies
run: |
pip install -e .
pip install pytest
- name: Test migrations on PostgreSQL
env:
DATABASE_URL: postgresql://testuser:testpass@localhost:5432/testdb
run: |
alembic upgrade head
pytest tests/test_migrations.py -v
- name: Test migrations on SQLite
run: |
export DATABASE_URL=sqlite:///test.db
alembic upgrade head
pytest tests/test_migrations.py -v
Common Migration Test Failures and Fixes
| Error | Cause | Fix |
|---|---|---|
| "table X already exists" | Migration applies twice in test | Reset database between tests; use a fresh fixture |
| "column X does not exist" | Downgrade didn't remove column | Fix downgrade() function in migration; verify reversibility |
| "null value in column X violates not-null constraint" | Adding non-nullable column to existing data | Make column nullable first, populate, then constrain |
| "foreign key constraint failed" | Migration drops table before removing foreign keys | Reorder: drop foreign keys first, then drop table |
| "no such table" | Migration references table that doesn't exist | Check migration order; verify parent migrations run first |
Verifying Schema Matches Models
After migrations, verify that the actual schema matches your SQLAlchemy models:
def test_schema_matches_models(test_db):
"""Verify migrated schema matches model definitions."""
from sqlalchemy import inspect
from app.models import User, Post
# Apply all migrations
config = Config('alembic.ini')
config.set_main_option('sqlalchemy.url', str(test_db.url))
from alembic.command import upgrade as alembic_upgrade
alembic_upgrade(config, 'head')
# Verify User model columns match database
inspector = inspect(test_db)
db_columns = {col['name'] for col in inspector.get_columns('users')}
model_columns = {col.name for col in User.__table__.columns}
assert db_columns == model_columns, (
f"Schema mismatch for users: "
f"DB has {db_columns}, model expects {model_columns}"
)
Key Takeaways
- Use pytest fixtures to create isolated test databases (in-memory SQLite or temporary files)
- Test each migration's
upgrade()anddowngrade()functions separately - For data migrations, verify both schema changes and data integrity
- Use parametrized tests to test all migrations apply successfully
- Test migrations in CI/CD on multiple databases (PostgreSQL, MySQL, SQLite)
- Verify the migrated schema matches your SQLAlchemy model definitions
- Common failures: duplicate tables, missing downgrade logic, nullable constraint violations
Frequently Asked Questions
How do I test a migration that depends on another migration?
Alembic automatically applies parent migrations first. When you call a migration's upgrade() function, you're assuming its parent has already been applied. In tests, either (1) manually call the parent's upgrade() first, or (2) use alembic upgrade <specific_revision> to apply a range of migrations in order.
Can I test migrations in-place on a real database?
Yes, but create a separate test database (not production). Use a database URL like postgresql://testuser:testpass@localhost/test_db. Always test on a copy, never on production.
What if my migration has complex SQL that's hard to test?
Break it into pieces. In your migration, use op.execute() for complex SQL; in tests, run the SQL directly and verify results. Or refactor complex SQL into a stored procedure and call it from the migration.
How do I test migrations for a new feature branch?
Create a fresh test database, apply migrations up to the base, then apply your new feature migration. Verify the schema change and any data transformations work correctly before merging to main.