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.
