Testing Data Pipelines in Python: Fixtures, Warehouse Fakes, and Property-Based Transforms

Most data pipeline test suites I inherit have one of two shapes. Either there are no tests, and every deploy is verified by watching a dashboard the next morning. Or there is a wall of tests that mock the warehouse client so thoroughly that they assert the mock was called, prove nothing about the SQL, and still pass when the transform is wrong.

The second is worse, because it produces confidence.

The pipeline that finally taught me a better approach had a 90-line function that opened a Snowflake connection, ran a query, reshaped the result, called an API for enrichment, and wrote back. Every attempt to test it turned into mocking four things. The fix was not a better mocking library. It was pulling the twelve lines of actual business logic into a function that takes a frame and returns a frame.

That separation is most of the battle. What follows is the layered strategy I use now: fast pure tests for transforms, real SQL against DuckDB for anything involving a warehouse, property-based tests for invariants, and a clear line between what CI can prove and what only production data checks can.

The four layers, and what each one is for

flowchart TD
  l1["Layer 1: pure transform tests"] --> fast["Milliseconds, no I/O, run on every save"]
  l2["Layer 2: SQL against DuckDB"] --> real["Real SQL semantics, seconds, runs in CI"]
  l3["Layer 3: contract and integration"] --> slow["Real warehouse, minutes, runs pre-merge"]
  l4["Layer 4: data quality in production"] --> prod["Runs on live data, alerts on drift"]
  fast --> ci["CI gate"]
  real --> ci
  slow --> ci
  prod --> pager["Pager, not CI"]

Layers 1 through 3 test the code. Layer 4 tests the data. Conflating them is why suites get slow and still miss incidents.

The distinction that matters most is the last one. A test that asserts “revenue is never negative” is a data quality check. It belongs in production against live data, on a schedule, with an alert. Putting it in CI against a fixture proves only that your fixture has no negative revenue. Both are needed; they are different systems with different failure responses.

Layer 1: make transforms pure

A pure transform takes data in and returns data out. No connections, no clock, no filesystem, no config lookup. Once code has that shape, testing is trivial and the test stays fast forever.

# Before: untestable without four mocks.
def process_orders(conn, api_key, run_date):
    df = pd.read_sql("select * from raw.orders", conn)
    df = df[df["order_date"] == run_date]
    df["tier"] = df["amount"].apply(lambda a: "gold" if a > 1000 else "standard")
    rates = requests.get(f"https://fx.example/rates?key={api_key}").json()
    df["amount_usd"] = df["amount"] * df["currency"].map(rates)
    df.to_sql("fct_orders", conn, if_exists="append")

# After: I/O at the edges, logic in the middle.
def assign_tier(df: pd.DataFrame) -> pd.DataFrame:
    """Tier is a pure function of amount. Boundary is inclusive at 1000."""
    return df.assign(
        tier=np.where(df["amount"] > 1000, "gold", "standard")
    )

def convert_currency(df: pd.DataFrame, rates: dict[str, float]) -> pd.DataFrame:
    """Rates are passed in, never fetched. Unknown currency -> null, not a crash."""
    return df.assign(amount_usd=df["amount"] * df["currency"].map(rates))

def run(conn, fx_client, run_date):
    """Thin orchestration. Nothing here needs a unit test."""
    df = read_orders(conn, run_date)
    df = convert_currency(assign_tier(df), fx_client.rates_for(run_date))
    write_orders(conn, df)

Now the tests are boring, which is the goal:

def test_tier_boundary_is_exclusive_at_1000():
    df = pd.DataFrame({"amount": [999.99, 1000.00, 1000.01]})
    out = assign_tier(df)
    assert out["tier"].tolist() == ["standard", "standard", "gold"]

def test_unknown_currency_yields_null_not_exception():
    df = pd.DataFrame({"amount": [100.0], "currency": ["XYZ"]})
    out = convert_currency(df, {"USD": 1.0, "EUR": 1.08})
    assert out["amount_usd"].isna().all()

That first test encodes a decision — is 1000 gold or standard? — that otherwise lives only in someone’s memory. Tests on boundaries are the highest-value tests in a data codebase, because boundaries are where the business rule and the implementation disagree.

run() gets an integration test at layer 3, or no test at all. It has no logic worth asserting.

Layer 2: DuckDB as a warehouse fake

Most pipeline logic is SQL, and SQL cannot be tested by mocking the connection. You need something that parses and executes it.

DuckDB does this in-process, with no server, in milliseconds, and its dialect is close enough to Postgres and Snowflake that the overwhelming majority of analytical SQL runs unchanged. It is the best testing tool the data ecosystem has produced in years, and it is worth reading the DuckDB local pipelines post for the broader case.

import duckdb
import pytest

@pytest.fixture
def warehouse():
    """In-memory warehouse with the schema our SQL expects."""
    con = duckdb.connect(":memory:")
    con.execute("""
        create table raw_orders (
            order_id    varchar,
            customer_id varchar,
            amount      decimal(12,2),
            status      varchar,
            updated_at  timestamp
        );
        create table dim_customers (
            customer_id varchar,
            segment     varchar,
            updated_at  timestamp
        );
    """)
    yield con
    con.close()

def test_dedup_keeps_latest_row_per_order(warehouse):
    warehouse.execute("""
        insert into raw_orders values
          ('O1', 'C1', 100.00, 'pending', '2026-01-01 10:00:00'),
          ('O1', 'C1', 150.00, 'shipped', '2026-01-01 11:00:00'),
          ('O2', 'C2',  50.00, 'shipped', '2026-01-01 10:00:00');
    """)
    result = warehouse.execute(DEDUP_SQL).fetchdf()

    assert len(result) == 2
    o1 = result[result["order_id"] == "O1"].iloc[0]
    assert o1["status"] == "shipped"
    assert float(o1["amount"]) == 150.00

def test_join_does_not_fan_out_on_duplicate_customer(warehouse):
    """Regression test: a duplicated dimension key must not multiply orders."""
    warehouse.execute("""
        insert into raw_orders values
          ('O1', 'C1', 100.00, 'shipped', '2026-01-01 10:00:00');
        insert into dim_customers values
          ('C1', 'enterprise', '2026-01-01'),
          ('C1', 'smb',        '2026-01-02');   -- the bug
    """)
    result = warehouse.execute(ENRICH_SQL).fetchdf()
    assert len(result) == 1, "join fanned out on a duplicate dimension key"

That second test is the one I care about most. It encodes the failure class from the fan-out postmortem as an executable check, and it runs in about 30 milliseconds. Every production incident should leave behind a test like this — three rows of setup and an assertion that the specific thing cannot happen again.

Where DuckDB will not help: warehouse-specific syntax. QUALIFY works, MERGE works, but Snowflake’s object_construct, lateral flatten, and dateadd argument order differ. Two options — keep the SQL portable, or use a dialect translation layer like SQLGlot in tests. I prefer portable SQL, since it also makes migrations cheaper, but a handful of translation shims is a fine compromise.

Layer 3: fixtures that look like production

Handcrafted fixtures are clean, and clean data is the data your pipeline already handles. The bugs live in the ugly rows.

My rule: build fixtures by sampling production and keeping every weird row you find. A committed fixture of 500 rows that includes the customer with an emoji in their name, the order with a null amount, the timestamp on a DST boundary, and the duplicate key from last March is worth more than 100,000 generated rows.

@pytest.fixture(scope="session")
def orders_fixture():
    """
    Sampled from production 2026-03-01, scrubbed, plus known edge cases.
    Do not regenerate without keeping the rows in edge_cases.csv.
    """
    return pd.read_parquet("tests/fixtures/orders_sample.parquet")

Two practices keep this maintainable. Scrub PII at generation time with a deterministic hash so the same source row always maps to the same fake value — otherwise the fixture churns and every diff is noise. And keep the deliberately weird rows in a separate, hand-edited file that is concatenated in, so a refresh of the sample cannot silently drop them.

For schema drift specifically, tests are the wrong tool. A test asserts today’s schema; what you want is to be notified when the producer changes theirs, which is what a data contract and a schema check in the ingestion path are for.

Layer 3b: property-based tests

Handcrafted cases test what you thought of. Property-based tests generate inputs and check that invariants hold, which finds the cases you did not think of. For data transforms this fits unusually well, because transforms have obvious invariants.

from hypothesis import given, settings
from hypothesis import strategies as st
import hypothesis.extra.pandas as hpd

order_frames = hpd.data_frames(
    columns=[
        hpd.column("order_id", elements=st.text(min_size=1, max_size=8)),
        hpd.column("amount", elements=st.floats(
            min_value=0, max_value=1e6, allow_nan=False)),
        hpd.column("currency", elements=st.sampled_from(["USD", "EUR", "XYZ"])),
    ],
    index=hpd.range_indexes(min_size=0, max_size=200),   # includes empty
)

@given(df=order_frames)
def test_tier_assignment_preserves_row_count(df):
    """A transform that classifies must never add or drop rows."""
    assert len(assign_tier(df)) == len(df)

@given(df=order_frames)
def test_tier_is_always_one_of_two_values(df):
    out = assign_tier(df)
    assert set(out["tier"].unique()) <= {"gold", "standard"}

@given(df=order_frames)
@settings(max_examples=200)
def test_dedup_is_idempotent(df):
    """Deduplicating twice equals deduplicating once."""
    once = dedup_orders(df)
    twice = dedup_orders(once)
    pd.testing.assert_frame_equal(
        once.reset_index(drop=True), twice.reset_index(drop=True)
    )

The invariants that pay off most in data work:

• Row count preservation for anything that maps or classifies.
• Idempotence for dedup, merge, and upsert logic. Running it twice must equal running it once, or your backfills are unsafe.
• Sum preservation for anything that splits, allocates, or pivots. Total in equals total out, within float tolerance.
• Key uniqueness at the declared output grain.
• Order independence — shuffling the input rows must not change the output, unless the transform is explicitly ordered.

That last one catches a specific, nasty bug: a ROW_NUMBER or drop_duplicates with a non-unique sort key, where the survivor depends on input order. It is the determinism problem from the deduplication post, and Hypothesis finds it in seconds by shuffling.

The empty frame is the single most valuable generated case. Hypothesis produces it immediately, and in my experience roughly one in four transforms crashes or returns the wrong dtype on an empty input. That happens in production the first quiet holiday.

Layer 4: data quality checks, which are not tests

Everything above runs against fixed inputs and answers “is the code correct.” Nothing above can tell you that yesterday’s file arrived with half the usual rows.

Data quality checks run against live data on a schedule, and they fail loudly to a person rather than to a CI status. They belong in the pipeline, not the test suite.

def check_freshness(con, table: str, ts_col: str, max_lag_hours: int):
    lag = con.execute(f"""
        select datediff('hour', max({ts_col}), current_timestamp())
        from {table}
    """).fetchone()[0]
    if lag is None or lag > max_lag_hours:
        raise DataQualityError(f"{table} is {lag}h stale (limit {max_lag_hours}h)")

def check_volume_within_band(con, table: str, date_col: str, tolerance=0.4):
    """Today's row count against the trailing 7-day median."""
    today, median = con.execute(f"""
        with daily as (
          select {date_col}::date as d, count(*) as n
          from {table}
          where {date_col} >= current_date - 8
          group by 1
        )
        select
          max(case when d = current_date then n end),
          median(case when d < current_date then n end)
        from daily
    """).fetchone()
    if median and abs(today - median) / median > tolerance:
        raise DataQualityError(f"volume {today} vs median {median}")

Keep the same discipline you would apply in dbt: freshness, volume, uniqueness, null rates, and referential integrity, each with an owner and a threshold. The framework choices are covered in the data quality checks post and the dbt testing strategy post; the point here is only that these do not belong in pytest.

Pitfalls

Mocking the warehouse client. You end up asserting that execute was called with a string. The SQL is never parsed, so a typo, a wrong join, or a broken window function passes. Use DuckDB.

Testing orchestration instead of logic. Airflow DAG tests that verify task dependencies catch import errors and little else. Test the callables; validate the DAG structure with a single import check.

Fixtures that are too clean. Hand-written test data has no nulls, no duplicates, no weird encodings, and no zero-row days. Sample production and keep the ugly rows.

Putting data quality assertions in CI. “Revenue is positive” against a fixture is tautological. Against production data on a schedule it is a real control.

No test for the empty input. Roughly a quarter of transforms break on zero rows, and it always surfaces on a holiday. One test, every transform.

Snapshot tests over full outputs. A committed CSV of expected output is brittle: every legitimate change produces a large, unreviewable diff and people regenerate it without reading. Assert specific properties instead.

Non-deterministic tests. Anything using now(), a random seed, or unordered group output will flake. Inject the clock, freeze the seed, sort before comparing.

FAQ

How much coverage should a data pipeline have?

Coverage percentage is a poor target here, since most lines are I/O glue. Aim for complete coverage of the pure transform functions and every business rule boundary, and be comfortable with near-zero coverage of connection and orchestration code.

Should I test my SQL or just my Python?

Test the SQL. It is where the logic lives in most pipelines, and DuckDB makes it fast enough to run on every commit. A Python suite that skips the SQL is testing the wrapper.

How do I test against a warehouse-specific feature?

Isolate it. If a model needs Snowflake’s flatten, split the warehouse-specific projection from the portable logic and test the portable part in DuckDB, with one integration test against a real Snowflake dev schema for the rest.

Is Hypothesis worth the setup cost?

For transform functions with clear invariants, yes — it finds empty frames, nulls, and ordering dependence with a few lines. For I/O-heavy code, no. I use it on maybe 20 percent of functions and it has found bugs in most of them.

How do I test incremental logic?

Simulate multiple runs against a DuckDB table: run the increment, assert output, run the same increment again, assert the result is unchanged. Then run an overlapping window and assert no duplicates appear. Idempotence and overlap handling are the two things incremental logic gets wrong.

Where do integration tests against the real warehouse fit?

A small number, on a dedicated dev schema, running pre-merge rather than on every commit. Their job is to catch dialect differences and permission problems, not logic errors. Keep the count low enough that the suite stays under a few minutes.

What this means for your pipelines

The reason data pipelines go untested is not that testing them is hard. It is that the code is usually written in a shape that cannot be tested, and once that shape exists, every attempt turns into a mocking exercise that produces nothing worth keeping.

Pull the logic out. A transform that takes a frame and returns a frame is testable in three lines, and the act of extracting it tends to reveal that the business rule was never written down anywhere. Then use DuckDB for the SQL, because a fake warehouse that actually executes SQL catches an entire class of bug that mocks structurally cannot. Add property-based tests for the handful of invariants that must always hold — row counts, idempotence, sum preservation — and let the generator find the empty frame before production does.

Then draw the line clearly. CI proves the code does what you meant. Production data checks prove the data is what you expected. Teams that blur these end up with slow test suites that still miss incidents, because the fixture always looks fine. Keep them separate, give the second one a pager, and every incident you resolve should leave behind a three-row test in the first.