Testable Transformations

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.

# 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.

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.

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.

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.