ETL Testing¶

loom.etl.testing provides in-memory harnesses for testing ETLStep subclasses without spinning up a real catalog or writing to Delta Lake.

Test a single step with Polars¶

PolarsStepRunner seeds source tables as plain Python tuples, runs one step, and returns a StepResult for assertions.

from loom.etl.testing import PolarsStepRunner

def test_clean_orders():
    runner = PolarsStepRunner()
    runner.seed("raw.orders", [(1, 10.0), (2, -5.0)], ["id", "amount"])
    result = runner.run(CleanOrdersStep, NoParams())

    result.assert_count(1)
    result.assert_schema({"id": "Int64", "amount": "Float64"})

StepResult methods: assert_schema, assert_count, assert_not_empty, show, to_polars.

Steps that declare `FromConfig` values¶

with_config gives the runner the config a step’s FromConfig values resolve from, so a test passes a fake token instead of setting an environment variable. The keys are checked when the step compiles, as ETLRunner does. SparkStepRunner has the same method.

def test_fetch_uses_the_token():
    runner = PolarsStepRunner().with_config({"respondio": {"api_token": "fake"}})
    result = runner.run(FetchMessages, DailyParams(run_date=date(2026, 9, 25)))
    result.assert_not_empty()

A step whose target the harness cannot capture (an IntoTemp, say) can be called directly: FetchMessages().execute(params, api_token="fake", ...).

Sources that declare `parse_json`¶

The harness passes seeded frames to the step as they are; it does not decode the columns a source declares with parse_json. To give the step what the Polars reader would, decode the seed with decode_json_columns, which runs the reader’s own decoding (including the normalisation of fields whose JSON type changes between rows) on an eager or lazy frame:

from loom.etl.backends.polars import decode_json_columns

events = decode_json_columns(raw_events, EnrichEvents.events.json_columns)
result = EnrichEvents().execute(params, events=events.lazy())

json_columns is the tuple of declarations on a FromTable or FromFile.

Targets that declare `replace(allow_empty=False)`¶

PolarsStepRunner captures the step’s result instead of writing it, and StubTargetWriter stores whatever it receives, so neither applies allow_empty: an empty result passes without EmptyReplaceError. Test the declaration on the compiled spec, and the guard itself with PolarsTargetWriter on a temporary directory if needed:

from loom.etl.compiler import ETLCompiler
from loom.etl.declarative.target import ReplaceSpec


def test_dim_dealer_refuses_an_empty_replace():
    spec = ETLCompiler().compile_step(BuildDimDealer).target_binding.spec
    assert isinstance(spec, ReplaceSpec)
    assert spec.allow_empty is False

Reusable seed datasets with `ETLScenario`¶

ETLScenario stores seed data as plain tuples so the same dataset can be used with any backend runner.

from loom.etl.testing import ETLScenario

ORDERS = (
    ETLScenario()
    .with_table("raw.orders", [(1, 10.0), (2, 20.0)], ["id", "amount"])
)

def test_double_amount(loom_polars_runner):
    ORDERS.apply(loom_polars_runner)
    result = loom_polars_runner.run(DoubleAmountStep, NoParams())
    result.assert_count(2)

Pytest fixtures: loom_polars_runner. Import loom.etl.testing.spark or register it explicitly in Spark-enabled suites to use loom_spark_runner.

Spark integration tests¶

Use SparkTestSession for a real PySpark + Delta session:

from loom.etl.testing.spark import SparkTestSession

with SparkTestSession.start() as spark:
    runner = SparkStepRunner(spark)
    runner.seed("raw.orders", [(1, 10.0)], ["id", "amount"])
    result = runner.run(CleanOrdersStep, NoParams())
    result.assert_count(1)

Compile-time stubs¶

StubCatalog — in-memory catalog.
StubSourceReader / StubTargetWriter — in-memory I/O.
StubRunObserver — records pipeline lifecycle events.

ETL Testing¶

Test a single step with Polars¶

Steps that declare FromConfig values¶

Sources that declare parse_json¶

Targets that declare replace(allow_empty=False)¶

Reusable seed datasets with ETLScenario¶