🐍 Python for Data Engineers · Chapter 13

Testing pipelines with pytest

Application bugs return a 500. Data bugs return a plausible wrong number that nobody questions for three weeks.

Study this lesson free in the app →

The basics

Testing a pipeline means two different jobs: proving your code transforms data correctly, and proving the data your pipeline produced is plausible every time it runs.

The problem it solves: an application bug is loud — someone gets a 500. A data bug is quiet: the dashboard loads, the number looks like a number, and it is wrong for three weeks before finance notices it is 22% below the payment provider. Code tests alone cannot catch this, because often the code is fine and the world changed — a new status value, a reformatted fare column.

The mental model: a bottling plant. Inspect the machine before the shift with a known bottle (unit test, in CI). Inspect the line continuously during production (data quality check, on every run). Neither replaces the other.

Explained

A broken API is loud: someone gets an error, someone pages you.

A broken pipeline is quiet. The dashboard still loads. The number is still a number. It is just wrong — and by the time anyone notices, three weeks of decisions have been made on it.

That asymmetry is why data engineers test differently: you test the code, and you also test the data the code produced.

Two inspections at the factory

A bottling plant checks two different things.

The machine inspection happens before the shift: is the capper torqued correctly, is the labeller aligned? You test the machine with a known bottle and a known cap. Fast, offline, no real product involved. That is your unit test.

The line inspection happens continuously during production: are bottles arriving at the expected rate, is the fill level within tolerance, did we somehow produce more caps than bottles? That is your data quality test — and no amount of machine inspection replaces it, because reality sends you things the machine was never shown.

Most teams do the first and skip the second. Then they find out in the quarterly review.

The five layers, cheapest first

Layers 1–3 run in CI, before merge. Layers 4–5 run in production, on every pipeline run — because they test facts about the world, not facts about your code.

The pytest you actually need

import pytest
from pipeline import validate, total_completed_fare

# ---- 1. plain assert, no boilerplate ----
def test_rejects_missing_ride_id():
    bad = {"event_id": "e1", "event_type": "ride_completed", "fare": 100}
    assert validate(bad).reason == "missing_ride_id"

# ---- 2. fixture: shared, reusable setup ----
@pytest.fixture
def events():
    return [
        {"event_id": "e1", "ride_id": "r1", "event_type": "ride_completed",
         "fare": 250, "event_time": "2026-07-31T10:00:00Z"},
        {"event_id": "e1", "ride_id": "r1", "event_type": "ride_completed",
         "fare": 250, "event_time": "2026-07-31T10:00:00Z"},   # duplicate!
    ]

def test_duplicates_are_not_double_counted(events):
    total, _ = total_completed_fare(events)
    assert total == 250          # not 500

# ---- 3. parametrize: one test, many cases ----
@pytest.mark.parametrize("fare, reason", [
    (None,  "invalid_fare"),
    (-50,   "invalid_fare"),
    ("250", "invalid_fare"),
    (0,     "invalid_fare"),
])
def test_bad_fares_rejected(fare, reason):
    e = {"event_id": "e", "ride_id": "r", "event_type": "ride_completed",
         "fare": fare, "event_time": "2026-07-31T10:00:00Z"}
    assert validate(e).reason == reason

# ---- 4. expected failures ----
def test_raises_on_unreadable_source():
    with pytest.raises(FileNotFoundError):
        load("does/not/exist.jsonl")

parametrize is the highest-value thing here: four edge cases, four separate reported results, one function.

The test that would have caught it

Remember the Pandas lesson: the upstream team reformatted fares as "1,250.00", to_numeric(errors="coerce") turned them all into NaN, and the report showed ₹0 with a perfectly normal row count.

No unit test would have caught it. The function was correct. The world changed.

What catches it is a data quality test that runs on real output, every run:

def test_output_is_plausible(daily_revenue, previous_run):
    assert daily_revenue.revenue.sum() > 0,                 "zero revenue"
    assert daily_revenue.revenue.isna().mean() < 0.01,      "too many nulls"
    assert len(daily_revenue) > 0,                          "empty output"

    drop = 1 - daily_revenue.revenue.sum() / previous_run.revenue.sum()
    assert drop < 0.30,  f"revenue fell {drop:.0%} vs previous run"

The last assertion is the one that matters. Absolute thresholds go stale as the business grows; comparison against the previous run stays honest, and catches the class of bug where the number is a number but the wrong one.

Fixtures that touch the outside world

# tmp_path — a real directory, cleaned up automatically
def test_writes_partitioned_output(tmp_path):
    write_rides(sample_df, tmp_path)
    assert (tmp_path / "city=Shillong").exists()

# monkeypatch — no network in unit tests, ever
def test_retries_then_succeeds(monkeypatch):
    calls = []

    def fake_get(url, **kw):
        calls.append(url)
        return FakeResponse(503) if len(calls) < 3 else FakeResponse(200, {"data": []})

    monkeypatch.setattr("pipeline.requests.get", fake_get)

    fetch("http://api/rides", {})
    assert len(calls) == 3          # proves the retry logic, in milliseconds

# a real database, once per test session
@pytest.fixture(scope="session")
def db():
    conn = connect(TEST_DSN)
    conn.execute(open("schema.sql").read())
    yield conn                       # everything before yield = setup
    conn.close()                     # everything after = teardown

The retry test is worth studying: it verifies genuinely important behaviour, with zero network and zero waiting. Testing time-based logic without sleeping is what separates fast suites from ones people skip.

What is wrong with this test?

def test_daily_revenue():
    df = pd.read_parquet("s3://prod/marts/daily_revenue/")
    assert df.revenue.sum() == 12_408_332.50

It passes today.

The six data quality dimensions

What you assert about production data — the checklist worth memorising:

1. Completeness — are rows missing? (row count vs the previous run, vs the source)
2. Uniqueness — is the primary key actually unique? (ride_id appearing twice)
3. Validity — do values obey the rules? (fare > 0, status in an allowed set)
4. Consistency — do related systems agree? (rides completed vs payments captured)
5. Timeliness — is the data fresh enough? (max event_time within the SLA)
6. Accuracy — does it match reality? (only reconciliation against a source of truth answers this)

In dbt these become unique, not_null, accepted_values, relationships and freshness tests, declared in YAML next to the model. In Python, Great Expectations or Soda. The tool matters far less than having a named list you check against — which is exactly what an interviewer is probing for.

Words to know

In two minutes

• Pipelines need two kinds of tests: code tests in CI and data checks in production on every run.
• Unit tests use 5–20 hand-written rows with your real edge cases — empty input, duplicates, nulls, wrong types, late data.
• pytest essentials: plain assert, fixtures (yield for teardown), parametrize for many cases, tmp_path, monkeypatch to remove network and clock.
• Keep the suite fast, deterministic and isolated, or people skip it; inject the clock instead of calling datetime.now().
• Data checks cover completeness, uniqueness, validity, consistency, timeliness, and accuracy (only reconciliation answers accuracy).
• Compare output to the previous run, not to a hard-coded number; a 30% drop should block publishing.
• A filter on an enum is an assumption; an accepted-values check at ingestion fires the day a new value appears.

Test yourself

There are 9 questions, 3 puzzles and 7 recall cards for this lesson in the app — the recall cards come back on later days, just before you would forget.