Most of what teams believe is untestable is untestable only because it sits behind a decorator. Move two things out and the interesting half becomes ordinary pytest.
This is the whole pipeline file for silver.sales_order_line. The transform and the rules come from elsewhere; the file is glue.
Why
The failure this prevents is a conclusion, not a bug: our pipelines can only be tested by running them. A team that believes this checks its rules by starting a pipeline, which costs ninety seconds and a warehouse, so the check moves to a nightly job, so nobody edits a rule with confidence and nobody deletes one at all.
With EXPECTATIONS as an importable dict, changing known_currency is a one-second test run. Without it the rule text is typed twice — once in the decorator, once in the test — and after the first edit the test guards a rule production no longer runs, while staying green. The same drift happens between pages: two documents publishing two versions of one dict is the identical failure at documentation scale.
How
pythonsrc/dwh/pipelines/silver_sales_order.py
from pyspark import pipelines as dp
from dwh.pipelines.transforms.silver_sales_order import EXPECTATIONS, to_silver_sales_order
# CI points this at dwh_test.run_<job_run_id>; prod points it at dwh_prod.bronze.
SOURCE = spark.conf.get("dwh.source_schema")
@dp.materialized_view(name="sales_order_line")
@dp.expect_all_or_drop(EXPECTATIONS)
def sales_order_line():
return to_silver_sales_order(
spark.read.table(f"{SOURCE}.sap_vbap"),
spark.read.table(f"{SOURCE}.fx_rate"),
)
Now try to import it from a test. There are four walls, and teams usually stop at the first one and conclude the platform is at fault.
textFour reasons this function is not the unit under test
1. spark is injected by the pipeline runtime as a global.
Outside an update: NameError: name 'spark' is not defined
— raised at IMPORT time, before any test body runs.
2. The decorators register a dataset into a pipeline graph.
With no graph to register into, decoration itself fails.
3. @dp.materialized_view does not hand your function back.
sales_order_line is a registered dataset, not a callable you can assert on.
4. Even if it were callable, it takes no arguments and names two
tables — so calling it needs a live catalog with data in it.
The fifth reason is the one that matters most and is the easiest to miss: the data quality rules are not in the function body at all. They are decorator arguments. A test that somehow executed sales_order_line() would still test none of them.
So two things move out of the pipeline file. The transform becomes a pure function — that is the standards route's first testability rule. The rules become a plain dict in the same transform module, imported by the pipeline and by the tests, so they cannot drift apart:
pythonsrc/dwh/pipelines/transforms/silver_sales_order.py — business rules 1, 2, 5 and 6 as data
EXPECTATIONS = {
"amount_positive": "amount_eur > 0 OR status = 'CANCELLED'",
"known_currency": "currency IN ('EUR', 'CHF')",
"fx_rate_resolved": "fx_rate IS NOT NULL",
"order_not_ancient": "order_ts >= TIMESTAMP '2020-01-01'",
"status_domain": "status IN ('OPEN', 'SHIPPED', 'CANCELLED')",
}
That dict is quoted, not authored here. The canonical copy is in the standards route, and where any page differs from it that page is the one that is wrong — which is the same argument as the dict itself, one level up.
Two rules are deliberately absent, for two different reasons:
Rule 3 — valid countries are DE, AT, CH, NL
The one people try to add. country lives on gold.dim_customer; an order line has no such column, and an expectation naming a column the dataset does not produce fails the pipeline at analysis. It is enforced in the customer transform, where the column exists.
Absent because an expectation asserts a predicate over the row it is handed, and a row whose fx_rate was silently defaulted satisfies that arithmetic exactly.
What an expectation can do is refuse the defaulting: fx_rate_resolved counts the rows the rate join left NULL. The value itself is checked by the reconciliation assertions in the SQL-testing route, and the gotcha at the end of the next topic is the reason.
Doing it
Move every constraint out of the decorator call into a module-level EXPECTATIONS dictIt belongs in the transform module. Mechanical commit first, logic changes second.
Read the source schema from spark.conf, set by the pipeline's configuration blockNever a literal catalog in the pipeline file.
Keep the pipeline file under ten linesIf it grows a branch, that branch has escaped the tested half.
Point the test suite at transforms/ onlyNothing under pipelines/ is ever imported by a test.
Embedding it
Ask someone to write a unit test for the decorated function without telling them it is impossibleTen minutes of hitting the four walls teaches the layout rule permanently; being told it takes and leaves nothing.
In review, ask one question of every pipeline file: 'which line here could be wrong?'If the honest answer is none, the glue is thin enough.
Have the team delete an expectation on purpose and find out which test catches itIf none does, that is today's work — see the event-log assertion below.
When someone proposes testing a pipeline by running it in dev and eyeballing the table, agreeThen ask what they will check tomorrow, and who will remember.
Defaults
Pipeline file = imports, one conf read, decorators. Nothing else
EXPECTATIONS is a dict in the transform module, imported by the pipeline, the tests and the CI assertion
@dp.table creates a streaming table; @dp.materialized_view creates a materialized viewsales_order_line is a batch rebuild, so it is a materialized view.
Source schema arrives through pipeline configurationThe same file then runs against seeded fixtures and against production.
Gotchas
Copying an old DLT example that uses @dlt.table for everythingUnder Lakeflow pipelines @table now means STREAMING TABLE, so a batch rebuild silently becomes a streaming table with a checkpoint and different restart semantics — and the first symptom is a full refresh that processes zero rows.
Constraints written inline in the decoratorThe test re-types the same SQL string, the two copies drift on the first edit, and both keep passing while guarding different rules.
A hardcoded source table in the pipeline fileThe CI pipeline then reads production bronze, the seeded bad row never enters, every assertion passes, and the harness proves nothing at all.
Importing anything from pipelines/ into a test 'just to check it parses'The import raises before collection and the whole test session errors out, which reads like a broken conftest and costs an hour.
An expectation is a filter with a counter attached. expect_all_or_drop removes the rows that fail and records how many. Nothing turns red. That inverts the usual risk: the dangerous defect is not a rule that fails, it is a rule that succeeds more often than it should, quietly deleting valid data.
Two tests, both pure pytest, both sub-second. The first is a corpus of rows that are legitimate business data and must survive every rule. The rows are shaped like the output of to_silver_sales_order, because that is what the expectations are evaluated against — fx_rate and amount_eur exist by the time the decorator sees the frame:
Why
Take the concrete case. Someone writes currency IN ('EUR') — CHF omitted, because Switzerland was added to the business after the list was typed. Every Swiss order line is dropped at silver. No job fails. No alert fires.
The event log recorded it faithfully, as known_currency: failed_records = 1284, in a table nobody had queried since the pipeline was built. What the business saw was Swiss revenue at zero for six days, and the first person to notice was a sales lead, not an engineer. The MUST_PASS corpus above turns that into a red build in under a second, because the converted CHF line is in it.
How
pythontests/test_expectations.py — no rule may drop a valid row
from datetime import datetime
from decimal import Decimal
import pytest
from pyspark.sql.types import (
DecimalType, IntegerType, StringType, StructField, StructType, TimestampType,
)
from dwh.pipelines.transforms.silver_sales_order import EXPECTATIONS
# Explicit, for the same reason the factories in the PySpark route are:
# inference gives order_ts STRING and the money columns DOUBLE, and then
# "order_ts >= TIMESTAMP '2020-01-01'" is silently tested as a LEXICOGRAPHIC
# string comparison that only works because ISO-8601 happens to sort correctly.
ROW_SCHEMA = StructType([
StructField("order_id", StringType(), False),
StructField("line_number", IntegerType(), False),
StructField("currency", StringType(), False),
StructField("status", StringType(), False),
StructField("quantity", DecimalType(12, 3), False),
StructField("unit_price", DecimalType(18, 2), False),
StructField("fx_rate", DecimalType(18, 6), True), # NULL is the case fx_rate_resolved exists for
StructField("order_ts", TimestampType(), False),
StructField("amount_eur", DecimalType(18, 2), True),
])
BASE = dict(
order_id="4711", line_number=1, currency="EUR", status="SHIPPED",
quantity=Decimal("2.000"), unit_price=Decimal("125.00"),
fx_rate=Decimal("1.000000"), order_ts=datetime(2026, 7, 1, 9, 0),
amount_eur=Decimal("250.00"),
)
MUST_PASS = [
("cancelled line, zero value", {**BASE, "status": "CANCELLED",
"amount_eur": Decimal("0.00")}),
("swiss order converted", {**BASE, "currency": "CHF",
"fx_rate": Decimal("0.940000"),
"quantity": Decimal("10.000"),
"unit_price": Decimal("12.00"),
"amount_eur": Decimal("112.80")}), # 10 x 12.00 x 0.94
("oldest permitted order", {**BASE, "order_ts": datetime(2020, 1, 1, 0, 0)}),
("line still open", {**BASE, "status": "OPEN"}),
]
@pytest.mark.parametrize("name,predicate", sorted(EXPECTATIONS.items()))
@pytest.mark.parametrize("case,row", MUST_PASS, ids=[c for c, _ in MUST_PASS])
def test_no_expectation_drops_a_valid_row(spark, name, predicate, case, row):
kept = spark.createDataFrame([row], ROW_SCHEMA).where(predicate).count()
assert kept == 1, f"{name} drops a valid row ({case}); predicate: {predicate}"
The schema is not tidiness. It is what makes the test evaluate the same predicate the warehouse evaluates. Write "2026-07-01 09:00:00" and 250.00 into BASE the way you would without it, and the frame never builds: TypeError: field order_ts: TimestampType can not accept object '2026-07-01 09:00:00' in type <class 'str'>. That error is the schema doing its job — the alternative is inference accepting both, order_not_ancient passing as a string comparison, and amount_positive tested in binary floating point against a warehouse doing DECIMAL with HALF_UP rounding.
The second test is the mirror. A rule that rejects nothing is not protection, it is decoration — and it is indistinguishable from protection until the day it is needed.
pythontests/test_expectations.py — every rule must reject something
MUST_FAIL = {
"amount_positive": {**BASE, "status": "SHIPPED", "amount_eur": Decimal("0.00")},
"known_currency": {**BASE, "currency": "USD"},
"fx_rate_resolved": {**BASE, "currency": "CHF", "fx_rate": None, "amount_eur": None},
"order_not_ancient": {**BASE, "order_ts": datetime(2019, 12, 31, 23, 59, 59)},
"status_domain": {**BASE, "status": "RETURNED"},
}
def test_every_rule_has_a_counterexample():
assert set(MUST_FAIL) == set(EXPECTATIONS), "a rule was added without a counterexample"
@pytest.mark.parametrize("name,row", sorted(MUST_FAIL.items()))
def test_expectation_rejects_its_counterexample(spark, name, row):
kept = spark.createDataFrame([row], ROW_SCHEMA).where(EXPECTATIONS[name]).count()
assert kept == 0, f"{name} no longer rejects the row it exists to reject"
Doing it
Add a MUST_PASS row for every value the business actually hasBoth currencies, all three statuses, the cancelled-with-zero case, and the exact date boundary.
Add the counterexample in the same commit as the ruleA rule without one is not reviewable — nobody can tell what it is meant to catch.
Write NULL handling explicitlyamount_eur > 0 drops every NULL silently, because in SQL three-valued logic the predicate is neither true nor false and the row is not retained.
Grep the event log for rules with zero failed records over ninety days and interrogate each oneEither the data is perfect or the rule is pointed at the wrong column.
Embedding it
Give the team the Swiss story with the numbers and let them find which rule caused itFive minutes, and the class of bug is understood for good.
Have each engineer add one MUST_PASS row for a case they have personally seen in the sourceTheir cases are better than yours — they have read the data.
Make 'where is the counterexample?' the standard review comment on any pull request touching EXPECTATIONSThen stop reviewing the rule text yourself, once they are asking it.
Run a five-minute kata: break one rule, run pytest, read which assertion caught itRepeat with a rule that no test covers — that gap is the next ticket.
Defaults
Every rule has at least one row that must pass and exactly one row that must fail
The MUST_PASS corpus covers the real domain: EUR and CHF; OPEN, SHIPPED, CANCELLED; the 2020-01-01 boundary
Rows are built with an explicit schema, in DECIMAL and TIMESTAMPThe predicate under test is then the predicate production evaluates.
expect_all warns and keeps; expect_all_or_drop removes; expect_all_or_fail stops the updatePick per rule and write down why.
Rules that can silently delete data are drop rules and get a MUST_PASS rowRules that are advisory are warn rules and do not.
Gotchas
NULLs: amount_eur > 0 OR status = 'CANCELLED' evaluates to NULL when amount_eur is NULLA NULL predicate keeps nothing — so a nullable column turns a value rule into a completeness rule nobody wrote. You notice it as a row count that drops the week the source starts sending nulls, with every rule reporting a failure you did not intend to write. Decide which one you meant and say so: amount_eur IS NULL OR amount_eur > 0, or the opposite.
Using expect_all when you meant expect_all_or_dropThe rows are counted and then written anyway, so the event log shows failures while the bad rows sit in silver — and everyone reading the dashboard believes the failures were dropped.
A rule with zero failed records for a yearIt is either redundant or wrong, and it costs nothing to keep, which is why nobody checks. status_domain over a column the source can only emit three values into guards a door that has no wall.
Calling createDataFrame without a schema in these testsInference gives order_ts STRING and the money columns DOUBLE, so order_ts >= TIMESTAMP '2020-01-01' is evaluated as a string comparison that happens to be right only because ISO-8601 sorts correctly — change the seed to '01.01.2020' and the test reports the rule as protective while the pipeline, which has a real TIMESTAMP, drops the row. Same class of defect for amount_positive: binary floating point in the test, DECIMAL with HALF_UP in the warehouse.
The blind spot no expectation covers: a transform that defaults a missing FX rate to 1.0F.coalesce(F.col('fx_rate'), F.lit(1.0)) turns a missing FX rate into a rate of 1.0. amount_eur is still positive, so amount_positive passes — and so does fx_rate_resolved, because the coalesce ran first and there is no NULL left to count. A rule cannot see a value the transform already invented. CHF revenue is overstated by roughly seven percent. Catch it by defaulting only the base currency by name, seeding a row whose currency has no rate that day, and asserting on the VALUE rather than on the rule.