Data Engineering Testing Patterns

Comprehensive testing strategies for data pipelines, SQL transformations, and data contracts.

Core Principles

1. Test data, not just code - Validate row counts, schema shapes, value distributions, and business rules alongside logic
2. Data contracts are boundaries - Every producer/consumer interface needs explicit schema and semantic contracts
3. Idempotent test fixtures - Tests must create, validate, and tear down their own state without side effects
4. Test at multiple granularities - Unit test individual transforms, integration test pipeline stages, end-to-end test full workflows
5. Fail fast with clear diagnostics - Store test failures with sample rows so debugging does not require re-running the full pipeline

Unit Testing SQL Transforms

Use pytest with DuckDB to validate individual transformations in memory

import pytest
import pandas as pd
from sqlalchemy import create_engine, text

@pytest.fixture(scope="module")
def test_engine():
    engine = create_engine("duckdb:///:memory:")
    yield engine
    engine.dispose()

@pytest.fixture
def seed_orders(test_engine):
    df = pd.DataFrame({
        "order_id": [1, 2, 3],
        "customer_id": [101, 101, 102],
        "amount": [50.00, 75.00, 200.00],
        "status": ["completed", "completed", "refunded"],
    })
    df.to_sql("orders", test_engine, if_exists="replace", index=False)
    return df

def test_revenue_excludes_refunds(test_engine, seed_orders):
    result = pd.read_sql(text("""
        SELECT customer_id, SUM(amount) AS total_revenue
        FROM orders WHERE status != 'refunded'
        GROUP BY customer_id
    """), test_engine)
    assert len(result) == 1
    assert result.iloc[0]["total_revenue"] == 125.00

See sql-transform-testing.md for:

• CTE isolation testing
• dbt unit test patterns with mock inputs
• Testing window functions and complex aggregations
• Snapshot and slowly changing dimension tests

Integration Testing Pipelines

Validate end-to-end pipeline execution against isolated staging schemas

@pytest.fixture(scope="session")
def staging_warehouse():
    engine = create_engine(os.environ["STAGING_DB_URL"])
    schema = f"test_{uuid.uuid4().hex[:8]}"
    engine.execute(f"CREATE SCHEMA {schema}")
    yield engine, schema
    engine.execute(f"DROP SCHEMA {schema} CASCADE")

def test_daily_revenue_end_to_end(staging_warehouse):
    engine, schema = staging_warehouse
    seed_test_data(engine, schema, fixture="daily_revenue_input.json")

    result = daily_revenue_pipeline(
        source_schema=schema, target_schema=schema, execution_date="2025-03-01",
    )
    assert result.status == "success"

    output = pd.read_sql(f"SELECT * FROM {schema}.revenue_daily", engine)
    assert len(output) > 0
    assert output["revenue"].sum() > 0

See integration-testing.md for:

• Staging environment provisioning and teardown
• Testing Airflow/Prefect DAGs locally
• Database transaction rollback patterns
• CI/CD integration with containerized warehouses

dbt Testing Patterns

Use generic tests, singular tests, and store_failures for production reliability

# models/staging/stg_orders.yml
version: 2
models:
  - name: stg_orders
    columns:
      - name: order_id
        tests: [unique, not_null]
      - name: status
        tests:
          - accepted_values:
              values: ["completed", "pending", "refunded", "cancelled"]
      - name: amount
        tests:
          - not_null
          - dbt_utils.expression_is_true:
              expression: ">= 0"
              config:
                store_failures: true
                schema: test_failures

See dbt-testing-patterns.md for:

• Writing custom generic tests
• Singular tests for complex business rules
• store_failures configuration and monitoring
• dbt unit tests with mock refs and sources
• Test macros for reusable validation logic

Data Contract Testing

Enforce schema and semantic contracts between producers and consumers

@dataclass
class DataContract:
    name: str
    version: str
    schema: dict
    semantic_rules: list

    def validate_schema(self, df: pd.DataFrame) -> list[str]:
        errors = []
        for col, spec in self.schema["columns"].items():
            if col not in df.columns:
                errors.append(f"Missing required column: {col}")
            elif not pd.api.types.is_dtype_equal(df[col].dtype, spec["dtype"]):
                errors.append(f"{col}: expected {spec['dtype']}, got {df[col].dtype}")
        return errors

    def validate_semantics(self, df: pd.DataFrame) -> list[str]:
        errors = []
        for rule in self.semantic_rules:
            failing = df.query(f"not ({rule['expression']})")
            if len(failing) > 0:
                errors.append(f"Rule '{rule['name']}' failed on {len(failing)} rows")
        return errors

def test_orders_contract(orders_df):
    contract = DataContract(
        name="orders", version="2.1.0",
        schema=load_contract("contracts/orders_v2.json"),
        semantic_rules=[
            {"name": "positive_amount", "expression": "amount >= 0"},
        ],
    )
    assert not contract.validate_schema(orders_df)
    assert not contract.validate_semantics(orders_df)

See data-contract-patterns.md for:

• JSON Schema and Avro contract definitions
• Backward and forward compatibility checks
• Contract versioning strategies
• Producer and consumer test responsibilities

Pipeline Regression Testing

Detect unexpected changes in pipeline output across code changes

import hashlib

def compute_output_fingerprint(df: pd.DataFrame, key_columns: list[str]) -> str:
    sorted_df = df.sort_values(key_columns).reset_index(drop=True)
    return hashlib.sha256(sorted_df.to_csv(index=False).encode()).hexdigest()

class TestRevenueRegression:
    EXPECTED_HASH = "a3f2b8c1d4e5..."

    def test_output_matches_baseline(self, pipeline_output):
        actual = compute_output_fingerprint(pipeline_output, ["date", "region"])
        assert actual == self.EXPECTED_HASH, "Output changed. Update hash if intentional."

    def test_row_count_within_bounds(self, pipeline_output):
        assert 900 <= len(pipeline_output) <= 1100

Test Data Management

Use factories and synthetic data for reproducible tests

from faker import Faker
import numpy as np

fake = Faker()
Faker.seed(42)
np.random.seed(42)

def generate_orders(n: int = 1000, anomaly_rate: float = 0.02) -> pd.DataFrame:
    orders = []
    for i in range(n):
        is_anomaly = np.random.random() < anomaly_rate
        orders.append({
            "order_id": i + 1,
            "customer_id": fake.random_int(min=1000, max=9999),
            "amount": round(np.random.lognormal(4, 1), 2) if not is_anomaly else -99.99,
            "status": np.random.choice(["completed", "pending", "refunded"], p=[0.7, 0.2, 0.1]),
        })
    return pd.DataFrame(orders)

See test-data-management.md for:

• Factory patterns for complex nested data
• Data anonymization for production-like test sets
• Fixture file organization (JSON, CSV, Parquet)
• Seeded random generation for reproducibility

Anti-Patterns

Performance

• Use DuckDB for local SQL tests - In-memory analytical queries without a database server
• Scope expensive fixtures to session - Database connections, schema provisioning, large data generation
• Parallelize with pytest-xdist - Run independent test modules across workers with pytest -n auto
• Use Parquet fixtures over CSV - Faster reads, type preservation, smaller files
• Limit integration test data volume - Use representative samples (500-1000 rows) rather than full datasets

source: dbt docs, Great Expectations docs, pytest-sql patterns, Data Contract specification