A factory returns a valid, boring row. Each argument is a delta from valid. That makes the test body a one-line statement of what is unusual about this case, which is also the only documentation the test will ever have.
Why
The alternative is a checked-in CSV or JSON fixture carrying every column of silver.sales_order_line. A reviewer cannot see which column mattered, because all forty are present and all forty look equally deliberate.
Then the model changes. Adding region_sk means editing thirty fixture files, and the ones you miss do not fail — the column is absent, arrives as NULL, and the test keeps passing while asserting less than it used to. A factory has one place to change.
How
pythontests/factories.py
from datetime import date, datetime
from decimal import Decimal
from typing import Any
from pyspark.sql.types import (
DateType, DecimalType, IntegerType, StringType, StructField, StructType,
TimestampType,
)
ORDER_LINE_SCHEMA = StructType([
StructField("order_id", StringType(), False),
StructField("line_number", IntegerType(), False),
StructField("customer_id", StringType(), False),
StructField("product_id", StringType(), False),
StructField("order_ts", TimestampType(), False),
StructField("quantity", DecimalType(12, 3), False),
StructField("unit_price", DecimalType(18, 2), False),
StructField("currency", StringType(), False),
StructField("status", StringType(), False),
])
# A boring line that satisfies all seven business rules.
_ORDER_LINE: dict[str, Any] = {
"order_id": "SO-1000",
"line_number": 10,
"customer_id": "C-0001",
"product_id": "P-0001",
"order_ts": datetime(2026, 3, 2, 9, 15),
"quantity": Decimal("2.000"),
"unit_price": Decimal("49.90"),
"currency": "EUR",
"status": "OPEN",
}
def order_line_rows(*overrides: dict[str, Any]) -> list[dict[str, Any]]:
"""One row per argument; each argument is the delta from a valid line.
order_line_rows() -> one valid line
order_line_rows({"status": "CANCELLED"}) -> one cancelled line
"""
return [{**_ORDER_LINE, **o} for o in (overrides or ({},))]
FX_RATE_SCHEMA = StructType([
StructField("currency", StringType(), False),
StructField("rate_date", DateType(), False),
StructField("fx_rate", DecimalType(18, 6), False),
])
# silver.fx_rate on the order date above. EUR is stored at 1 so the join in
# build_fct_order_line needs no special case for the reporting currency.
_FX_RATES: list[dict[str, Any]] = [
{"currency": "EUR", "rate_date": date(2026, 3, 2), "fx_rate": Decimal("1.000000")},
{"currency": "CHF", "rate_date": date(2026, 3, 2), "fx_rate": Decimal("0.940000")},
]
def fx_rate_rows(*overrides: dict[str, Any]) -> list[dict[str, Any]]:
"""No arguments: both currencies. With arguments: one row per delta from CHF."""
return [{**_FX_RATES[1], **o} for o in overrides] or [dict(r) for r in _FX_RATES]
pythontests/gold/test_fct_order_line.py — the test body is the delta
def test_cancelled_lines_are_kept_in_the_fact(spark, rates):
"""Rule 1 excludes cancellations from REVENUE, so the fact must keep the row."""
lines = spark.createDataFrame(
order_line_rows(
{"order_id": "SO-1", "status": "SHIPPED"},
{"order_id": "SO-2", "status": "CANCELLED"},
),
ORDER_LINE_SCHEMA,
)
fct = build_fct_order_line(lines, rates)
assertDataFrameEqual(
fct.select("order_id", "status"),
[Row(order_id="SO-1", status="SHIPPED"),
Row(order_id="SO-2", status="CANCELLED")],
)
Note the explicit schema on createDataFrame. It is not politeness — it is the difference between testing DECIMAL arithmetic and testing floating point.
Doing it
One factory per canonical grain: order_line_rows, customer_rows, fx_rate_rows. Nothing else
Always pass the explicit schema to createDataFrameInference is how DECIMAL becomes DOUBLE.
Keep defaults valid against all seven business rulesAny invalid seed is then visible in the test body.
Import the same factories from the SQL-assertion testsBoth kinds of test then seed identical data.
Embedding it
The first time someone checks in a CSV fixture, ask them to rewrite one test with the factoryDo it in the review; the diff makes the argument better than you can.
Have the team write the factory for the next table before the table existsIt forces the grain conversation while it is still free.
In review, ask which line of the test is the thing under testIf it takes more than a second to point at, the setup is too loud.
Defaults
Defaults are boring and valid; overrides are the interesting bit
Money and quantities are Decimal, never float
Factories return plain dicts, so they serve pytest, SQL seeding and a pipeline fixture table alike
Deterministic values onlyA test that depends on the wall clock fails once, at night, and is never reproduced.
Gotchas
createDataFrame inference turns 2.5 into DOUBLE and 2 into LONGProduction has DECIMAL(12,3) and INT, so the suite exercises binary floating point while the warehouse does decimal arithmetic with HALF_UP rounding. A one-cent error in rule 2 is then invisible in every test and visible in the first reconciliation against SAP.
A factory that computes derived values, for example filling amount_eur itselfThe expected value then comes from the same logic as the actual. The test proves that a function equals itself, and it stays green through the bug it was written to catch.
Returning a shared dict rather than a copy lets one test mutate the data of anotherThe spread in the comprehension is what prevents it; a factory that returns a module-level list does not.
datetime.now in a defaultThe test passes for months and fails the night an order_ts crosses a day boundary or an SLA window, and nobody can reproduce it in the morning.
#assertDataFrameEqual and what it actually compares
What
PySpark ships the comparison functions; there is no reason to hand-roll one. assertDataFrameEqual for data, assertSchemaEqual for contracts, and they are not interchangeable.
Why
A hand-written assert actual.collect() == expected fails on row order, fails on Row versus tuple, and prints two walls of text that a reader has to diff by eye. assertDataFrameEqual prints the differing rows and only those — a two-minute fix instead of a fifteen-minute squint.
The schema assertion earns its place separately. gold.fct_order_line is read by Power BI and by gold.mv_sales, so a column rename — or a column that stops being NOT NULL — is a breaking change to a published interface. You want it as a red test in a pull request, not as a broken report on a Monday. Which is precisely why the default ignoreNullable=True has to be overridden here: left alone, the test covers the rename and not the nullability, and half a contract reads exactly like a whole one.
How
pythontests/gold/test_fct_order_line.py
from decimal import Decimal
from pyspark.sql import Row
from pyspark.testing import assertDataFrameEqual, assertSchemaEqual
from dwh.gold.fct_order_line import build_fct_order_line
from tests.factories import ORDER_LINE_SCHEMA, order_line_rows
from tests.schemas import FCT_ORDER_LINE_SCHEMA
def test_amount_eur_is_rounded_to_the_cent(spark, rates):
"""Rule 2: amount_eur = ROUND(quantity * unit_price * fx_rate, 2)."""
lines = spark.createDataFrame(
order_line_rows({
"quantity": Decimal("3.000"),
"unit_price": Decimal("19.99"),
"currency": "CHF", # rates fixture has CHF -> EUR at 0.940000
}),
ORDER_LINE_SCHEMA,
)
actual = build_fct_order_line(lines, rates).select("order_id", "amount_eur")
# 3.000 * 19.99 * 0.94 = 56.3718 -> 56.37
assertDataFrameEqual(actual, [Row(order_id="SO-1000", amount_eur=Decimal("56.37"))])
def test_fct_order_line_schema_is_the_published_contract(spark, rates):
"""gold.fct_order_line is read by Power BI. Its schema is an interface."""
actual = build_fct_order_line(
spark.createDataFrame(order_line_rows(), ORDER_LINE_SCHEMA), rates
)
# ignoreNullable defaults to True from PySpark 4.0. Left at the default, a
# NOT NULL -> NULL change in a published interface passes silently.
assertSchemaEqual(actual.schema, FCT_ORDER_LINE_SCHEMA, ignoreNullable=False)
expected may be a DataFrame or a list of Row. For a handful of rows the list is better: it fits on screen and needs no second createDataFrame whose schema you then have to keep in step. Row order is not compared by default, which is correct — Spark promises no ordering without an ORDER BY. Set checkRowOrder=True only when order is part of the contract, which in a warehouse is almost never.
Know what that choice costs. When expected is a list of Row there is no schema on the expected side, so no schema comparison happens at all — test_amount_eur_is_rounded_to_the_cent above asserts one value and nothing about types. That is fine, because it is a value test and the schema test sits next to it, but it is only fine while both exist. Passing a DataFrame as expected is what makes assertDataFrameEqual compare schemas too; a list of Row never does.
And assertSchemaEqual is not strict by default any more. From PySpark 4.0 both it and assertDataFrameEqual take ignoreNullable, defaulting to True: the nullable property of the compared columns is simply not taken into account. For a value test that default is a mercy. For the published-interface test it is a hole — a gold column that quietly goes from NOT NULL to nullable is exactly the change Power BI notices and the contract test would not — so that one call passes ignoreNullable=False. The keyword arrived with 4.0; on PySpark 3.5 and earlier (DBR 16.x and below) the comparison is already strict and there is no parameter to relax it, so write the assertion for the version you actually pin rather than trying to satisfy both.
Doing it
assertDataFrameEqual for values, assertSchemaEqual for interfacesDo not let one stand in for the other.
Declare the expected gold schemas once, in tests/schemas.pyAssert against them from every test that touches the table. One file to diff when the DDL moves.
Pass ignoreNullable=False on every contract assertion, and nowhere elseStrict where the schema is the interface, relaxed where you are testing a value.
Select the columns you are asserting aboutComparing all thirty makes every unrelated change a failure in this test.
Embedding it
Break one value on purpose and let the team read the failure outputThey will stop writing collect-and-compare within the week.
Ask for the assertion first in review, before the transformationA test whose assertion is hard to write is usually a function doing two jobs.
Have the team add the schema-contract test for the table BI already readsThen rename a column and watch it fire. That demonstration is what buys the practice.
Follow it immediately with the quieter one: drop ignoreNullable=False and make a gold column nullableLet them watch the same test stay green. A team that has seen a contract test pass through a breaking change stops trusting green as a state and starts reading what was compared.
Defaults
One behaviour per test, named in the test name, with the business rule number in the docstring
Exact comparison on DECIMAL; no tolerance on money, ever
Schema contracts asserted for every gold table, in one file, so the interface is visible in one placeAssert them with ignoreNullable=False, or the contract does not cover nullability.
Gotchas
On PySpark 3.5 and earlier — that is, DBR 16.x and below — a nullability mismatch fails the comparisonA frame from createDataFrame carries nullable=False on the fields you declared; the same data after a join comes back nullable=True, so the test fails over a difference that changes no number and no report, and the usual reaction is to weaken the assertion rather than to build expected the same way. From PySpark 4.0 this specific failure is gone, because ignoreNullable defaults to True — which means an upgrade silently retires a class of red tests, and nobody notices that the contract test went quiet with them.
assertSchemaEqual on PySpark 4.0+ will NOT fail when a gold column changes from NOT NULL to nullableThat is the other half of the same default: assertSchemaEqual(actual.schema, FCT_ORDER_LINE_SCHEMA) leaves the contract test green through a real breaking change to a published interface, and the first report of it is a Power BI refresh error or a downstream model that starts emitting NULL keys. Pass ignoreNullable=False on every contract assertion.
rtol and atol relax FLOAT and DOUBLE comparisonIf amount_eur is a double, a tolerance makes a wrong cent pass. Keeping money in DECIMAL removes the temptation, because decimals compare exactly.
Both sides are materialised to the driverPointing this at a real gold table under a Connect run collects millions of rows into the client process and either hangs or dies. Use it on seeded data; use the zero-rows SQL pattern for production-scale checks.
Comparing the output of a function against a value computed by importing the same functionIt is common, it is always green, and it is worth nothing. The expected value is a constant, written by hand, or it is not an expected value.