Testing Data Pipelines: Unit, Integration, Schema Validation, and Contracts

A data pipeline with no tests is a pipeline where the only feedback loop is production. A bug in the transformation logic, a schema change from an upstream source, a NULL propagating through a join that should have filtered it: none of these produce a loud failure. They produce wrong output that reaches dashboards and reports, where the only signal is a stakeholder noticing that a number looks wrong. That is the most expensive feedback loop available.

The testing pyramid for data pipelines is different from the one for application code, but the principle is the same: fast, isolated tests at the bottom of the pyramid cover the most ground at the lowest cost, and slower integration tests at the top confirm that the fast tests are testing the right thing.

Unit tests: transformation logic in isolation

Transformation functions (parsing, cleaning, filtering, joining, aggregating) should be pure functions that take data in and return data out, with no side effects and no dependencies on external systems. A pure transformation function is trivially unit-testable: pass known input, assert known output.

# transforms/revenue.py
def apply_currency_conversion(
    rows: list[dict],
    rates: dict[str, float],
    base_currency: str = "USD"
) -> list[dict]:
    result = []
    for row in rows:
        rate = rates.get(row["currency"])
        if rate is None:
            raise ValueError(f"No rate for currency: {row['currency']}")
        result.append({**row, f"amount_{base_currency.lower()}": round(row["amount"] * rate, 2)})
    return result

# tests/test_revenue.py
import pytest
from transforms.revenue import apply_currency_conversion

RATES = {"BRL": 0.20, "EUR": 1.08}

def test_converts_brl_to_usd():
    rows = [{"order_id": "1", "amount": 500.0, "currency": "BRL"}]
    result = apply_currency_conversion(rows, RATES)
    assert result[0]["amount_usd"] == pytest.approx(100.0)

def test_raises_on_unknown_currency():
    rows = [{"order_id": "2", "amount": 10.0, "currency": "JPY"}]
    with pytest.raises(ValueError, match="No rate for currency: JPY"):
        apply_currency_conversion(rows, RATES)

def test_preserves_original_fields():
    rows = [{"order_id": "3", "amount": 100.0, "currency": "EUR", "status": "completed"}]
    result = apply_currency_conversion(rows, RATES)
    assert result[0]["status"] == "completed"  # no field dropped

These tests run in milliseconds, require no database, no Airflow, and no external service. They are the cheapest tests to write and provide the most immediate feedback when a transformation logic change breaks an expected behavior.

Integration tests: end-to-end on representative samples

Integration tests run the pipeline end-to-end against a representative sample of production data (or a synthetic dataset that covers edge cases) and assert on the output in a test database. They are slower than unit tests and require more infrastructure, but they catch errors that unit tests cannot: join cardinality mistakes, incorrect column references, filter interactions that cancel each other out.

# tests/test_pipeline_integration.py
import pandas as pd
import pytest
from sqlalchemy import create_engine
from pipelines.orders import run_orders_pipeline

@pytest.fixture(scope="module")
def test_db():
    engine = create_engine("postgresql://test:test@localhost:5433/test_db")
    yield engine
    engine.dispose()

def test_orders_pipeline_produces_correct_revenue(test_db):
    # Load test fixture into source table
    fixture = pd.read_parquet("tests/fixtures/orders_sample.parquet")
    fixture.to_sql("raw_orders", test_db, if_exists="replace", index=False)

    # Run the pipeline against the test database
    run_orders_pipeline(conn=test_db, order_date="2024-08-01")

    # Assert on the output
    result = pd.read_sql("SELECT * FROM fct_orders WHERE order_date = '2024-08-01'", test_db)

    assert len(result) == 47  # known count from fixture
    assert result["revenue_usd"].sum() == pytest.approx(12_450.80, rel=1e-3)
    assert result["order_id"].nunique() == len(result)  # no duplicates
    assert result["revenue_usd"].ge(0).all()  # no negative revenue

The fixture is a small, carefully constructed dataset (50 to 200 rows) that covers the specific edge cases the pipeline must handle: orders in multiple currencies, orders with NULL fields, orders in the boundary condition of the date filter. It is committed to the repository alongside the test.

Schema validation: catching upstream changes

The most common failure mode in production data pipelines is an upstream schema change that the pipeline was not designed to handle. A column is renamed, a new required field appears with no default, a numeric column changes type to string. None of these produce a useful error from the pipeline itself: the failure shows up as a type error buried in a stack trace, or as a NULL in a field that should not be NULL, or as wrong output with no error at all.

# Schema contract with Pandera
import pandera as pa
from pandera import Column, DataFrameSchema, Check

orders_schema = DataFrameSchema({
    "order_id": Column(str, nullable=False, unique=True),
    "order_date": Column(pa.DateTime, nullable=False),
    "amount": Column(float, Check.greater_than(0), nullable=False),
    "currency": Column(str, Check.isin(["BRL", "USD", "EUR", "GBP"]), nullable=False),
    "status": Column(str, Check.isin(["completed", "cancelled", "pending"]), nullable=False),
}, strict=True)  # strict=True: fail on unexpected extra columns

def validate_raw_orders(df: pd.DataFrame) -> pd.DataFrame:
    try:
        return orders_schema.validate(df, lazy=True)
    except pa.errors.SchemaErrors as e:
        raise ValueError(
            f"Schema validation failed for raw_orders:\n{e.failure_cases}"
        ) from e

Running validate_raw_orders as the first step of the pipeline converts schema drift from a silent data quality issue into an explicit, immediately visible failure with a specific error message indicating which columns failed and why.

Great Expectations for continuous data quality

Schema validation checks structure. Data quality validation checks content: are the values in the expected range, are there more nulls than expected, did the row count drop precipitously? Great Expectations provides a framework for defining these expectations as code and running them as part of the pipeline.

import great_expectations as gx

context = gx.get_context()

# Define expectations for the orders dataset
suite = context.add_expectation_suite("orders.raw")
batch = context.get_batch({"path": "data/raw/orders_2024-08-01.parquet"}, suite)

batch.expect_column_values_to_not_be_null("order_id")
batch.expect_column_values_to_be_unique("order_id")
batch.expect_column_values_to_be_between("amount", min_value=0.01, max_value=1_000_000)
batch.expect_table_row_count_to_be_between(min_value=1000, max_value=500_000)
batch.expect_column_proportion_of_unique_values_to_be_between(
    "currency", min_value=0.001, max_value=0.01  # 4-5 distinct currencies from ~50k orders
)

results = context.run_validation_operator(
    "action_list_operator",
    assets_to_validate=[batch]
)

if not results["success"]:
    raise RuntimeError("Data quality validation failed. See Great Expectations Data Docs.")

The full test coverage picture

All four layers run in CI on every pull request. Unit and schema tests run on every commit. Integration tests run on the merge to main. Data quality expectations run in production as part of each pipeline execution. Together they close the gap between "the code is correct" and "the pipeline is producing correct data."