Test Data Management: Fixtures, Factories, and Data Strategies

# fixtures.py - static data loaded once
import json
import os

FIXTURES_DIR = os.path.join(os.path.dirname(__file__), "fixtures")


def load_fixture(name):
    """Load a static fixture file."""
    path = os.path.join(FIXTURES_DIR, f"{name}.json")
    with open(path) as f:
        return json.load(f)


# fixtures/users.json
"""
[
    {"id": 1, "email": "alice@example.com", "name": "Alice", "role": "admin"},
    {"id": 2, "email": "bob@example.com", "name": "Bob", "role": "user"},
    {"id": 3, "email": "charlie@example.com", "name": "Charlie", "role": "user"}
]
"""

# Usage in tests:
def test_get_admin_users(api_client):
    # Seed known admin user from fixture
    seed_users(load_fixture("users"))
    response = api_client.get("/users?role=admin")
    assert len(response.json()) == 1
    assert response.json()[0]["email"] == "alice@example.com"

# factories.py - dynamic data generation
import factory
from factory import Faker, Sequence, SubFactory

class UserFactory(factory.Factory):
    class Meta:
        model = dict

    id = Sequence(lambda n: n + 100)  # Start at 100 to avoid fixture IDs
    email = Faker("email")
    name = Faker("name")
    role = "user"
    created_at = Faker("iso8601")

class AdminFactory(UserFactory):
    role = "admin"

class OrderFactory(factory.Factory):
    class Meta:
        model = dict

    id = Sequence(lambda n: n + 1000)
    user_id = Sequence(lambda n: n + 100)
    product = Faker("word")
    quantity = Faker("random_int", min=1, max=5)
    total = Faker("pydecimal", left_digits=3, right_digits=2, positive=True)
    status = "pending"

# Usage in tests:
def test_create_order(api_client):
    user = UserFactory()
    seed_user(user)  # Insert user into test database

    order_data = OrderFactory(user_id=user["id"])
    response = api_client.post("/orders", json=order_data)

    assert response.status_code == 201
    assert response.json()["user_id"] == user["id"]

def test_admin_has_access(api_client):
    admin = AdminFactory()
    seed_user(admin)

    response = api_client.get("/admin/dashboard")
    assert response.status_code == 200

# Generate unique test data on the fly
@pytest.fixture
def unique_user():
    return UserFactory(email=f"test-{uuid.uuid4().hex[:8]}@example.com")

# seed.py - database seeding for integration tests
class TestDataSeeder:
    def __init__(self, db_connection):
        self.db = db_connection

    def seed_default_data(self):
        """Seed reference data required by all tests."""
        self.db.execute("""
            INSERT INTO roles (id, name) VALUES (1, 'admin'), (2, 'user')
        """)
        self.db.execute("""
            INSERT INTO categories (id, name)
            VALUES (1, 'Electronics'), (2, 'Books'), (3, 'Clothing')
        """)
        print("Default data seeded")

    def seed_users(self, count=10):
        """Seed a specified number of test users."""
        users = [UserFactory() for _ in range(count)]
        for user in users:
            self.db.execute(
                "INSERT INTO users (email, name, role) VALUES (?, ?, ?)",
                (user["email"], user["name"], user["role"])
            )
        print(f"Seeded {count} users")
        return users

    def seed_orders(self, user_ids, orders_per_user=3):
        """Seed orders for given users."""
        orders = []
        for uid in user_ids:
            for _ in range(orders_per_user):
                order = OrderFactory(user_id=uid)
                self.db.execute(
                    "INSERT INTO orders (user_id, product, quantity, total, status) "
                    "VALUES (?, ?, ?, ?, ?)",
                    (order["user_id"], order["product"],
                     order["quantity"], order["total"], order["status"])
                )
                orders.append(order)
        print(f"Seeded {len(orders)} orders")
        return orders

    def cleanup_all(self):
        """Remove all test data."""
        self.db.execute("DELETE FROM orders")
        self.db.execute("DELETE FROM users")
        # Keep reference data (roles, categories)
        print("Test data cleaned up")

import pytest

# Strategy 1: Transaction rollback
@pytest.fixture
def db_transaction(connection):
    """Use database transaction that rolls back after test."""
    connection.begin()
    yield connection
    connection.rollback()


# Strategy 2: Dedicated test schema
@pytest.fixture(scope="session")
def test_schema(connection):
    """Create a dedicated test schema and drop it after the session."""
    connection.execute("CREATE SCHEMA IF NOT EXISTS test_data")
    yield "test_data"
    connection.execute("DROP SCHEMA test_data CASCADE")


# Strategy 3: Unique prefix per test
@pytest.fixture
def unique_test_id():
    """Generate a unique prefix for each test's data."""
    import uuid
    return f"test_{uuid.uuid4().hex[:8]}"


# Strategy 4: Factory-based cleanup
@pytest.fixture
def tracked_data():
    """Track all created records for cleanup."""
    created = {"users": [], "orders": [], "products": []}
    yield created
    # Cleanup in reverse dependency order
    for table in ["orders", "products", "users"]:
        ids = created[table]
        if ids:
            # Bulk delete
            pass  # delete from {table} where id in {ids}


# Strategy 5: Parallel-safe data isolation
@pytest.fixture
def isolated_db(db_pool, worker_id):
    """Each parallel worker gets its own database schema."""
    schema = f"test_{worker_id}"
    db_pool.execute(f"CREATE SCHEMA IF NOT EXISTS {schema}")
    yield schema
    db_pool.execute(f"DROP SCHEMA {schema} CASCADE")