Database migration testing in Testing Fundamentals - Build an Automation Script

Verify data integrity after database migration

Preconditions (3)

Step 1: Connect to the source database and export the test data from the 'users' table

Step 2: Run the migration script/tool to migrate data from source to target database

Step 3: Connect to the target database and query the 'users' table

Step 4: Compare the data in the source and target 'users' tables for all columns

Step 5: Verify that the number of records in both tables is the same

Step 6: Verify that no data is lost or altered during migration

✅ Expected Result: All records in the 'users' table are migrated correctly with identical data and record count in the target database

Automation Requirements - Python with pytest and psycopg2 for PostgreSQL

Assertions Needed:

Assert that record counts in source and target tables are equal

Assert that each record's data matches exactly between source and target

Best Practices:

Use database connection pooling or context managers to manage connections

Use parameterized queries to avoid SQL injection

Fetch data in a consistent order for comparison

Use clear and descriptive assertion messages

Handle exceptions and close connections properly

Automated Solution

Testing Fundamentals

import psycopg2
import pytest

SOURCE_DB_CONFIG = {
    'host': 'source-db-host',
    'database': 'source_db',
    'user': 'source_user',
    'password': 'source_pass'
}

TARGET_DB_CONFIG = {
    'host': 'target-db-host',
    'database': 'target_db',
    'user': 'target_user',
    'password': 'target_pass'
}

def get_users_data(db_config):
    with psycopg2.connect(**db_config) as conn:
        with conn.cursor() as cur:
            cur.execute('SELECT id, username, email, created_at FROM users ORDER BY id')
            return cur.fetchall()

@pytest.fixture(scope='module')
def source_users():
    return get_users_data(SOURCE_DB_CONFIG)

@pytest.fixture(scope='module')
def target_users():
    return get_users_data(TARGET_DB_CONFIG)


def test_user_data_migration(source_users, target_users):
    assert len(source_users) == len(target_users), f"Record count mismatch: source={len(source_users)}, target={len(target_users)}"
    for i, (src_row, tgt_row) in enumerate(zip(source_users, target_users), start=1):
        assert src_row == tgt_row, f"Data mismatch at record {i}: source={src_row}, target={tgt_row}"

This test script connects to both source and target PostgreSQL databases using psycopg2. It fetches all records from the users table ordered by id to ensure consistent comparison.

Fixtures source_users and target_users provide the data sets for the test function test_user_data_migration.

The test first asserts that the number of records is the same in both databases. Then it compares each record one by one to ensure data integrity.

Using context managers ensures connections and cursors are properly closed. Assertions include clear messages to help identify issues.

Common Mistakes - 4 Pitfalls

{'mistake': 'Not ordering the data before comparison', 'why_bad': 'Data fetched from databases can be in any order, causing false mismatches', 'correct_approach': "Always use ORDER BY clause on a unique column like 'id' to get consistent order"}

Not closing database connections

{'mistake': 'Comparing only record counts without data validation', 'why_bad': 'Record counts can match but data might be corrupted or incomplete', 'correct_approach': "Compare each record's data fields to ensure full integrity"}

Hardcoding credentials in test code

Bonus Challenge

Now add data-driven testing with 3 different tables: 'users', 'orders', and 'products' to verify migration for each

Show Hint