Data Pipeline Testing — Catching Bugs Before They Hit Production

Software engineers have spent decades perfecting testing practices. Data engineers inherited almost none of it. The reason is simple — data pipeline bugs are different. A software bug crashes the application. A data bug silently corrupts a dashboard, and nobody notices until the CEO asks why revenue dropped 40% (it did not — a JOIN duplicated rows).

Testing data pipelines requires different tools, different strategies, and a different mindset than testing application code.

Why data testing is different

In software testing, you control the inputs. You write a unit test, pass in known values, and assert the output. The function is deterministic — same input, same output, every time.

Data pipelines break this model in several ways:

The data itself is the variable. Your transformation logic might be perfect, but if the source system starts sending null customer IDs or changes a column type from integer to string, your pipeline breaks. You cannot unit test against every possible data anomaly.

Scale makes testing expensive. You cannot run your full pipeline against production data in CI. A pipeline that processes 500 million rows takes hours — you need strategies for testing with representative subsets.

Correctness is domain-specific. A software test checks “does the function return the right value?” A data test checks “does this revenue number match what finance expects?” That requires business context no test framework can provide automatically.

Failures are silent. A broken API returns a 500 error. A broken pipeline returns wrong numbers that look plausible. The dashboard still loads. Nobody complains until the quarterly board meeting.

The data testing pyramid

Like the traditional testing pyramid, data testing works in layers. More tests at the bottom (fast, cheap, focused), fewer at the top (slow, expensive, broad).

        ╱ ╲
       ╱ E2E ╲         End-to-end pipeline validation
      ╱────────╲        (slowest, most expensive)
     ╱Integration╲     Cross-stage consistency checks
    ╱──────────────╲    (moderate speed and cost)
   ╱   Unit Tests    ╲  Transformation logic tests
  ╱────────────────────╲ (fastest, cheapest)
 ╱   Data Quality Gates  ╲  Schema, freshness, volume
╱──────────────────────────╲ (continuous, automated)

Layer 1: Data quality gates

These run continuously, often before transformations even start. They catch upstream problems early.

def validate_source_data(df, table_name):
    """Run before any transformation. Fail fast on bad input."""
    checks = {
        'not_empty': len(df) > 0,
        'schema_match': set(df.columns) == EXPECTED_SCHEMAS[table_name],
        'freshness': df['_loaded_at'].max() > datetime.now() - timedelta(hours=24),
        'no_full_nulls': not df.isnull().all(axis=0).any(),
    }

    failures = [name for name, passed in checks.items() if not passed]
    if failures:
        raise DataQualityError(
            f"{table_name} failed quality gates: {', '.join(failures)}"
        )

Layer 2: Unit tests

Test individual transformation functions in isolation, with controlled inputs and expected outputs.

Layer 3: Integration tests

Verify that stages connect correctly — the output schema of stage A matches the expected input of stage B.

Layer 4: End-to-end tests

Run the full pipeline on a known dataset and validate the final output against expected results. These are slow but catch emergent bugs that unit tests miss.

Unit testing transformation logic

Testing Python transformations

Python transformations are the easiest to unit test. Isolate your logic into pure functions, then test with pytest:

# transformations.py
import pandas as pd

def calculate_customer_lifetime_value(orders_df):
    """Calculate CLV as total revenue minus refunds per customer."""
    return (
        orders_df
        .groupby('customer_id')
        .agg(
            total_orders=('order_id', 'count'),
            total_revenue=('amount', lambda x: x[orders_df.loc[x.index, 'status'] != 'refunded'].sum()),
            first_order=('ordered_at', 'min'),
            last_order=('ordered_at', 'max'),
        )
        .assign(
            clv=lambda df: df['total_revenue'],
            tenure_days=lambda df: (df['last_order'] - df['first_order']).dt.days
        )
        .reset_index()
    )

# tests/test_transformations.py
import pandas as pd
import pytest
from transformations import calculate_customer_lifetime_value

@pytest.fixture
def sample_orders():
    return pd.DataFrame({
        'order_id': [1, 2, 3, 4],
        'customer_id': ['A', 'A', 'B', 'B'],
        'amount': [100, 200, 150, 50],
        'status': ['completed', 'completed', 'completed', 'refunded'],
        'ordered_at': pd.to_datetime([
            '2024-01-01', '2024-06-01', '2024-03-01', '2024-03-15'
        ])
    })

def test_clv_excludes_refunds(sample_orders):
    result = calculate_customer_lifetime_value(sample_orders)
    customer_b = result[result['customer_id'] == 'B']
    assert customer_b['total_revenue'].values[0] == 150  # Excludes $50 refund

def test_clv_counts_all_orders(sample_orders):
    result = calculate_customer_lifetime_value(sample_orders)
    customer_a = result[result['customer_id'] == 'A']
    assert customer_a['total_orders'].values[0] == 2

def test_tenure_calculation(sample_orders):
    result = calculate_customer_lifetime_value(sample_orders)
    customer_a = result[result['customer_id'] == 'A']
    assert customer_a['tenure_days'].values[0] == 152  # Jan 1 to Jun 1

The key insight: extract transformation logic into pure functions that take DataFrames in and return DataFrames out. No database connections, no file I/O, no side effects. This makes them trivially testable.

Testing SQL transformations

SQL is harder to unit test because it runs inside a database. You have three strategies:

Strategy 1: dbt unit tests (dbt 1.8+)

dbt now supports native unit testing. You define mock inputs and expected outputs:

# models/marts/_marts.yml
unit_tests:
  - name: test_order_total_calculation
    model: fct_orders
    given:
      - input: ref('stg_orders')
        rows:
          - { order_id: 1, subtotal: 100, tax: 8, discount: 10 }
          - { order_id: 2, subtotal: 200, tax: 16, discount: 0 }
      - input: ref('stg_payments')
        rows:
          - { order_id: 1, payment_status: 'paid' }
          - { order_id: 2, payment_status: 'paid' }
    expect:
      rows:
        - { order_id: 1, total_amount: 98 }
        - { order_id: 2, total_amount: 216 }

Strategy 2: SQL transpilation with sqlglot

Parse SQL and validate it without a database:

import sqlglot

def test_model_uses_ref_not_hardcoded_tables():
    """Ensure models use ref() instead of hardcoded table names."""
    with open('models/marts/fct_orders.sql') as f:
        sql = f.read()

    # After Jinja rendering, check for hardcoded schema references
    parsed = sqlglot.parse(sql)
    for statement in parsed:
        for table in statement.find_all(sqlglot.exp.Table):
            assert table.db is None or table.db.startswith('{{'), \
                f"Hardcoded table reference found: {table}"

Strategy 3: Test against a development database

Run your dbt models in a CI-specific schema with a subset of data:

# CI environment: build models in a temporary schema
dbt run --target ci --select my_model+
dbt test --target ci --select my_model+

Data quality checks in depth

Quality checks validate the data itself, not just the transformation logic.

Freshness checks

def check_freshness(table, column, max_age_hours):
    """Alert if data is stale."""
    query = f"""
    SELECT MAX({column}) as latest,
           CURRENT_TIMESTAMP - MAX({column}) as age
    FROM {table}
    """
    result = execute_query(query)
    age_hours = result['age'].total_seconds() / 3600

    if age_hours > max_age_hours:
        raise StaleDataError(
            f"{table} is {age_hours:.1f}h old (threshold: {max_age_hours}h)"
        )

Volume checks

def check_volume(table, expected_min, expected_max=None, partition_column=None):
    """Detect abnormal row counts — either data loss or duplication."""
    if partition_column:
        query = f"""
        SELECT COUNT(*) as row_count
        FROM {table}
        WHERE {partition_column} = CURRENT_DATE - 1
        """
    else:
        query = f"SELECT COUNT(*) as row_count FROM {table}"

    count = execute_query(query)['row_count']

    if count < expected_min:
        raise VolumeError(
            f"{table}: {count} rows (expected >= {expected_min}). "
            "Possible data loss or source outage."
        )
    if expected_max and count > expected_max:
        raise VolumeError(
            f"{table}: {count} rows (expected <= {expected_max}). "
            "Possible duplication."
        )

Distribution checks

The most sophisticated quality check — detect when data drifts statistically:

import numpy as np
from scipy import stats

def check_distribution(current_values, historical_values, column_name,
                       threshold=0.05):
    """Use KS test to detect distribution drift."""
    statistic, p_value = stats.ks_2samp(current_values, historical_values)

    if p_value < threshold:
        raise DistributionDriftError(
            f"{column_name}: distribution changed significantly "
            f"(KS statistic={statistic:.4f}, p-value={p_value:.4f}). "
            f"Current mean={np.mean(current_values):.2f} vs "
            f"historical mean={np.mean(historical_values):.2f}"
        )

This catches subtle bugs that row-level checks miss — like a currency conversion error that shifts all amounts by 10%.

Contract testing between pipeline stages

Contract testing ensures that when stage A changes its output, it does not break stage B. Think of it like API contracts between microservices, but for data.

# contracts/order_pipeline_contract.py
from dataclasses import dataclass
from typing import Set

@dataclass
class StageContract:
    required_columns: Set[str]
    column_types: dict
    non_nullable: Set[str]
    primary_key: str

STAGING_TO_MARTS_CONTRACT = StageContract(
    required_columns={
        'order_id', 'customer_id', 'ordered_at',
        'total_amount_dollars', 'status'
    },
    column_types={
        'order_id': 'int64',
        'customer_id': 'object',
        'total_amount_dollars': 'float64',
    },
    non_nullable={'order_id', 'customer_id', 'total_amount_dollars'},
    primary_key='order_id',
)

def validate_contract(df, contract: StageContract, stage_name: str):
    """Validate DataFrame against a stage contract."""
    errors = []

    # Check required columns
    missing = contract.required_columns - set(df.columns)
    if missing:
        errors.append(f"Missing columns: {missing}")

    # Check types
    for col, expected_type in contract.column_types.items():
        if col in df.columns and str(df[col].dtype) != expected_type:
            errors.append(
                f"{col}: expected {expected_type}, got {df[col].dtype}"
            )

    # Check nullability
    for col in contract.non_nullable:
        if col in df.columns and df[col].isnull().any():
            null_count = df[col].isnull().sum()
            errors.append(f"{col}: {null_count} null values (not allowed)")

    # Check primary key uniqueness
    if contract.primary_key in df.columns:
        dupes = df[contract.primary_key].duplicated().sum()
        if dupes > 0:
            errors.append(f"Primary key {contract.primary_key}: {dupes} duplicates")

    if errors:
        raise ContractViolation(
            f"Stage '{stage_name}' contract violated:\n" +
            "\n".join(f"  - {e}" for e in errors)
        )

In dbt, contract testing is built-in with model contracts:

models:
  - name: fct_orders
    config:
      contract:
        enforced: true
    columns:
      - name: order_id
        data_type: int
      - name: total_amount_dollars
        data_type: float

When contract.enforced: true, dbt will fail the build if the model output does not match the declared schema. This prevents accidental breaking changes.

Testing idempotency

An idempotent pipeline produces the same result whether you run it once or ten times. This is critical for reliability — you need to safely rerun failed pipelines without duplicating data.

def test_pipeline_idempotency(pipeline, test_data):
    """Run the pipeline twice and verify identical output."""
    # First run
    pipeline.run(test_data, target_table='test_output')
    result_1 = read_table('test_output')

    # Second run — same input
    pipeline.run(test_data, target_table='test_output')
    result_2 = read_table('test_output')

    # Results must be identical
    assert len(result_1) == len(result_2), \
        f"Row count changed: {len(result_1)} → {len(result_2)} (duplicates?)"

    pd.testing.assert_frame_equal(
        result_1.sort_values('id').reset_index(drop=True),
        result_2.sort_values('id').reset_index(drop=True),
    )

Common idempotency patterns:

-- Pattern 1: DELETE + INSERT (atomic replacement)
DELETE FROM target_table WHERE partition_date = '2024-01-15';
INSERT INTO target_table SELECT * FROM staging WHERE date = '2024-01-15';

-- Pattern 2: MERGE (upsert)
MERGE INTO target_table t
USING staging_table s ON t.id = s.id
WHEN MATCHED THEN UPDATE SET t.amount = s.amount, t.updated_at = s.updated_at
WHEN NOT MATCHED THEN INSERT (id, amount, updated_at) VALUES (s.id, s.amount, s.updated_at);

Automating tests in CI/CD

GitHub Actions for dbt

# .github/workflows/dbt-ci.yml
name: dbt CI
on:
  pull_request:
    paths: ['models/**', 'macros/**', 'tests/**']

jobs:
  dbt-test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'

      - name: Install dependencies
        run: pip install dbt-snowflake

      - name: dbt deps
        run: dbt deps

      - name: Lint SQL
        run: sqlfluff lint models/ --dialect snowflake

      - name: dbt compile
        run: dbt compile --target ci

      - name: dbt run (modified models only)
        run: dbt run --target ci --select state:modified+
        env:
          DBT_STATE: ./target

      - name: dbt test
        run: dbt test --target ci --select state:modified+

The state:modified+ selector is key — it only builds and tests models changed in the PR, plus their downstream dependencies. This keeps CI fast even in large projects.

Real-world example: testing a revenue pipeline

Here is a complete test suite for a revenue pipeline:

# tests/test_revenue_pipeline.py
import pytest

class TestRevenuePipeline:
    """Comprehensive tests for the revenue calculation pipeline."""

    def test_revenue_matches_payments(self, db):
        """Total revenue must equal sum of successful payments."""
        revenue = db.query(
            "SELECT SUM(revenue) FROM fct_revenue WHERE period = '2024-Q1'"
        )
        payments = db.query(
            "SELECT SUM(amount) FROM stg_payments "
            "WHERE status = 'success' AND period = '2024-Q1'"
        )
        assert abs(revenue - payments) < 0.01, \
            f"Revenue ({revenue}) != Payments ({payments})"

    def test_no_future_revenue(self, db):
        """No revenue should be recorded for future dates."""
        future = db.query(
            "SELECT COUNT(*) FROM fct_revenue WHERE date > CURRENT_DATE"
        )
        assert future == 0

    def test_currency_conversion_applied(self, db):
        """Non-USD orders should be converted to USD."""
        unconverted = db.query("""
            SELECT COUNT(*) FROM fct_revenue r
            JOIN stg_orders o ON r.order_id = o.order_id
            WHERE o.currency != 'USD'
              AND r.revenue_usd = o.amount
        """)
        assert unconverted == 0, \
            f"{unconverted} orders not currency-converted"

    def test_refunds_reduce_revenue(self, db):
        """Refunded orders should have negative or zero revenue."""
        bad_refunds = db.query("""
            SELECT COUNT(*) FROM fct_revenue
            WHERE order_status = 'refunded' AND revenue_usd > 0
        """)
        assert bad_refunds == 0

Next steps

• dbt Fundamentals — learn the tool that powers most of these testing patterns.
• Pipeline CI/CD — automate these tests in your deployment workflow.
• Data Quality & Governance — the broader quality framework.
• Data Observability — continuous monitoring beyond point-in-time tests.