Test Data Management Complete Guide

import { faker } from '@faker-js/faker';

interface TestUser {
  id: string;
  name: string;
  email: string;
  age: number;
  role: 'admin' | 'user' | 'moderator';
  createdAt: Date;
  metadata: Record<string, unknown>;
}

function generateUser(overrides: Partial<TestUser> = {}): TestUser {
  return {
    id: faker.string.uuid(),
    name: faker.person.fullName(),
    email: faker.internet.email(),
    age: faker.number.int({ min: 18, max: 80 }),
    role: faker.helpers.arrayElement(['admin', 'user', 'moderator']),
    createdAt: faker.date.past({ years: 2 }),
    metadata: {
      signupSource: faker.helpers.arrayElement(['web', 'mobile', 'referral']),
      timezone: faker.location.timeZone(),
      preferences: {
        newsletter: faker.datatype.boolean(),
        theme: faker.helpers.arrayElement(['light', 'dark']),
      },
    },
    ...overrides,
  };
}

function generateBulkUsers(count: number): TestUser[] {
  return Array.from({ length: count }, () => generateUser());
}

// Generate 1000 test users
const testUsers = generateBulkUsers(1000);

// Generate users with specific characteristics
const adminUsers = generateBulkUsers(10).map(u => ({ ...u, role: 'admin' as const }));
const europeanUsers = generateBulkUsers(50).map(u => ({
  ...u,
  metadata: { ...u.metadata, timezone: 'Europe/Berlin' },
}));

"""Generate test data with Faker in Python."""

from faker import Faker
from typing import Any
import json

fake = Faker()


def generate_order() -> dict[str, Any]:
    """Generate a single test order."""
    return {
        "order_id": fake.uuid4(),
        "customer": {
            "name": fake.name(),
            "email": fake.email(),
            "address": {
                "street": fake.street_address(),
                "city": fake.city(),
                "country": fake.country(),
                "postcode": fake.postcode(),
            },
        },
        "items": [
            {
                "product_id": fake.uuid4(),
                "name": fake.word().capitalize(),
                "price": round(fake.random_number(digits=4) / 100, 2),
                "quantity": fake.random_int(min=1, max=5),
            }
            for _ in range(fake.random_int(min=1, max=5))
        ],
        "total": 0.0,
        "status": fake.random_element(["pending", "completed", "cancelled", "refunded"]),
        "created_at": fake.date_time_this_year().isoformat(),
    }


def finalize_order(order: dict) -> dict:
    """Calculate total after generating items."""
    total = sum(item["price"] * item["quantity"] for item in order["items"])
    order["total"] = round(total, 2)
    return order


orders = [finalize_order(generate_order()) for _ in range(100)]

# Export to JSON for test consumption
with open("test-orders.json", "w") as f:
    json.dump(orders, f, indent=2)

-- Subsetting with referential integrity
BEGIN;

-- Extract active users (last 90 days)
CREATE TEMP TABLE active_users AS
SELECT * FROM users
WHERE last_login >= CURRENT_DATE - INTERVAL '90 days';

-- Extract orders for active users only
CREATE TEMP TABLE user_orders AS
SELECT o.* FROM orders o
INNER JOIN active_users u ON o.user_id = u.id;

-- Extract order items for those orders
CREATE TEMP TABLE order_items AS
SELECT oi.* FROM order_items oi
INNER JOIN user_orders o ON oi.order_id = o.id;

-- Export subsetted tables
\COPY active_users TO 'subset_users.csv' CSV HEADER;
\COPY user_orders TO 'subset_orders.csv' CSV HEADER;
\COPY order_items TO 'subset_items.csv' CSV HEADER;

COMMIT;

# Database clone configuration (PostgreSQL)
clone_config:
  source: production
  target: staging
  method: pg_basebackup
  schedule:
    frequency: daily
    time: "03:00 UTC"
  post_processing:
    - run: masking_scripts/mask_pii.sql
    - run: scripts/anonymize.sql
    - run: scripts/truncate_audit_logs.sql

-- PostgreSQL data masking functions
CREATE OR REPLACE FUNCTION mask_email(email TEXT)
RETURNS TEXT AS $$
BEGIN
  RETURN regexp_replace(
    email,
    '^[^@]+',
    'user.' || floor(random() * 9999)::TEXT
  );
END;
$$ LANGUAGE plpgsql;

CREATE OR REPLACE FUNCTION mask_phone(phone TEXT)
RETURNS TEXT AS $$
BEGIN
  RETURN '+1-555-' || lpad(floor(random() * 9999)::TEXT, 4, '0');
END;
$$ LANGUAGE plpgsql;

CREATE OR REPLACE FUNCTION mask_name(name TEXT)
RETURNS TEXT AS $$
DECLARE
  first_names TEXT[] := ARRAY['Alice', 'Bob', 'Charlie', 'Diana', 'Eve'];
  last_names TEXT[] := ARRAY['Smith', 'Jones', 'Brown', 'Taylor', 'Wilson'];
BEGIN
  RETURN first_names[1 + floor(random() * array_length(first_names, 1))]
    || ' ' ||
    last_names[1 + floor(random() * array_length(last_names, 1))];
END;
$$ LANGUAGE plpgsql;

-- Apply masking to a production clone
BEGIN;

UPDATE customers SET
  first_name = mask_name(first_name),
  last_name = mask_name(last_name),
  email = mask_email(email),
  phone = mask_phone(phone),
  credit_card = '4111-1111-1111-1111'  -- Test card number
WHERE email NOT LIKE '%@testcompany.com';

COMMIT;

-- PostgreSQL dynamic masking with views
CREATE OR REPLACE VIEW customers_masked AS
SELECT
  id,
  CASE
    WHEN current_user = 'support_role' THEN mask_name(first_name)
    ELSE first_name
  END AS first_name,
  CASE
    WHEN current_user = 'support_role' THEN mask_name(last_name)
    ELSE last_name
  END AS last_name,
  CASE
    WHEN current_user IN ('admin_role', 'audit_role') THEN email
    ELSE mask_email(email)
  END AS email,
  CASE
    WHEN current_user = 'admin_role' THEN phone
    ELSE mask_phone(phone)
  END AS phone,
  created_at,
  status
FROM customers;

"""Tokenization service example."""

import hashlib
import secrets
from typing import Optional
import redis


class TokenizationService:
    """Map sensitive values to tokens and back."""

    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
        self.namespace = "token:"

    def tokenize(self, value: str) -> str:
        """Create a deterministic token for a value."""
        hash_key = hashlib.sha256(value.encode()).hexdigest()[:16]
        token = f"tok_{hash_key}"

        # Store mapping (idempotent)
        self.redis.setex(
            f"{self.namespace}{token}",
            86400,  # 24-hour TTL
            value,
        )
        return token

    def detokenize(self, token: str) -> Optional[str]:
        """Retrieve original value from token."""
        return self.redis.get(f"{self.namespace}{token}")

    def batch_tokenize(self, values: list[str]) -> list[str]:
        """Tokenize multiple values."""
        return [self.tokenize(v) for v in values]


service = TokenizationService(redis.Redis())
email_token = service.tokenize("[email protected]")
print(f"Token: {email_token}")
print(f"Original: {service.detokenize(email_token)}")

# test-data-manifest.yml
version: "2.1"
schema_version: "2026-05"
generated_at: "2026-05-24T10:00:00Z"

datasets:
  unit_tests:
    source: synthetic
    size: 100_records
    compatibility:
      min_schema: "2026-01"
      max_schema: "2026-06"

  integration_tests:
    source: production_subset
    production_snapshot: "2026-05-23"
    mask_applied: true
    size: 10000_records
    compatibility:
      min_schema: "2026-03"

  performance_tests:
    source: production_clone
    production_snapshot: "2026-05-22"
    mask_applied: false  # Internal test environment
    size: 1000000_records

# .github/workflows/data-refresh.yml
name: Test Data Refresh

on:
  schedule:
    - cron: "0 4 * * *"  # Daily at 4 AM
  workflow_dispatch:

jobs:
  refresh-data:
    runs-on: ubuntu-latest
    services:
      postgres:
        image: postgres:16-alpine
        env:
          POSTGRES_PASSWORD: test
        ports:
          - 5432:5432

    steps:
      - uses: actions/checkout@v4

      - name: Pull production snapshot
        run: |
          pg_dump \
            --host=${{ secrets.PROD_DB_HOST }} \
            --username=${{ secrets.PROD_DB_USER }} \
            --dbname=production \
            --format=custom \
            --file=prod_dump.dump

      - name: Restore to staging
        run: |
          pg_restore \
            --host=localhost \
            --username=test \
            --dbname=test \
            --jobs=4 \
            prod_dump.dump

      - name: Apply masking
        run: |
          psql -h localhost -U test -d test -f scripts/mask_data.sql

      - name: Run data validation
        run: |
          psql -h localhost -U test -d test -f scripts/validate_data.sql

      - name: Export subsetted data
        run: |
          psql -h localhost -U test -d test -c "\COPY active_users TO 'test_data/users.csv' CSV HEADER"

      - name: Upload artifacts
        uses: actions/upload-artifact@v4
        with:
          name: test-data
          path: test_data/

// Test cleanup patterns
import { PrismaClient } from '@prisma/client';

const prisma = new PrismaClient();

describe('Order processing', () => {
  // Create test data once, clean up after all tests
  beforeAll(async () => {
    await prisma.user.create({
      data: { id: 'test-user-1', name: 'Test', email: '[email protected]' },
    });
  });

  afterAll(async () => {
    await prisma.orderItem.deleteMany({ where: { order: { userId: 'test-user-1' } } });
    await prisma.order.deleteMany({ where: { userId: 'test-user-1' } });
    await prisma.user.delete({ where: { id: 'test-user-1' } });
    await prisma.$disconnect();
  });

  it('creates an order for the test user', async () => {
    const order = await prisma.order.create({
      data: {
        userId: 'test-user-1',
        total: 29.99,
        status: 'pending',
      },
    });
    expect(order.id).toBeDefined();
    expect(order.status).toBe('pending');
  });
});

// Transaction rollback pattern for integration tests
import { PrismaClient } from '@prisma/client';

const prisma = new PrismaClient();

describe('Database operations', () => {
  beforeEach(async () => {
    await prisma.$transaction([
      prisma.$executeRawUnsafe('BEGIN'),
    ]);
  });

  afterEach(async () => {
    await prisma.$transaction([
      prisma.$executeRawUnsafe('ROLLBACK'),
    ]);
  });

  it('inserts a user within a transaction', async () => {
    const user = await prisma.user.create({
      data: { name: 'Alice', email: '[email protected]' },
    });
    expect(user.name).toBe('Alice');
    // Transaction rolls back after this test
  });

  it('starts with a clean database', async () => {
    const users = await prisma.user.findMany();
    expect(users).toHaveLength(0); // Previous user was rolled back
  });
});

# PostgreSQL point-in-time clone
pg_basebackup -D /data/test_clone -X stream -P
pg_ctl -D /data/test_clone start -o "-p 5433"

# Clone to specific timestamp
pg_restore --dbname=test_db \
  --jobs=4 \
  --schema=public \
  --data-only \
  --use-list=table_list.txt \
  prod_dump.dump

"""Custom test data pipeline."""

#!/usr/bin/env python3

import json
import subprocess
from pathlib import Path
from datetime import datetime
from typing import Optional

import psycopg2


class TestDataPipeline:
    """Automated test data provisioning pipeline."""

    def __init__(self, config_path: str):
        with open(config_path) as f:
            self.config = json.load(f)

    def pull_production_snapshot(self) -> Path:
        """Pull latest production snapshot."""
        output = Path(f"/tmp/testdata/prod_dump_{datetime.now():%Y%m%d}.dump")
        output.parent.mkdir(parents=True, exist_ok=True)

        subprocess.run([
            "pg_dump",
            f"--host={self.config['prod_host']}",
            f"--dbname={self.config['prod_db']}",
            f"--file={output}",
            "--format=custom",
            "--no-owner",
        ], check=True)
        return output

    def restore_to_test_db(self, dump_path: Path) -> None:
        """Restore dump to test database."""
        conn = psycopg2.connect(
            host=self.config["test_host"],
            dbname=self.config["test_db"],
            user=self.config["test_user"],
            password=self.config["test_password"],
        )
        conn.autocommit = True

        with conn.cursor() as cur:
            cur.execute("SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname = 'test'")
            cur.execute("DROP DATABASE IF EXISTS test")
            cur.execute("CREATE DATABASE test")

        conn.close()

        subprocess.run([
            "pg_restore",
            f"--dbname=postgresql://{self.config['test_user']}:{self.config['test_password']}@{self.config['test_host']}/test",
            f"--jobs=4",
            str(dump_path),
        ], check=True)

    def apply_masking(self) -> None:
        """Apply masking rules to test database."""
        subprocess.run([
            "psql",
            f"--host={self.config['test_host']}",
            f"--dbname=test",
            f"--file=scripts/mask_pii.sql",
        ], check=True)

    def run(self) -> bool:
        """Execute full data refresh pipeline."""
        try:
            dump = self.pull_production_snapshot()
            self.restore_to_test_db(dump)
            self.apply_masking()
            print(f"[SUCCESS] Test data refreshed from {dump}")
            return True
        except subprocess.CalledProcessError as e:
            print(f"[ERROR] Pipeline failed: {e}")
            return False


if __name__ == "__main__":
    pipeline = TestDataPipeline("config/testdata.json")
    success = pipeline.run()
    exit(0 if success else 1)

-- Data profiling queries
SELECT
  column_name,
  COUNT(*) AS total_rows,
  COUNT(DISTINCT value) AS unique_values,
  AVG(LENGTH(value::TEXT)) AS avg_length,
  MIN(LENGTH(value::TEXT)) AS min_length,
  MAX(LENGTH(value::TEXT)) AS max_length
FROM users
GROUP BY column_name;

"""Test data health monitoring."""

#!/usr/bin/env python3

import psycopg2
from datetime import datetime, timedelta


def check_data_freshness(conn) -> dict:
    """Check test data age."""
    with conn.cursor() as cur:
        cur.execute("SELECT MAX(created_at) FROM orders")
        max_date = cur.fetchone()[0]
        age_days = (datetime.now() - max_date).days
        return {
            "metric": "data_age_days",
            "value": age_days,
            "status": "ok" if age_days < 7 else "stale",
        }


def check_referential_integrity(conn) -> list[dict]:
    """Check for orphan rows."""
    checks = [
        ("SELECT COUNT(*) FROM order_items WHERE order_id NOT IN (SELECT id FROM orders)", "orphan_order_items"),
        ("SELECT COUNT(*) FROM orders WHERE user_id NOT IN (SELECT id FROM users)", "orphan_orders"),
    ]

    results = []
    for query, name in checks:
        with conn.cursor() as cur:
            cur.execute(query)
            count = cur.fetchone()[0]
            results.append({
                "metric": name,
                "value": count,
                "status": "ok" if count == 0 else "violation",
            })
    return results