lesson 247 Production Readiness Capstone · advanced

Test the whole pipeline and its properties

Unit tests check each function; an integrated test runs the whole pipeline, files and all, in a temporary folder. The strongest pipeline tests check properties that must hold for any input: every row is accounted for, the totals agree, and running twice changes nothing.

You'll be able to

Write an end-to-end test that runs the real pipeline on fixture files in tempfile.TemporaryDirectory
State a pipeline's invariants and assert them: loaded + rejected = rows read, totals match, reruns are identical
Check invariants on several inputs, including empty and broken ones, so a defect can't hide behind one happy example

The idea

The Testing & Code Quality module tested functions with known inputs and outputs, and the Software Architecture module introduced integration tests through the public entry point. For a data pipeline, comparing one output file with an expected file is a start, but it only covers the inputs you thought of. Pipelines also have invariants: properties that must be true for *any* input.

Accounting: every data row is either loaded or rejected; loaded + rejected == rows read. A pipeline that silently drops rows breaks it.
Consistency: the report's totals equal the sum over the loaded rows. Double-counting or lost rounding breaks it.
Idempotence (see Reliable Data Pipelines): running twice on the same input produces byte-identical output.
Recovery (see Reliable Data Pipelines): a crashed-and-resumed run equals a clean one.

An integrated test runs the real code end to end in a temporary folder (tempfile.TemporaryDirectory() creates one and deletes it after the with block), so tests never depend on files left over from earlier runs, and checks the invariants on several inputs: the normal export, an export with bad rows, an empty one.

Run each example to see what it prints. Running examples doesn't count toward progress.

example-1.py · An end-to-end test in a temporary folder

import csv
import io
import json
import os
import tempfile
import unittest

ZONES = {39: "Brooklyn", 70: "Queens", 186: "Manhattan"}

def run(input_path, output_path):
    with open(input_path, encoding="utf-8") as f:
        rows = list(csv.DictReader(f))
    revenue, rejected = {}, []
    for number, row in enumerate(rows, start=2):
        try:
            total = float(row["total_amount"])
            zone = int(row["pickup_zone_id"])
        except ValueError:
            rejected.append(number)
            continue
        borough = ZONES.get(zone, "Unknown")
        revenue[borough] = round(revenue.get(borough, 0) + total, 2)
    with open(output_path, "w", encoding="utf-8") as f:
        json.dump({"rows": len(rows), "rejected": rejected, "revenue": revenue}, f, sort_keys=True)

EXPORT = "trip_id,pickup_zone_id,total_amount\n16126,186,64.19\n16127,186,n/a\n16128,39,18.7\n16132,70,66.71\n"

class PipelineTest(unittest.TestCase):
    def run_pipeline(self, folder, text):
        source = os.path.join(folder, "trips.csv")
        target = os.path.join(folder, "report.json")
        with open(source, "w", encoding="utf-8") as f:
            f.write(text)
        run(source, target)
        with open(target, encoding="utf-8") as f:
            return f.read()

    def test_report_and_invariants(self):
        with tempfile.TemporaryDirectory() as folder:
            first = self.run_pipeline(folder, EXPORT)
            second = self.run_pipeline(folder, EXPORT)
        report = json.loads(first)
        self.assertEqual(report["revenue"], {"Brooklyn": 18.7, "Manhattan": 64.19, "Queens": 66.71})
        loaded = report["rows"] - len(report["rejected"])
        self.assertEqual(loaded + len(report["rejected"]), 4)
        self.assertEqual(round(sum(report["revenue"].values()), 2), 149.6)
        self.assertEqual(first, second)

result = unittest.TextTestRunner(stream=io.StringIO()).run(unittest.defaultTestLoader.loadTestsFromTestCase(PipelineTest))
print("ran", result.testsRun, "ok", result.wasSuccessful())
print(os.listdir("."))

An invariant is only as good as the inputs it is checked on. A pipeline that drops bad rows instead of reporting them passes every test that uses clean data:

example-2.py · The accounting invariant catches a silent drop

import csv
import io

def summarise(text):
    loaded, rejected = 0, []
    for row in csv.DictReader(io.StringIO(text)):
        try:
            float(row["total_amount"])
        except ValueError:
            continue
        loaded += 1
    return loaded, rejected

text = "trip_id,total_amount\n16126,64.19\n16127,n/a\n16128,18.7\n"
rows = len(text.strip().splitlines()) - 1
loaded, rejected = summarise(text)
assert loaded + len(rejected) == rows, f"{loaded} loaded + {len(rejected)} rejected != {rows} rows"
print("accounted for")

Writing the invariants as one reusable function makes it cheap to check them on many inputs at once. Each input below is a small export built to stress one thing: a broken number, an empty file, a repeated trip.

example-3.py · Invariants over several inputs

import csv
import io

def summarise(text):
    revenue, rejected, seen = {}, [], set()
    for number, row in enumerate(csv.DictReader(io.StringIO(text)), start=2):
        try:
            total = float(row["total_amount"])
        except ValueError:
            rejected.append(number)
            continue
        if row["trip_id"] in seen:
            rejected.append(number)
            continue
        seen.add(row["trip_id"])
        revenue[row["zone"]] = round(revenue.get(row["zone"], 0) + total, 2)
    return revenue, rejected

def broken_invariants(text, revenue, rejected):
    rows = list(csv.DictReader(io.StringIO(text)))
    problems = []
    loaded_rows = [row for n, row in enumerate(rows, start=2) if n not in rejected]
    if len(loaded_rows) + len(rejected) != len(rows):
        problems.append("accounting")
    if round(sum(revenue.values()), 2) != round(sum(float(r["total_amount"]) for r in loaded_rows), 2):
        problems.append("consistency")
    return problems

cases = {
    "normal": "trip_id,zone,total_amount\n16126,Manhattan,64.19\n16128,Brooklyn,18.7\n",
    "bad number": "trip_id,zone,total_amount\n16126,Manhattan,64.19\n16127,Manhattan,n/a\n",
    "empty": "trip_id,zone,total_amount\n",
    "repeated trip": "trip_id,zone,total_amount\n16126,Manhattan,64.19\n16126,Manhattan,64.19\n",
}
for name, text in cases.items():
    revenue, rejected = summarise(text)
    print(f"{name:14}", broken_invariants(text, revenue, rejected) or "all hold")

When to use it: for every pipeline before it ships, and again in CI on every push (see Application Delivery). Mistakes to avoid: tests that read or write the real data folder; one happy-path fixture only; asserting exact floating-point sums (round both sides); and tests that pass because they check nothing about rows that were dropped.

Review

Integrated tests run the real pipeline end to end, in a temporary folder.
Pipeline invariants: every row accounted for, totals consistent with the loaded rows, reruns identical, resumes equal to clean runs.
Check invariants on many small inputs built to stress one thing each; round money before comparing.
Next: the pipeline is correct; is it fast enough for a full night of trips?