Database Testing

Database Schema Testing and Validation: Drift Detection and Tools

HelpMeTest

HelpMeTest

— 6 min read
Database Schema Testing and Validation: Drift Detection and Tools

Schema drift is one of the most dangerous forms of technical debt in database-driven applications. It happens gradually: a developer adds a column directly to staging but forgets to write a migration, another runs a manual ALTER TABLE on production to fix an emergency, and a third adds an index locally that never gets documented. Over time, your databases diverge from each other and from what your code expects.

This guide covers the tools and strategies for testing and validating database schemas — catching drift before it causes failures, comparing environments, and validating that APIs align with database structure.

What Is Schema Drift?

Schema drift occurs when the actual database schema diverges from what the application code, migrations, or other databases expect. Common causes:

  • Direct changes made to production or staging without migrations
  • Migrations applied out of order across environments
  • Failed migration rollbacks that left the schema in a partial state
  • Environment-specific manual fixes that were never propagated
  • Backups restored to databases with different schema versions

The symptom is often subtle: a query works in development but fails in production because a column doesn't exist, an index is missing causing performance degradation, or a constraint is present in one environment but not another.

Schema Comparison Tools

MySQL Schema Compare

For MySQL, mysqldiff from the MySQL Utilities package compares schemas:

# Install
pip install mysql-connector-python

# Compare two databases
mysqldiff --server1=user:pass@dev-host/mydb \
          --server2=user:pass@prod-host/mydb \
          --difftype=SQL

# Output shows SQL needed to make server1 match server2:
# ALTER TABLE `orders` ADD COLUMN `cancelled_at` DATETIME DEFAULT NULL;
# CREATE INDEX `idx_orders_status` ON `orders` (`status`);

PostgreSQL Schema Compare

For PostgreSQL, migra is the standard tool:

# Install
pip install migra psycopg2

# Compare schemas
migra postgresql://user:pass@dev/mydb postgresql://user:pass@prod/mydb

# Output: SQL to migrate prod to match dev
# ALTER TABLE orders ADD COLUMN cancelled_at TIMESTAMPTZ;
# CREATE INDEX CONCURRENTLY idx_orders_status ON orders(status);

migra is particularly useful in CI because it exits with code 0 when schemas match and non-zero when they differ, making it easy to fail a pipeline on drift.

Liquibase Diff

Liquibase has built-in diff capabilities that work across multiple database types:

# Install Liquibase
brew install liquibase

# Compare two databases
liquibase \
  --driver=com.mysql.cj.jdbc.Driver \
  --url="jdbc:mysql://dev-host/mydb" \
  --username=user \
  --password=pass \
  diff \
  --referenceUrl="jdbc:mysql://prod-host/mydb" \
  --referenceUsername=user \
  --referencePassword=pass

# Generate a changelog from the diff
liquibase \
  --url="jdbc:mysql://dev-host/mydb" \
  --username=user \
  --password=pass \
  diffChangeLog \
  --referenceUrl="jdbc:mysql://prod-host/mydb" \
  --referenceUsername=user \
  --referencePassword=pass \
  --changeLogFile=diff.xml

Automated Schema Validation in CI

Checking for Unapplied Migrations

The fastest schema validation: verify that no migrations are pending:

# Flyway
mvn flyway:info | grep "Pending" && exit 1 || echo "All migrations applied"

# Liquibase
liquibase status --verbose
# Exits non-zero if there are unapplied changesets

In GitHub Actions:

- name: Validate database schema
  run: |
    mvn flyway:validate
    mvn flyway:info | grep -v "Pending" || (echo "Pending migrations found" && exit 1)

Schema Snapshot Testing

Snapshot the expected schema and compare it on every CI run:

# Generate current schema snapshot
pg_dump --schema-only -h localhost mydb > expected_schema.sql

# In CI: compare current state to committed snapshot
pg_dump --schema-only -h $TEST_DB_HOST $TEST_DB > current_schema.sql
diff expected_schema.sql current_schema.sql

Or use a dedicated schema snapshot tool like skeema:

# Initialize skeema in your repo
skeema init --host=localhost --user=root --password=pass --dir=schema

# After changes, push the schema directory to git
git add schema/
git commit -m "Update schema snapshot"

# In CI, validate schema matches snapshot
skeema diff --host=localhost --user=root --password=pass
# Exits non-zero if actual schema differs from snapshot files

Schemathesis for API-Schema Validation

Schemathesis is primarily an API testing tool, but it's also excellent for validating that your API responses match your database schema. It generates test cases from OpenAPI specifications and runs them against your live API:

# Install
pip install schemathesis

# Run against a live API
schemathesis run http://localhost:8080/openapi.json --checks all

# Or against an OpenAPI file
schemathesis run path/to/openapi.yaml \
  --base-url http://localhost:8080 \
  --checks not_a_server_error,response_schema_conformance

What Schemathesis Catches

  • API responses that don't match the declared response schema
  • Missing fields that the OpenAPI spec says should be present
  • Type mismatches (database returns integer, API declares string)
  • Null values in fields declared as required
  • 500 errors caused by unexpected input combinations

This is particularly valuable when your API is a thin wrapper over database queries — Schemathesis finds the edge cases where the database returns something unexpected.

Custom Schema Validation with Schemathesis

import schemathesis
from schemathesis import DataGenerationMethod

schema = schemathesis.from_path("openapi.yaml", base_url="http://localhost:8080")

@schema.parametrize()
def test_api_responses_match_schema(case):
    response = case.call()
    case.validate_response(response)

@schema.parametrize(method="GET", endpoint="/api/users/{user_id}")
def test_user_endpoint_handles_edge_cases(case):
    response = case.call()
    # All 200 responses should have valid structure
    if response.status_code == 200:
        case.validate_response(response)
    # Only 404 is acceptable for user not found
    assert response.status_code in [200, 404], \
        f"Unexpected status code: {response.status_code}"

Writing Schema Validation Tests

JUnit Schema Tests (Java)

@SpringBootTest
@ActiveProfiles("test")
class DatabaseSchemaValidationTest {

    @Autowired
    private JdbcTemplate jdbcTemplate;

    @Test
    void requiredTablesShouldExist() {
        List<String> requiredTables = List.of(
            "users", "orders", "order_items", "products", "categories"
        );
        
        for (String table : requiredTables) {
            assertDoesNotThrow(
                () -> jdbcTemplate.execute("SELECT 1 FROM " + table + " LIMIT 1"),
                "Required table missing: " + table
            );
        }
    }

    @Test
    void criticalColumnsShouldExist() {
        // Map of table -> required columns
        Map<String, List<String>> requiredColumns = Map.of(
            "users", List.of("id", "email", "created_at", "status"),
            "orders", List.of("id", "user_id", "total_amount", "status", "created_at")
        );
        
        for (var entry : requiredColumns.entrySet()) {
            String table = entry.getKey();
            for (String column : entry.getValue()) {
                assertDoesNotThrow(
                    () -> jdbcTemplate.execute("SELECT " + column + " FROM " + table + " LIMIT 1"),
                    "Required column missing: " + table + "." + column
                );
            }
        }
    }

    @Test
    void uniqueConstraintsShouldBeEnforced() {
        // email must be unique in users
        jdbcTemplate.update(
            "INSERT INTO users (email, name) VALUES ('constraint_test@test.com', 'Test')"
        );
        
        assertThrows(DataIntegrityViolationException.class, () ->
            jdbcTemplate.update(
                "INSERT INTO users (email, name) VALUES ('constraint_test@test.com', 'Duplicate')"
            )
        );
    }

    @Test
    void foreignKeyConstraintsShouldBeEnforced() {
        assertThrows(DataIntegrityViolationException.class, () ->
            jdbcTemplate.update(
                "INSERT INTO orders (user_id, total_amount) VALUES (999999, 100.00)"
            )
        );
    }
}

Python Schema Validation

import pytest
import sqlalchemy
from sqlalchemy import inspect

@pytest.fixture(scope='session')
def engine():
    return sqlalchemy.create_engine('mysql+pymysql://user:pass@localhost/mydb')

@pytest.fixture(scope='session')
def inspector(engine):
    return inspect(engine)

def test_required_tables_exist(inspector):
    required_tables = ['users', 'orders', 'order_items', 'products']
    existing_tables = inspector.get_table_names()
    
    for table in required_tables:
        assert table in existing_tables, f"Required table '{table}' is missing"

def test_users_table_columns(inspector):
    columns = {col['name']: col for col in inspector.get_columns('users')}
    
    assert 'id' in columns
    assert 'email' in columns
    assert not columns['email']['nullable'], "email should be NOT NULL"
    assert columns['email']['type'].__class__.__name__ in ['VARCHAR', 'String']

def test_orders_has_user_foreign_key(inspector):
    fks = inspector.get_foreign_keys('orders')
    
    user_fk = next((fk for fk in fks if fk['referred_table'] == 'users'), None)
    assert user_fk is not None, "orders.user_id should reference users table"
    assert 'user_id' in user_fk['constrained_columns']

def test_critical_indexes_exist(inspector):
    # Check that the email index exists on users (needed for login performance)
    indexes = inspector.get_indexes('users')
    email_index = next((idx for idx in indexes 
                        if 'email' in idx['column_names']), None)
    assert email_index is not None, "Index on users.email is required for login performance"

Schema Drift Detection in Production

For production databases, run drift detection on a schedule:

# drift_detector.py
import subprocess
import sys
from datetime import datetime

def check_schema_drift():
    """Compare production schema against the committed snapshot."""
    
    # Dump current production schema
    result = subprocess.run([
        'pg_dump', '--schema-only',
        '-h', 'prod-host', '-U', 'prod-user', 'prod-db'
    ], capture_output=True, text=True)
    
    current_schema = result.stdout
    
    # Load committed snapshot
    with open('schema/expected_schema.sql') as f:
        expected_schema = f.read()
    
    if current_schema.strip() != expected_schema.strip():
        print(f"[{datetime.now()}] SCHEMA DRIFT DETECTED")
        
        # Generate diff
        subprocess.run(['diff', 
                        '/tmp/expected_schema.sql', 
                        '/tmp/current_schema.sql'])
        
        # Alert (Slack, PagerDuty, etc.)
        send_alert("Schema drift detected on production database")
        sys.exit(1)
    else:
        print(f"[{datetime.now()}] Schema matches expected state")

Run this as a scheduled job:

# .github/workflows/schema-drift-check.yml
on:
  schedule:
    - cron: '0 */6 * * *'  # Every 6 hours

jobs:
  drift-check:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Check schema drift
        run: python drift_detector.py
        env:
          PROD_DB_HOST: ${{ secrets.PROD_DB_HOST }}
          PROD_DB_PASSWORD: ${{ secrets.PROD_DB_PASSWORD }}

Liquibase Schema Validation

Liquibase provides built-in snapshot and diff capabilities for ongoing validation:

# Take a snapshot of the current state
liquibase snapshot \
  --snapshotFormat=json \
  --outputFile=schema_snapshot.json

# Later, compare current state to snapshot
liquibase diff \
  --referenceUrl=offline:mysql?snapshot=schema_snapshot.json \
  --url="jdbc:mysql://prod-host/mydb"

# Generate a report
liquibase diffChangeLog \
  --referenceUrl=offline:mysql?snapshot=schema_snapshot.json \
  --url="jdbc:mysql://prod-host/mydb" \
  --changeLogFile=drift_report.xml

Summary: Schema Validation Strategy

A complete schema validation setup has four layers:

  1. Migration validation — flyway validate or liquibase status in every CI pipeline run. Catches edited migrations and pending changes.
  2. Schema comparison — migra, mysqldiff, or Liquibase diff between environments. Catches drift between dev, staging, and production.
  3. Constraint testing — automated tests that verify foreign keys, unique constraints, NOT NULL constraints, and check constraints are enforced. Catches constraints that were added to one environment but not others.
  4. Scheduled drift detection — periodic comparison between production and the committed schema snapshot. Catches manual changes made outside the migration process.

Together, these layers make schema drift detectable within hours rather than months — and give you the data you need to fix it before it causes failures.

Read more

Start now free