Testing Data Pipelines

Write four tests for two small transform functions, watch them pass, then let a “harmless” refactor slip past review and see which test catches it.

Two transforms, four tests

def clean_email(raw: str) -> str:
    return raw.strip().lower()

def dedupe_orders(rows: list[dict]) -> list[dict]:
    seen = set()
    out = []
    for row in rows:
        if row["order_id"] not in seen:
            seen.add(row["order_id"])
            out.append(row)
    return out

def test_clean_email_trims_whitespace():
    assert clean_email("  Ada@Example.com  ") == "ada@example.com"

def test_clean_email_lowercases():
    assert clean_email("GRACE@EXAMPLE.COM") == "grace@example.com"

def test_dedupe_orders_keeps_first_occurrence():
    rows = [{"order_id": 1, "amount": 10}, {"order_id": 1, "amount": 999}, {"order_id": 2, "amount": 20}]
    result = dedupe_orders(rows)
    assert len(result) == 2
    assert result[0]["amount"] == 10, f"expected first occurrence (amount=10), got amount={result[0]['amount']}"

def test_dedupe_orders_empty_input():
    assert dedupe_orders([]) == []

A tiny runner that reports pass/fail without pulling in a test framework:

def run_tests(tests):
    passed = failed = 0
    for name, fn in tests:
        try:
            fn()
            print(f"PASS  {name}")
            passed += 1
        except AssertionError as e:
            print(f"FAIL  {name}: {e}")
            failed += 1
    print(f"{passed} passed, {failed} failed")

run_tests([
    ("clean_email trims whitespace", test_clean_email_trims_whitespace),
    ("clean_email lowercases", test_clean_email_lowercases),
    ("dedupe_orders keeps first occurrence", test_dedupe_orders_keeps_first_occurrence),
    ("dedupe_orders handles empty input", test_dedupe_orders_empty_input),
])

$ python pipeline_testing.py
PASS  clean_email trims whitespace
PASS  clean_email lowercases
PASS  dedupe_orders keeps first occurrence
PASS  dedupe_orders handles empty input
4 passed, 0 failed

The “optimization” that changes behavior

Someone notices dedupe_orders builds a set and a list by hand and “simplifies” it to a dict comprehension — same row count in, same row count out, code review approves it on sight:

def dedupe_orders(rows: list[dict]) -> list[dict]:
    # "optimization": keep the last-seen row per order_id instead of the first
    by_id = {row["order_id"]: row for row in rows}
    return list(by_id.values())

== after the "optimization" ==
PASS  clean_email trims whitespace
PASS  clean_email lowercases
FAIL  dedupe_orders keeps first occurrence: expected first occurrence (amount=10), got amount=999
PASS  dedupe_orders handles empty input
3 passed, 1 failed

Both versions have the same row count, same input, same output shape — a reviewer scanning the diff has no reason to suspect anything changed. The dict comprehension keeps whichever row overwrites last, which for a duplicate order (a common CDC replay artifact — see the incremental-loads-and-cdc lesson) is the exact opposite of “first occurrence wins,” and no one asked which one was actually correct. The test did not know why it mattered; it only had to know what the function was supposed to do, stated once, in advance.

Practice

def test_clean_email_idempotent():
    assert clean_email("ada@example.com") == "ada@example.com"

PASS  clean_email idempotent
1 passed, 0 failed

def test_dedupe_orders_no_duplicates():
    rows = [{"order_id": 1, "amount": 10}, {"order_id": 2, "amount": 20}]
    assert dedupe_orders(rows) == rows

PASS  dedupe_orders no duplicates
1 passed, 0 failed

PASS  clean_email trims whitespace
PASS  clean_email lowercases
PASS  dedupe_orders keeps first occurrence
PASS  dedupe_orders handles empty input
4 passed, 0 failed

Next: streaming, exactly-once semantics — the same correctness problems with the clock running.