Database Testing

Testing Database Migrations: Flyway, Liquibase, Alembic, and Prisma Migrate

HelpMeTest

16 May 2026 — 6 min read

Database migrations are the most dangerous deployments you'll make. A bad migration can corrupt data, lock tables, or bring down production — and rollbacks are often impossible once users interact with the new schema. This guide covers how to test migrations thoroughly across the four most popular migration tools.

Why Migration Testing Is Different

Application code bugs are usually recoverable — deploy a fix. Migration bugs are different:

Data loss: DROP COLUMN or TRUNCATE can't be undone after the fact
Lock escalation: ALTER TABLE locks entire tables in PostgreSQL without CONCURRENTLY
Irreversible transforms: Data normalization that modifies existing records
Long-running ops: Index creation on large tables can block queries for minutes

Testing migrations means verifying three things:

Forward migration applies cleanly on the current schema
Rollback works (if your tool supports it) and returns to a known state
Data integrity is preserved — existing rows survive the migration

Testing Flyway Migrations

Flyway uses versioned SQL files (V1__create_users.sql, V2__add_email_index.sql). Tests should run in a fresh schema each time.

Project Structure

src/
  main/resources/db/migration/
    V1__create_schema.sql
    V2__add_users.sql
    V3__add_email_index.sql
test/
  java/com/example/
    MigrationTest.java
  resources/
    application-test.properties

Spring Boot Integration

// application-test.properties
spring.datasource.url=jdbc:tc:postgresql:16:///testdb
spring.flyway.enabled=true
spring.flyway.clean-disabled=false  # Allow clean in tests

// MigrationTest.java
@SpringBootTest
@Testcontainers
class MigrationTest {
    
    @Container
    static PostgreSQLContainer<?> postgres = new PostgreSQLContainer<>("postgres:16-alpine");
    
    @Autowired
    Flyway flyway;
    
    @Autowired
    JdbcTemplate jdbc;
    
    @Test
    void migrationsApplyCleanly() {
        MigrationInfoService info = flyway.info();
        MigrationInfo[] applied = info.applied();
        
        assertThat(applied).isNotEmpty();
        assertThat(Arrays.stream(applied).allMatch(m -> 
            m.getState() == MigrationState.SUCCESS
        )).isTrue();
    }
    
    @Test
    void schemaHasExpectedTables() {
        List<String> tables = jdbc.queryForList(
            "SELECT tablename FROM pg_tables WHERE schemaname = 'public'",
            String.class
        );
        
        assertThat(tables).containsExactlyInAnyOrder(
            "users", "orders", "products", "flyway_schema_history"
        );
    }
    
    @Test
    void usersTableHasCorrectColumns() {
        var columns = jdbc.queryForList("""
            SELECT column_name, data_type, is_nullable
            FROM information_schema.columns
            WHERE table_name = 'users'
            ORDER BY ordinal_position
        """);
        
        assertThat(columns).extracting("column_name")
            .contains("id", "email", "name", "created_at");
        
        // Email must not be nullable
        var emailColumn = columns.stream()
            .filter(c -> c.get("column_name").equals("email"))
            .findFirst().orElseThrow();
        assertThat(emailColumn.get("is_nullable")).isEqualTo("NO");
    }
    
    @Test  
    void rollbackToV2AndReapply() {
        // Test idempotency: clean + migrate should always succeed
        flyway.clean();
        flyway.migrate();
        
        // Verify count after re-migration
        int schemaHistoryCount = jdbc.queryForObject(
            "SELECT COUNT(*) FROM flyway_schema_history", Integer.class
        );
        assertThat(schemaHistoryCount).isGreaterThan(0);
    }
}

Flyway Migration Guard Tests

Test that migrations don't lose data:

@Test
void v3MigrationPreservesExistingData() {
    // Migrate to V2 only
    Flyway flywayV2 = Flyway.configure()
        .dataSource(dataSource)
        .target("2")
        .load();
    flywayV2.clean();
    flywayV2.migrate();
    
    // Insert data at V2 state
    jdbc.execute("""
        INSERT INTO users (email, name) 
        VALUES ('preserve@test.com', 'Preserved User')
    """);
    
    // Apply V3
    Flyway flywayV3 = Flyway.configure()
        .dataSource(dataSource)
        .load();
    flywayV3.migrate();
    
    // Verify data survived migration
    Integer count = jdbc.queryForObject(
        "SELECT COUNT(*) FROM users WHERE email = 'preserve@test.com'",
        Integer.class
    );
    assertThat(count).isEqualTo(1);
}

Testing Liquibase Migrations

Liquibase uses changelog files (XML, YAML, JSON, or SQL) with rollback support built in.

Changelog Example (YAML)

# db/changelog/db.changelog-003.yaml
databaseChangeLog:
  - changeSet:
      id: 003-add-phone-column
      author: dev-team
      changes:
        - addColumn:
            tableName: users
            columns:
              - column:
                  name: phone
                  type: varchar(20)
                  constraints:
                    nullable: true
      rollback:
        - dropColumn:
            tableName: users
            columnName: phone

JUnit 5 Tests

@ExtendWith(LiquibaseExtension.class)
class LiquibaseMigrationTest {
    
    @RegisterExtension
    static PostgreSQLContainerExtension postgres = new PostgreSQLContainerExtension();
    
    @Test
    void allChangeSetsApplySuccessfully() throws Exception {
        Database database = DatabaseFactory.getInstance()
            .findCorrectDatabaseImplementation(
                new JdbcConnection(dataSource.getConnection())
            );
        
        Liquibase liquibase = new Liquibase(
            "db/changelog/db.changelog-master.yaml",
            new ClassLoaderResourceAccessor(),
            database
        );
        
        // Apply all changes
        liquibase.update(new Contexts(), new LabelExpression());
        
        // Verify
        assertTableExists("users");
        assertTableExists("orders");
        assertColumnExists("users", "phone");
    }
    
    @Test
    void rollbackWorksCorrectly() throws Exception {
        Liquibase liquibase = createLiquibase();
        liquibase.update(new Contexts(), new LabelExpression());
        
        // Rollback the last changeset
        liquibase.rollback(1, new Contexts(), new LabelExpression());
        
        // phone column should be gone
        assertColumnNotExists("users", "phone");
        
        // But earlier schema should still exist
        assertTableExists("users");
        assertColumnExists("users", "email");
    }
    
    @Test
    void generateRollbackScript() throws Exception {
        Liquibase liquibase = createLiquibase();
        
        // Preview rollback SQL without executing
        StringWriter writer = new StringWriter();
        liquibase.rollback(1, new Contexts(), new LabelExpression(), writer);
        
        String rollbackSql = writer.toString();
        assertThat(rollbackSql).containsIgnoringCase("DROP COLUMN phone");
    }
}

Testing Alembic Migrations (Python)

Alembic is the migration tool for SQLAlchemy. Its --sql mode generates SQL for preview, and downgrade handles rollbacks.

Migration File Structure

# alembic/versions/003_add_phone_to_users.py
"""add phone to users

Revision ID: a1b2c3d4e5f6
Revises: 9z8y7x6w5v4u
Create Date: 2024-01-15 10:30:00
"""
from alembic import op
import sqlalchemy as sa

revision = 'a1b2c3d4e5f6'
down_revision = '9z8y7x6w5v4u'

def upgrade():
    op.add_column('users', sa.Column('phone', sa.String(20), nullable=True))

def downgrade():
    op.drop_column('users', 'phone')

pytest-based Migration Tests

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

@pytest.fixture(scope="session")
def pg_container():
    with PostgresContainer("postgres:16-alpine") as container:
        yield container

@pytest.fixture
def db_url(pg_container):
    return pg_container.get_connection_url()

@pytest.fixture
def alembic_config(db_url):
    config = Config("alembic.ini")
    config.set_main_option("sqlalchemy.url", db_url)
    return config

def test_migrations_apply_cleanly(alembic_config, db_url):
    command.upgrade(alembic_config, "head")
    
    engine = create_engine(db_url)
    with engine.connect() as conn:
        result = conn.execute(text(
            "SELECT version_num FROM alembic_version"
        ))
        version = result.scalar()
    
    # Should be at the head revision
    script = ScriptDirectory.from_config(alembic_config)
    head = script.get_current_head()
    assert version == head

def test_all_revisions_have_downgrade(alembic_config):
    """Ensure every migration has a rollback path"""
    script = ScriptDirectory.from_config(alembic_config)
    
    for rev in script.walk_revisions():
        # Check that downgrade function exists and isn't just `pass`
        module = rev.module
        downgrade_source = inspect.getsource(module.downgrade)
        assert 'pass' not in downgrade_source.strip().split('\n')[-1], \
            f"Revision {rev.revision} has empty downgrade"

def test_roundtrip_migration(alembic_config, db_url):
    """Upgrade to head, downgrade to base, upgrade again — should be idempotent"""
    command.upgrade(alembic_config, "head")
    command.downgrade(alembic_config, "base")
    command.upgrade(alembic_config, "head")
    
    engine = create_engine(db_url)
    inspector = inspect(engine)
    tables = inspector.get_table_names()
    
    assert "users" in tables
    assert "orders" in tables

def test_data_preservation_across_migration(alembic_config, db_url):
    """Migrate to previous version, insert data, then apply next migration"""
    # Migrate to revision before the phone column
    command.upgrade(alembic_config, "9z8y7x6w5v4u")
    
    engine = create_engine(db_url)
    with engine.connect() as conn:
        conn.execute(text(
            "INSERT INTO users (email, name) VALUES ('test@example.com', 'Test')"
        ))
        conn.commit()
    
    # Apply next migration
    command.upgrade(alembic_config, "a1b2c3d4e5f6")
    
    with engine.connect() as conn:
        result = conn.execute(text("SELECT COUNT(*) FROM users"))
        assert result.scalar() == 1  # Data preserved

def test_no_missing_revisions(alembic_config):
    """Detect broken revision chains"""
    script = ScriptDirectory.from_config(alembic_config)
    
    revisions = list(script.walk_revisions())
    revision_ids = {rev.revision for rev in revisions}
    
    for rev in revisions:
        if rev.down_revision:
            parents = rev.down_revision if isinstance(rev.down_revision, tuple) else (rev.down_revision,)
            for parent in parents:
                assert parent in revision_ids, \
                    f"Revision {rev.revision} references missing parent {parent}"

Testing Prisma Migrate

Prisma uses prisma migrate dev locally and prisma migrate deploy in production. Testing migrations means running the deploy command against a test database.

Setup with Vitest/Jest

// tests/migrations.test.ts
import { execSync } from 'child_process';
import { PrismaClient } from '@prisma/client';
import { PostgreSqlContainer, StartedPostgreSqlContainer } from '@testcontainers/postgresql';

let container: StartedPostgreSqlContainer;
let prisma: PrismaClient;

beforeAll(async () => {
  container = await new PostgreSqlContainer('postgres:16-alpine').start();
  
  const dbUrl = container.getConnectionUri();
  
  // Apply all migrations
  execSync('npx prisma migrate deploy', {
    env: { ...process.env, DATABASE_URL: dbUrl }
  });
  
  prisma = new PrismaClient({ datasources: { db: { url: dbUrl } } });
}, 60000);

afterAll(async () => {
  await prisma.$disconnect();
  await container.stop();
});

test('schema matches Prisma client expectations', async () => {
  // If schema is wrong, Prisma operations will throw
  const user = await prisma.user.create({
    data: { email: 'schema@test.com', name: 'Schema Test' }
  });
  
  expect(user.id).toBeDefined();
  expect(user.createdAt).toBeInstanceOf(Date);
});

test('all indexes exist', async () => {
  const result = await prisma.$queryRaw<Array<{ indexname: string }>>`
    SELECT indexname 
    FROM pg_indexes 
    WHERE tablename = 'User'
  `;
  
  const indexNames = result.map(r => r.indexname);
  expect(indexNames).toContain('User_email_key'); // unique index
});

Prisma Migration Shadow Database

Prisma's migrate dev uses a shadow database to detect drift. In CI, ensure the shadow database URL is configured:

# .env.test
DATABASE_URL="postgresql://test:test@localhost:5432/app_test"
SHADOW_DATABASE_URL="postgresql://test:test@localhost:5432/app_test_shadow"

CI/CD Integration

GitHub Actions Example

name: Migration Tests
on: [push, pull_request]

jobs:
  migrations:
    runs-on: ubuntu-latest
    
    services:
      postgres:
        image: postgres:16-alpine
        env:
          POSTGRES_PASSWORD: testpass
          POSTGRES_DB: migtest
        ports:
          - 5432:5432
        options: --health-cmd pg_isready --health-interval 10s
    
    steps:
      - uses: actions/checkout@v4
      
      - name: Test Flyway migrations
        run: |
          flyway -url=jdbc:postgresql://localhost:5432/migtest \
                 -user=postgres -password=testpass \
                 -locations=filesystem:src/main/resources/db/migration \
                 migrate
        
      - name: Verify schema
        run: python tests/verify_schema.py
      
      - name: Test rollback
        run: |
          flyway -url=jdbc:postgresql://localhost:5432/migtest \
                 -user=postgres -password=testpass \
                 undo  # If using Flyway Teams

Zero-Downtime Migration Patterns

Some migrations are inherently risky. Test these patterns:

Adding a nullable column (safe):

ALTER TABLE users ADD COLUMN phone VARCHAR(20);
-- Safe: existing rows get NULL, no lock on data

Renaming a column (risky — requires multi-step):

-- Step 1: Add new column
ALTER TABLE users ADD COLUMN phone_number VARCHAR(20);

-- Step 2: Deploy code that writes to both columns

-- Step 3: Backfill
UPDATE users SET phone_number = phone WHERE phone_number IS NULL;

-- Step 4: Deploy code that reads from new column only

-- Step 5: Drop old column
ALTER TABLE users DROP COLUMN phone;

Test each step independently, with data in the database, to verify no data is lost.

Monitoring Post-Migration

After a migration deploys to production, your tests are done — but the risk isn't. Migrations can cause:

Slow queries from missing indexes
Application errors from schema mismatches
Data integrity issues from partial migrations

HelpMeTest runs your end-to-end tests continuously after deployment. Set up plain-English health checks that verify critical database-backed workflows, and get alerted immediately if a migration caused unexpected behavior — not when users start complaining.

Summary

Migration testing is not optional — it's the difference between confident deployments and production fire drills. Test forward migration, rollback, data preservation, and revision chain integrity. Run these tests in CI against a real database (Testcontainers makes this easy). And monitor production after every migration deploy to catch the issues that pre-deploy testing can't predict.