08 - Data Pipeline Testing Strategy

21 min read

Data Pipeline Testing Strategy

Quote

“Data quality is not a technical problem. It is a trust problem — one bad number erodes months of credibility.”

— Chad Sanderson

“Write tests until fear is transformed into boredom.”

— Kent Beck, Test-Driven Development (2002)

The Data Engineering Testing Pyramid

block-beta
    columns 7
    space:2 E2E["E2E<br/>Nightly"]:3 space:2
    space:1 INT["Integration<br/>On merge"]:5 space:1
    CONTRACT["Contract + Quality<br/>Every PR and every load"]:7
    UNIT["Unit Tests<br/>Every PR"]:7

    style E2E fill:#cc4125,stroke:#cc4125,color:#fff
    style INT fill:#e8b84d,stroke:#e8b84d,color:#1a1a2e
    style CONTRACT fill:#4285f4,stroke:#4285f4,color:#fff
    style UNIT fill:#34a853,stroke:#34a853,color:#fff

The pyramid reads bottom-to-top: the base (unit tests) runs the most tests at the lowest cost, while the peak (E2E) runs the fewest tests at the highest cost. Each layer catches a different class of failure that the layers below cannot.

Where to Invest First

Most pipeline teams have zero tests. If you’re starting from nothing, add in this order: (1) dbt generic tests on gold models (unique, not_null on key columns), (2) row count assertions after every load, (3) pytest for transform functions. These three catch 80% of production data issues.

Pyramid Layers — Detailed Breakdown

Unit Tests — transform logic

Unit tests validate individual functions, SQL transforms, and dbt models in isolation. They are the fastest, cheapest, and most numerous tests in the pyramid.

How Unit Tests Work

Feed a known input DataFrame or SQL result into a transform function and assert the output matches expected values. No database, no network, no external dependencies. Pure logic verification.

Skip Unit Tests And...

Broken transforms reach production. A z-score function that divides by the wrong column, a deduplication query that keeps the wrong row, an aggregation that double-counts — all of these produce silently wrong data that the dashboard displays with full confidence. Nobody notices until a business user challenges a number weeks later.

Add Unit Tests for Every Transform Function

For every transform function, write at least one pytest test with a hand-crafted DataFrame fixture that includes known inputs and expected outputs you can verify manually. For dbt models, declare unique, not_null, and accepted_values tests in schema.yml. These run in seconds with no infrastructure and catch logic errors before any data moves.

Data Quality Assertions — data properties

Quality assertions validate data properties at layer boundaries: nulls, counts, ranges, uniqueness, and freshness. They run after every pipeline execution, not just in CI.

How Quality Assertions Work

After each pipeline stage writes its output, run a set of checks against the result:

  • Row count > 0 (table not empty after load)
  • Row count ± 20% of yesterday (no explosion from bad join, no loss from filter bug)
  • Null percentage < threshold per column
  • Value ranges (no negative prices, no future dates)
  • Uniqueness on business keys (no duplicates from failed dedup)
  • Freshness (latest date within SLA window)

Tools: dbt generic tests, dbt-expectations, custom Python/SQL assertions. See data-quality-framework > Data Quality Dimensions. When: every pipeline run in production + every PR in CI.

Skip Quality Assertions And...

Silent data degradation. A source API starts returning NULLs for a field — your pipeline loads them without error. Over weeks, 50% of your gold table is NULL. Nobody notices because there’s no row count drop, no schema change, no error in the logs. The dashboard just gradually becomes wrong.

Assert Data Properties After Every Pipeline Stage

After each layer writes its output, run: row count > 0, row count within ±20% of the prior load, null percentage below threshold per required column, and uniqueness on the business key. Use dbt not_null + dbt-expectations expect_column_proportion_of_unique_values_to_be_between in CI, and the same checks as Python assertions in production after every Airflow task.

Contract Tests — schema conformance

Contract tests verify that source data matches the expected schema before any transform runs. They are the only defense against upstream changes.

How Contract Tests Work

Define the expected schema (column names, types, nullability, value ranges) in a contract file or dbt source YAML. On every ingestion, validate the incoming data against the contract. If a column is missing, renamed, or has a new type, the test fails before the data enters bronze.

Skip Contract Tests And...

An upstream API renames price to current_price. Your pipeline loads NULLs into the price column — every row, every day. The pipeline reports success, row counts match, no errors in logs. Nobody notices until a business user asks why the dashboard shows zero for everything. This is the #1 silent pipeline killer.

Validate Source Schema on Every Ingestion

Define the expected schema in a contract YAML or dbt source YAML. At the start of every ingestion job, compare the actual incoming column names and types against the contract using a Pydantic model or a custom Python validator. Fail fast and quarantine the load if any column is missing, renamed, or has a type mismatch — before a single row enters bronze.

Integration Tests — cross-system flow

Integration tests run the pipeline end-to-end with real connections but test data. They catch the “works in dev, breaks in prod” failures that unit tests cannot.

How Integration Tests Work

Spin up a real database (Docker SQL Server in GitHub Actions), load test fixtures (100-1000 rows of known data), run the full bronze → silver → gold flow, and assert the output shape and values.

Skip Integration Tests And...

“Works in dev, breaks in prod” — the #1 pipeline failure mode. Your SQL runs perfectly against your local Docker database but fails on the production SQL Server because of a different collation, a missing index, or a permission issue. Mocking the database doesn’t test your SQL against a real query engine.

Spin Up Real Infrastructure in CI

In GitHub Actions, use a mcr.microsoft.com/mssql/server Docker container as the test SQL Server. Seed it with 100–1000 rows of fixture data, run the full bronze → silver → gold transform pipeline against it, and assert output shape and row counts. This catches collation mismatches, missing indexes, and permission issues on every merge to main, not on first production deploy.

E2E Pipeline Validation — full flow

E2E tests validate the complete pipeline from data fetch through gold output, including orchestration, quality gates, and export. They catch multi-step regressions that no single-layer test can detect.

How E2E Tests Work

Run the full pipeline (or a representative subset) on a fixed test dataset. Compare the gold output against a “golden file” — a known-correct reference output. Flag any difference above a tolerance threshold (exact match for integers, ±0.01 for floats).

  • Use a dedicated test database/dataset with seeded fixture data
  • Run nightly or weekly — too slow and expensive for every PR
  • Compare: pd.testing.assert_frame_equal() with atol for numeric tolerance, or dbt custom tests against a seeded reference table

Skip E2E Tests And...

Multi-step regressions go undetected. A subtle change in a silver transform produces correct-looking silver data, but when gold aggregates it, the composite scores shift by 5%. Unit tests pass (the function works). Quality tests pass (no nulls, no duplicates). Only the full end-to-end comparison catches the drift — because it checks the final answer, not intermediate steps.

Maintain a Golden File for E2E Comparison

Save the known-correct gold output for a representative test dataset as a Parquet or CSV fixture (“golden file”). After every nightly pipeline run against the test dataset, compare actual output against the golden file using pd.testing.assert_frame_equal() with atol=0.01 for numeric columns. Any difference beyond the tolerance triggers a review — even if all lower-layer tests passed.

What to Test at Each Pipeline Layer

This table maps testing to the medallion architecture — what specific checks apply at each data layer, which tool runs them, and where the implementation lives in the vault.

Pipeline LayerWhat to TestPrimary ToolVault Reference
BronzeSchema matches source, row count > 0, no all-NULL columns, source freshnessdbt source tests, Python assertionsdata-quality-framework > Medallion Bronze Quality Gate — Landing / Raw
SilverBusiness logic correctness, dedup worked, SCD2 integrity, gap-fill completenessdbt tests, pytest with fixturesdbt-testing-framework > dbt Built-in Generic Tests
GoldOutput shape matches expectations, z-scores in bounds, no NaN in scores, ranks are contiguousdbt tests, pandas assertionsdata-quality-framework > Medallion Gold Quality Gate — Consumption / Publication
Cross-layerRow count preservation (bronze → silver minus expected dedup), referential integritydbt cross-model tests, SQL assertionsdata-quality-framework > Data Quality Quarantine Pattern

Gold Quality Is Last Defense

Gold tables feed dashboards, APIs, and downstream consumers. A quality bug in gold is visible to the business. Every gold model must have at minimum: unique + not_null on the primary key, accepted_values on categorical columns, and a row count assertion comparing today vs yesterday (detect data loss or explosion).

Mandatory Gold Quality Gates

For every gold model, declare in schema.yml: unique + not_null on the primary key, accepted_values on all categorical columns, and a dbt-expectations row count test comparing today’s result against yesterday’s count within a ±20% band. Set all gold-layer tests to severity: error (not warn) so any failure halts publication before bad data reaches clients.

Test Types — Detailed Breakdown

Unit tests — transform logic

Test individual functions, SQL transforms, and dbt models in isolation. These are the fastest tests and should make up the majority of your test suite.

Python (pytest): Test transform functions with sample DataFrames as fixtures. See 10_py_testing_migration for Pandas/Polars testing patterns.

C# (xUnit): Test transform methods with test DataFrames. See 10_cs_testing_migration for Deedle/Polars.NET patterns.

dbt: Schema tests in schema.yml — unique, not_null, accepted_values, relationships. See dbt-testing-framework > dbt Built-in Generic Tests for the full list.

The unit test rule

Every transform function must have at least one test that verifies correct output for known input. If a function calculates a z-score, test it with a known dataset where you can verify the expected z-score by hand. If a function deduplicates, test it with a dataset containing known duplicates and verify the count drops correctly.

Data quality assertions

Validate data properties at layer boundaries. These run after every pipeline execution, not just in CI.

Checks to implement:

CheckWhat It CatchesTool
Row count > 0Empty table after failed loaddbt dbt_utils.at_least_one, SQL assertion
Row count ± 20% of yesterdayData explosion from bad join or data loss from filter bugCustom dbt test, Python assertion
Null percentage < thresholdSilent schema change producing NULLsdbt not_null, custom threshold test
Value rangesOut-of-bounds scores, negative prices, future datesdbt accepted_range, custom SQL
Uniqueness on business keyDuplicates from failed dedup or bad mergedbt unique, unique_combination_of_columns
FreshnessStale data — source hasn’t updateddbt source freshness, custom metric

See data-quality-framework > Data Quality Dimensions for the full quality taxonomy and gcp-pipeline-health-and-sla > Row Count Validation for production monitoring integration.

Quality Checks Need Production Too

Data quality degrades between deploys — source schemas change, APIs return different data, volumes shift. dbt tests in CI catch code bugs. Production quality checks catch data bugs. You need both.

Run Quality Checks in Both CI and Production

In CI: run dbt test --select state:modified+ on every PR to catch code-level regressions. In production: after every Airflow pipeline run, execute the same row count, null percentage, and freshness assertions as Python pipeline steps (not just dbt). Wire failures to PagerDuty or a Slack alert channel so production data bugs surface immediately, not when a business user notices.

Contract tests — schema conformance

Verify that source data matches the expected schema before any transform runs.

The problem: An upstream API changes a field name from price to current_price. Your pipeline loads NULLs into the price column — every row, every day — until someone notices the dashboard is wrong.

Implementation:

Schema Changes Kill Silently

Code bugs raise exceptions. Data volume changes are visible in monitoring. But a renamed column silently loads NULLs — the pipeline reports success, row counts match, and nobody notices until a business user asks why the dashboard shows zero. Contract tests are the only defense against this.

Detect Schema Drift at Ingestion Time

At the top of every ingestion function, validate the incoming payload’s column names against the defined contract using a Pydantic model or an explicit column-presence check. If any expected column is absent or any unexpected column appears under a different name, raise a SchemaDriftError, quarantine the load, and alert immediately. Do not let schema-drifted data silently populate downstream tables.

Integration tests

Test the pipeline end-to-end with real connections but test data.

Scope: Full bronze → silver → gold flow with a small dataset (100-1000 rows).

Pattern: Dedicated test database/dataset, seeded with fixture data, torn down after each test run.

GitHub Actions implementation: Spin up a SQL Server Docker container, load test fixtures, run the pipeline, assert output shape and values.

See github-actions-data-engineering > Full Python Lint + Test Workflow for the CI workflow and github-actions-data-engineering > dbt Build Against Dev Schema for dbt integration tests in CI.

Integration Tests Need Real Infra

Mocking a database doesn’t test your SQL. Mocking GCS doesn’t test your upload logic. Integration tests must use real connections — even if that’s a Docker SQL Server in GitHub Actions. The cost is minutes of CI time; the payoff is catching “works in dev, breaks in prod” before it reaches prod.

Use Docker Services in GitHub Actions for Real Connections

In .github/workflows, declare a services block with mcr.microsoft.com/mssql/server as the SQL Server container. Run bcp or pyodbc against localhost,1433 in the test job — these are real connections against a real query engine. For GCS, use a dedicated test-* bucket in the CI service account with lifecycle rules to auto-delete objects after 1 day.

Regression tests — snapshot comparison

Verify today’s output matches yesterday’s expected output after a code change.

The problem: A code change alters the z-score calculation subtly. Unit tests pass (they test the new formula). But production output is different from what downstream consumers expect — scores shift, ranks change, dashboards look wrong.

Implementation:

  • Save expected output as a “golden file” (CSV/Parquet fixture)
  • After pipeline run, compare actual output against golden file
  • Flag differences above a threshold (exact match for integers, ±0.01 for floats)

dbt: Custom test comparing current model output against a seeded reference table.

Python: pytest with golden file fixtures — pd.testing.assert_frame_equal() with atol for numeric tolerance.

Test Data Management

ApproachWhen to UseGotcha
Hand-crafted fixturesUnit tests — known edge cases (NULLs, duplicates, boundary dates)Time-consuming to maintain as schema evolves
Sampled from productionIntegration tests — realistic data shape and volumeMust anonymize PII; may miss edge cases
Synthetic generationLoad testing, edge case coverageMay not reflect real data distributions

Test Data Edge Case Rule

Test data must include at least one example of every edge case your code handles: NULL values, duplicate keys, missing dates, out-of-range values, Unicode characters, empty strings, and zero-length files. If your code has a try/except for a specific condition, your test data must trigger that condition.

Production Data Only Is Insufficient

Production data tests the happy path — the data your pipeline already handles correctly. It never tests the edge cases that will crash your pipeline when they first appear. Always supplement production samples with hand-crafted edge case fixtures.

Build a Fixture Library Covering All Edge Cases

Maintain a tests/fixtures/ directory with hand-crafted CSV/Parquet files that explicitly cover: rows with NULLs in every nullable column, duplicate keys before deduplication, dates at boundary values (min date, max date, future date), out-of-range numeric values, empty strings vs NULL, and zero-length files. Update fixtures whenever a new try/except branch or business rule check is added to the code.

CI/CD Test Automation

The complete testing pipeline on every PR and merge. For GitHub Actions workflow syntax, see github-actions-data-engineering > End-to-End Pipeline: PR to Lint to Test to Build to Deploy to Verify.

# Conceptual workflow — link to implementation pages for full YAML
on:
  pull_request:
    branches: [main]
 
jobs:
  lint-and-type-check:     # ruff, mypy, sqlfluff
  unit-tests:              # pytest -m "not integration"
  dbt-tests:               # dbt test --select state:modified+
  schema-validation:       # contract tests against source schemas
  # --- runs only on merge to main ---
  integration-test:        # Docker SQL Server + test data + pipeline run
  data-quality:            # assertions on integration test output

When each test type runs

Test TypeOn Every PROn MergeNightlyOn Every Pipeline Run
Unit testsYesYes——
dbt generic testsYesYes—Yes (in production)
Contract testsYesYes—Yes (on ingestion)
Integration tests—Yes——
Regression tests——Yes—
Data quality assertions———Yes

Test Timing Cost-Benefit

Unit tests and contract tests are cheap (seconds, no infrastructure) — run on every PR. Integration tests need infrastructure (Docker containers, real connections) — run on merge only. Regression tests compare large datasets — run nightly to avoid slowing CI. Data quality assertions run in production after every load — they catch data problems that code tests can’t.

Production Monitoring Is Not Testing

ConcernTesting (pre-deployment)Monitoring (post-deployment)
QuestionDoes the code work correctly?Is the pipeline healthy in production?
WhenBefore code reaches productionAfter every production run
Toolpytest, dbt test, GitHub ActionsDatadog, Cloud Monitoring, custom metrics
CatchesLogic bugs, schema violations, regressionData drift, volume changes, latency, failures

Data quality checks are the exception — they span both worlds. Run them in CI (testing) AND after every production load (monitoring).

Anti-Patterns

Pipeline Testing Anti-Patterns

Anti-PatternConsequenceFix
No tests at all (“it worked in my notebook”)Every deploy is a gambleStart with dbt not_null + unique on gold PKs
Testing only the happy pathFirst NULL crashes productionAdd edge case fixtures: NULLs, duplicates, empty, boundary dates
Testing transforms but not schemasCode correct, but input changed — garbage in, garbage outAdd contract tests on every source
Tests only in CI, never in productionData quality degrades silently between deploysRun dbt tests + row count checks after every production load
Production data as only test dataNever tests edge casesSupplement with synthetic fixtures
Testing by visual inspection”I looked at it and it seems right” — not reproducibleAutomated assertions with explicit expected values
dbt tests without thresholdsTest passes with 99% NULLs (technically not ALL NULL)Use dbt-expectations with accepted_range and percentage thresholds
Mocking the databaseSQL logic never tested against a real engineUse Docker SQL Server in CI for integration tests

Minimum Viable Test Suite for a New Pipeline

If starting from zero: (1) add unique + not_null dbt tests on every gold model primary key; (2) add a row count assertion comparing today vs yesterday as an Airflow task after every load; (3) write pytest unit tests for every transform function with hand-crafted fixtures covering NULLs and duplicates; (4) add a contract schema check at ingestion. These four steps cover the most common production failures and take less than a day to implement.

Graph View

Backlinks