Seeding & Fixtures: Realistic Test Data for Warehouse Integration Tests

Seeding and fixtures are the two halves of the same job: building the world a warehouse test needs before it runs, and tearing that world down after — because the moment you stop testing a lone Python function and start testing a SQL model, a dbt transform, or a join across three tables, the thing under test is no longer your code but the data plus the query, and a query is only as trustworthy as the test data you fed it. A transform that passes against two hand-typed rows will still corrupt production if the real data has a null foreign key, a duplicate business key, or a customer with no orders — cases your toy fixture never contained. The hard part of testing a pipeline was never "call the function"; it was manufacturing a realistic, referentially-consistent input and getting back to a clean slate before the next test.

This guide is the walkthrough for doing that well — treating the warehouse as the system under test and building integration tests that seed it with realistic data, run the real transform, and assert on the output. It is framed the way the topic actually comes up in senior data-engineering interviews and code review: why unit tests cannot cover a SQL/dbt pipeline, how pytest fixtures scope and tear down the database (function versus session, transactional rollback versus truncate), how a factory mints valid rows and keeps a seed graph's referential integrity intact while staying deterministic, how the ephemeral-schema and DuckDB patterns let you spin up a throwaway warehouse per test, and how isolation, parallelism, teardown, and CI keep the suite fast and non-flaky. Each section pairs a teaching block with a worked interview answer — code, a step-by-step trace, an output table, then a concept-by-concept breakdown of why it works.

When you want hands-on reps immediately after reading, drill the data validation practice library →, rehearse pipeline scenarios on the ETL practice library →, and sharpen the test-architecture axis with the system design practice library →.

---

On this page

• Why integration tests need realistic seed data
• Fixtures — pytest scopes, setup, teardown, and rollback
• Factories and seeding — FK-consistent, deterministic data
• Integration test patterns — ephemeral schema, DuckDB, assert
• Isolation, parallelism, teardown, and CI
• Cheat sheet — seeding & fixtures
• Frequently asked questions
• Practice on PipeCode

---

1. Why integration tests need realistic seed data

Unit tests check a function; the warehouse is the system under test, and only realistic seed data exercises it

The one-sentence invariant: an integration test for a data pipeline puts the warehouse under test — the real SQL, the real dbt model, the real joins — so its result is only as trustworthy as the seed data behind it, which means the whole discipline of seeding and fixtures is about manufacturing realistic, referentially-consistent test data, running the actual transform against it, asserting on the output, and returning to a known-clean state — because a transform that passes on two hand-typed rows will still break on production data that has a null foreign key, a duplicate key, or an empty group your toy fixture never contained. Unit-test the pure Python you own; integration-test the SQL, because the SQL is the logic and the data is the input.

Why unit tests are not enough for a pipeline.

• The logic lives in the SQL, not in Python. A dbt model or a warehouse view is where the aggregation, the join, and the window function live. A unit test that mocks the database tests your glue code and asserts nothing about the query that actually produces the numbers.
• Mocks assert your assumptions, not reality. A mocked cursor returns whatever you told it to; it can never surface a real LEFT JOIN that fans out, a GROUP BY that drops a null key, or a SUM that overflows. The bugs that reach production are exactly the ones a mock cannot reproduce.
• Data shape is the input space. The interesting cases — a customer with zero orders, an order with a deleted product, two rows with the same natural key, a late-arriving dimension — are properties of the data, so you can only cover them by seeding them.

The four axes interviewers actually probe.

• Fidelity. Is the seed data realistic and referentially consistent? A seed graph where orders.customer_id points at a customer that does not exist tests nothing a real join would hit. The senior answer builds a valid graph — parents before children, foreign keys satisfied — and seeds the edge cases on purpose.
• Isolation. Does each test start from a known-clean state, independent of every other test? A test that passes alone but fails in the suite (or vice versa) has a state-leak bug. The senior answer names transactional rollback, truncation, or schema-per-test as the isolation mechanism.
• Determinism. Does the same seed produce the same result every run? Random data and wall-clock timestamps make assertions flap. The senior answer reseeds the random generator and freezes the clock so the output is reproducible.
• Teardown and cost. Is the world cleaned up afterward, and is the suite fast enough to run on every push? The senior answer guarantees teardown even on failure and picks an isolation strategy (rollback, DuckDB) that keeps CI in seconds, not minutes.

The 2026 reality — the toolkit is small and well-worn.

• pytest fixtures are the setup/teardown backbone: a yield-based function that builds a resource (an engine, a connection, a seeded schema), hands it to the test, and cleans up afterward — composed and scoped to control cost.
• factory_boy (or a small builder) manufactures valid rows on demand, wiring foreign keys with SubFactory so every seed graph is referentially consistent, and overriding only the fields a given test cares about.
• DuckDB or an ephemeral schema gives each test a throwaway warehouse — DuckDB in-process with zero setup, or a CREATE SCHEMA on a real Postgres/warehouse that you DROP on teardown.
• dbt seeds load static reference data (a region map, a date spine) from CSV so the model's lookups resolve, complementing the row-level factories.

What interviewers listen for.

• Do you say the warehouse (the SQL) is the system under test and explain why a mock cannot cover it? — senior signal.
• Do you build a referentially-consistent seed graph (parents before children, valid foreign keys) rather than two isolated rows? — required answer.
• Do you make seed data deterministic (reseed the RNG, freeze the clock) so assertions do not flap? — senior signal.
• Do you name an isolation and teardown strategy so tests are independent and the world is left clean? — required answer.

Worked example — the unit-vs-integration decision table

Detailed explanation. The single most useful artifact for a pipeline-testing interview is a memorised mapping of what to test at which level. Every senior discussion converges on it: pure transformation logic you own is a unit test; anything whose behaviour depends on the database engine, the SQL, or the data shape is an integration test. Walk through classifying the pieces of a typical dbt-plus-Python pipeline.

• The unit-testable. A pure Python function that parses a row, a helper that formats a currency, a validation predicate — no database, no SQL.
• The integration-testable. A dbt model, a warehouse view, a MERGE/upsert, a join across tables, an incremental model's late-arriving logic — the behaviour is in the SQL and the data.
• The rule. If a mock could fake the answer, it is a unit test; if the answer depends on the engine executing SQL over seeded data, it is an integration test.

Question. For each pipeline component, name the test level and what you must seed (if anything).

Input.

Component	Behaviour depends on	Test level
parse_amount("$4.20") helper	pure Python	unit — no seed
stg_orders dbt model	SQL over raw table	integration — seed raw
daily-sales aggregate view	GROUP BY semantics	integration — seed orders
incremental model (late data)	merge + watermark	integration — seed two batches
a Pydantic row validator	pure Python	unit — no seed

Code.

# UNIT: pure logic you own — no database, fast, exhaustive on edge inputs.
def parse_amount(raw: str) -> int:
    """'$4.20' -> 420 cents."""
    return round(float(raw.strip().lstrip("$")) * 100)

def test_parse_amount():
    assert parse_amount("$4.20") == 420
    assert parse_amount("0") == 0                 # no mock, no seed, microseconds

# INTEGRATION: the LOGIC is in the SQL, so the test must run the SQL over seeded rows.
def test_daily_sales_rollup(seeded_warehouse):   # fixture seeds orders + runs transform
    got = seeded_warehouse.query(
        "SELECT region, revenue_cents FROM daily_sales ORDER BY region")
    # the GROUP BY, the SUM, the null handling — none of it is Python you can unit-test.
    assert got == [("EU", 5000), ("US", 990)]

Step-by-step explanation.

1. parse_amount is pure Python with no I/O, so it is a unit test: call it with representative and edge inputs and assert the return value. It needs no seed data and runs in microseconds — the right level for logic you own.
2. daily_sales, by contrast, is produced by a SQL GROUP BY ... SUM(...). There is no Python function to call — the behaviour is the query — so the only way to test it is to seed the input rows and run the actual SQL.
3. The integration test's assertion (EU -> 5000) is a claim about what the engine does with the seeded data: how SUM treats nulls, whether the GROUP BY drops an empty region, whether a LEFT JOIN fans out. A mock returning a fixed list would test none of that.
4. The incremental-model row is the sharpest example: its correctness depends on running the model twice with two seeded batches and asserting the second run merges rather than duplicates — a purely data-and-engine behaviour.
5. The mistake is unit-testing the pipeline by mocking the warehouse: the test goes green while the SQL is wrong, because the mock faked the very answer the SQL was supposed to compute. The decision table is the antidote — level follows where the behaviour lives.

Output.

Component	Right level	Wrong choice (common mistake)
Pure parse/format/validate	unit, no seed	spinning up a DB (slow, pointless)
dbt model / view / join	integration, seed input	mock the cursor (tests nothing)
Aggregate / window logic	integration, seed groups	assert on hand-typed mock rows
Incremental / merge	integration, seed two batches	single-batch unit test

Rule of thumb. Draw the line by where the behaviour lives: pure Python you own is a unit test with no seed; anything whose result depends on the engine executing SQL over data is an integration test that must be seeded. Mocking the warehouse to "unit test" a transform tests your assumptions, not the query.

Worked example — arrange-act-assert for a warehouse transform

Detailed explanation. Every warehouse integration test has the same three-beat shape: arrange a seeded input, act by running the real transform, assert on the output. Naming the beats keeps the test readable and forces you to seed a realistic input rather than reverse-engineering rows to make an assertion pass. Build the skeleton for a daily-sales rollup.

• Arrange. Seed orders with a referentially-consistent, edge-case-bearing set of rows.
• Act. Run the actual transform SQL (the same statement dbt/production runs), not a paraphrase.
• Assert. Query the output table and compare it to the expected result.

Question. Write the arrange-act-assert skeleton for a transform that rolls orders up to daily revenue by region, and identify what each beat must contain.

Input.

Beat	Contains	Failure if skipped
Arrange	seeded orders (incl. edge cases)	tests an empty/unreal input
Act	the real transform SQL	tests a paraphrase, not prod
Assert	output compared to expected	test passes without checking
Teardown	reset to clean state	next test sees leaked rows

Code.

def test_daily_sales_by_region(db):
    # ---- ARRANGE: seed a realistic, referentially-consistent input --------------
    db.execute("""
        INSERT INTO orders (id, region, total_cents, order_date) VALUES
          (1, 'EU', 4200, DATE '2026-01-01'),
          (2, 'EU',  800, DATE '2026-01-01'),   -- same day+region -> must aggregate
          (3, 'US',  990, DATE '2026-01-01'),
          (4, 'EU', 1500, DATE '2026-01-02')    -- different day -> separate group
    """)

    # ---- ACT: run the REAL transform (identical to the production model) ---------
    db.execute("""
        CREATE TABLE daily_sales AS
        SELECT order_date, region,
               SUM(total_cents) AS revenue_cents,
               COUNT(*)         AS n_orders
        FROM orders
        GROUP BY order_date, region
    """)

    # ---- ASSERT: query the output and compare to the expected result ------------
    rows = db.execute("""
        SELECT order_date, region, revenue_cents, n_orders
        FROM daily_sales ORDER BY order_date, region
    """).fetchall()

    assert rows == [
        (date(2026, 1, 1), 'EU', 5000, 2),   # 4200 + 800 aggregated
        (date(2026, 1, 1), 'US',  990, 1),
        (date(2026, 1, 2), 'EU', 1500, 1),
    ]
    # teardown handled by the `db` fixture (rollback/drop) — see section 2.

Step-by-step explanation.

1. The arrange beat seeds four rows chosen to exercise the transform: two EU rows on the same day (to prove aggregation), a US row (a second group), and a second day (to prove the grain). The input is deliberately shaped to make the assertion meaningful.
2. The act beat runs the same SQL the production model runs. Copying the statement verbatim (or invoking the dbt model) is what makes this an integration test — a rephrased query would test a different transform than the one that ships.
3. The assert beat reads the output table back and compares it to an explicit expected result. The EU total of 5000 is the point of the test: it proves the GROUP BY ... SUM collapsed the two same-day rows correctly.
4. Ordering the output (ORDER BY order_date, region) before comparing makes the assertion stable — SQL does not guarantee row order, so comparing an unordered result to an ordered list is a classic flaky-test cause.
5. The teardown is implicit here but essential: the db fixture (next section) rolls back or drops the seeded rows and the daily_sales table, so the next test starts clean. Without it, the leaked daily_sales table would collide with the next test's CREATE TABLE.

Output.

order_date	region	revenue_cents	n_orders
2026-01-01	EU	5000	2
2026-01-01	US	990	1
2026-01-02	EU	1500	1

Rule of thumb. Structure every warehouse test as arrange (seed a realistic input) → act (run the real transform) → assert (compare the output to an explicit expected result), and always order the output before comparing. If the arrange step is two isolated rows, the test proves almost nothing — seed the cases that make the assertion earn its keep.

Worked example — a referentially-consistent seed graph versus toy data

Detailed explanation. The most common seeding bug is data that is not referentially consistent: an order whose customer_id matches no customer, or a fact row whose dimension key is missing. A LEFT JOIN silently returns nulls, an INNER JOIN silently drops the row, and the test passes on data that could never exist in production. Build a valid graph — customers, then their orders, then their items — so the transform's joins actually fire.

• The toy trap. Insert an orders row with customer_id = 99 when no customer 99 exists; the join produces a null name and the bug hides.
• The consistent graph. Insert the customer first, then the order referencing it, then the items referencing the order — every foreign key resolves.
• The payoff. The transform's JOIN to customers returns a real name, so a test on "revenue by customer segment" actually exercises the join.

Question. Contrast a toy seed (orphan foreign keys) with a referentially-consistent seed graph, and show how each behaves under the transform's join.

Input.

Seed style	orders.customer_id	Join result	Test value
Toy (orphan FK)	points at no customer	null / dropped row	hides bugs
Consistent graph	points at a seeded customer	real joined row	exercises the join
Consistent + edge	one customer with zero orders	tests the LEFT JOIN null	covers reality

Code.

def seed_graph(db):
    # 1. PARENTS FIRST — a customer must exist before an order can reference it.
    db.execute("""INSERT INTO customers (id, name, segment) VALUES
        (1, 'Acme Retail', 'enterprise'),
        (2, 'Globex',      'smb'),
        (3, 'Initech',     'enterprise')""")     # customer 3: intentionally NO orders

    # 2. CHILDREN REFERENCE REAL PARENTS — every customer_id below exists above.
    db.execute("""INSERT INTO orders (id, customer_id, total_cents) VALUES
        (10, 1, 4200),
        (11, 1,  800),
        (12, 2,  990)""")                          # customer 3 -> tests the LEFT JOIN

    # 3. GRANDCHILDREN REFERENCE REAL ORDERS.
    db.execute("""INSERT INTO order_items (order_id, sku, qty) VALUES
        (10, 'WIDGET-1', 3), (10, 'GADGET-9', 1), (11, 'WIDGET-1', 1)""")

def test_revenue_by_segment(db):
    seed_graph(db)
    db.execute("""
        CREATE TABLE seg_revenue AS
        SELECT c.segment, COALESCE(SUM(o.total_cents), 0) AS revenue_cents
        FROM customers c
        LEFT JOIN orders o ON o.customer_id = c.id    -- LEFT JOIN: keep customer 3
        GROUP BY c.segment
    """)
    rows = db.execute(
        "SELECT segment, revenue_cents FROM seg_revenue ORDER BY segment").fetchall()
    # enterprise = Acme(5000) + Initech(0); smb = Globex(990)
    assert rows == [('enterprise', 5000), ('smb', 990)]

Step-by-step explanation.

1. The seed inserts parents before children: customers, then orders referencing those customers, then items referencing those orders. Every foreign key resolves, so the graph is a valid instance of the schema — exactly what production data looks like.
2. Customer 3 (Initech) is seeded with no orders on purpose. This is the edge case that a toy seed of "one customer, one order" would miss, and it is precisely what the transform's LEFT JOIN plus COALESCE(..., 0) has to handle.
3. The transform LEFT JOINs customers to orders so a customer with zero orders survives with revenue_cents = 0. The test asserts enterprise = 5000 — Acme's 5000 plus Initech's 0 — which only means something because Initech was seeded with no orders.
4. Had the seed used an orphan customer_id (say order 12 pointing at a non-existent customer 99), the LEFT JOIN's parent side would be missing and the row would silently vanish from a customers-driven query, or produce a null segment — a bug the assertion might not catch because the data was impossible to begin with.
5. The senior discipline: seed a graph, not rows. Model the parent/child/grandchild structure, satisfy every foreign key, and deliberately include the reality-bearing edges (a parent with no children, a child with many) so the joins under test are actually exercised.

Output.

segment	revenue_cents	why
enterprise	5000	Acme 5000 + Initech 0 (LEFT JOIN kept it)
smb	990	Globex 990
(orphan-FK toy)	wrong/null	order pointed at no customer

Rule of thumb. Seed a referentially-consistent graph — parents before children, every foreign key satisfied — and deliberately include edge shapes like a parent with zero children so the transform's joins are genuinely exercised. Orphan foreign keys produce data that cannot exist in production, so a test that passes on them proves nothing.

Senior interview question on why pipelines need integration tests

A senior interviewer often opens with: "Your team unit-tests its dbt pipeline by mocking the warehouse, and the tests are green — but a revenue model shipped a bug that double-counted refunds. Explain why the unit tests missed it, what an integration test would have caught, and how you would structure a test that seeds realistic data, runs the real model, and asserts on the output — including the referential integrity and edge cases you would seed."

Solution Using the warehouse as the system under test with a seeded graph and a real transform

# 1. THE DIAGNOSIS: the bug is in the SQL (a JOIN to refunds fans out), so a mock
#    that fakes the cursor's return value can never surface it. The logic under test
#    is the query itself — it must run against seeded data.

# 2. ARRANGE — a referentially-consistent graph WITH the edge that triggers the bug:
def seed(db):
    db.execute("INSERT INTO customers (id, name) VALUES (1, 'Acme')")
    db.execute("""INSERT INTO orders (id, customer_id, gross_cents) VALUES
        (10, 1, 5000)""")
    # the trap: TWO refund rows for one order -> a naive JOIN multiplies gross by 2.
    db.execute("""INSERT INTO refunds (order_id, refund_cents) VALUES
        (10, 500), (10, 300)""")

-- 3. ACT — the REAL model (the buggy shape, so the test can prove the fix).
--    BUGGY: JOIN to refunds fans out; gross is counted once PER refund row.
--    CREATE TABLE net_revenue AS
--      SELECT o.id, o.gross_cents - r.refund_cents AS net
--      FROM orders o JOIN refunds r ON r.order_id = o.id;   -- gross double-counted

-- FIXED: pre-aggregate refunds to one row per order BEFORE joining.
CREATE TABLE net_revenue AS
SELECT o.id,
       o.gross_cents - COALESCE(ref.total_refund, 0) AS net_cents
FROM orders o
LEFT JOIN (
    SELECT order_id, SUM(refund_cents) AS total_refund
    FROM refunds GROUP BY order_id           -- collapse many refunds -> one row
) ref ON ref.order_id = o.id;

# 4. ASSERT — the seeded case pins the correct net: 5000 - (500 + 300) = 4200.
def test_net_revenue_handles_multiple_refunds(db):
    seed(db)
    # ... run the FIXED transform above ...
    net = db.execute("SELECT net_cents FROM net_revenue WHERE id = 10").fetchone()[0]
    assert net == 4200        # buggy fan-out would give 4000 or 4500, never 4200

Step-by-step trace.

Step	Buggy pipeline	Integration test catches it
Seed	(mock returned a fixed net)	1 order, 2 refunds — the trigger
Run	JOIN fans out per refund	runs the REAL SQL over the seed
Result	gross counted twice	net = 4000/4500, not 4200
Assert	no real assertion	assert net == 4200 fails on bug
Fix	pre-aggregate refunds	test goes green on correct SQL

After adopting the integration test, the mocked unit test is replaced by one that seeds a single order with two refunds — the exact shape that makes a naive orders JOIN refunds fan out and double-count gross revenue. The test runs the real model and asserts net == 4200; the buggy fan-out yields the wrong number and the test fails, forcing the fix (pre-aggregate refunds to one row per order before joining). The warehouse is the system under test, the seed graph is referentially consistent, and the edge case is seeded on purpose.

Output:

Metric	Mocked unit test	Integration test
What is under test	glue code	the actual SQL
Refund fan-out bug	invisible (mock faked net)	caught (assert fails)
Referential integrity	none (no real rows)	enforced (seeded graph)
Confidence in prod	false green	real green
Edge coverage	whatever the mock returns	seeded on purpose

Why this works — concept by concept:

• Warehouse as the system under test — the transform's logic lives in the SQL, so the test runs the real query over seeded rows instead of mocking its result; a mock can only confirm the answer you already assumed, never the answer the engine computes.
• Referentially-consistent seed graph — seeding parents before children with valid foreign keys makes the input a real instance of the schema, so the join under test behaves exactly as it would in production rather than on impossible data.
• Seeding the edge on purpose — deliberately seeding one order with two refunds recreates the fan-out that triggers the bug, turning an invisible defect into a failing assertion; the test data is designed around the failure mode.
• An explicit expected output — asserting the exact net (4200) rather than "some number" is what distinguishes a test that catches double-counting from one that passes on any value the buggy SQL happens to return.
• Cost — one seeded order plus a real transform run against a throwaway warehouse, versus a mocked test that ships a revenue bug to production. The eliminated cost is a financial-reporting incident — O(rows seeded) to run a true integration test instead of O(revenue) to clean up a double-count in prod.

ETL
Topic — etl
ETL problems on transforms, joins, and pipeline correctness

Practice →

Data validation Topic — data-validation Data validation problems on referential integrity and edge cases

Practice →

---

2. Fixtures — pytest scopes, setup, teardown, and rollback

A fixture builds the world a test needs and tears it down — scope decides how often, transactions decide how clean

The mental model in one line: a pytest fixture is a yield-based function that runs setup before the test, hands the test whatever it built (a connection, a seeded schema, a factory), and runs teardown after — and two knobs control it: the scope (function/class/module/session) decides how often the setup runs, trading isolation for speed, while the reset strategy (transactional rollback versus TRUNCATE/DROP SCHEMA) decides how the database returns to clean between tests — so the expensive, immutable things (the engine, the schema DDL) live at session scope and the cheap, mutable things (the seeded rows) are undone per function, usually by wrapping each test in a transaction that is rolled back. Get the scopes right and the suite is fast and isolated; get them wrong and you either rebuild the world every test (slow) or leak state across tests (flaky).

What a fixture is and how it runs.

• Setup / yield / teardown. Everything before yield is setup; the value yielded is injected into any test (or fixture) that names the fixture as an argument; everything after yield is teardown, and it runs even if the test fails.
• Dependency injection. A test declares a fixture by putting its name in the signature (def test_x(db):); pytest resolves and runs the fixture, then passes its value in. Fixtures can depend on other fixtures the same way.
• conftest.py. Fixtures placed in a conftest.py are available to every test in that directory tree without importing — the standard home for the database, engine, and seed fixtures.

Fixture scope — how often setup runs.

• function (default). Setup/teardown around every test. Maximum isolation, maximum cost — the right scope for the mutable per-test state (seeded rows, the transaction).
• module / class. Once per module or class. Good for moderately expensive, read-mostly setup shared by a group of tests.
• session. Once for the whole test run. The right scope for the truly expensive, immutable things: creating the engine, spinning up the container, applying the schema DDL/migrations.
• The rule. Push each piece of setup to the widest scope at which it stays correct: build the engine once (session), reset the data every test (function).

Reset strategy — rollback versus truncate.

• Transactional rollback. Wrap each test in a transaction and roll it back in teardown; the database never actually persists the test's writes, so reset is instant. The fastest isolation, and the default choice when the code under test does not manage its own transactions.
• Truncate / delete between tests. TRUNCATE ... RESTART IDENTITY CASCADE (or targeted DELETE) after each test. Needed when the code under test commits (so a rollback would not undo it) or spans multiple connections.
• Drop / recreate schema. DROP SCHEMA ... CASCADE; CREATE SCHEMA — the heaviest reset, required when the transform issues DDL (CREATE TABLE, dbt building models) that a transaction cannot cleanly contain.
• The trade. Rollback is fastest but breaks if the SUT commits or uses a separate connection; truncate/drop always works but is slower — pick the lightest reset that survives what your transform actually does.

The failure modes senior engineers pre-empt.

• Session-scoped mutable state. Seeding rows in a session fixture means test 2 sees test 1's writes. Mitigation: keep mutable seed data at function scope; reserve session scope for immutable setup.
• Rollback that does not roll back. If the transform commits (or dbt opens its own connection), the outer transaction cannot undo it and rows leak. Mitigation: detect the commit boundary and switch that suite to truncate/drop-schema teardown.
• Teardown skipped on failure. Putting cleanup after the assertion (not in a fixture's teardown) means a failing test leaks its world. Mitigation: always clean up in the fixture's post-yield block, which pytest runs regardless of outcome.

Common interview probes on fixtures.

• "What scope for the database engine versus the seeded rows?" — engine at session (expensive, immutable), rows at function (cheap, mutable).
• "How do you reset the database between tests?" — transactional rollback by default; truncate or drop-schema when the SUT commits or issues DDL.
• "Where does teardown go so it runs on failure?" — after yield in the fixture; pytest runs it whether the test passes or raises.
• "How do fixtures share setup?" — a fixture depends on another by naming it as an argument; common ones live in conftest.py.

Worked example — a session engine plus a per-function rollback fixture

Detailed explanation. The canonical two-tier fixture setup: build the SQLAlchemy engine once at session scope, then give every test a connection wrapped in a transaction that is rolled back on teardown. The expensive thing happens once; the isolation happens every test, for free. Wire it in conftest.py.

• Session tier. engine — create once, dispose at the end of the run.
• Function tier. db — open a connection, begin a transaction, yield it, roll back.
• The payoff. Each test sees a pristine database and writes nothing durable.

Question. Write a conftest.py that builds the engine once per session and gives each test a rolled-back transactional connection.

Input.

Fixture	Scope	Setup	Teardown
engine	session	create engine, apply schema	dispose
db	function	connect, begin transaction	rollback, close
net effect	—	pristine DB per test	nothing persisted

Code.

# conftest.py — shared by every test in the tree, no import needed.
import pytest
from sqlalchemy import create_engine, text

@pytest.fixture(scope="session")
def engine():
    # SESSION scope: the engine + schema are expensive and immutable — build ONCE.
    eng = create_engine("postgresql+psycopg://test:test@localhost:5432/test")
    with eng.begin() as conn:
        conn.execute(text("CREATE TABLE IF NOT EXISTS orders "
                          "(id INT PRIMARY KEY, region TEXT, total_cents BIGINT)"))
    yield eng
    eng.dispose()                       # teardown runs after the whole session

@pytest.fixture(scope="function")
def db(engine):
    # FUNCTION scope: every test gets a connection inside a transaction we ROLL BACK,
    # so nothing the test writes is ever committed — the DB is pristine next test.
    conn = engine.connect()
    txn = conn.begin()
    try:
        yield conn                      # the test uses `db` to seed + query
    finally:
        txn.rollback()                  # undo EVERYTHING, even if the test failed
        conn.close()

def test_seed_is_isolated(db):
    db.execute(text("INSERT INTO orders VALUES (1, 'EU', 4200)"))
    assert db.execute(text("SELECT count(*) FROM orders")).scalar() == 1

def test_starts_clean(db):
    # proves isolation: the previous test's INSERT was rolled back, so this sees 0.
    assert db.execute(text("SELECT count(*) FROM orders")).scalar() == 0

Step-by-step explanation.

1. The engine fixture is scope="session", so creating the engine and applying the schema DDL happens exactly once for the entire test run — the expensive, immutable setup is not repeated per test.
2. The db fixture is scope="function" (the default, stated for clarity) and depends on engine. Because it names engine as an argument, pytest resolves the session engine first, then runs db's setup per test.
3. db opens a connection and begins a transaction before yield. The test seeds and queries through this connection, seeing its own writes — but they live only inside the open transaction.
4. In teardown (finally), txn.rollback() discards everything the test wrote and conn.close() returns the connection. Putting this in finally guarantees it runs even when the test raises, so a failing test never leaks rows.
5. The two tests prove the isolation: the first inserts a row and sees count 1; the second sees count 0 because the first test's insert was rolled back, never committed. Reset is instant — no delete, no truncate, just a discarded transaction.

Output.

Test	Sees before	Writes	Sees after (next test)
test_seed_is_isolated	0 rows	1 row	rolled back
test_starts_clean	0 rows	—	0 rows
engine setup	once (session)	—	disposed at end

Rule of thumb. Build the engine and schema once at session scope, and wrap every test in a function-scoped transaction you roll back in teardown. It gives per-test isolation at near-zero cost, and putting the rollback in a finally (post-yield) block guarantees the world is reset even when a test fails.

Worked example — truncate or drop-schema teardown when the transform commits

Detailed explanation. Transactional rollback quietly fails when the code under test commits or issues DDL — a dbt run, a CREATE TABLE AS, a stored procedure with COMMIT. The outer transaction cannot undo a committed write, so rows and tables leak into the next test. The fix is a heavier reset: truncate the data, or drop and recreate the schema. Build a fixture for a transform that materialises tables.

• Why rollback fails. The transform commits (or dbt opens its own connection), so the outer BEGIN/ROLLBACK has nothing to undo.
• The truncate reset. TRUNCATE ... RESTART IDENTITY CASCADE empties tables and resets sequences between tests.
• The drop reset. DROP SCHEMA ... CASCADE; CREATE SCHEMA — required when the transform creates new tables the test did not pre-declare.

Question. Write a teardown fixture for a transform that runs CREATE TABLE AS (which a rollback cannot cleanly undo), keeping each test isolated.

Input.

Reset	Works when	Cost
rollback	SUT never commits	cheapest
truncate	SUT commits into known tables	medium
drop schema	SUT creates new tables (DDL)	heaviest

Code.

import pytest
from sqlalchemy import create_engine, text

@pytest.fixture(scope="session")
def engine():
    eng = create_engine("postgresql+psycopg://test:test@localhost:5432/test")
    yield eng
    eng.dispose()

@pytest.fixture(scope="function")
def clean_schema(engine):
    # The transform issues DDL (CREATE TABLE AS) and COMMITs, so a rollback can't
    # undo it. Give each test a freshly DROP+CREATE'd schema instead.
    with engine.begin() as c:
        c.execute(text('DROP SCHEMA IF EXISTS it CASCADE'))
        c.execute(text('CREATE SCHEMA it'))
        c.execute(text('CREATE TABLE it.orders (id INT, region TEXT, total_cents BIGINT)'))
    yield "it"
    # teardown: drop whatever the transform created, no matter what it named.
    with engine.begin() as c:
        c.execute(text('DROP SCHEMA IF EXISTS it CASCADE'))

def test_transform_materialises_a_table(engine, clean_schema):
    with engine.begin() as c:                     # the SUT commits here
        c.execute(text("INSERT INTO it.orders VALUES (1,'EU',4200),(2,'EU',800)"))
        c.execute(text("""CREATE TABLE it.daily AS
                          SELECT region, SUM(total_cents) rev FROM it.orders
                          GROUP BY region"""))     # DDL — rollback couldn't undo this
    rev = engine.connect().execute(
        text("SELECT rev FROM it.daily WHERE region='EU'")).scalar()
    assert rev == 5000
    # `it.daily` is dropped by clean_schema teardown -> next test won't collide.

Step-by-step explanation.

1. The transform under test runs CREATE TABLE it.daily AS ... and commits it. A transactional-rollback fixture would leave it.daily behind, because a committed CREATE TABLE is not undone by rolling back the test's transaction — the classic reason rollback isolation silently breaks.
2. clean_schema sidesteps that by dropping and recreating the whole it schema in setup, guaranteeing the test starts with only the tables it declares (it.orders) and none left over from a prior test.
3. The yield "it" hands the schema name to the test; the test seeds it.orders, runs the committing transform, and asserts on it.daily.
4. Teardown runs DROP SCHEMA IF EXISTS it CASCADE, which removes everything in the schema — including it.daily, which the setup did not know about. Dropping the schema (not individual tables) is what makes this robust to a transform that creates arbitrary new objects.
5. The trade-off is cost: drop/create is heavier than a rollback, so you reserve it for suites where the SUT genuinely commits or issues DDL, and keep the fast rollback fixture for everything that does not. Matching the reset to the transform is the senior call.

Output.

Test action	Rollback fixture	Drop-schema fixture
INSERT (committed)	leaks	reset
CREATE TABLE AS	leaks (table remains)	dropped with schema
next test collides?	yes (dirty)	no (clean schema)
relative cost	cheapest	heavier but correct

Rule of thumb. When the transform commits or issues DDL, transactional rollback cannot isolate it — switch to TRUNCATE ... CASCADE for known tables or DROP SCHEMA ... CASCADE; CREATE SCHEMA when the transform creates new objects. Match the reset to what the SUT actually does, and keep the fast rollback fixture for the suites that never commit.

Worked example — composing a seed fixture on top of a database fixture

Detailed explanation. Fixtures compose: a seed fixture depends on the db fixture, seeds a standard graph, and yields handles the test can use. This keeps the arrange step out of every test body and lets a test override just the piece it cares about. Build a reusable seeded fixture and a factory-style fixture on top of it.

• The base. db — a clean, isolated connection (from the rollback fixture above).
• The seed. seeded — depends on db, inserts a baseline graph, yields ids.
• The override. A test that needs a special case adds its own row via db on top of the baseline.

Question. Write a seeded fixture that layers a baseline graph onto db, and show a test using it plus a test overriding it.

Input.

Fixture	Depends on	Provides
db	engine	isolated connection
seeded	db	baseline graph + ids
test	seeded (+ db)	baseline, optionally extended

Code.

import pytest
from sqlalchemy import text

@pytest.fixture
def seeded(db):
    # Compose ON TOP of the isolated `db` fixture: insert a baseline graph once,
    # yield the ids so tests can reference them without re-seeding.
    db.execute(text("INSERT INTO customers (id, name) VALUES (1,'Acme'),(2,'Globex')"))
    db.execute(text("""INSERT INTO orders (id, customer_id, total_cents) VALUES
        (10,1,4200),(11,1,800),(12,2,990)"""))
    yield {"customer_ids": [1, 2], "order_ids": [10, 11, 12]}
    # no explicit teardown needed: `db` rolls the whole transaction back.

def test_uses_baseline(seeded, db):
    total = db.execute(text(
        "SELECT SUM(total_cents) FROM orders WHERE customer_id = 1")).scalar()
    assert total == 5000                       # 4200 + 800, from the baseline

def test_overrides_with_an_edge_case(seeded, db):
    # start from the baseline, then ADD the special row this test cares about.
    db.execute(text("INSERT INTO customers (id, name) VALUES (3, 'Initech')"))  # no orders
    rows = db.execute(text("""
        SELECT c.id, COALESCE(SUM(o.total_cents), 0) AS rev
        FROM customers c LEFT JOIN orders o ON o.customer_id = c.id
        GROUP BY c.id ORDER BY c.id""")).fetchall()
    assert rows == [(1, 5000), (2, 990), (3, 0)]   # Initech survives with 0

Step-by-step explanation.

1. The seeded fixture names db as an argument, so pytest first builds the isolated transactional connection, then runs seeded's setup — fixture composition through dependency injection.
2. seeded inserts a baseline graph (two customers, three orders) and yields a small dict of the ids it created, so tests reference customer 1 without hard-coding or re-inserting.
3. test_uses_baseline consumes the baseline unchanged and asserts on it — the common case, where a test needs "some realistic data" and does not care about the specifics.
4. test_overrides_with_an_edge_case starts from the same baseline and adds its own row (Initech, a customer with no orders) via db. Layering the edge case on top of the shared baseline keeps the test focused on the one thing it exercises.
5. Neither test writes teardown code: because seeded sits on top of the rollback db fixture, the whole transaction — baseline plus any per-test additions — is discarded at the end. Composition means the isolation is inherited, not re-implemented.

Output.

Test	Baseline from seeded	Adds	Asserts
test_uses_baseline	2 customers, 3 orders	—	customer 1 = 5000
test_overrides...	2 customers, 3 orders	Initech (0 orders)	3 rows incl. (3, 0)
teardown	inherited from db	—	all rolled back

Rule of thumb. Layer a seeded fixture on top of the isolated db fixture to keep the arrange step out of test bodies, and let individual tests add the edge case they care about on top of the shared baseline. Composition means each fixture does one job and inherits the isolation of the fixtures it depends on.

Senior interview question on fixture scope and isolation

A senior interviewer might ask: "Design the fixture layer for a warehouse test suite. Explain what you put at session scope versus function scope and why, how you reset the database between tests, what you change when the transform commits or a dbt model issues DDL, and how you guarantee the world is cleaned up even when a test fails — all while keeping the suite fast enough to run on every push."

Solution Using session-scoped setup, function-scoped rollback, and a commit-aware fallback

# conftest.py — the fixture layer, tiered by cost and mutability.
import pytest
from sqlalchemy import create_engine, text

# 1. SESSION scope: expensive + immutable — engine, schema, migrations. Built ONCE.
@pytest.fixture(scope="session")
def engine():
    eng = create_engine("postgresql+psycopg://test:test@localhost:5432/test")
    with eng.begin() as c:
        c.execute(text("CREATE SCHEMA IF NOT EXISTS it"))
        c.execute(text("CREATE TABLE IF NOT EXISTS it.orders "
                      "(id INT, customer_id INT, total_cents BIGINT)"))
    yield eng
    eng.dispose()

# 2. FUNCTION scope (default): fast rollback isolation for the common case.
@pytest.fixture
def db(engine):
    conn = engine.connect()
    txn = conn.begin()
    try:
        yield conn
    finally:
        txn.rollback()          # teardown ALWAYS runs (finally) -> clean on failure too
        conn.close()

# 3. COMMIT-AWARE fallback: for suites whose transform COMMITs / issues DDL.
@pytest.fixture
def clean_schema(engine):
    with engine.begin() as c:
        c.execute(text("DROP SCHEMA IF EXISTS it CASCADE"))
        c.execute(text("CREATE SCHEMA it"))
        c.execute(text("CREATE TABLE it.orders (id INT, customer_id INT, total_cents BIGINT)"))
    yield "it"
    with engine.begin() as c:
        c.execute(text("DROP SCHEMA IF EXISTS it CASCADE"))

# 4. Tests pick the fixture that matches what the SUT does.
def test_readonly_transform(db):          # no commit -> fast rollback fixture
    db.execute(text("INSERT INTO it.orders VALUES (1,1,4200)"))
    assert db.execute(text("SELECT count(*) FROM it.orders")).scalar() == 1

def test_materialising_transform(engine, clean_schema):   # commits DDL -> drop-schema
    with engine.begin() as c:
        c.execute(text("INSERT INTO it.orders VALUES (1,1,4200),(2,1,800)"))
        c.execute(text("CREATE TABLE it.daily AS "
                      "SELECT customer_id, SUM(total_cents) r FROM it.orders GROUP BY 1"))
    assert engine.connect().execute(text("SELECT r FROM it.daily")).scalar() == 5000

Step-by-step trace.

Tier	Fixture	Scope	Reset	Used for
Engine/schema	engine	session	dispose at end	expensive, immutable
Fast isolation	db	function	transaction rollback	non-committing tests
Commit-safe	clean_schema	function	drop + recreate schema	committing / DDL tests
Guarantee	finally / post-yield	—	runs on pass or fail	no leaks

After the design, the engine and schema are built exactly once per run (session scope), the common read/write tests use the function-scoped db fixture whose rolled-back transaction resets state in microseconds, and the minority of tests whose transform commits or runs CREATE TABLE AS use clean_schema, which drops and recreates the schema so committed DDL cannot leak. Every teardown lives after yield (or in a finally), so a failing test still cleans up. The suite is fast because the expensive setup is amortised and the common reset is a rollback; it is correct because the commit-aware fixture handles the cases a rollback cannot.

Output:

Metric	Naive (all function scope, rollback only)	Tiered design
Engine builds per run	once per test (slow)	once per session
Reset for non-committing test	rollback	rollback (fast)
Reset when SUT commits/DDL	leaks (rollback no-op)	drop schema (correct)
Cleanup on test failure	risk of leak	guaranteed (finally)
Suite speed	slow	fast

Why this works — concept by concept:

• Scope by cost and mutability — the engine and schema are expensive and immutable, so they live at session scope and are built once; the seeded rows are cheap and mutable, so they reset per function — the split that makes the suite both fast and isolated.
• Transactional rollback — wrapping each test in a transaction and discarding it resets the database in microseconds without deleting anything, the fastest possible isolation for any transform that does not manage its own commits.
• Commit-aware fallback — when the SUT commits or issues DDL a rollback becomes a no-op, so those suites drop and recreate the schema, guaranteeing isolation for exactly the cases rollback cannot handle.
• Teardown after yield — placing cleanup in the fixture's post-yield (or a finally) block means pytest runs it whether the test passes or raises, so a failing test never leaks its world into the next.
• Cost — one engine build per session, a rollback per common test, and a schema drop only where required, versus rebuilding the world every test or leaking state between them. The eliminated cost is a slow, flaky suite — O(1) session setup plus O(1) per-test reset instead of O(tests) full rebuilds.

Data validation
Topic — data-validation
Data validation problems on setup, state, and isolation

Practice →

Defensive coding Topic — defensive-coding Defensive coding problems on teardown and cleanup guarantees

Practice →

---

3. Factories and seeding — FK-consistent, deterministic data

A factory mints one valid row on demand; seeding assembles a referentially-consistent graph the transform can actually join

The mental model in one line: a factory is a builder that produces one valid row with sensible defaults so a test overrides only the field it cares about, and seeding composes factories into a referentially-consistent graph — SubFactory wires a child's foreign key to a freshly-built parent so the keys always resolve, Sequence and Faker fill the boring fields, and reseeding the random generator plus freezing the clock makes the whole graph deterministic so assertions do not flap — while dbt seeds load the static reference data (a region map, a date spine) the model looks up — so between factories for the volatile facts and seeds for the stable dimensions, every foreign key in the input is satisfied and every run produces the same rows. Hand-typed INSERTs rot; a factory encodes "what a valid row looks like" in one place and keeps the graph consistent as the schema evolves.

Why factories beat literal INSERTs.

• Valid defaults, targeted overrides. A factory fills every required field with a sensible default, so a test writes OrderFactory(total_cents=0) to exercise the zero case and ignores the twenty other columns it does not care about.
• One place to evolve. When the schema gains a NOT NULL column, you add a default to the factory once, instead of editing every hand-typed INSERT across the suite.
• Readable intent. OrderFactory(status="refunded") says what the test is about; a 12-column INSERT buries the one meaningful value in noise.

factory_boy essentials.

• Factory + Meta.model. A class declaring how to build one instance of a model (a SQLAlchemy/Django model, a dataclass, or a dict via a plain factory).
• SubFactory. A field whose value is another factory — the mechanism that builds a parent and wires the child's foreign key to it, guaranteeing referential integrity.
• Sequence and Faker. Sequence(lambda n: ...) gives unique, ordered values (ids, emails); Faker("company") fills realistic-looking strings.
• build vs create. build() returns an unsaved instance (in-memory); create() persists it to the session/database — use create for integration tests that then query the data back.

FK-consistent seed graphs.

• Parents via SubFactory. A child factory's customer = SubFactory(CustomerFactory) builds (and persists) a customer first, then sets the child's customer_id — so the foreign key always points at a real row.
• Children via RelatedFactory / post-generation. To give a parent N children, a RelatedFactory or a @post_generation hook creates order items after the order exists.
• Sharing a parent. Pass an explicit parent (OrderFactory(customer=acme)) so several orders reference the same customer rather than each minting a new one.

Deterministic seeds.

• Reseed the generator. factory.random.reseed_random(12345) (and Faker's seed) pins the pseudo-random stream so the same run produces the same values — a prerequisite for asserting on generated data.
• Freeze the clock. Wrap the test in freezegun.freeze_time("2026-01-01") (or inject a fixed now) so created_at defaults and date logic are reproducible.
• Stable ids. Use Sequence from a known start, and reset sequences on teardown, so ids do not drift between runs and cross-test.

dbt seeds for reference data.

• What they are. CSV files in the seeds/ directory that dbt seed loads into tables — the idiomatic home for small, static reference data (a country map, a currency table, a date spine).
• When to use them. For dimensions and lookups the model joins to but the test does not vary: seed them once as CSV rather than manufacturing them per test.
• Typing. Declare +column_types in dbt_project.yml so a seed's region_code is varchar(8), not an inferred type — seeds are data and a lightweight schema contract.

The failure modes senior engineers pre-empt.

• Random data breaking assertions. Asserting == "Acme" on a Faker("company") value flaps every run. Mitigation: reseed the generator, or assert on structure/counts rather than the random value, and pin the fields you assert on.
• Orphan foreign keys. Building a child with a hard-coded customer_id that no factory created leaves a dangling key. Mitigation: always wire parents via SubFactory or pass an explicit persisted parent.
• Unseeded reference tables. The model joins to dim_region but the test never seeded it, so every lookup is null. Mitigation: load reference dimensions via dbt seeds (or a session fixture) before the model runs.

Common interview probes on factories and seeding.

• "Why a factory instead of INSERT statements?" — valid defaults, one place to evolve, override only what the test cares about.
• "How do you keep foreign keys consistent?" — SubFactory builds and wires the parent; share a parent by passing it explicitly.
• "How do you make generated data deterministic?" — reseed the RNG/Faker and freeze the clock so the same run reproduces the same rows.
• "Where does static reference data come from?" — dbt CSV seeds (or a session fixture), typed via +column_types.

Worked example — factory_boy with SubFactory for referential integrity

Detailed explanation. The core of consistent seeding is SubFactory: a child factory declares its parent as a sub-factory, so building the child first builds a valid parent and wires the foreign key. Build Customer -> Order -> OrderItem factories where every key resolves automatically.

• The chain. OrderItem.order -> Order.customer -> Customer.
• SubFactory. Each child's parent field is the parent's factory.
• Sharing. Pass an explicit parent to make siblings share it.

Question. Define factories so that creating an OrderItem automatically creates a valid Order and Customer, and show how to make several orders share one customer.

Input.

Factory	FK field	Wired by
CustomerFactory	— (root)	Sequence / Faker
OrderFactory	customer	SubFactory(CustomerFactory)
OrderItemFactory	order	SubFactory(OrderFactory)
shared parent	pass customer=acme	explicit override

Code.

import factory
from factory.alchemy import SQLAlchemyModelFactory
from models import Customer, Order, OrderItem
from db import Session                      # a scoped SQLAlchemy session

class CustomerFactory(SQLAlchemyModelFactory):
    class Meta:
        model = Customer
        sqlalchemy_session = Session
        sqlalchemy_session_persistence = "flush"     # persist so FKs resolve
    id = factory.Sequence(lambda n: n + 1)
    name = factory.Faker("company")
    segment = factory.Iterator(["enterprise", "smb"])

class OrderFactory(SQLAlchemyModelFactory):
    class Meta:
        model = Order
        sqlalchemy_session = Session
        sqlalchemy_session_persistence = "flush"
    id = factory.Sequence(lambda n: 1000 + n)
    customer = factory.SubFactory(CustomerFactory)   # builds a REAL parent, sets customer_id
    total_cents = 4200
    status = "paid"

class OrderItemFactory(SQLAlchemyModelFactory):
    class Meta:
        model = OrderItem
        sqlalchemy_session = Session
        sqlalchemy_session_persistence = "flush"
    order = factory.SubFactory(OrderFactory)          # chains up: item -> order -> customer
    sku = factory.Sequence(lambda n: f"SKU-{n}")
    qty = 1

# Usage --------------------------------------------------------------------------
item = OrderItemFactory()                 # ONE call -> a customer, an order, an item
assert item.order.customer.id is not None # every FK in the chain is satisfied

# Share one customer across several orders (siblings), not one each:
acme = CustomerFactory(name="Acme")
orders = OrderFactory.create_batch(3, customer=acme, total_cents=1000)
assert {o.customer_id for o in orders} == {acme.id}   # all three point at Acme

Step-by-step explanation.

1. Each factory declares Meta.model and the SQLAlchemy session, with sqlalchemy_session_persistence = "flush" so a built parent is flushed (gets its primary key) before a child references it — the mechanism that makes the foreign key resolve.
2. OrderFactory.customer = SubFactory(CustomerFactory) means building an order first builds a customer via CustomerFactory, then sets order.customer_id to that customer's id. The child cannot have an orphan key because the parent is created as part of building the child.
3. OrderItemFactory.order = SubFactory(OrderFactory) chains the whole graph: a single OrderItemFactory() call builds a customer, an order referencing it, and an item referencing the order — three referentially-consistent rows from one line.
4. To make siblings share a parent, pass it explicitly: OrderFactory.create_batch(3, customer=acme) reuses the one acme customer for all three orders instead of minting three customers. This is how you seed "one customer with three orders," a common realistic shape.
5. Sequence gives each row a unique, ordered id/sku so there are no primary-key collisions across the batch, and Faker/Iterator fill the fields the test does not care about — keeping the test body focused on the one attribute it overrides.

Output.

Call	Rows created	Foreign keys
OrderItemFactory()	customer + order + item	all resolved
OrderFactory()	customer + order	order.customer_id set
create_batch(3, customer=acme)	3 orders (1 shared customer)	all point at acme
hard-coded customer_id=99	order only	orphan (avoid)

Rule of thumb. Wire every foreign key with SubFactory so building a child always builds a valid parent, and pass an explicit parent when you want siblings to share one. A single factory call should yield a referentially-consistent slice of the graph — never hard-code a foreign key to a row no factory created.

Worked example — deterministic seeding so assertions are stable

Detailed explanation. Generated data is only useful in a test if it is reproducible: the same run must produce the same rows, or an assertion on a value flaps. Pin the random generator, freeze the clock, and control the sequence starts so the seed is deterministic. Make a factory-driven test that asserts on exact generated values.

• Reseed. factory.random.reseed_random(seed) fixes the pseudo-random stream.
• Freeze time. freeze_time fixes created_at defaults and any date math.
• Assert safely. Either pin the fields you assert on, or assert on counts/structure, not on an unpinned random string.

Question. Configure a test so the same seed produces identical rows every run, and choose assertions that stay stable.

Input.

Source of nondeterminism	Fix
Faker random values	reseed_random(12345)
datetime.now() defaults	freeze_time("2026-01-01")
auto-increment ids	Sequence from a fixed start + reset
row order	ORDER BY in the assertion query

Code.

import factory
import pytest
from freezegun import freeze_time

@pytest.fixture(autouse=True)
def _deterministic():
    # Runs before EVERY test: pin the RNG so Faker/Iterator reproduce the same stream.
    factory.random.reseed_random(12345)
    yield

@freeze_time("2026-01-01 00:00:00")           # freeze the clock for date-dependent defaults
def test_seed_is_reproducible(db):
    orders = OrderFactory.create_batch(3)      # deterministic: same 3 rows every run
    # Assert on STRUCTURE and PINNED fields, not on unpinned random strings:
    assert len(orders) == 3
    assert [o.id for o in orders] == [1000, 1001, 1002]     # Sequence from fixed start
    assert all(o.created_at.date().isoformat() == "2026-01-01" for o in orders)
    assert all(o.status == "paid" for o in orders)          # a pinned default, stable

def test_assert_the_value_you_control(db):
    # If you must assert an exact figure, PIN it in the factory call — don't rely on Faker.
    o = OrderFactory(total_cents=4200)
    assert o.total_cents == 4200               # stable because the test set it

Step-by-step explanation.

1. The autouse=True fixture reseeds factory.random before every test, so Faker and Iterator draw from the same pseudo-random stream each run — the same "random" company name and segment appear every time, making generated data reproducible.
2. @freeze_time("2026-01-01") pins the clock, so any factory default that calls datetime.now() (like created_at) resolves to a fixed timestamp instead of the wall clock — otherwise a date assertion would fail a millisecond later.
3. The test asserts on structure and pinned fields: the row count, the Sequence-generated ids (deterministic from a fixed start), the frozen date, and a hard-coded default (status == "paid"). None of these depend on an unpinned random value.
4. It deliberately does not assert on the Faker("company") name, because even reseeded that is fragile across Faker versions — the discipline is to assert on what you control (pinned fields, counts) and treat generated filler as scenery.
5. The second test shows the safe pattern for an exact figure: pin total_cents=4200 in the factory call and assert on it. The value is stable because the test set it, not because you got lucky with the generator — an assertion should test the transform, not the RNG.

Output.

Run	Ids	created_at	status	Faker name
1st	1000,1001,1002	2026-01-01	paid	(reproducible)
2nd	1000,1001,1002	2026-01-01	paid	(same)
without reseed/freeze	drift	wall clock	paid	flaps

Rule of thumb. Make seeds deterministic by reseeding the generator and freezing the clock, then assert on the fields you pinned (overrides, sequences, counts) rather than on unpinned generated values. Test data should be reproducible run to run, and your assertions should test the transform, not the random generator.

Worked example — dbt CSV seeds for reference tables

Detailed explanation. Row-level factories are right for the volatile facts, but static reference data — a region map, a currency list, a date spine — is better loaded as a dbt seed: a CSV in seeds/ that dbt seed materialises into a table, typed via dbt_project.yml. Seed a dim_region the model joins to.

• The CSV. seeds/dim_region.csv — small, static, version-controlled.
• The typing. +column_types pins the column types instead of inferring.
• The use. The model JOINs dim_region; the test seeds it once, then seeds facts per test.

Question. Add a typed dbt seed for a region dimension and show how a model and its test consume it.

Input.

Piece	Value
Seed file	seeds/dim_region.csv
Loader	dbt seed --select dim_region
Typing	+column_types in dbt_project.yml
Consumer	a model JOIN dim_region

Code.

# seeds/dim_region.csv  — static reference data, checked into the repo.
region_code,region_name,manager
EU,Europe,Ada
US,United States,Grace
APAC,Asia Pacific,Lin

# dbt_project.yml — type the seed columns explicitly (don't let dbt infer).
seeds:
  my_project:
    dim_region:
      +column_types:
        region_code: varchar(8)
        region_name: varchar(64)
        manager:     varchar(64)

-- models/marts/regional_sales.sql — the model JOINs the seeded dimension.
SELECT r.region_name, r.manager,
       SUM(o.total_cents) AS revenue_cents
FROM {{ ref('stg_orders') }} o
JOIN {{ ref('dim_region') }} r ON r.region_code = o.region   -- ref() a seed like a model
GROUP BY r.region_name, r.manager

# The test's arrange step: load the static dimension ONCE, then seed facts per test.
dbt seed --select dim_region        # materialises seeds/dim_region.csv -> table
dbt run  --select stg_orders regional_sales
# then assert on the built `regional_sales` table (see section 4).

Step-by-step explanation.

1. seeds/dim_region.csv holds the static region dimension as version-controlled data. Because it never varies between tests, it belongs in a seed, not in a per-test factory — you load it once and every test joins to the same stable dimension.
2. The +column_types block in dbt_project.yml pins each column's type, so region_code is varchar(8) rather than whatever dbt infers from the CSV. Typing seeds makes them a lightweight schema contract, not just data.
3. In the model, {{ ref('dim_region') }} references the seed exactly like a model, so the JOIN to the dimension resolves through dbt's dependency graph — the seed is built before the model that depends on it.
4. dbt seed --select dim_region materialises the CSV into a table; the test then seeds the facts (stg_orders) per case with factories and runs the model. Static dimension via seed, volatile facts via factory — the right tool for each.
5. Without the seed, the model's JOIN dim_region would return nulls (or drop rows) because the dimension table would be empty — a silent bug where the region name is null. Seeding reference data is what makes the model's lookups resolve in the test.

Output.

Data	Loaded by	Varies per test?	Why
dim_region	dbt seed (CSV)	no	static reference
stg_orders facts	factory	yes	the case under test
model output	dbt run	yes	derived from facts
unseeded dimension	—	—	join returns null (bug)

Rule of thumb. Load static reference data (dimensions, maps, date spines) as typed dbt seeds and manufacture the volatile facts with factories. Type your seed columns in dbt_project.yml, and always seed the dimensions a model joins to — an empty reference table turns a JOIN into silent nulls.

Senior interview question on building a consistent, deterministic seed graph

A senior interviewer might ask: "You are testing a dbt mart that joins facts to a region dimension and rolls up revenue by segment. Design the seed strategy: how you manufacture a referentially-consistent fact graph, how you keep foreign keys valid, how you load the static dimension, how you make the whole seed deterministic so assertions do not flap, and how a single test overrides just the case it exercises."

Solution Using SubFactory graphs, dbt seeds for dimensions, and a reseeded, frozen clock

# 1. FACTORIES for the volatile facts — SubFactory keeps every FK valid.
import factory
from factory.alchemy import SQLAlchemyModelFactory
from models import Customer, Order
from db import Session

class CustomerFactory(SQLAlchemyModelFactory):
    class Meta: model = Customer; sqlalchemy_session = Session
    id = factory.Sequence(lambda n: n + 1)
    segment = factory.Iterator(["enterprise", "smb"])

class OrderFactory(SQLAlchemyModelFactory):
    class Meta: model = Order; sqlalchemy_session = Session
    id = factory.Sequence(lambda n: 1000 + n)
    customer = factory.SubFactory(CustomerFactory)     # FK wired -> no orphans
    region = factory.Iterator(["EU", "US"])
    total_cents = 1000

# 2. DETERMINISM: reseed + freeze so the same seed reproduces the same rows.
import pytest
from freezegun import freeze_time

@pytest.fixture(autouse=True)
def _deterministic():
    factory.random.reseed_random(12345)
    with freeze_time("2026-01-01"):
        yield

# 3. THE TEST: static dimension via seed (arrange once), facts via factory (per test),
#    override only the case under test, assert on the rolled-up output.
def test_revenue_by_segment(seeded_dim_region, db):     # fixture ran `dbt seed dim_region`
    acme = CustomerFactory(segment="enterprise")
    OrderFactory.create_batch(2, customer=acme, region="EU", total_cents=2500)  # 5000
    CustomerFactory(segment="smb")                       # a customer with NO orders (edge)
    run_dbt(["run", "--select", "regional_sales"])       # ACT: the real model

    rows = db.execute("""SELECT segment, revenue_cents
                         FROM regional_sales_by_segment ORDER BY segment""").fetchall()
    assert rows == [("enterprise", 5000), ("smb", 0)]    # smb kept via LEFT JOIN

Step-by-step trace.

Concern	Mechanism	Guarantees
FK consistency	SubFactory builds parents	no orphan keys
Shared parent	pass customer=acme	2 orders, 1 customer
Static dimension	dbt seed (CSV)	join resolves
Determinism	reseed + freeze_time	same rows every run
Edge case	smb customer, 0 orders	LEFT JOIN tested
Focus	override only segment/total	readable intent

After the design, the volatile facts come from factories whose SubFactory fields guarantee every foreign key resolves, the static dim_region is loaded once as a typed dbt seed, and an autouse fixture reseeds the generator and freezes the clock so the same inputs reproduce byte-for-byte each run. The test shares one customer across two orders (a realistic shape), seeds an enterprise customer with revenue and an smb customer with none (the LEFT-JOIN edge), runs the real model, and asserts the exact rollup — enterprise = 5000, smb = 0. Every value the assertion checks is one the test pinned.

Output:

Metric	Hand-typed INSERTs	Factory + seed graph
Foreign-key integrity	manual, error-prone	guaranteed (SubFactory)
Schema change cost	edit every INSERT	one default in the factory
Determinism	wall-clock/random drift	reseeded + frozen
Reference data	forgotten (null joins)	typed dbt seed
Test readability	12-column noise	one meaningful override

Why this works — concept by concept:

• SubFactory graphs — declaring a child's parent as a sub-factory builds and wires the foreign key automatically, so every seeded fact references a real dimension row and the transform's joins behave exactly as they would in production.
• Shared explicit parents — passing one persisted customer to a batch of orders seeds the realistic "one customer, many orders" shape instead of minting a parent per child, which matters for any per-customer rollup.
• dbt seeds for dimensions — loading static reference data as a typed CSV seed makes the model's lookups resolve and pins the column types, keeping the dimension stable across every test while the facts vary.
• Reseed plus freeze — pinning the random stream and the clock makes the entire seed deterministic, so assertions test the transform's logic rather than flapping on a generated value or a wall-clock timestamp.
• Cost — one factory definition per table plus one seed per dimension, versus hand-typed inserts re-edited on every schema change and flaky assertions on random data. The eliminated cost is a brittle, high-maintenance fixture layer — O(tables) factories written once instead of O(tests × columns) literal rows maintained forever.

Data processing
Topic — data-processing
Data processing problems on building and shaping input data

Practice →

ETL Topic — etl ETL problems on dimensions, reference data, and joins

Practice →

---

4. Integration test patterns — ephemeral schema, DuckDB, assert

Spin up a throwaway warehouse, seed it, run the real transform, and assert on the output

The mental model in one line: a warehouse integration test spins up a throwaway warehouse — an in-process DuckDB database or an ephemeral CREATE SCHEMA on a real engine — seeds it with the fixture graph, runs the actual transform (the same SQL or dbt model production runs), and asserts on the output table by comparing it to an expected result, then drops the throwaway on teardown — so the system under test is the real query over real seeded data, the test is naturally isolated because the warehouse is created and destroyed per test, and the assertion is a frame-to-frame or row-to-row comparison rather than a mock's echo. DuckDB gives you this with zero setup for local and CI runs; an ephemeral schema on Postgres/Snowflake gives you the same isolation on the engine you actually ship to.

The pattern in three beats.

• Arrange — a throwaway warehouse plus seed. Create an ephemeral schema/database and load the fixture graph. Because it is throwaway, the test starts from a known-empty state with no cross-test leakage.
• Act — the real transform. Run the exact SQL or invoke the dbt model that production runs. The whole point is that the SUT is the shipped query, not a paraphrase.
• Assert — output versus expected. Read the output table into a frame (pandas/polars) or a list of rows and compare it to an explicit expected result, ordered for stability.

Ephemeral schema per test.

• Create + search_path. CREATE SCHEMA test_ab12; SET search_path TO test_ab12 scopes every unqualified table to that schema, so the test's tables are namespaced away from every other test.
• Unique name. A random suffix (test_{uuid}) or the xdist worker id makes the schema name collision-free, which is what enables parallelism (section 5).
• Drop on teardown. DROP SCHEMA test_ab12 CASCADE removes the schema and everything the transform created in it — the isolation and the cleanup in one statement.

DuckDB as a fast local warehouse.

• In-process, zero setup. duckdb.connect(":memory:") is a full SQL engine in the test process — no container, no service, milliseconds to start. Ideal for transform logic that is standard SQL.
• Naturally isolated. Each connection is its own database, so a per-test connection is inherently isolated — no shared state, trivially parallel across processes.
• dbt-duckdb. The dbt-duckdb adapter runs dbt models against DuckDB, so you can test real dbt models locally and in CI without a warehouse.
• The caveat. DuckDB is not byte-identical to Snowflake/BigQuery, so engine-specific SQL (proprietary functions, exact type coercions) still needs a smoke test on the real engine — DuckDB covers the logic, not the dialect edge cases.

Asserting on outputs.

• Exact frame comparison. pandas.testing.assert_frame_equal(got, expected) (or polars' equivalent) is the crispest assertion — but sort both frames and reset indexes first, since SQL row order is not guaranteed.
• Business invariants. Assert properties (revenue >= 0, count matches, no duplicate keys) when the exact frame is large or partly generated.
• Referential and null checks. Assert that a LEFT JOIN kept the parentless row, that no key is null, that a rollup's total matches the input sum — the behaviours a mock could never verify.

Testing a dbt model end-to-end.

• Seed → run → assert. Load seeds/fixtures, dbt run --select the_model (or dbt build), then query the built table and assert. This tests the compiled SQL exactly as it ships.
• dbt's own tests. dbt test runs schema tests (unique, not_null, relationships, accepted_values) — assertions expressed as data contracts, complementary to Python assertions on values.
• Programmatic invocation. dbtRunner().invoke([...]) drives dbt from pytest so the arrange/act/assert lives in one test function.

The failure modes senior engineers pre-empt.

• Asserting on unstable order. Comparing an unordered SQL result to an ordered expected frame flaps. Mitigation: ORDER BY in the query or sort both frames before comparing.
• Comparing floats exactly. assert 0.1 + 0.2 == 0.3 fails; monetary and averaged columns drift. Mitigation: compare in integer cents, or use a tolerance (check_exact=False/rtol).
• Hidden state from a prior test. Reusing a schema/connection leaks rows. Mitigation: a fresh ephemeral schema or DuckDB connection per test, dropped on teardown.

Common interview probes on integration patterns.

• "How do you give each test a clean warehouse?" — ephemeral CREATE SCHEMA (dropped on teardown) or a per-test in-process DuckDB connection.
• "How do you test a dbt model?" — seed, dbt run --select model, assert on the built table; add dbt test schema contracts.
• "How do you assert on the output?" — sort and compare frames, or assert invariants; compare money in integer cents, not floats.
• "Why not just use the real warehouse?" — cost, speed, and isolation; DuckDB/ephemeral schema gives the same logic coverage in milliseconds, with a smoke test on the real engine for dialect edges.

Worked example — a DuckDB integration test for a SQL transform

Detailed explanation. DuckDB makes an integration test as cheap as a unit test: an in-process database, seeded and torn down in the test, running the real transform SQL. Build a test for a daily-sales rollup that seeds orders, runs the transform, and compares the output frame to an expected frame.

• Arrange. duckdb.connect(":memory:"), create and seed orders.
• Act. Run the real CREATE TABLE daily_sales AS SELECT ....
• Assert. assert_frame_equal on the sorted output versus expected.

Question. Write a DuckDB integration test that seeds orders, runs the rollup transform, and asserts the output equals the expected frame.

Input.

Beat	DuckDB call
arrange	connect(":memory:") + seed orders
act	run the transform SQL
assert	.df() + assert_frame_equal
teardown	connection closes with the test

Code.

import duckdb
import pandas as pd
from pandas.testing import assert_frame_equal

# The transform under test — the SAME SQL production runs (import it, don't retype it).
DAILY_SALES_SQL = """
CREATE TABLE daily_sales AS
SELECT order_date, region,
       SUM(total_cents) AS revenue_cents,
       COUNT(*)         AS n_orders
FROM orders
GROUP BY order_date, region
"""

def test_daily_sales_rollup():
    con = duckdb.connect(":memory:")          # ARRANGE: throwaway in-process warehouse
    con.execute("CREATE TABLE orders "
                "(id INT, region VARCHAR, total_cents BIGINT, order_date DATE)")
    con.execute("""INSERT INTO orders VALUES
        (1,'EU',4200,DATE '2026-01-01'),
        (2,'EU', 800,DATE '2026-01-01'),      -- same day+region -> must aggregate
        (3,'US', 990,DATE '2026-01-01')""")

    con.execute(DAILY_SALES_SQL)              # ACT: run the REAL transform

    got = con.execute(                        # ASSERT: read output, ORDER for stability
        "SELECT order_date, region, revenue_cents, n_orders "
        "FROM daily_sales ORDER BY order_date, region").df()
    expected = pd.DataFrame({
        "order_date":   pd.to_datetime(["2026-01-01", "2026-01-01"]).date,
        "region":       ["EU", "US"],
        "revenue_cents":[5000, 990],          # 4200 + 800 aggregated for EU
        "n_orders":     [2, 1],
    })
    assert_frame_equal(got, expected, check_dtype=False)
    con.close()                               # TEARDOWN: in-memory DB vanishes

Step-by-step explanation.

1. duckdb.connect(":memory:") creates a complete SQL warehouse inside the test process in milliseconds — no container, no service. Because it is in-memory and per-test, it starts empty and vanishes when closed, so isolation is automatic.
2. The arrange step creates and seeds orders with the same aggregation-triggering shape as before (two EU rows on one day, one US row), so the transform's GROUP BY has something meaningful to collapse.
3. The act step runs DAILY_SALES_SQL — the actual transform, imported as a constant rather than retyped, so the test exercises the shipped SQL. This is the line that makes it an integration test.
4. The assert step reads the output with an explicit ORDER BY and compares it to an expected pandas frame via assert_frame_equal. Ordering the query makes the comparison stable, and check_dtype=False tolerates harmless int64/int32 differences while still verifying the values.
5. con.close() drops the in-memory database, so there is nothing to clean up and nothing to leak. The whole test — spin up, seed, transform, assert, tear down — runs in milliseconds, cheap enough to keep hundreds of them in the suite.

Output.

order_date	region	revenue_cents	n_orders
2026-01-01	EU	5000	2
2026-01-01	US	990	1

Rule of thumb. Use an in-process DuckDB database as a throwaway warehouse: seed it, run the imported transform SQL, and compare the output to an expected frame with assert_frame_equal — sorting the query for stability and comparing money in integer cents. It is as fast as a unit test but actually exercises the SQL, and the connection's close is the entire teardown.

Worked example — an ephemeral Postgres schema with search_path isolation

Detailed explanation. When the transform uses engine-specific features (or you want to test on the engine you ship), give each test an ephemeral schema on a real Postgres and scope its tables with search_path. The schema is created in setup and dropped in teardown, so tests are isolated even though they share a database. Build the fixture.

• Create. CREATE SCHEMA test_<uuid> and SET search_path to it.
• Run. All unqualified tables resolve inside the schema; run the real transform.
• Drop. DROP SCHEMA ... CASCADE on teardown removes everything.

Question. Write a fixture that gives each test an isolated, uniquely-named Postgres schema and drops it afterward.

Input.

Step	Statement
create	CREATE SCHEMA "test_ab12"
scope	SET search_path TO "test_ab12"
run	transform's unqualified tables land here
drop	DROP SCHEMA "test_ab12" CASCADE

Code.

import uuid
import pytest
from sqlalchemy import create_engine, text

@pytest.fixture(scope="session")
def engine():
    eng = create_engine("postgresql+psycopg://test:test@localhost:5432/test")
    yield eng
    eng.dispose()

@pytest.fixture
def schema(engine):
    # A unique schema per test -> isolation on a SHARED database (and parallel-safe).
    name = f"test_{uuid.uuid4().hex[:8]}"
    conn = engine.connect()
    conn.execute(text(f'CREATE SCHEMA "{name}"'))
    conn.execute(text(f'SET search_path TO "{name}"'))   # unqualified tables -> this schema
    conn.commit()
    try:
        yield conn
    finally:
        # DROP CASCADE removes the schema AND every table the transform created in it.
        conn.execute(text(f'DROP SCHEMA IF EXISTS "{name}" CASCADE'))
        conn.commit()
        conn.close()

def test_transform_in_isolated_schema(schema):
    schema.execute(text("CREATE TABLE orders (id INT, region TEXT, total_cents BIGINT)"))
    schema.execute(text("INSERT INTO orders VALUES (1,'EU',4200),(2,'EU',800)"))
    schema.execute(text("CREATE TABLE daily AS "
                        "SELECT region, SUM(total_cents) rev FROM orders GROUP BY region"))
    schema.commit()
    rev = schema.execute(text("SELECT rev FROM daily WHERE region='EU'")).scalar()
    assert rev == 5000
    # `daily` and `orders` live in test_xxxx and are DROPped on teardown -> no leak.

Step-by-step explanation.

1. The schema fixture generates a unique name (test_ plus a random hex suffix), so two tests — even running concurrently — never share a schema. Uniqueness is what lets this isolation strategy scale to parallel workers.
2. CREATE SCHEMA plus SET search_path TO the new schema means every unqualified table the transform creates (orders, daily) lands inside this test's schema, namespaced away from every other test's tables in the same database.
3. The test seeds orders, runs the real transform (CREATE TABLE daily AS ...), commits, and asserts on daily. Because the transform commits, a rollback fixture would not isolate it — but a per-test schema does, since the whole schema is disposable.
4. Teardown runs DROP SCHEMA ... CASCADE in a finally, removing the schema and everything in it — both the seeded orders and the transform-created daily — regardless of whether the test passed. One statement is the entire cleanup.
5. This gives real-engine fidelity (you are testing on Postgres, with its exact SQL dialect and types) while keeping per-test isolation on a single shared database — the pattern for when DuckDB's approximation of the target engine is not enough.

Output.

Aspect	Shared schema (naive)	Ephemeral schema
Isolation	leaks across tests	per-test namespace
Commit/DDL support	leaks committed tables	dropped with schema
Parallel-safe	no (name clashes)	yes (unique names)
Engine fidelity	real Postgres	real Postgres
Cleanup	manual/none	one DROP CASCADE

Rule of thumb. For real-engine fidelity with per-test isolation, give each test a uniquely-named ephemeral schema, SET search_path to it so unqualified tables land inside, and DROP SCHEMA ... CASCADE on teardown. Unique names make it parallel-safe, and dropping the schema cleans up committed tables a rollback never could.

Worked example — testing a dbt model end to end

Detailed explanation. The highest-fidelity test runs the actual dbt model: seed the inputs, invoke dbt run on the model, then assert on the built table — plus dbt test for the schema contracts. Drive it from pytest with the dbt programmatic API against dbt-duckdb. Test a stg_orders staging model.

• Seed. Load raw input (a seed CSV or factory-inserted rows) into the source table.
• Run. dbtRunner().invoke(["run", "--select", "stg_orders"]).
• Assert. Query the built stg_orders table; run dbt test for not_null/unique.

Question. Write a pytest test that seeds raw orders, runs the stg_orders dbt model, and asserts on the built table plus its schema tests.

Input.

Beat	dbt call
seed	dbt seed --select raw_orders
run	dbt run --select stg_orders
assert (values)	query the built table
assert (contracts)	dbt test --select stg_orders

Code.

import duckdb
import pytest
from dbt.cli.main import dbtRunner

@pytest.fixture
def dbt():
    return dbtRunner()      # drives the REAL dbt against the dbt-duckdb profile

def test_stg_orders_model(dbt):
    # ARRANGE — load the raw source (a seed CSV of raw orders) into the warehouse.
    r = dbt.invoke(["seed", "--select", "raw_orders", "--profiles-dir", "ci"])
    assert r.success

    # ACT — build the REAL staging model exactly as it ships.
    r = dbt.invoke(["run", "--select", "stg_orders", "--profiles-dir", "ci"])
    assert r.success

    # ASSERT (values) — query the built table in the dbt-duckdb database.
    con = duckdb.connect("target/ci.duckdb", read_only=True)
    n, bad = con.execute(
        "SELECT COUNT(*), COUNT(*) FILTER (WHERE order_id IS NULL) FROM stg_orders"
    ).fetchone()
    assert n > 0 and bad == 0            # rows built, no null business keys

    # ASSERT (contracts) — dbt's own schema tests: unique, not_null, relationships.
    r = dbt.invoke(["test", "--select", "stg_orders", "--profiles-dir", "ci"])
    assert r.success                     # the model's data contracts hold

Step-by-step explanation.

1. The dbtRunner from dbt's programmatic API lets pytest invoke real dbt commands, so the arrange/act/assert of a model test lives in one Python function instead of a shell script — and the model that runs is exactly the one that ships.
2. The arrange step loads the raw source via dbt seed --select raw_orders, materialising a CSV of raw orders into the dbt-duckdb database. (You could equally insert rows with factories; the seed is convenient for a fixed raw input.)
3. The act step runs only the model under test with --select stg_orders, so the test is scoped and fast — dbt compiles and executes the real staging SQL against the seeded source.
4. The value assertions query the built stg_orders table directly in the DuckDB file: it verifies rows were produced and that no order_id (the business key) is null — a property of the transform's cleaning logic that only a real run reveals.
5. The contract assertions run dbt test --select stg_orders, executing the model's declared schema tests (unique, not_null, relationships, accepted_values). These are assertions expressed as data contracts, complementing the Python value checks — together they cover both "the numbers are right" and "the shape is right."

Output.

Assertion	Checks	Catches
r.success (run)	model compiles + runs	broken SQL
n > 0	rows produced	empty output
bad == 0	no null business key	cleaning bug
dbt test	unique/not_null/relationships	contract violation

Rule of thumb. Test a dbt model by seeding its source, running only that model with --select, and asserting on the built table — then run dbt test to enforce the schema contracts. Driving dbt from pytest via dbtRunner puts the whole arrange-act-assert in one function and exercises the exact SQL that ships.

Senior interview question on structuring warehouse integration tests

A senior interviewer might ask: "Design the integration test setup for a dbt warehouse pipeline that must run on every pull request. Cover how each test gets an isolated warehouse, whether you use DuckDB or the real engine and why, how you seed and run the actual transform, how you assert on the output without flakiness, and how you test both the values and the schema contracts."

Solution Using a per-test DuckDB warehouse, the real dbt model, and stable frame assertions

# 1. ISOLATION: a throwaway DuckDB database per test -> naturally isolated + parallel.
import duckdb, pandas as pd, pytest
from pandas.testing import assert_frame_equal
from dbt.cli.main import dbtRunner

@pytest.fixture
def warehouse(tmp_path):
    path = tmp_path / "test.duckdb"          # unique temp path per test (pytest tmp_path)
    con = duckdb.connect(str(path))
    yield con
    con.close()                              # teardown: file + connection gone

# 2. SEED + ACT + ASSERT for a plain-SQL transform (fast path, most tests).
DAILY = ("CREATE TABLE daily AS SELECT region, SUM(total_cents) revenue_cents "
         "FROM orders GROUP BY region")

def test_rollup_values(warehouse):
    warehouse.execute("CREATE TABLE orders (region VARCHAR, total_cents BIGINT)")
    warehouse.execute("INSERT INTO orders VALUES ('EU',4200),('EU',800),('US',990)")
    warehouse.execute(DAILY)                                     # the REAL transform
    got = warehouse.execute("SELECT region, revenue_cents FROM daily "
                            "ORDER BY region").df()              # ORDER -> stable
    expected = pd.DataFrame({"region": ["EU", "US"], "revenue_cents": [5000, 990]})
    assert_frame_equal(got, expected, check_dtype=False)        # cents (int) -> no float drift

# 3. HIGH-FIDELITY path: run the actual dbt model + its contracts on dbt-duckdb.
def test_stg_orders_model():
    dbt = dbtRunner()
    assert dbt.invoke(["seed", "--select", "raw_orders", "--profiles-dir", "ci"]).success
    assert dbt.invoke(["run",  "--select", "stg_orders", "--profiles-dir", "ci"]).success
    assert dbt.invoke(["test", "--select", "stg_orders", "--profiles-dir", "ci"]).success

Step-by-step trace.

Layer	Choice	Why
Isolation	per-test DuckDB file (tmp_path)	clean world, parallel-safe
Engine	DuckDB for logic; real engine smoke test	fast + faithful enough
Act	imported SQL / real dbt model	tests the shipped transform
Assert (values)	sorted assert_frame_equal, int cents	stable, no float drift
Assert (contracts)	dbt test schema tests	unique / not_null / relationships
Teardown	close connection / temp file removed	no leak

After the design, each test gets its own DuckDB database on a pytest tmp_path, so tests are isolated and safely parallel with no shared schema to clash on. The common, logic-focused tests seed a small graph, run the imported transform SQL, and compare a sorted output frame to an expected one in integer cents — stable because the query is ordered and money is not a float. The high-fidelity tests drive real dbt (seed/run/test) against dbt-duckdb, exercising the exact model and its schema contracts. A nightly smoke test re-runs a subset on the real warehouse to catch dialect edges DuckDB approximates. Teardown is a connection close and a temp file that pytest removes.

Output:

Metric	Real-warehouse-only tests	DuckDB + dbt design
Per-test setup	minutes (warehouse)	milliseconds (in-process)
Isolation	shared, leak-prone	per-test DB, clean
Parallel-safe	hard	trivial
Flakiness	order/float drift	sorted + integer cents
Fidelity	full	logic in CI + real-engine smoke

Why this works — concept by concept:

• Throwaway warehouse per test — a per-test DuckDB file (or ephemeral schema) starts empty and is destroyed on teardown, so every test is isolated by construction and the suite parallelises with no shared state to collide on.
• DuckDB for speed, real engine for fidelity — running the logic in an in-process engine keeps CI in seconds, while a smaller smoke test on the real warehouse catches the dialect and type edges DuckDB only approximates — coverage where it is cheap, fidelity where it matters.
• Running the real transform — importing the shipped SQL or invoking the actual dbt model makes the SUT the production query, so the test fails when the transform is wrong rather than when a paraphrase is wrong.
• Stable assertions — ordering the output and comparing money as integer cents removes the two classic flakiness sources (unstable row order and float drift), and dbt test adds contract-level assertions on top of the value checks.
• Cost — milliseconds of in-process setup and a sorted frame comparison per test, versus minutes of warehouse provisioning and flaky order/float assertions. The eliminated cost is a slow, unreliable PR pipeline — O(ms) DuckDB runs instead of O(minutes) warehouse spin-ups per test.

ETL
Topic — etl
ETL problems on transforms and end-to-end pipeline tests

Practice →

Data processing Topic — data-processing Data processing problems on output validation and comparisons

Practice →

---

5. Isolation, parallelism, teardown, and CI

Every test starts from a known-clean world and leaves none behind — that is what lets you run them in parallel and trust the result

The mental model in one line: a warehouse test suite is trustworthy only if every test is isolated (it neither sees nor leaves shared state), which is what lets you run it in parallel to keep CI fast — and the mechanisms are a small set of choices: transactional rollback or per-test truncate/schema for isolation, a schema-or-database per xdist worker so parallel processes never collide, a guaranteed teardown (in a finally/post-yield block) so a failing test still cleans up, and a CI setup that provides the warehouse (a Postgres service container, or DuckDB with no service at all) and reseeds efficiently — because a suite that leaks state is flaky, and a flaky suite is one nobody trusts or keeps green. Isolation and teardown are not hygiene niceties; they are the precondition for parallelism, and parallelism is what keeps the suite fast enough to run on every push.

Isolation strategies, cheapest to heaviest.

• Transactional rollback. Fastest; per-function transaction discarded on teardown. Use when the SUT does not commit.
• Truncate between tests. TRUNCATE ... RESTART IDENTITY CASCADE on the touched tables. Use when the SUT commits into known tables.
• Schema per test. CREATE SCHEMA + DROP ... CASCADE. Use when the SUT issues DDL or creates arbitrary tables.
• Database per worker. A separate database per parallel worker (below) — the coarsest isolation, the basis for parallelism.

Parallelism with pytest-xdist.

• -n auto. pytest -n auto (pytest-xdist) runs tests across as many worker processes as there are cores, cutting wall-clock time roughly linearly.
• worker_id. Each worker exposes a worker_id fixture (gw0, gw1, …, or master when not distributed) — the key to giving each worker its own database or schema namespace.
• Per-worker resources. Create one database (or schema prefix) per worker so workers never write to the same table; DuckDB per-process is inherently isolated and needs no coordination.

Teardown discipline.

• Always in teardown. Cleanup goes after yield (or in a finally), never after the assertion — so it runs whether the test passes, fails, or errors.
• Idempotent + defensive. DROP ... IF EXISTS, TRUNCATE guarded, so teardown succeeds even if setup half-failed.
• Leak detection. A session-end check that no test_* schemas or temp databases remain catches fixtures that forgot to clean up before they accumulate.

CI setup.

• Provide the warehouse. A Postgres service container for engine-fidelity tests, or nothing at all for DuckDB — DuckDB needs no service, which is why it is the default for PR pipelines.
• Seed strategy. Apply schema/migrations once per job (session fixture), reseed the mutable data per test; do not re-run migrations per test.
• Fixtures in the repo. Keep factories, seed CSVs, and expected frames version-controlled next to the tests so the input is reviewable and reproducible.
• Keep it fast. -n auto, DuckDB where possible, --select to scope dbt, and split slow real-engine tests into a nightly job so the PR gate stays in seconds.

The failure modes senior engineers pre-empt.

• Order-dependent flakiness. A test that passes only after another ran shares state. Mitigation: full isolation (rollback/schema-per-test) and running in random order to surface the coupling.
• Cross-worker collisions. Parallel workers writing the same table corrupt each other. Mitigation: a database/schema per worker_id, or per-process DuckDB.
• Leaked schemas/databases. A skipped teardown accumulates test_* schemas until the server clogs. Mitigation: teardown in finally, DROP IF EXISTS, and a leak-detection sweep.

Common interview probes on isolation and CI.

• "How do you run warehouse tests in parallel?" — pytest-xdist with a database/schema per worker_id; DuckDB per process needs no coordination.
• "How do you guarantee cleanup on failure?" — teardown in a finally/post-yield block, idempotent DROP IF EXISTS.
• "How do you keep CI fast?" — DuckDB (no service), -n auto, migrate once/reseed per test, nightly job for slow real-engine tests.
• "How do you catch flaky, order-dependent tests?" — full isolation plus randomised order; a leak sweep at session end.

Worked example — parallel-safe isolation with a database per xdist worker

Detailed explanation. To run warehouse tests in parallel without collisions, give each xdist worker its own database, keyed on worker_id. Each worker builds and tears down its own database once per session; tests within a worker share it and isolate with rollback/schema. Build the per-worker fixture.

• The key. worker_id — gw0, gw1, … per worker (master when serial).
• The resource. One database per worker, created at session scope.
• The isolation. Within a worker, tests still rollback/schema-isolate.

Question. Write a session fixture that provisions a separate database per xdist worker so pytest -n auto never collides, and drops it afterward.

Input.

Run mode	worker_id	Database
serial	master	test_master
-n auto worker 0	gw0	test_gw0
-n auto worker 1	gw1	test_gw1
collisions	—	impossible (unique db)

Code.

# conftest.py — one database PER worker so parallel processes never share tables.
import pytest
from sqlalchemy import create_engine, text

@pytest.fixture(scope="session")
def worker_db(worker_id):
    # worker_id is 'master' (serial) or 'gw0','gw1',... under `pytest -n auto`.
    db_name = f"test_{worker_id}"
    admin = create_engine(
        "postgresql+psycopg://test:test@localhost:5432/postgres",
        isolation_level="AUTOCOMMIT")                 # CREATE/DROP DATABASE can't run in a txn
    with admin.connect() as c:
        c.execute(text(f"DROP DATABASE IF EXISTS {db_name}"))
        c.execute(text(f"CREATE DATABASE {db_name}"))

    eng = create_engine(f"postgresql+psycopg://test:test@localhost:5432/{db_name}")
    with eng.begin() as c:                            # apply schema ONCE per worker
        c.execute(text("CREATE TABLE orders (id INT, region TEXT, total_cents BIGINT)"))
    yield eng

    # teardown: dispose, then drop the worker's whole database.
    eng.dispose()
    with admin.connect() as c:
        c.execute(text(f"DROP DATABASE IF EXISTS {db_name}"))
    admin.dispose()

@pytest.fixture
def db(worker_db):
    # within a worker, still isolate per test with a rolled-back transaction.
    conn = worker_db.connect()
    txn = conn.begin()
    try:
        yield conn
    finally:
        txn.rollback()
        conn.close()

def test_runs_in_its_own_database(db):
    db.execute(text("INSERT INTO orders VALUES (1,'EU',4200)"))
    assert db.execute(text("SELECT count(*) FROM orders")).scalar() == 1

Step-by-step explanation.

1. The worker_db fixture reads the built-in worker_id (from pytest-xdist), which is gw0, gw1, … per parallel worker or master when running serially — a unique token per process.
2. It derives a per-worker database name (test_gw0) and creates it with an AUTOCOMMIT engine, because CREATE DATABASE/DROP DATABASE cannot run inside a transaction. Each worker thus owns a physically separate database — no two workers can touch the same table.
3. The schema is applied once per worker (session scope), not per test, so the expensive DDL is amortised across every test that worker runs.
4. Inside a worker, the db fixture still wraps each test in a rolled-back transaction, so tests within a worker are isolated from each other too — per-worker database for cross-process isolation, per-test rollback for within-process isolation.
5. Teardown disposes the worker's engine and drops its database, so nothing leaks. Running pytest -n auto now spreads the suite across cores with each worker in its own database — linear speedup with zero collisions.

Output.

Workers	Databases	Collisions	Wall-clock
1 (serial)	test_master	none	1x
4 (-n auto)	test_gw0..3	none	~4x faster
shared db (naive)	one	frequent	flaky
teardown	dropped per worker	—	no leak

Rule of thumb. Parallelise with pytest-xdist by giving each worker its own database keyed on worker_id, applying the schema once per worker and still rolling back per test inside it. A physically separate database per process makes collisions impossible, which is what lets -n auto deliver near-linear speedup on a warehouse suite.

Worked example — guaranteed teardown even when a test fails

Detailed explanation. The bug that quietly rots a suite is teardown that only runs on success: put cleanup after the assertion and a failing test leaks its schema, which accumulates until the server clogs or the next test collides. The fix is teardown in a finally/post-yield block plus idempotent drops. Build a leak-proof ephemeral-schema fixture and a session-end leak check.

• The rule. Cleanup after yield, so pytest runs it on pass, fail, or error.
• Idempotent. DROP SCHEMA IF EXISTS ... CASCADE succeeds even if setup half-failed.
• The sweep. A session-end assertion that no test_* schemas remain.

Question. Write an ephemeral-schema fixture whose teardown runs even when the test raises, plus a session-end check that no schemas leaked.

Input.

Scenario	Naive (cleanup after assert)	Guaranteed (finally)
test passes	schema dropped	schema dropped
test fails	schema LEAKS	schema dropped
setup half-fails	dangling schema	IF EXISTS drop still safe
session end	leaks accumulate	sweep asserts none remain

Code.

import uuid
import pytest
from sqlalchemy import create_engine, text

@pytest.fixture(scope="session")
def engine():
    eng = create_engine("postgresql+psycopg://test:test@localhost:5432/test")
    yield eng
    eng.dispose()

@pytest.fixture
def ephemeral_schema(engine):
    name = f"test_{uuid.uuid4().hex[:8]}"
    with engine.begin() as c:
        c.execute(text(f'CREATE SCHEMA "{name}"'))
    try:
        yield name
    finally:
        # runs on PASS, FAIL, or ERROR -> a failing test can never leak its schema.
        with engine.begin() as c:
            c.execute(text(f'DROP SCHEMA IF EXISTS "{name}" CASCADE'))   # idempotent

def test_that_raises(ephemeral_schema, engine):
    with engine.begin() as c:
        c.execute(text(f'CREATE TABLE "{ephemeral_schema}".t (x INT)'))
    assert 1 == 2          # FAILS — but the finally-block teardown still drops the schema

@pytest.fixture(scope="session", autouse=True)
def _leak_sweep(engine):
    # after the whole session, assert no test_* schemas were left behind.
    yield
    with engine.begin() as c:
        leaked = c.execute(text(
            "SELECT schema_name FROM information_schema.schemata "
            "WHERE schema_name LIKE 'test_%'")).fetchall()
    assert not leaked, f"leaked schemas: {[r[0] for r in leaked]}"

Step-by-step explanation.

1. ephemeral_schema creates a uniquely-named schema in setup, then yields the name inside a try. The finally block is the crux: pytest guarantees the post-yield teardown runs whether the test returns normally, fails an assertion, or raises an unexpected error.
2. test_that_raises deliberately fails (assert 1 == 2) after creating a table in its schema. Because the drop lives in the fixture's finally, the schema is still cleaned up — a naive fixture that dropped the schema after the assertion would leak it, since the assertion aborts the test before reaching the cleanup.
3. The drop is DROP SCHEMA IF EXISTS ... CASCADE — idempotent and defensive, so it succeeds even if setup half-failed (e.g. the schema was never created) and it removes every object the transform created, not just the ones the test declared.
4. The _leak_sweep session fixture runs after the entire suite and queries information_schema.schemata for any surviving test_* schemas, asserting there are none. This catches any fixture that forgot to clean up before the leaks accumulate across CI runs.
5. Together, per-fixture finally teardown and a session-end sweep make leaks both prevented and detected — the test author cannot silently ship a fixture that leaks, because the sweep fails the build if one does.

Output.

Event	Schema after	Suite result
passing test	dropped	green
failing test	dropped (finally)	red (assert), no leak
half-failed setup	safe (IF EXISTS)	no dangling schema
session end	none remain	sweep passes

Rule of thumb. Put teardown in a finally/post-yield block with idempotent DROP ... IF EXISTS, so a failing test cleans up as reliably as a passing one, and add a session-end leak sweep that fails the build if any test_* schema survives. Cleanup that only runs on success is not cleanup — it is a slow leak.

Senior interview question on isolation, parallelism, and CI for warehouse tests

A senior interviewer might ask: "Your warehouse integration suite is slow and occasionally flaky — tests pass locally but fail in CI depending on order. Design the fix: the isolation strategy, how you run it in parallel without collisions, how you guarantee teardown so nothing leaks, and how you structure CI so the PR gate stays fast while still covering the real engine."

Solution Using per-worker databases, per-test isolation, guaranteed teardown, and a tiered CI

# 1. ISOLATION + PARALLELISM: a database per xdist worker, rollback per test inside it.
@pytest.fixture(scope="session")
def worker_db(worker_id):                       # gw0/gw1/... under `-n auto`
    name = f"test_{worker_id}"
    admin = create_engine("postgresql+psycopg://test@localhost/postgres",
                          isolation_level="AUTOCOMMIT")
    with admin.connect() as c:
        c.execute(text(f"DROP DATABASE IF EXISTS {name}"))
        c.execute(text(f"CREATE DATABASE {name}"))
    eng = create_engine(f"postgresql+psycopg://test@localhost/{name}")
    apply_migrations(eng)                        # ONCE per worker, not per test
    yield eng
    eng.dispose()
    with admin.connect() as c:
        c.execute(text(f"DROP DATABASE IF EXISTS {name}"))

@pytest.fixture
def db(worker_db):
    conn = worker_db.connect(); txn = conn.begin()
    try:
        yield conn
    finally:
        txn.rollback(); conn.close()             # per-test isolation + guaranteed cleanup

# 2. CI — tiered: fast DuckDB gate on every PR, real-engine + nightly for fidelity.
# .github/workflows/tests.yml
jobs:
  pr-fast:                       # every push/PR: DuckDB, no service, parallel, seconds
    steps:
      - run: pip install -e ".[test]"
      - run: pytest -n auto -p randomly tests/unit tests/integration_duckdb

  pr-postgres:                   # every PR: real engine for dialect fidelity
    services:
      postgres:
        image: postgres:16
        env: { POSTGRES_USER: test, POSTGRES_PASSWORD: test }
        ports: ["5432:5432"]
    steps:
      - run: pytest -n auto tests/integration_pg

  nightly-warehouse:            # nightly only: slow real-warehouse smoke tests
    if: github.event.schedule
    steps:
      - run: pytest -m warehouse_smoke

Step-by-step trace.

Concern	Mechanism	Effect
Cross-worker isolation	database per worker_id	no table collisions
Within-worker isolation	rollback per test	order-independent
Migration cost	apply once per worker	fast
Teardown	finally + DROP IF EXISTS	no leaks on failure
Order flakiness	pytest -p randomly	surfaces coupling
CI speed	DuckDB PR gate, -n auto	seconds
Fidelity	Postgres PR + nightly warehouse	dialect covered

After the redesign, each xdist worker owns a physically separate database (so parallel tests cannot collide), and within a worker every test runs in a rolled-back transaction (so order cannot matter); migrations apply once per worker rather than per test. Teardown lives in finally blocks with idempotent drops, so a failing test leaks nothing, and pytest-randomly shuffles order to surface any residual coupling. CI is tiered: a DuckDB gate runs on every push in seconds with no service, a Postgres job covers dialect fidelity on every PR, and the slow real-warehouse smoke tests run nightly — so the PR gate stays fast while the real engine is still exercised.

Output:

Metric	Before (slow, flaky)	After (isolated, tiered)
Order dependence	fails by order	order-independent
Parallel-safe	collisions	database per worker
Cleanup on failure	leaks	guaranteed (finally)
PR gate time	minutes	seconds (DuckDB, -n auto)
Real-engine coverage	none/slow	PR Postgres + nightly warehouse

Why this works — concept by concept:

• Database per worker — a physically separate database per xdist worker_id makes cross-process collisions impossible, which is the precondition for running a warehouse suite in parallel and getting near-linear speedup.
• Rollback per test — wrapping each test in a discarded transaction inside its worker's database makes tests order-independent, so a green run does not depend on which test happened to run first.
• Guaranteed teardown — cleanup in finally blocks with idempotent DROP IF EXISTS means a failing test cleans up as reliably as a passing one, and a leak sweep turns any lapse into a red build instead of a slow accumulation.
• Tiered CI — a DuckDB gate for speed on every PR, a Postgres job for dialect fidelity, and a nightly real-warehouse smoke test put coverage where it is cheap and fidelity where it matters, keeping the PR gate in seconds.
• Cost — one database and one migration per worker plus a rollback per test, versus a shared, leak-prone database re-migrated per test and a minutes-long serial run. The eliminated cost is a slow, flaky pipeline nobody trusts — O(workers) setup with O(1) per-test reset instead of O(tests) rebuilds run serially.

Defensive coding
Topic — defensive-coding
Defensive coding problems on isolation, cleanup, and flakiness

Practice →

Design
Topic — design
Design problems on test architecture and CI pipelines

Practice →

---

Cheat sheet — seeding & fixtures

• The testing gap. Unit-test pure Python you own; integration-test the SQL/dbt, because the logic lives in the query and the input is the data. A mock can only echo the answer you assumed — the warehouse is the system under test, and only real seeded data exercises it.
• Arrange-act-assert. Every warehouse test: arrange a referentially-consistent seed (parents before children, edge cases on purpose) → act by running the real transform (imported SQL or the dbt model, not a paraphrase) → assert on the output, ordered for stability. If arrange is two isolated rows, the test proves nothing.
• Fixture scope. Push each piece of setup to the widest scope where it stays correct: build the engine/schema once at session scope (expensive, immutable); reset the seeded rows every test at function scope (cheap, mutable). Never seed mutable data at session scope — test 2 will see test 1's writes.
• Reset strategy. Transactional rollback is fastest — wrap each test in a transaction and discard it — but it is a no-op if the SUT commits or issues DDL. Then use TRUNCATE ... RESTART IDENTITY CASCADE (known tables) or DROP SCHEMA ... CASCADE; CREATE SCHEMA (transform creates new objects). Match the reset to what the transform does.
• Factories. A factory mints one valid row so a test overrides only the field it cares about; SubFactory wires a child's foreign key to a freshly-built parent so the graph is always referentially consistent; pass an explicit parent to make siblings share one. create() persists, build() stays in memory.
• Determinism. Reseed the generator (factory.random.reseed_random(seed)) and freeze the clock (freeze_time) so the same seed reproduces the same rows; assert on the fields you pinned (overrides, sequences, counts), not on unpinned Faker values. Money in integer cents, never floats.
• dbt seeds. Load static reference data (dimensions, maps, date spines) as typed CSV seeds (+column_types in dbt_project.yml) and ref() them like models; manufacture the volatile facts with factories. An unseeded dimension turns a JOIN into silent nulls.
• Throwaway warehouse. Give each test an in-process DuckDB database (zero setup, naturally isolated, dbt-duckdb for real models) or an ephemeral schema (CREATE SCHEMA test_<uuid> + search_path, DROP ... CASCADE on teardown) on the real engine for dialect fidelity.
• Asserting on output. Sort the query (ORDER BY) or the frame before comparing — SQL row order is not guaranteed. Use assert_frame_equal on integer-cents columns; assert invariants (no null keys, totals match) for large/partly-generated outputs; add dbt test for unique/not_null/relationships contracts.
• Isolation + parallelism. Rollback/schema-per-test isolates within a process; a database (or schema) per worker_id isolates across pytest -n auto workers so parallel runs never collide. DuckDB per process is inherently isolated. Randomise order (pytest-randomly) to surface coupling.
• Teardown. Cleanup goes after yield / in a finally, never after the assertion — so it runs on pass, fail, and error — with idempotent DROP IF EXISTS. Add a session-end leak sweep that fails the build if any test_* schema/database survives.
• CI. Tier it: a DuckDB gate on every PR (no service, -n auto, seconds), a Postgres service-container job for dialect fidelity, and slow real-warehouse smoke tests on a nightly schedule. Apply migrations once per job; reseed mutable data per test; version-control factories, seed CSVs, and expected frames next to the tests.

Frequently asked questions

What are seeding and fixtures in the context of integration tests?

A fixture is the setup-and-teardown scaffolding a test needs — in pytest, a yield-based function that builds a resource (a database connection, an isolated schema, a factory), hands it to the test, and cleans up afterward. Seeding is the act of loading realistic input data into that resource before the test runs. Together they solve the core problem of testing a data pipeline: the warehouse is the system under test, so you must construct a realistic, referentially-consistent input (seed it), run the real transform, assert on the output, and return to a known-clean state (teardown). Fixtures control how often setup runs (scope) and how cleanly the world resets (rollback vs truncate vs drop-schema); seeding — via factories for volatile facts and dbt seeds for static reference data — controls what the transform sees. A test with good fixtures but toy seed data still proves nothing; a test with realistic seed data but leaky fixtures becomes flaky. You need both.

Why aren't unit tests enough for a SQL or dbt pipeline?

Because the logic under test is the SQL, not Python you can call in isolation. A dbt model's aggregation, its joins, its null handling, its window functions — that behaviour lives in the query and only appears when a real engine executes it over real data. A unit test that mocks the database asserts on whatever the mock was told to return, so it can confirm your assumptions but never surface a LEFT JOIN that fans out, a GROUP BY that drops a null key, or a SUM over a duplicated row. Those are exactly the bugs that reach production, and they are properties of the data shape, which means you can only cover them by seeding that shape and running the actual transform. Unit-test the pure functions you own (parsing, formatting, validation); integration-test the transforms, because a green mock over broken SQL is a false green that ships the bug.

pytest fixture scope — function, module, or session?

Push each piece of setup to the widest scope at which it stays correct. Expensive, immutable setup — creating the database engine, spinning up a container, applying the schema or migrations — belongs at session scope so it runs once for the whole test run. Cheap, mutable state — the seeded rows, the per-test transaction — belongs at function scope (the default) so every test starts clean and independent. The trap is putting mutable data at a wide scope: if you seed rows in a session fixture, the second test sees the first test's writes, and now the suite is order-dependent and flaky. module/class scope is a middle ground for moderately expensive, read-mostly setup shared by a group of related tests. The rule of thumb: build the world once (session), reset the data every test (function).

Rollback or truncate between tests?

Default to transactional rollback: wrap each test in a transaction and discard it on teardown, so nothing is ever persisted and the reset is instant — the fastest possible isolation. But rollback silently becomes a no-op when the code under test commits (a stored procedure with COMMIT, a dbt run, a CREATE TABLE AS) or uses a separate connection, because the test's outer transaction has nothing to undo. In those cases switch to a heavier reset: TRUNCATE ... RESTART IDENTITY CASCADE when the transform commits into tables you pre-declared, or DROP SCHEMA ... CASCADE; CREATE SCHEMA when it creates new objects a truncate would not know about. The decision is entirely about what the SUT does: if it never commits, rollback; if it commits into known tables, truncate; if it issues DDL, drop and recreate the schema. Keep the fast rollback fixture for the majority of tests and reserve the heavy reset for the minority that need it.

How do I keep seed data referentially consistent and deterministic?

For consistency, seed a graph, not rows: insert parents before children and satisfy every foreign key. Factories make this automatic — a child factory declares its parent as a SubFactory, so building the child first builds a valid parent and wires the key; pass an explicit parent when several children should share one. Load the static dimensions a transform joins to as dbt seeds so those lookups resolve. For determinism, pin the two sources of randomness: reseed the generator (factory.random.reseed_random(seed) and Faker's seed) so "random" values reproduce, and freeze the clock (freeze_time) so timestamp defaults and date math are stable. Then assert only on the fields you controlled — overrides, sequences, counts — rather than on an unpinned generated string, and compare monetary values in integer cents to avoid float drift. The result is a seed that is a realistic instance of the schema and reproduces byte-for-byte on every run, so an assertion tests the transform rather than flapping on the data.

How do I run warehouse integration tests in parallel without flakiness?

Isolation is the precondition for parallelism, so establish it at two levels. Across processes, use pytest-xdist (pytest -n auto) and give each worker its own database (or schema namespace) keyed on the worker_id fixture, so two workers can never write to the same table. Within a worker, keep per-test isolation with a rolled-back transaction or a fresh ephemeral schema, so order does not matter. DuckDB makes this trivial — a per-process in-memory database is inherently isolated and needs no coordination — which is why it is the default for fast CI. Guarantee teardown in finally/post-yield blocks with idempotent DROP IF EXISTS so a failing worker leaks nothing, and add a session-end sweep that fails the build if any test_* schema survives. Finally, run tests in randomised order (pytest-randomly) to surface any hidden coupling before it becomes an intermittent CI failure. Isolated tests parallelise safely; parallel tests keep the suite fast enough to run on every push.

Practice on PipeCode

• Drill the data validation practice library → for the referential-integrity, edge-case, and output-assertion problems that seeding and fixtures make concrete.
• Rehearse pipeline scenarios on the ETL practice library → for the transform, join, and dbt-model cases where an integration test earns its keep over a mock.
• Sharpen the test-architecture axis with the system design practice library → for the isolation, parallelism, teardown, and CI trade-offs a warehouse test suite must get right.
• Stack the prerequisites against PipeCode's broader 450+ data-engineering catalogue to anchor the fixture-scope, factory, and arrange-act-assert patterns against real graded inputs — seeding, referential integrity, deterministic data, and clean teardown.

Lock in seeding-and-fixtures muscle memory

Docs explain pytest and factory_boy. PipeCode drills explain the decision — when a mock hides a `refund` fan-out bug, when `SubFactory` beats a hand-typed foreign key, when transactional `rollback` isolates and when only `DROP SCHEMA` will, and when a per-worker database is the difference between a fast suite and a flaky one. Pipecode.ai is Leetcode for Data Engineering — pipeline-testing practice tuned for the production trade-offs data engineers actually face.

Practice data validation problems →
Practice ETL problems →