ETL / ELT
Intermediate 13 min read ETL / ELT

Pipeline Testing

Test the transform, and test the data it emits

Data pipelines deserve tests just like code: unit tests on transform logic, fixture-based integration tests, and data tests on outputs (uniqueness, grain, freshness). The cheap wins are the ones you run in CI before deploy.

unit testintegration testfixturedata testCIregression
Prerequisites: dbt for Transformation Schema Evolution
Before we start

What you will learn

  • Write unit tests for transformation functions
  • Build fixture-driven integration tests for models
  • Add data tests (uniqueness, grain, freshness) in the pipeline
Introduction

What is this?

Pipeline testing spans three levels: unit tests on pure transform logic (functions, macros, SQL expressions), integration tests that run models against small fixture tables, and data tests that assert on real output (uniqueness of grain keys, not-null, row counts, freshness). Most value comes from fast tests in CI plus data assertions at deploy time.

Motivation

Why do you need it?

  • Silent schema drifts and grain violations are data bugs; tests catch them before the dashboard does.
  • Tests encode the contract: 'what does correct look like' becomes executable.
  • CI-friendliness is the difference between tests that run and tests that rot.
Think about it

Mental model

Imagine

A pilot's pre-flight checklist

Think of

Pipeline tests

You don't 'hope' the straps and fuel are right — you check them before takeoff (unit/integration in CI) and verify the readings in the air (data tests on outputs).

Mechanics

How it works

  • Unit: pure functions/macros take inputs -> expected outputs; run in seconds with pytest/JS test harness.
  • Integration: a fixture (small CSV/SQL dataset) runs the real model; assert the exact output rows.
  • Data tests: uniqueness, not_null, relationships, accepted_values, custom SQL run against built outputs.
  • Freshness tests: assert max(updated_at) is recent, per table, so staleness alerts before consumers do.
  • CI wiring: run unit + fixture integration on each PR; run data tests on deploy stage and on schedule.

Three test layers and where they run.

Unit tests
Integration tests
Data tests
CI + schedule
Reference

Key commands

pytest tests/ -q

Run the unit+fixture suite fast in CI.

dbt test --select fct_order_lines

Run every declared data test on that model.

dbt build

Run models then tests in dependency order, failing fast per node.

dbt source freshness

Assert sources are fresh within the SLA (staleness test).

Hands-on

Examples

A unit test for a transform helper

A unit test for a transform helper
# transform.py
def normalize_amount(raw):
    return round(Decimal(str(raw)), 2)
 
# test_transform.py (pytest)
import pytest
def test_normalize_amount_handles_strings():
    assert normalize_amount("10.005") == Decimal("10.01")
 
def test_normalize_amount_rejects_none():
    with pytest.raises(TypeError):
        normalize_amount(None)

Fast, deterministic unit tests encode the contract of the transform before it ever touches a warehouse.

Real world

In production

  • A PR that adds a column triggers fixture integration: one test proves the new column flows through.
  • Grain tests log 'dup order_line_key' before the number reaches a dashboard.
  • Freshness tests on raw sources alert an hour before consumers do, pointing at the real gap.
Guidelines

Best practices

  • Keep unit tests fast and pure; never touch real warehouses in unit scope.
  • Data-test every grain key (unique + not_null) and every partition boundary you model.
  • Run tests in CI on PR and again on deploy; schedule a daily full test pass too.
  • Treat a failing test as a contract breach: fix or formally change the contract, not the test.
Watch out

Common mistakes

  • Testing only happy paths, so nulls and type drift sail through.
  • Data tests on outputs but zero fixture tests, so logic regressions still pass.
  • Deleting a failing test to unblock a deploy without a root-cause follow-up.
When things break

Debugging tips

  • A transform slot changing under test = the fixture drift; regenerating fixtures prints the intended shift.
  • When a data test fails in prod but not CI, diff the environment (schema drift, permissions, catalog pointer).
  • Keep a 'golden dataset' of expected outputs; a regression prints exactly which row changed.
Interviews

Interview checkpoint

Test yourself

Quick quiz

0/2
1

Which test catches a filter that quietly drops half the rows?

2

Uniqueness and not-null tests on grain keys are data tests best run on every deploy.

Go deeper

Official documentation

dbt data tests
docs
pytest docs
docs
Great Expectations docs
docs