Testing Data Pipelines with pytest
What to test in a pipeline, how to build tiny hand-checked fixtures, and the tests that catch real production bugs: replays, duplicates and bad rows.
PythonIntermediate2 min readUpdated 11 Oct 2026
Test behaviour, not lines of code
The bugs that hurt in production are rarely syntax errors. They are behaviours:
| Behaviour | Test |
|---|---|
| Re-running a batch | Output is identical (idempotent) |
| Duplicate source rows | Counted once |
| Bad rows | Rejected or quarantined, not silently dropped |
| Failure half-way | Nothing partial is published |
| Late data | Lands in the right partition |
A fixture you can check by hand
import pytest
@pytest.fixture
def orders():
return [
{"order_id": "O1", "qty": 2, "price": 500}, # 1000
{"order_id": "O1", "qty": 2, "price": 500}, # duplicate resend
{"order_id": "O2", "qty": 1, "price": 300}, # 300
]
def test_duplicates_counted_once(orders):
assert total_sales(dedupe(orders)) == 1300Keep fixtures small enough that the expected number fits in a comment. If you need a calculator to check a test, the fixture is too big.
Test idempotency directly
def test_replay_changes_nothing(tmp_path):
run_pipeline(tmp_path, batch="2026-01-05")
first = read_facts(tmp_path)
run_pipeline(tmp_path, batch="2026-01-05")
assert read_facts(tmp_path) == firsttmp_path gives each test its own directory, so tests never share state.
Inject failures
def test_failure_rolls_back(tmp_path, monkeypatch):
def boom(*args, **kwargs):
raise RuntimeError("injected")
monkeypatch.setattr(pipeline, "publish", boom)
with pytest.raises(RuntimeError):
run_pipeline(tmp_path, batch="2026-01-05")
assert read_facts(tmp_path) == []Finished this note?