test-data-management

Test Data Management Skill

Expertise: Factory functions, database isolation, seed data strategies, test pollution prevention.

Factory Pattern (Python — pytest)

# tests/factories.py
from faker import Faker
from decimal import Decimal
import pytest_asyncio

fake = Faker()

def build_user(**overrides) -> dict:
    """Build a user dict — does NOT write to DB"""
    return {
        "email": fake.email(domain="example-test.com"),  # Never real domains
        "name": fake.name(),
        "role": "viewer",
        "password_hash": "hashed_test_password",
        **overrides,
    }

def build_order(**overrides) -> dict:
    return {
        "status": "pending",
        "total_amount": Decimal("99.99"),
        "currency": "USD",
        **overrides,
    }

# Async factory fixture — writes to DB
@pytest_asyncio.fixture
async def create_user(db_session):
    created = []
    async def _create(**overrides):
        user = User(**build_user(**overrides))
        db_session.add(user)
        await db_session.flush()  # Get ID without committing
        created.append(user)
        return user
    yield _create
    # Cleanup is handled by transaction rollback (see isolation below)

# Usage in test
async def test_user_can_view_own_profile(create_user, client):
    user = await create_user(role="viewer")
    response = await client.get(f"/users/{user.id}", headers=auth_headers(user))
    assert response.status_code == 200
    assert response.json()["email"] == user.email

Database Isolation Strategies

Option 1: Transaction rollback (fastest — no cleanup needed)

# conftest.py
@pytest_asyncio.fixture
async def db_session(engine):
    async with engine.connect() as conn:
        transaction = await conn.begin()
        session = AsyncSession(bind=conn)
        yield session
        await transaction.rollback()   # Rollback after each test — zero pollution
        await session.close()

Option 2: Truncate tables (compatible with most ORM features)

@pytest_asyncio.fixture(autouse=True)
async def clean_tables(db_session):
    yield
    # After test: truncate in reverse FK order
    await db_session.execute(text("TRUNCATE order_items, orders, users RESTART IDENTITY CASCADE"))
    await db_session.commit()