--- title: 'Testable Transformations' source: 'https://academia.sh/en/courses/data-pipelines/testable-transformations' course: 'Data Pipelines and Orchestration' language: en updated: '2026-10-21T05:02:59+00:00' license: 'CC BY-SA 4.0' objectives: - 'Defines the condition that makes a transformation testable and writes a unit test with a fixed input against a hand-computed output.' - "Runs the six expectations placed on the transformation's own output over the six corruption classes and counts what is caught and what is not." - 'Writes by name the residue no expectation sees and shows that a rule error that shifts labels leaves the serving exactly the same.' --- # Testable Transformations Counting what expectations placed on the transformation's own output catch and what they do not: six expectations never fire on the sound pipeline and catch four of the six corruption classes, while duplication and type drift are swallowed by the transformation itself and never appear in the output. The unit change is caught by an expectation on the output at the third stage, and no baseline is needed for it, although the serving stays exactly the same. A rule error that shifts the period label back by one period leaves 2200 rows, 55 cells and five serving values exactly the same, so it trips none of the six expectations; only a unit test on a fixed input catches it. Three lessons wrote down the **place** of the transformation: whether it is done at the source or at the target, in which order it runs, and in which layer it stops. The answer to all three questions was an arrangement, and all three assumed the same thing — that the rule itself is correct and the data arrives in good shape. At no stage was it tested that the transformation **produces the correct result**. This lesson asks that question and puts two separate answer instruments side by side: a unit test that runs on a hand-written **fixed input**, and an expectation placed on the **actual output** of the run. What they catch is different and, more importantly, there is a residue that neither of them can see. - **TF41.** The source, the five-stage pipeline and the corruption classes are **fictional**; no real store, transformation framework or data test library is called. The seed is **20260218**, and the output does not depend on the run. - **TF42.** In this lesson the **transformation** is the composition of the first three stages: `s1_screen`, `s2_dedupe` and consumption from the period difference. Its input is the raw row, its output the consumption row. - **TF43.** An **expectation** is a routine that returns, as true or false, an invariant the output must carry. Here `True` means **broken**. - **TF44.** The unit test routine was measured in the Unit Testing and Test-Driven Development course of the Software Quality and Testing curriculum, and the translation of quality dimensions into checks in the Introduction to Data Engineering course; neither **is repeated**. The difference here is at a single point: those checks looked at the **source**, these look at the **output of the transformation**. - **TF45.** The baseline is **the transformation without expectations**: broken expectations **0**, because no expectation has been written. - **TF46.** The corruption classes are not attacks but accidents on the producer side, and each corrupts **a single period** (`2026-07`). - **TF47.** The consumption ceiling **200** is a physical choice; it is not the 49 known from the source's generator. The one who writes the expectation knows the meter, not the generator. - **TF48.** Resolution: the step of the consumption row ratio is **1/2200 = 0.0005**, the summary **55 cells**, the serving **5 values**. **No ratio is printed over the serving output**, the value itself is written. ## The Condition of a Testable Transformation A transformation is testable only when it is a **pure function**: for the same input it gives the same output on every run, carries no hidden state, does not look at the clock, reads nothing from outside. This condition is the direct counterpart of the rule of the course — a transformation that does not give the same result when rerun has no result to test, because every run makes a different answer defensible. Once the condition is met, two separate kinds of tests become possible, and they look at different questions: | Type | Input | Oracle: where the expected result comes from | Catches | |---|---|---|---| | unit test | hand-written fixed input | hand-computed output | an error in the **rule** | | data test (expectation) | actual output of the run | invariant the output must carry | a corruption of the **data** | The distinction lives in the oracle column. In a unit test the correct answer comes from the person writing the test, and since the input is kept small it can be computed by hand. An expectation has no such answer; nobody can know by hand the correct total of two thousand small rows. An expectation can only write what **must be true of the output itself on every run**: the key is unique, consumption cannot be negative, the cell count is complete. The block below sets up the source, the five stages and the six corruption classes. It is the definition used throughout the lesson. ```python # FICTIONAL. The meter readings of the measurement network, the five-stage pipeline and the corruption # classes are defined in the lesson; no real store, framework or orchestrator is called. SEED, M32 = 20260218, 0xFFFFFFFF def rng(t): x = ((t ^ (t >> 16)) * 2246822507) & M32 x = ((x ^ (x >> 13)) * 3266489909) & M32 s = [(x ^ (x >> 16)) & M32] def next_value(): s[0] = (s[0] * 1664525 + 1013904223) & M32 return s[0] / 4294967296 return next_value REGION = ["east", "west", "north", "south", "central"] TYPE = ["residential", "commercial"] PERIODS = [f"2026-{a:02d}" for a in range(1, 13)] SUBSCRIBERS = {} for i in range(200): r = rng(SEED + 37 * i) SUBSCRIBERS[f"A{i + 1:03d}"] = {"region": REGION[int(r() * 5)], "type": TYPE[int(r() * 2)], "start": 10000 + int(r() * 60000)} def generate_source(): records = [] for i, (sub, a) in enumerate(SUBSCRIBERS.items()): e = a["start"] for p in PERIODS: r = rng(SEED + 101 * i + 7 * PERIODS.index(p)) e += 4 + int(r() * 46) records.append({"subscriber": sub, "period": p, "region": a["region"], "type": a["type"], "index": e, "unit": "m3"}) return records CORRUPTIONS = ["late_arrival", "duplicated", "schema_changed", "unit_changed", "missing_partition", "type_drift"] def corrupt(records, cls, period="2026-07"): """Corrupts a single period , the remaining periods stay intact.""" out, counter = [], 0 for k in records: if k["period"] != period: out.append(k) continue if cls == "late_arrival": if k["subscriber"] <= "A040": # 40 subscribers never arrived in this period counter += 1 continue out.append(k) elif cls == "duplicated": out.append(k) if k["subscriber"] <= "A060": # 60 records arrived twice out.append(dict(k)) counter += 1 elif cls == "schema_changed": y = dict(k) y["index_value"] = y.pop("index") # the field name changed counter += 1 out.append(y) elif cls == "unit_changed": y = dict(k) y["index"] = k["index"] * 1000 # liters instead of m3 y["unit"] = "m3" # THE LABEL DID NOT CHANGE , silent corruption counter += 1 out.append(y) elif cls == "missing_partition": if k["region"] == "north": # one region's partition was never written counter += 1 continue out.append(k) elif cls == "type_drift": y = dict(k) y["index"] = str(k["index"]) # the number turned into text counter += 1 out.append(y) return out, counter def s1_screen(records): """A row that does not carry the required fields is dropped.""" REQUIRED = ("subscriber", "period", "region", "index", "unit") keep = [k for k in records if all(a in k for a in REQUIRED)] return keep, len(records) - len(keep) def s2_dedupe(records): seen, keep = set(), [] for k in records: key = (k["subscriber"], k["period"]) if key in seen: continue seen.add(key) keep.append(k) return keep, len(records) - len(keep) def s3_consumption(records): """Consumption from the period difference. If the previous period does not exist , the row cannot be computed.""" table = {(k["subscriber"], k["period"]): k for k in records} out, dropped = [], 0 for k in records: i = PERIODS.index(k["period"]) if i == 0: continue prev = table.get((k["subscriber"], PERIODS[i - 1])) if prev is None: dropped += 1 continue try: f = int(k["index"]) - int(prev["index"]) except (TypeError, ValueError): dropped += 1 continue out.append(dict(k, m3=f)) return out, dropped def s4_summary(records): totals = {} for k in records: key = (k["region"], k["period"]) totals[key] = totals.get(key, 0) + k["m3"] return totals, 0 def s5_serve(summary): """Serving: the period average per region.""" region = {} for (b, d), v in summary.items(): region.setdefault(b, []).append(v) return {b: round(sum(v) / len(v), 2) for b, v in region.items()}, 0 ``` ## Unit Test on a Fixed Input The second block writes an eight-row input and computes the correct output by hand. The input has two subscribers, three periods each, one duplicated row and one row missing a required field; therefore all three stages have to do their work. Next to it a rule is placed that does the same calculation but writes the result with **the label of the previous period**. ```python def transform(records): """From raw row to consumption row: screen , dedupe , period difference.""" v, _ = s1_screen(records) v, _ = s2_dedupe(v) v, _ = s3_consumption(v) return v def transform_shifted(records): """THE SAME calculation , the only difference: the result is written with the previous period's label.""" return [dict(k, period=PERIODS[PERIODS.index(k["period"]) - 1]) for k in transform(records)] def row(sub, p, e, type="residential"): return {"subscriber": sub, "period": p, "region": "east", "type": type, "index": e, "unit": "m3"} # Fixed input: eight rows , one duplicate and one row with a missing field. FIXED = [row("T1", "2026-01", 100), row("T1", "2026-02", 112), row("T1", "2026-02", 112), row("T1", "2026-03", 130), row("T2", "2026-01", 500, "commercial"), row("T2", "2026-02", 540, "commercial"), row("T2", "2026-03", 545, "commercial"), {"subscriber": "T3", "period": "2026-02"}] # Hand-computed correct output: the oracle of the test. EXPECTED = [("T1", "2026-02", 12), ("T1", "2026-03", 18), ("T2", "2026-02", 40), ("T2", "2026-03", 5)] def essence(v): return sorted((k["subscriber"], k["period"], k["m3"]) for k in v) def report(name, v): print(f"{name:<11}", " ".join(f"{a}/{d[-2:]}={m}" for a, d, m in essence(v)), " passed:", essence(v) == sorted(EXPECTED)) print("fixed input:", len(FIXED), "rows , expected output:", len(EXPECTED), "rows") report("correct", transform(FIXED)) report("shifted", transform_shifted(FIXED)) print("same input run twice , is the output the same:", essence(transform(FIXED)) == essence(transform(FIXED))) ``` ``` fixed input: 8 rows , expected output: 4 rows correct T1/02=12 T1/03=18 T2/02=40 T2/03=5 passed: True shifted T1/01=12 T1/02=18 T2/01=40 T2/02=5 passed: False same input run twice , is the output the same: True ``` Eight rows fall to four: the duplicated row in the `s2_dedupe` stage, the row with the missing field in the `s1_screen` stage, and the first period of each subscriber by the definition of consumption. All four numbers can be verified by hand: for T1, 112 − 100 = 12 and 130 − 112 = 18; for T2, 540 − 500 = 40 and 545 − 540 = 5. The shifted rule produces **the same four numbers** and fails the test immediately, because the labels have slipped back by one period. This is the strength of the test: the calculation itself is right, only the period it is written to is wrong, and this distinction is visible to the eye on a fixed input. The last line verifies the rule of the course on the fixed input — when the same input is run twice, the output is exactly the same. ## Expectations on the Output The third block defines six expectations. All of them look at the **transformation's own output**; none goes back to the source and none needs a correct result known in advance. The expectations are run on the six corruption classes one by one. ```python import statistics # All the expectations look at the transformation's OWN output , not at the source. ROWS, CELLS, CEILING = 2200, 55, 200 def e_key_unique(v, sm): key = [(k["subscriber"], k["period"]) for k in v] return len(key) != len(set(key)) def e_no_negative(v, sm): return any(k["m3"] < 0 for k in v) def e_ceiling(v, sm): return any(k["m3"] > CEILING for k in v) def e_row_ratio(v, sm): return abs(len(v) - ROWS) / ROWS > 0.02 def e_cell_completeness(v, sm): return len(sm) != CELLS def e_volatility(v, sm): """A region's period cells must not stray far from their own median.""" for b in REGION: vals = [x for (bb, _), x in sm.items() if bb == b] median = statistics.median(vals) if vals else 0 if not vals or any(abs(x - median) / abs(median) > 0.30 for x in vals): return True return False EXPECTATIONS = [("key_unique", e_key_unique), ("no_negative", e_no_negative), ("ceiling", e_ceiling), ("row_ratio", e_row_ratio), ("cell_completeness", e_cell_completeness), ("volatility", e_volatility)] def run_checks(v): """Names of the broken expectations , summary and serving.""" sm = s4_summary(v)[0] return [a for a, f in EXPECTATIONS if f(v, sm)], sm, s5_serve(sm)[0] SOURCE = generate_source() CLEAN, _, SERVING = run_checks(transform(SOURCE)) print("expectations:", len(EXPECTATIONS), " broken on the sound pipeline:", len(CLEAN)) print(f"{'corruption':<19}{'rows':>6}{'cells':>7}{'broken expectations':>34}{'serving':>9}") for s in CORRUPTIONS: v = transform(corrupt(SOURCE, s)[0]) bz, sm, sn = run_checks(v) print(f"{s:<19}{len(v):>6}{len(sm):>7}{','.join(bz) or 'none':>34}" f"{'same' if sn == SERVING else 'changed':>9}") ``` ``` expectations: 6 broken on the sound pipeline: 0 corruption rows cells broken expectations serving late_arrival 2120 55 row_ratio,volatility changed duplicated 2200 55 none same schema_changed 1800 45 row_ratio,cell_completeness changed unit_changed 2200 55 no_negative,ceiling,volatility same missing_partition 2108 53 row_ratio,cell_completeness changed type_drift 2200 55 none same ``` The first line shows there is no cost: on the sound pipeline none of the six expectations fires, so no good row is wrongly blocked. **Six expectations catch four of the six classes.** The `unit_changed` row is the most important number of this lesson. In the Introduction to Data Engineering course the same corruption was visible only through the distribution check that looks at the summary of the fourth stage; here the `no_negative` and `ceiling` expectations catch it **in the output of the transformation**, at the third stage. The difference is not merely one stage: the distribution check needs **a sound baseline summary** to work, these three expectations need no baseline at all. A newly built pipeline has no history to compare against, whereas an invariant can be written on the first run. The serving column, on the other hand, still says **same** for `unit_changed` — although the five served values stay exactly in place, two periods are garbage, and the only thing that says so is the expectation. The remaining three catches are ordinary: `late_arrival` and `missing_partition` lower the row count and break the cell band, while `schema_changed` makes the whole period vanish at the `s1_screen` stage and leaves 1800 rows and 45 cells. ## The Residue No Expectation Sees Two lines in the table are empty: `duplicated` and `type_drift`. The fourth block writes two rules next to them — one faulty, one deliberate — and runs the same six expectations on them. ```python SHIFTED = transform_shifted(SOURCE) FACTORED = [dict(k, m3=k["m3"] * (1.05 if k["type"] == "commercial" else 1.0)) for k in transform(SOURCE)] for name, v in (("shifted rule", SHIFTED), ("factor rule", FACTORED)): bz, sm, sn = run_checks(v) print(f"{name:<17} rows={len(v)} cells={len(sm)} broken={','.join(bz) or 'none'}") print(" serving:", sn) print(" baseline:", SERVING) ``` ``` shifted rule rows=2200 cells=55 broken=none serving: {'central': 1321.45, 'north': 1212.55, 'east': 881.36, 'west': 833.09, 'south': 1065.55} factor rule rows=2200 cells=55 broken=none serving: {'central': 1357.32, 'north': 1235.57, 'east': 902.13, 'west': 859.22, 'south': 1095.81} baseline: {'central': 1321.45, 'north': 1212.55, 'east': 881.36, 'west': 833.09, 'south': 1065.55} ``` The residue has three parts, and the mechanism of each is separate. **The two classes the transformation swallows.** `duplicated` is repaired in the `s2_dedupe` stage and `type_drift` in the type conversion; an expectation looking at the output cannot tell a repaired corruption from a corruption that never happened. The emptiness of these two lines is proof of the transformation's soundness, but nobody learns that the producer is still sending text, and the text of a coming period may not be convertible to a number. **An expectation placed on the output does not take the place of a check placed on the input**; the two look at different questions. **The label-shifting rule.** The shifted rule produces 2200 rows and 55 cells, breaks no expectation and leaves the serving **exactly the same**. The reason is structural: the set of calculated numbers has not changed at all, only the periods they are written to have slipped; the average of the eleven cells per region is the average of the same numbers. No invariant that can be written about the output itself can see this, because the output itself is flawless — what is wrong is **which fact it corresponds to**. The only thing that catches this error is the fixed-input test at the start of the lesson. **The deliberate rule change.** The factor rule reports commercial consumption multiplied by 1.05. None of the six expectations breaks, because there is no corruption; the serving, however, moves up in all five values. Expectations cannot say whether this difference is **a wanted change or an accident**. The only thing that makes the distinction is writing the rule version to the record. ## Summary - A transformation is testable only if it is a pure function; a rule that does not give the same output for the same input on every run has no result to test. - A unit test catches an error in the rule, an expectation catches a corruption of the data; the distinction lies in where the oracle comes from, and the two do not take each other's place. - On the sound pipeline none of the six expectations fires, and they catch four of the six corruption classes; `duplicated` and `type_drift` are swallowed by the transformation itself and never appear in the output. - `unit_changed` is caught by an expectation on the output at the third stage and **without any baseline**, although the five served values stay exactly the same. - The rule that shifts the period label back by one period produces 2200 rows, 55 cells and the same five serving values, so it trips none of the six expectations; only the unit test with a fixed input sees it. ## Next Step The factor rule of the last block opens the question this lesson cannot close. The rule changed and the new runs will produce the correct result — but eleven of the twelve periods stand computed with the **old rule**. Recomputing all the periods means run cost, while recomputing only a few periods leaves one part of the report with the new rule and the rest with the old one. The next lesson measures backfill over these two options: how many periods are recomputed, how far the mixed report deviates from the full application, and which checks that report passes.