Pytest for Data Engineering: Fixtures, Parametrization & Docker-Compose Integration Tests

pytest for data engineering is the load-bearing test framework for every serious 2026 data platform — the one tool that lets a senior engineer wire dbt model contracts, Airflow DAG smoke tests, Spark transform property tests, and end-to-end Postgres+Kafka+MinIO integration suites into a single pytest invocation that runs on a laptop, on a Codespace, and on GitHub Actions with the exact same behaviour. Every ingestion pipeline you ship pushes rows through code paths whose failure modes (schema drift, null explosion, off-by-one window boundaries, transaction leakage) can only be caught by tests that exercise the real behaviour against a real database — and pytest's fixture model, parametrization primitives, and testcontainers ecosystem are what turn that from a Herculean manual chore into a pytest -n auto one-liner.

This guide is the senior-DE walkthrough you wished existed the first time an interviewer asked "walk me through your pytest fixtures scope strategy for a suite with 300 DB-touching tests," or "how do you pytest parametrize a Spark UDF across the null / edge / valid axis without exploding your test count," or "your integration suite spins up pytest docker-compose per-worker under pytest-xdist — how do you keep the Postgres port from colliding?" It walks through the four canonical test layers — unit (pure Python transforms), contract (dbt/Spark schema + not-null assertions), component-with-DB (pytest-postgresql transactional fixtures), and integration-with-compose (pytest testcontainers running Postgres + Kafka + MinIO) — the four axes senior interviewers actually probe (fixture scope, parametrization discipline, container hygiene, CI reproducibility), the canonical config for each, and the CI wiring that keeps a 1000-test data-engineering suite under three minutes on GitHub Actions. Each section pairs a teaching block with a Solution-Tail 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 Python practice library →, rehearse on the ETL practice library →, and sharpen the validation axis with the data-validation practice library →.

---

On this page

• Why pytest is the load-bearing test framework for data engineering in 2026
• Fixtures — session scope, factory patterns, DB seed lifecycles
• Parametrization — table-driven tests over dbt/Spark transforms
• Docker-Compose + Testcontainers integration tests
• CI wiring — coverage, xdist parallelism, flaky-test discipline
• Cheat sheet — pytest recipes for data engineering
• Frequently asked questions
• Practice on PipeCode

---

1. Why pytest is the load-bearing test framework for data engineering in 2026

Four test layers, one framework — why pytest wins the DE stack over unittest, nose, and hand-rolled runners

The one-sentence invariant: pytest for data engineering is the framework where every meaningful test layer (unit transforms, dbt/Spark schema contracts, component tests against a real Postgres, and end-to-end integration suites spun up via pytest docker-compose or pytest testcontainers) shares one runner, one fixture graph, one parametrization primitive, and one CI story — which is what makes a 1000-test data-engineering suite feasible on a laptop, in a Codespace, and on GitHub Actions with identical behaviour. The choice of pytest over unittest, nose2, or Bazel-native runners is not aesthetic; it is structural. Fixtures compose. Parametrization is declarative. Plugins (pytest-xdist, pytest-cov, pytest-postgresql, pytest-mock, pytest-rerunfailures) plug in without ceremony. Every other framework in the Python ecosystem has receded, and in 2026 there is no serious data-platform team that does not run pytest as the entry point.

The four test layers every serious DE team ships.

• Layer 1 — unit tests. Pure-Python transforms with no I/O — a normalize_email function, a parse_iso_timestamp helper, a chunk_iterable utility. Sub-millisecond per test; hundreds per file; run on every save. This is the base of the pyramid.
• Layer 2 — contract tests. dbt schema tests (not_null, unique, accepted_values), Great Expectations suites, JSON-schema validators for API payloads, pydantic model validation. Fast; deterministic; catch schema drift before it reaches production.
• Layer 3 — component tests with a real DB. SQL-emitting Python code executed against a real Postgres (via pytest-postgresql template DB) or DuckDB (in-process). Catch off-by-one window boundaries, transaction-leakage bugs, and index-usage regressions that unit tests cannot. ~100 ms per test.
• Layer 4 — integration tests with docker-compose. Airflow scheduler + Postgres + Redis + a Kafka broker + MinIO, all spun up via docker-compose or testcontainers-python, then a DAG runs end-to-end. ~5–30 s per test; run on PR and nightly. Catches the failure modes that only appear when networked services interact.

The four axes senior interviewers probe.

• Fixture scope discipline. Do you know when to use session (expensive setup like a container) versus module (per-file setup) versus function (per-test isolation)? Do you know that autouse=True at session scope is a footgun that couples every test to that fixture whether it needs it or not? This is the single most-probed pytest topic.
• Parametrization hygiene. Do you know how to use ids= to keep parametrized test names readable? Do you know when to use indirect=True to run params through a fixture? Do you know that pytest_generate_tests lets you parametrize dynamically from a manifest? Senior teams reject test files where parametrized test IDs are test_transform[test_transform0] instead of test_transform[null-input].
• Container hygiene. Do you spin one container per session (fast, shared state) or per-test (slow, hermetic)? Do you know the wait-for-ready health-check idiom? Do you clean up volumes at teardown? Do you know how to make containers survive pytest-xdist parallelism without port collisions? Senior signal: name testcontainers-python and the network= parameter unprompted.
• CI reproducibility. Does the suite pass on your laptop, in Codespaces, and in GitHub Actions the same way — or does it flake on CI only? Do you have a coverage gate (--cov-fail-under)? Do you have flaky-test discipline (pytest-rerunfailures + quarantine + nightly retriage)? Do you know how to shard across xdist workers without breaking DB fixtures?

The 2026 reality — pytest + testcontainers is the default; the alternatives are legacy.

• unittest ships in the stdlib and remains for legacy codebases, but nobody starts a greenfield DE project with it in 2026. The lack of fixture composition, the boilerplate setUp / tearDown, and the absence of parametrization make it a dead end for the kind of matrix-driven testing DE work demands.
• nose / nose2 are historical curiosities; the maintainers have moved on. Any codebase still on nose should schedule a migration.
• pytest + pytest-xdist + pytest-cov + pytest-postgresql (or testcontainers-python) is the canonical stack. Every dbt-adapter, every Airflow provider, every managed-Spark toolkit publishes fixtures and helpers targeting this stack.
• Bazel / pants / native build runners wrap pytest rather than replacing it — even in monorepos, the actual test execution goes through pytest.

What interviewers listen for.

• Do you name all four test layers without prompting? — senior signal.
• Do you say "session-scope for the expensive container, function-scope for the test data" in the first sentence when scope comes up? — required answer.
• Do you push back on "just use unittest.mock everywhere" with the argument that "mocks don't catch schema drift; component tests against a real Postgres do"? — senior signal.
• Do you name testcontainers-python (or pytest-docker-compose) as the integration primitive rather than "we shell out to docker-compose"? — senior signal.
• Do you describe pytest as "a fixture graph plus a parametrization matrix" rather than as "a test framework"? — required answer.

Worked example — the four-layer test pyramid for a dbt + Airflow + Spark platform

Detailed explanation. The single most useful artifact for a pytest-for-DE interview is a memorised four-layer pyramid table. Every senior test-strategy discussion converges on this table within the first ten minutes; having it in your head is what separates a fluent answer from a stumbling one. Walk through building the table for a hypothetical retail data platform that runs dbt in Snowflake, Airflow on Kubernetes, and Spark on EMR.

• Platform. dbt-core targeting Snowflake, Airflow 2.9 on Kubernetes, Spark 3.5 on EMR.
• Test count budget. ~800 unit + ~300 contract + ~150 component + ~40 integration = ~1,290 tests total.
• Speed budget. Unit ~10 s total, contract ~30 s, component ~2 min, integration ~5 min. Full suite under 8 minutes on CI.
• Coverage target. ≥85% line coverage, ≥70% branch coverage on the Python packages.

Question. Build the four-layer test pyramid for the platform, name the pytest plugin for each layer, and pick the run-frequency.

Input.

Layer	Tests	Speed	Plugins	Frequency
Unit	800	~10 s	pytest, pytest-mock	on save
Contract	300	~30 s	pytest, pydantic, dbt-core	on save
Component	150	~2 min	pytest-postgresql, pytest-mock	on PR
Integration	40	~5 min	testcontainers-python, pytest-xdist	on PR + nightly

Code.

# tests/conftest.py — one file wires all four layers
import pytest
from pathlib import Path

# ------------------------------------------------------------------
# Layer 1 (unit) — no fixtures needed; pure functions
# ------------------------------------------------------------------


# ------------------------------------------------------------------
# Layer 2 (contract) — pydantic + manifest-driven parametrization
# ------------------------------------------------------------------
@pytest.fixture(scope="session")
def dbt_manifest():
    """Loaded once per session; used by parametrized contract tests."""
    import json
    manifest_path = Path("target/manifest.json")
    return json.loads(manifest_path.read_text())


# ------------------------------------------------------------------
# Layer 3 (component-with-DB) — pytest-postgresql template
# ------------------------------------------------------------------
from pytest_postgresql import factories

postgresql_proc = factories.postgresql_proc(port=None, unixsocketdir="/tmp")
postgresql = factories.postgresql("postgresql_proc", dbname="test_db")


# ------------------------------------------------------------------
# Layer 4 (integration) — testcontainers-python
# ------------------------------------------------------------------
@pytest.fixture(scope="session")
def compose_stack():
    """Spin up Postgres + Kafka + MinIO once per session."""
    from testcontainers.compose import DockerCompose
    with DockerCompose(".", compose_file_name="docker-compose.test.yml") as compose:
        compose.wait_for("http://localhost:9001/minio/health/ready")
        yield compose

Step-by-step explanation.

1. Layer 1 (unit) needs no fixtures — pure Python transforms are called directly with in-memory args. This is the fastest layer and where the bulk of the tests live. Sub-millisecond per test.
2. Layer 2 (contract) uses one session-scoped dbt_manifest fixture loaded once, then parametrized tests iterate every model in the manifest. This gives you 300 tests for the cost of ~30 lines of code — the parametrization amplifier is what makes contract testing tractable.
3. Layer 3 (component) uses pytest-postgresql's postgresql_proc (session-scoped process fixture that starts one Postgres binary) and postgresql (function-scoped database that gets a fresh template-cloned DB per test). This gives per-test DB isolation at the cost of ~100 ms per test — template-DB cloning is what makes the pattern fast.
4. Layer 4 (integration) uses testcontainers-python's DockerCompose fixture at session scope — the compose stack starts once per pytest run, and every integration test shares it. The wait_for health check is non-optional; without it, tests race the container startup and flake.
5. The four layers are declared in a single conftest.py at the repo root and automatically discovered by pytest — you don't import them, you don't wire them; they just appear in every test file that names them. This is fixture-graph magic.

Output.

Layer	Fixture scope	Per-test cost	Isolation
Unit	none	<1 ms	pure functions
Contract	session (manifest)	~1 ms per model	in-memory
Component	session (proc) + function (db)	~100 ms	fresh DB per test
Integration	session (compose)	~5–30 s	shared containers, isolated schemas

Rule of thumb. Never mix scopes across layers by accident. If a component test starts to depend on a session-scoped compose fixture, you have promoted it to Layer 4 — accept the ~5 s per-test cost or refactor the dependency out.

Worked example — what interviewers actually probe about pytest for DE

Detailed explanation. The senior data-engineering pytest interview has a predictable structure: the interviewer opens with an ambiguous question ("how would you test a dbt project?"), then progressively narrows to test whether you know the axes. The candidates who name the four layers in sentence one score highest; the candidates who describe "we write unit tests with mocks for everything" score lowest. Walk through the interview grading rubric.

• Ambiguous opener. "How would you build a test suite for our dbt + Airflow + Spark platform?" — invites you to name the four layers.
• Follow-up 1. "How do you make the Postgres fixture fast enough for 150 tests?" — probes fixture scope + template-DB knowledge.
• Follow-up 2. "How do you parametrize a schema-contract test over every table in the schema?" — probes pytest_generate_tests.
• Follow-up 3. "How do you run the suite in parallel without Postgres fixtures colliding?" — probes xdist + port allocation.
• Follow-up 4. "One test in the integration suite flakes 5% of the time. What's your discipline?" — probes rerunfailures + quarantine.

Question. Draft a 5-minute senior pytest-for-DE answer that covers all four axes without waiting to be asked.

Input.

Interview signal	Weak answer	Senior answer
Layers named	"we write pytest tests"	"unit, contract, component-with-DB, integration-with-compose"
Fixture scope	"we use setUp"	"session for containers, function for data, template-DB clone per test"
Parametrize discipline	"we loop inside the test"	"@pytest.mark.parametrize with ids=; pytest_generate_tests for dynamic"
Container story	"we shell out to docker-compose"	"testcontainers-python with wait_for health checks; --xdist-group by port"
Flaky discipline	"we rerun the pipeline"	"pytest-rerunfailures with reruns=2; quarantine marker; nightly retriage"

Code.

Senior pytest-for-DE answer template (5 minutes)
================================================

Minute 1 — name the four layers up front
  "I'd build four layers: unit for pure transforms, contract for
   dbt/pydantic schema, component-with-DB via pytest-postgresql for
   SQL-emitting code, and integration-with-compose via
   testcontainers-python for the end-to-end Airflow DAG."

Minute 2 — fixture scope discipline
  "Session-scoped fixtures for the expensive things — the Postgres
   process, the compose stack, the dbt manifest. Function-scoped
   fixtures for the isolation-critical things — the DB itself
   (template-cloned per test), the tmp_path, the monkeypatched env."

Minute 3 — parametrization discipline
  "Every contract test uses @pytest.mark.parametrize with explicit
   ids= so test names read like test_not_null[orders.customer_id]. For
   dynamic axes — every model in the dbt manifest — I use
   pytest_generate_tests to expand at collection time."

Minute 4 — container hygiene
  "testcontainers-python for integration; DockerCompose fixture at
   session scope with a wait_for health-check. Under xdist, I group
   by --xdist-group so all tests hitting the same container land on
   the same worker; container port is randomised via published_port=0."

Minute 5 — CI reproducibility + flaky discipline
  "GitHub Actions matrix on 3.11/3.12; pytest -n auto with
   --cov-fail-under=85; --junit-xml uploaded to a flaky-test dashboard.
   Flaky tests get @pytest.mark.flaky(reruns=2) plus a quarantine
   label; a nightly job retriages them so nothing stays flaky forever."

Step-by-step explanation.

1. Minute 1 is the crucial framing. Naming the four layers immediately — "unit, contract, component-with-DB, integration-with-compose" — signals you're a strategy-shaper, not a task-runner. Weak candidates dive into tools ("we'd use pytest-mock and…") before naming the layers.
2. Minute 2 addresses fixture scope before the interviewer asks. This preempts the common trap where you commit to "all fixtures at function scope" and then have to defend a 40-minute test run. Naming session/module/function up front is senior signal.
3. Minute 3 is the parametrization probe. Every senior DE test suite is 80% parametrized cases; the fluent answer names ids=, indirect=True, and pytest_generate_tests unprompted.
4. Minute 4 is the container hygiene question. Naming testcontainers-python and the wait_for health-check pattern shows you've built the pattern, not read about it. Naming --xdist-group shows you've run it under parallel CI.
5. Minute 5 covers CI and flakes — the reliability axis. The rerunfailures + quarantine + nightly retriage story shows you've inherited a flaky suite and rescued it, which is the operational credential senior teams hire for.

Output.

Grading criterion	Weak score	Senior score
Names four layers in minute 1	rare	mandatory
Names fixture scope discipline	rare	required
Names parametrize + ids	occasional	mandatory
Names testcontainers-python	rare	senior signal
Names flaky discipline	rare	senior signal

Rule of thumb. The senior pytest-for-DE answer is a 5-minute monologue that covers all four axes without waiting for the follow-ups. Rehearse it once; deploy it every time.

Worked example — the "pick the test layer" decision tree

Detailed explanation. Given a new piece of code, the senior engineer runs a 4-question decision tree to place it on the right test layer. Codifying the tree makes the interview answer reproducible: any interviewer can hand you a code snippet and you can walk the tree out loud. Walk through the tree with three canonical inputs: a pure normalize_phone function, a SQL-emitting find_orders_in_window helper, and an Airflow DAG that reads from Postgres and writes to S3.

• Q1. Does the code touch any I/O (network, disk, DB)? → no = Layer 1 unit; yes = go to Q2.
• Q2. Is the I/O purely SQL against one database? → yes = Layer 3 component with pytest-postgresql; no = go to Q3.
• Q3. Does the code coordinate multiple services (broker, DB, storage)? → yes = Layer 4 integration with compose; no = mock the one external dependency with pytest-mock and stay at Layer 1.
• Q4 (parallel branch). Is the code a schema definition (dbt model, pydantic model, JSON schema)? → yes = Layer 2 contract with parametrized assertion; N/A to Q1.

Question. Walk the decision tree for the three code snippets and record the layer each ends up on.

Input.

Code	Q1 (I/O?)	Q2 (SQL only?)	Q3 (multi-service?)	Layer
normalize_phone	no	—	—	Layer 1 unit
find_orders_in_window	yes	yes	—	Layer 3 component
ingest_daily_orders DAG	yes	no	yes	Layer 4 integration

Code.

# Decision-tree helper (illustrative)
def pick_test_layer(touches_io: bool,
                     is_sql_only: bool,
                     multi_service: bool,
                     is_schema_def: bool) -> str:
    """Return the pytest layer for a code artifact."""
    if is_schema_def:
        return "Layer 2 contract"
    if not touches_io:
        return "Layer 1 unit"
    if is_sql_only:
        return "Layer 3 component"
    if multi_service:
        return "Layer 4 integration"
    return "Layer 1 unit (mock the one external)"


# Walk the three code snippets
print(pick_test_layer(False, False, False, False))
# -> Layer 1 unit

print(pick_test_layer(True,  True,  False, False))
# -> Layer 3 component

print(pick_test_layer(True,  False, True,  False))
# -> Layer 4 integration

print(pick_test_layer(False, False, False, True))
# -> Layer 2 contract

Step-by-step explanation.

1. Scenario 1 — normalize_phone(raw: str) -> str is a pure function with no I/O. The tree short-circuits at Q1 → Layer 1 unit. Sub-millisecond, hundreds of these, run on save.
2. Scenario 2 — find_orders_in_window(conn, start, end) executes SQL against Postgres and returns rows. Q1 = yes, Q2 = yes → Layer 3 component with pytest-postgresql. Per-test template DB clone; ~100 ms; catches window-boundary bugs unit tests cannot.
3. Scenario 3 — ingest_daily_orders_dag reads from Postgres, writes JSONL to MinIO, publishes a completion event to Kafka. Q1 = yes, Q2 = no, Q3 = yes → Layer 4 integration with testcontainers-python. Session-scoped compose; ~10 s; catches inter-service failures.
4. Scenario 4 — a pydantic.BaseModel for the API payload → Layer 2 contract. Parametrized test_payload_shape[orders-good], test_payload_shape[orders-missing-total], etc. Fast; deterministic.
5. If the code fails Q1-Q3 but interacts with exactly one external dependency (say, an HTTP API), mock the dependency with pytest-mock's mocker.patch(...) and keep it at Layer 1. This preserves the pyramid shape — thousands of unit tests, dozens of integration tests, not the inverted pyramid that kills CI budgets.

Output.

Code	Layer	Fixture needed	Speed
normalize_phone	Layer 1 unit	none	<1 ms
find_orders_in_window	Layer 3 component	postgresql (function-scope)	~100 ms
ingest_daily_orders_dag	Layer 4 integration	compose_stack (session-scope)	~10 s
OrdersPayload (pydantic)	Layer 2 contract	none	<1 ms

Rule of thumb. The four-question decision tree is a whiteboard-friendly answer. Practice walking it end-to-end so an interviewer can hand you any code snippet and get a layer name in under 30 seconds.

Senior interview question on pytest test-layer strategy

A senior interviewer often opens with: "You inherit a data platform with 900 tests, most of them Layer 1 unit tests with unittest.mock mocking Postgres, and a CI run that takes 25 minutes because the 40 integration tests each spin up their own docker-compose. Walk me through the test-layer strategy you'd migrate to, the fixture-scope changes, and the CI wiring you'd expect to bring the run under 8 minutes."

Solution Using a four-layer pytest pyramid with session-scoped compose + template-DB fixtures + xdist parallelism

# 1. tests/conftest.py — one fixture graph for all four layers
import pytest
from pathlib import Path
from pytest_postgresql import factories

# Layer 3 — pytest-postgresql: one process per session, fresh DB per test
postgresql_proc = factories.postgresql_proc(
    port=None,                      # OS-allocated port (xdist-safe)
    unixsocketdir="/tmp",
    load=[Path("schema.sql").read_text()],  # template DB is pre-seeded
)
postgresql = factories.postgresql(
    "postgresql_proc",
    dbname="test_db",
    load=[],                        # per-test DB is a template clone
)


# Layer 4 — testcontainers: one compose stack per session
@pytest.fixture(scope="session")
def compose_stack():
    from testcontainers.compose import DockerCompose
    with DockerCompose(
        ".",
        compose_file_name="docker-compose.test.yml",
        pull=True,
    ) as compose:
        compose.wait_for("http://localhost:9001/minio/health/ready")
        # Also wait for Postgres + Kafka
        compose.wait_for("http://localhost:9092/kafka-health")
        yield compose

# 2. tests/unit/test_normalize.py — Layer 1
import pytest
from mypkg.transforms import normalize_phone

@pytest.mark.parametrize("raw,expected", [
    ("[phone]", "[phone]"),
    ("415.555.1212",     "[phone]"),
    ("[phone]",   "[phone]"),
], ids=["plus-space", "dot-sep", "paren-sep"])
def test_normalize_phone(raw, expected):
    assert normalize_phone(raw) == expected

# 3. tests/component/test_find_orders.py — Layer 3
import pytest
from mypkg.orders import find_orders_in_window

def test_find_orders_in_window_inclusive_start(postgresql):
    """Window is [start, end) — start is inclusive, end is exclusive."""
    with postgresql.cursor() as cur:
        cur.execute("INSERT INTO orders(id, created_at) VALUES (1, '2026-01-01 00:00:00')")
        cur.execute("INSERT INTO orders(id, created_at) VALUES (2, '2026-01-02 00:00:00')")

    rows = find_orders_in_window(
        postgresql,
        start="2026-01-01 00:00:00",
        end  ="2026-01-02 00:00:00",
    )
    assert [r.id for r in rows] == [1]   # id=1 is inside; id=2 is boundary-exclusive

# 4. tests/integration/test_dag_end_to_end.py — Layer 4
import pytest

@pytest.mark.integration
def test_ingest_daily_orders_dag_end_to_end(compose_stack):
    """Full DAG: read Postgres, write MinIO, publish Kafka completion."""
    from mypkg.dags.ingest import ingest_daily_orders_dag

    # (arrange test data in compose_stack.postgres; elided for brevity)
    result = ingest_daily_orders_dag.test()

    assert result.state == "success"
    # Assert MinIO object landed
    assert compose_stack.get_service_host("minio", 9000) is not None
    # Assert Kafka completion event
    # ...

# 5. .github/workflows/tests.yml — CI matrix
name: tests
on: [push, pull_request]
jobs:
  pytest:
    runs-on: ubuntu-24.04
    strategy:
      matrix:
        python: ["3.11", "3.12"]
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: ${{ matrix.python }} }
      - run: pip install -e '.[test]'
      - run: |
          pytest \
            -n auto \
            --dist=loadgroup \
            --cov=mypkg \
            --cov-branch \
            --cov-fail-under=85 \
            --junit-xml=junit.xml \
            tests/
      - if: always()
        uses: actions/upload-artifact@v4
        with: { name: junit-${{ matrix.python }}, path: junit.xml }

Step-by-step trace.

Step	Before (mock-heavy, per-test compose)	After (four-layer pyramid, session compose + xdist)
Unit tests	900 heavy mocks	800 pure-function tests
Contract tests	0	300 parametrized schema tests
Component tests	0 (mocked)	150 pytest-postgresql tests
Integration tests	40 per-test compose	40 sharing one session compose
Postgres startup	none (all mocked)	one per session (~1 s)
Compose startup	40 × ~10 s = 6.7 min	1 × ~10 s
Parallelism	serial	xdist -n auto (8 workers)
Total CI time	~25 min	~7 min

After the migration, the suite shape has inverted from mock-heavy (unit tests full of mocker.patch("psycopg2.connect") that never caught the off-by-one window bug) to real-behaviour-heavy (component tests against a real Postgres that catch it every time), and the CI budget has dropped from 25 minutes to 7 minutes despite adding 450 tests. The pytest-xdist parallelism amortises the compose startup across all workers.

Output:

Metric	Before	After
Total tests	940	1,290
CI runtime	25 min	7 min
Postgres coverage	mocked (0% real)	100% real behaviour
Compose spin-ups	40 per run	1 per run
Flakes / week	~15	~1
Bug catch rate on PR	~40%	~85%

Why this works — concept by concept:

• Four-layer pyramid — Unit / Contract / Component-with-DB / Integration-with-compose is the shape the industry converged on. Each layer catches a different class of bug at a different cost. Mixing scopes across layers by accident is the anti-pattern this shape prevents.
• Session-scoped compose — The compose stack starts once per pytest run, shared across all Layer-4 tests. This is the ~40-minute-to-40-second win. The wait_for health check makes the fixture reliable.
• Template-DB cloning — pytest-postgresql runs one Postgres process per session, but every test gets a fresh DB by cloning the template. Cloning is ~100 ms; starting a fresh Postgres is ~1 s. This is what makes 150 Layer-3 tests feasible.
• xdist parallelism — -n auto spawns one worker per CPU. Session-scoped fixtures are per-worker, so each worker gets its own Postgres process and the compose stack is shared across workers via --dist=loadgroup. This is the ~7-minute-to-~2-minute latent win when hardware allows.
• Cost — one Postgres binary (~50 MB), one compose stack (~2 GB with Postgres + Kafka + MinIO), coverage tooling (pytest-cov ~3% CPU overhead), and xdist (one worker per CPU). The eliminated cost is the 6.7 minutes of compose spin-ups and the false confidence of mock-heavy unit tests. Net O(1) startup per run versus O(N) startup per test.

Python
Topic — python
Python testing and pytest patterns

Practice →

ETL Topic — etl ETL problems on pipeline testing

Practice →

---

2. Fixtures — session scope, factory patterns, DB seed lifecycles

@pytest.fixture is a dependency-injection graph — and scope discipline is the difference between a 2-minute suite and a 20-minute one

The mental model in one line: pytest fixtures are a directed dependency-injection graph where every fixture declares its own construction cost via a scope (function / class / module / session) and its own teardown via yield or a finalizer, and the difference between a fast suite and a slow one is almost always a fixture that was silently promoted from session to function scope by a well-meaning refactor. Every senior data engineer has debugged the "why did the test suite go from 90 s to 12 min overnight" mystery, and roughly nine times out of ten the answer is that someone changed scope="session" to the default scope="function" on the Postgres process fixture, and now every one of 400 tests starts its own Postgres.

The four scopes for pytest fixtures.

• function (default). Constructed and torn down for every test. Best for data that mutates during the test (a fresh DB row, a temp file, a monkeypatched env var). Cheapest to reason about; most expensive at scale.
• class. Constructed once per test class, torn down after. Rarely used in modern DE codebases because pytest style discourages class TestFoo: grouping.
• module. Constructed once per test file. Useful when a group of related tests share the same setup — e.g. all tests in test_orders_analytics.py reuse the same seeded orders dataset.
• session. Constructed once per pytest run, torn down at exit. Mandatory for expensive setup: the Postgres process, the compose stack, the dbt manifest, an HTTP mock server. The rule of thumb: if setup takes >1 second, it belongs at session scope.

Yield-teardown vs finalizer — pick one style and stick to it.

• yield-teardown. The modern, Pythonic style. yield the_thing at the point where the test receives the fixture; anything after the yield runs at teardown. Reads top-to-bottom like a contextmanager. This is what >95% of 2026 pytest code uses.
• request.addfinalizer(fn). The older style. Register a callback that runs at teardown. Useful when teardown depends on runtime state that doesn't exist yet at yield time (e.g. you need to tear down N containers where N is determined mid-test).
• autouse=True. The footgun. Marks a fixture as automatically active for every test in scope without being explicitly requested. Legitimate use: cross-cutting concerns like resetting a global logger, seeding a UTC timezone, patching datetime.now. Illegitimate use: silently coupling every test to a Postgres fixture — which then breaks every unit test that doesn't need a DB.

The factory-as-fixture pattern — the single most useful pattern for DE test data.

• The problem. Tests need "an order row for customer X with N line items" — but every test needs a different shape. A single orders_seed fixture that returns a hardcoded row is either too specific (only works for one test) or too general (returns a dict every test has to reshape).
• The pattern. The fixture returns a factory function that tests call with the shape they need. def make_order(customer_id=1, total_cents=1000, status='pending'): ... — every test calls make_order(status='shipped') or make_order(customer_id=99) and gets exactly what it needs.
• Why it wins. Test data stays close to the test that needs it (readability), the factory centralises the DB-write logic (DRY), and the factory can produce multiple rows in one test without re-invoking the fixture.
• Composability. Factories compose — make_shipment can call make_order internally. This is how you build hierarchical test data without ceremony.

The DB-seed lifecycle — three patterns, one right answer per use case.

• Truncate-per-test. Fixture truncates every table at teardown. Slow (~50 ms per test); simple; works for small schemas. Suitable for <50 tests total.
• Rollback-per-test (savepoint). Fixture opens a savepoint at setup; rolls back at teardown. Fast (<5 ms per test); requires the test to not commit; works only when the code under test doesn't call COMMIT. Suitable for read-only tests and code that uses the injected connection.
• Template-DB clone-per-test. pytest-postgresql clones a pre-seeded template DB for every test (~100 ms per test); test gets a fresh DB it can COMMIT into; teardown drops the clone. Suitable for tests that must exercise real commit behaviour. This is the pattern for the majority of DE component tests.

Common interview probes on pytest fixtures.

• "What scope should the Postgres fixture be?" — required answer is "the Postgres process at session; the DB itself at function via template-clone."
• "When would you use autouse=True?" — cross-cutting concerns only; never for I/O fixtures.
• "How do you avoid teardown order bugs?" — declare explicit dependencies (fixture A requests fixture B); pytest tears down in reverse dependency order.
• "How do factory fixtures compose?" — one factory calls another; both are yielded as callables from separate fixtures.

Worked example — session-scoped Postgres + function-scoped template DB

Detailed explanation. The canonical DE Postgres fixture setup: one Postgres process per pytest session (~1 s startup, amortised across N tests) plus one template-cloned database per test (~100 ms clone, hermetic). This gives real Postgres behaviour with per-test isolation, at ~1/10th the cost of spinning up a fresh Postgres per test. Walk through the fixture wiring.

• Session fixture. postgresql_proc — one binary, one datadir, one Postgres process. Started once, killed at exit.
• Session fixture. template_db — a database created inside postgresql_proc, pre-seeded with schema + reference data. Cloned by every function-scoped postgresql.
• Function fixture. postgresql — a fresh database cloned from template_db for every test.
• Teardown. Function-scoped fixture drops its clone; session-scoped fixture kills the process.

Question. Wire the three fixtures and write one Layer-3 component test that inserts a row and asserts the schema is present.

Input.

Component	Scope	Setup cost	Teardown
postgresql_proc	session	~1 s	kill process
template_db	session	~500 ms (schema load)	drop DB
postgresql	function	~100 ms (clone)	drop DB

Code.

# tests/conftest.py — the three-fixture pattern
import pytest
from pathlib import Path
from pytest_postgresql import factories

# Session — one Postgres process for the entire pytest run
postgresql_proc = factories.postgresql_proc(
    port=None,                          # OS-allocated (xdist-safe)
    unixsocketdir="/tmp",
    load=[],                            # no schema on the process itself
)

# Session — the template database, pre-seeded with schema.sql
postgresql_template = factories.postgresql(
    "postgresql_proc",
    dbname="template_db",
    load=[Path(__file__).parent / "schema.sql"],
)

# Function — a fresh clone of template_db for every test
postgresql = factories.postgresql(
    "postgresql_proc",
    dbname="test_db",
    load=[Path(__file__).parent / "schema.sql"],
)

-- tests/schema.sql — the schema every test starts with
CREATE TABLE public.orders (
    id            BIGSERIAL PRIMARY KEY,
    customer_id   BIGINT       NOT NULL,
    total_cents   BIGINT       NOT NULL,
    status        TEXT         NOT NULL DEFAULT 'pending',
    created_at    TIMESTAMPTZ  NOT NULL DEFAULT now()
);

CREATE INDEX idx_orders_customer_id ON public.orders (customer_id);
CREATE INDEX idx_orders_status      ON public.orders (status);

CREATE TABLE public.customers (
    id      BIGSERIAL PRIMARY KEY,
    email   TEXT      NOT NULL UNIQUE,
    name    TEXT      NOT NULL
);

# tests/component/test_orders_schema.py — Layer-3 test
def test_orders_schema_has_indexes(postgresql):
    """Every test gets a fresh DB with the schema pre-loaded."""
    with postgresql.cursor() as cur:
        cur.execute("""
            SELECT indexname
            FROM   pg_indexes
            WHERE  schemaname = 'public'
              AND  tablename  = 'orders'
            ORDER  BY indexname
        """)
        indexes = [r[0] for r in cur.fetchall()]

    assert "idx_orders_customer_id" in indexes
    assert "idx_orders_status"      in indexes
    assert "orders_pkey"            in indexes


def test_orders_insert_visible(postgresql):
    """Fresh DB per test — no leakage from previous tests."""
    with postgresql.cursor() as cur:
        cur.execute("""
            INSERT INTO public.orders(customer_id, total_cents)
            VALUES (1, 1000)
            RETURNING id
        """)
        order_id = cur.fetchone()[0]
    postgresql.commit()

    with postgresql.cursor() as cur:
        cur.execute("SELECT COUNT(*) FROM public.orders")
        assert cur.fetchone()[0] == 1     # exactly one — no leakage

Step-by-step explanation.

1. postgresql_proc starts one real Postgres binary on an OS-allocated port with a temp datadir. The port=None is critical for xdist safety — every worker gets a distinct port. This runs once per pytest session.
2. postgresql_template (session) creates one database inside that process, loads schema.sql, and stays around for the duration. It is never mutated during tests — it's the "clean stamp" that every function-scoped clone starts from.
3. postgresql (function) creates a fresh database per test by cloning the template (CREATE DATABASE test_db TEMPLATE template_db). Cloning is a Postgres-native O(1) operation that copies file blocks; ~100 ms even for a large template.
4. Every test receives a psycopg2.Connection to its own fresh DB. Tests can INSERT / COMMIT / DELETE freely; teardown drops the entire database, so leakage between tests is impossible.
5. The alternative — one long-lived DB with a rollback-per-test savepoint — is faster (~5 ms per test) but breaks any code under test that itself calls COMMIT (nested savepoints don't behave the same as top-level transactions). The template-clone pattern is the more forgiving default.

Output.

Test	Postgres process	Template DB	Function DB	Duration
test_orders_schema_has_indexes	reused	reused	freshly cloned	~110 ms
test_orders_insert_visible	reused	reused	freshly cloned	~115 ms
(subsequent tests)	reused	reused	freshly cloned	~100–120 ms
First test in run	started	seeded	cloned	~1.6 s

Rule of thumb. Postgres process at session scope; the database at function scope via template clone. Never start a Postgres per test; never share a database across tests.

Worked example — factory fixture for dbt seed data

Detailed explanation. DE tests routinely need to seed varied data shapes: "an order with 3 line items", "a customer with a 5-year history", "an inventory row at the safety-stock boundary". A single orders_seed fixture can't cover all shapes without becoming a config-driven monster. The factory-as-fixture pattern solves this: the fixture yields a callable that tests invoke with the shape they need. Walk through the pattern.

• The factory function. make_order(customer_id=1, total_cents=1000, status='pending', **kwargs) -> Order. Every kwarg has a default; tests override only what they care about.
• The fixture that yields the factory. Function-scoped — the fixture closes over the DB connection and returns the factory. Test invokes the factory zero or more times.
• Cleanup. The DB connection is torn down by the underlying postgresql fixture; created rows disappear with the cloned DB. No manual cleanup needed.
• Composability. make_order_with_items(customer_id, items) can call make_order internally, then insert line items.

Question. Write the make_order factory fixture and a test that uses it to seed three orders with distinct statuses.

Input.

Component	Value
Factory signature	make_order(customer_id=1, total_cents=1000, status='pending')
Return type	dataclass Order(id, customer_id, total_cents, status)
DB	function-scoped postgresql fixture
Cleanup	automatic (template-clone drops at teardown)

Code.

# tests/conftest.py — factory-as-fixture pattern
import pytest
from dataclasses import dataclass


@dataclass
class Order:
    id: int
    customer_id: int
    total_cents: int
    status: str


@pytest.fixture
def make_order(postgresql):
    """Return a factory function that inserts an order and returns it."""
    def _make(customer_id: int = 1,
              total_cents: int = 1000,
              status: str = "pending") -> Order:
        with postgresql.cursor() as cur:
            cur.execute("""
                INSERT INTO public.orders(customer_id, total_cents, status)
                VALUES (%s, %s, %s)
                RETURNING id
            """, (customer_id, total_cents, status))
            order_id = cur.fetchone()[0]
        postgresql.commit()
        return Order(order_id, customer_id, total_cents, status)

    return _make        # yield the callable, not the row


@pytest.fixture
def make_customer(postgresql):
    """Composable partner factory — used by higher-level make_order_full."""
    counter = {"n": 0}

    def _make(email: str | None = None, name: str = "Test User"):
        counter["n"] += 1
        email = email or f"user-{counter['n']}@example.com"
        with postgresql.cursor() as cur:
            cur.execute("""
                INSERT INTO public.customers(email, name)
                VALUES (%s, %s)
                RETURNING id
            """, (email, name))
            cid = cur.fetchone()[0]
        postgresql.commit()
        return cid

    return _make

# tests/component/test_orders_factory.py — using the factories
def test_status_filter_returns_only_matching(postgresql, make_order):
    """Seed three orders; query by status; assert selectivity."""
    make_order(status="pending")
    make_order(status="shipped")
    make_order(status="cancelled")

    with postgresql.cursor() as cur:
        cur.execute("SELECT COUNT(*) FROM public.orders WHERE status = 'shipped'")
        assert cur.fetchone()[0] == 1


def test_customer_order_join(postgresql, make_customer, make_order):
    """Compose two factories — customer + order — in one test."""
    alice = make_customer(email="[e-mail]", name="Alice")
    make_order(customer_id=alice, status="pending")
    make_order(customer_id=alice, status="pending")

    with postgresql.cursor() as cur:
        cur.execute("""
            SELECT c.name, COUNT(o.id)
            FROM   public.customers c
            JOIN   public.orders   o ON o.customer_id = c.id
            GROUP  BY c.name
        """)
        assert cur.fetchall() == [("Alice", 2)]

Step-by-step explanation.

1. make_order (fixture) opens a closure over postgresql (the function-scoped DB connection) and returns an inner _make function. The fixture itself runs once per test that requests it; the inner function can be called any number of times by the test.
2. Every call to _make(...) executes one INSERT with defaulted-or-overridden values, commits, and returns an Order dataclass carrying the generated id. The test can compose arbitrary sequences of factory calls.
3. make_customer shows the composability pattern — it also uses a counter closure to generate unique emails so tests don't collide on the UNIQUE(email) constraint. This is a common factory idiom: side-effect-free defaults with per-call uniqueness.
4. In test_customer_order_join, both factories are requested; pytest resolves both against the same underlying postgresql connection because that fixture is function-scoped and the two factory fixtures both request it. Fixture-graph sharing is automatic.
5. Cleanup is trivial: the underlying postgresql fixture tears down the entire cloned database at test end, so every row inserted by every factory call vanishes with it. Factories never need explicit cleanup.

Output.

Test	Factory calls	DB rows	Duration
test_status_filter_returns_only_matching	make_order × 3	3 orders	~115 ms
test_customer_order_join	make_customer × 1, make_order × 2	1 cust + 2 orders	~120 ms
(fresh DB per test)	inputs isolated	outputs isolated	(no leakage)

Rule of thumb. Every DE test suite needs at least one factory per major table. Return a callable, defaulted-and-overridable, closing over the DB fixture. This is the pattern the industry converged on because it scales: 5 factories cover 500 tests worth of data-shape variation.

Worked example — tmp_path + monkeypatch for filesystem and env isolation

Detailed explanation. DE code routinely reads env vars (AWS_REGION, DATABASE_URL, KAFKA_BOOTSTRAP), writes to disk (parquet dumps, JSONL landing), and shells out to external tools (dbt run). Tests must isolate these side effects: an env var set in test A cannot leak into test B; a file written in test C cannot pollute test D. Pytest ships two built-in fixtures for this: tmp_path (per-test temp directory that auto-cleans) and monkeypatch (per-test env-var / attribute patcher that auto-restores).

• tmp_path. Yields a pathlib.Path to a fresh directory. Auto-deleted after the test. Alternative: tmp_path_factory at session scope for cross-test reuse.
• monkeypatch. Method-rich — monkeypatch.setenv, monkeypatch.setattr, monkeypatch.delenv. Every change auto-reverts at teardown.
• monkeypatch.chdir. Changes the working directory for the test; reverts at teardown. Critical for dbt tests that assume dbt_project.yml in cwd.
• When to reach for these. Any test that touches os.environ, os.chdir, pathlib.Path for writes, or global module-level attributes. Never modify these directly — always through the fixture.

Question. Write a test that reads env vars for its DB config, writes a JSONL landing file, and asserts both without leaking to sibling tests.

Input.

Concern	Fixture	Restoration
DATABASE_URL env var	monkeypatch.setenv	automatic
KAFKA_BOOTSTRAP env var	monkeypatch.setenv	automatic
/tmp/landing directory	tmp_path	auto-deleted
cwd for dbt run	monkeypatch.chdir	reverted

Code.

# tests/component/test_env_and_fs.py — tmp_path + monkeypatch
import json
from pathlib import Path
from mypkg.landing import write_orders_landing


def test_write_orders_landing_creates_jsonl(tmp_path, monkeypatch):
    """Isolated env + filesystem; every assertion is post-test cleaned."""
    # 1. Env — inject config the code reads via os.environ
    monkeypatch.setenv("LANDING_ROOT", str(tmp_path))
    monkeypatch.setenv("AWS_REGION", "us-west-2")

    # 2. Filesystem — everything under tmp_path is auto-cleaned
    orders = [
        {"id": 1, "total_cents": 1000},
        {"id": 2, "total_cents": 2500},
    ]

    # 3. Exercise the code under test
    output_path = write_orders_landing(orders)

    # 4. Assert the file materialised where we expected
    assert output_path == tmp_path / "orders.jsonl"
    lines = output_path.read_text().splitlines()
    assert len(lines) == 2
    assert json.loads(lines[0]) == {"id": 1, "total_cents": 1000}


def test_env_var_does_not_leak_to_next_test(monkeypatch):
    """Sanity check — env from the previous test must be gone."""
    import os
    assert os.environ.get("LANDING_ROOT") is None      # cleared by monkeypatch teardown
    assert os.environ.get("AWS_REGION")   != "us-west-2"  # or wasn't set at all


def test_dbt_test_uses_chdir(tmp_path, monkeypatch):
    """dbt requires cwd == project root; monkeypatch.chdir reverts."""
    # Create a fake dbt_project.yml in tmp_path
    (tmp_path / "dbt_project.yml").write_text("name: test_project\nversion: 1.0\n")

    monkeypatch.chdir(tmp_path)
    # Anything the code does with Path.cwd() sees tmp_path
    from pathlib import Path
    assert Path.cwd() == tmp_path
    # After the test, cwd reverts automatically

Step-by-step explanation.

1. tmp_path is a function-scoped built-in fixture that yields a fresh temp dir. Pytest deletes the directory at test end (subject to the --basetemp retention setting, typically "keep the last 3"). This makes filesystem tests hermetic by default.
2. monkeypatch.setenv("LANDING_ROOT", str(tmp_path)) injects the env var for the duration of this test only. Pytest records the original value at set time and restores it at teardown — no test can leak an env var into a sibling.
3. Step 3 exercises write_orders_landing, which reads LANDING_ROOT from os.environ and writes orders.jsonl under it. Because both the env var and the target directory are per-test, this test can be re-run under pytest-xdist in parallel with 10 copies of itself without collision.
4. test_env_var_does_not_leak_to_next_test is the sanity check — proves monkeypatch cleaned up. In CI, this pattern is worth its weight in gold for catching accidental global-state leakage.
5. test_dbt_test_uses_chdir shows the monkeypatch.chdir idiom for tools that require a specific working directory. Without this, sibling tests would start in the wrong cwd and fail cryptically.

Output.

Test	tmp_path	env vars set	env vars visible after
test_write_orders_landing	/tmp/pytest-.../test_0	LANDING_ROOT, AWS_REGION	(cleared)
test_env_var_does_not_leak	/tmp/pytest-.../test_1	(none)	(still cleared)
test_dbt_test_uses_chdir	/tmp/pytest-.../test_2	(none, chdir only)	cwd restored

Rule of thumb. For any test that touches env vars, cwd, or filesystem writes, use monkeypatch and tmp_path — never modify globals directly and never rely on test-ordered cleanup. Test isolation is non-negotiable.

Senior interview question on fixture scope discipline

A senior interviewer might ask: "You inherit a suite with 400 pytest tests where every test spins up its own Postgres via docker.run(...) in a function-scoped fixture. Full CI run takes 42 minutes. Walk me through the fixture-scope migration you'd do, the template-DB pattern, the factory pattern for test data, and the exact time budget you'd expect on 8-core CI."

Solution Using session-scoped Postgres process + template-clone per test + factory fixtures

# 1. tests/conftest.py — session Postgres, function template clone, factories
import pytest
from dataclasses import dataclass
from pathlib import Path
from pytest_postgresql import factories


# ------------------------------------------------------------------
# Session — one Postgres binary for the whole test run
# ------------------------------------------------------------------
postgresql_proc = factories.postgresql_proc(
    port=None,                          # OS-allocated port; xdist-safe
    unixsocketdir="/tmp",
    postgres_options="-c fsync=off "    # turn off durability for speed
                     "-c synchronous_commit=off "
                     "-c full_page_writes=off",
)


# ------------------------------------------------------------------
# Function — a fresh clone of the schema-loaded template
# ------------------------------------------------------------------
postgresql = factories.postgresql(
    "postgresql_proc",
    dbname="test_db",
    load=[Path(__file__).parent / "schema.sql"],
)


# ------------------------------------------------------------------
# Factory — reusable data shape generator
# ------------------------------------------------------------------
@dataclass
class Order:
    id: int
    customer_id: int
    total_cents: int
    status: str


@pytest.fixture
def make_order(postgresql):
    """Callable that inserts an order with defaulted overrides."""
    def _make(customer_id=1, total_cents=1000, status="pending") -> Order:
        with postgresql.cursor() as cur:
            cur.execute("""
                INSERT INTO public.orders(customer_id, total_cents, status)
                VALUES (%s, %s, %s) RETURNING id
            """, (customer_id, total_cents, status))
            oid = cur.fetchone()[0]
        postgresql.commit()
        return Order(oid, customer_id, total_cents, status)
    return _make

# 2. A representative Layer-3 test using both fixtures
def test_bulk_status_update(postgresql, make_order):
    """Seed 5 pending orders; bulk-update to shipped; assert count."""
    for _ in range(5):
        make_order(status="pending")

    with postgresql.cursor() as cur:
        cur.execute("""
            UPDATE public.orders
            SET    status = 'shipped'
            WHERE  status = 'pending'
        """)
        rowcount = cur.rowcount
    postgresql.commit()

    assert rowcount == 5

    with postgresql.cursor() as cur:
        cur.execute("SELECT COUNT(*) FROM public.orders WHERE status='shipped'")
        assert cur.fetchone()[0] == 5

# 3. pytest.ini — test-discovery + speed-critical options
[pytest]
addopts = -ra -q --strict-markers --strict-config
markers =
    integration: end-to-end tests that require docker-compose
    slow: tests > 1 s that we want to skip in the fast lane
testpaths = tests

Step-by-step trace.

Step	Before (function-scope Postgres)	After (session + template clone)
Postgres startup	400 × ~1 s = 6.7 min	1 × ~1 s
DB clone	none	400 × ~100 ms = ~40 s
Schema load	400 × ~200 ms = 80 s	1 × ~200 ms (into template)
Test body	400 × ~50 ms = 20 s	400 × ~50 ms = 20 s
Full run (serial)	~42 min	~2 min
Full run (8-core xdist)	~6 min	~30 s

After the migration, the same 400 tests run in 2 minutes serially and 30 seconds with 8-core xdist. The Postgres process starts once per session (per xdist worker), the template DB is created once with the schema, and every test gets a fresh clone via Postgres's native CREATE DATABASE ... TEMPLATE machinery.

Output:

Metric	Before	After
CI runtime (serial)	42 min	2 min
CI runtime (8-core xdist)	~6 min	~30 s
Postgres binary starts	400	1 (× N workers)
Docker image pulls	400	0 (pytest-postgresql uses local binary)
Test isolation	full (own container)	full (own DB)
Memory pressure	~400 × 100 MB = 40 GB peak	~1 × 100 MB per worker

Why this works — concept by concept:

• Session-scoped process fixture — the Postgres binary starts once and is reused. Amortises ~1 s of startup across all N tests. This is the O(1) versus O(N) win.
• Template-DB clone — Postgres's CREATE DATABASE ... TEMPLATE is a file-block copy at the OS level. ~100 ms even for GB-scale schemas. Every test gets a fresh DB with the schema pre-loaded, no per-test schema loading needed.
• Speed-critical Postgres flags — fsync=off, synchronous_commit=off, full_page_writes=off trade durability for speed. Safe in tests because the DB is thrown away at exit. Cuts commit latency by ~5×.
• Factory-as-fixture — one factory per major table, closure over postgresql, callable defaults. Scales to arbitrary data shapes without exploding the fixture count.
• Cost — one Postgres process per xdist worker (~100 MB RAM), one template DB (~50 MB), ~100 ms per test for the clone. The eliminated cost is 400 container startups + 400 schema loads = ~7 minutes. Net O(1) startup per worker versus O(N) startup per test.

Python
Topic — oop/python
OOP + fixtures + factories in Python

Practice →

SQL Topic — sql SQL problems on schema seeding and test data

Practice →

---

3. Parametrization — table-driven tests over dbt/Spark transforms

@pytest.mark.parametrize turns a 100-test suite into a 3-file suite — as long as your ids= are readable

The mental model in one line: pytest parametrize is a decorator that runs the same test function once per row of a parameter table, giving you N logical tests for the cost of one function definition — but the readability of the resulting test IDs (test_transform[null-input] vs test_transform[0]) is the single largest predictor of whether your team maintains the parametrized style or refactors it into 30 copy-pasted test functions six months later. Every senior data engineer has inherited a suite with test_transform[0] through test_transform[47] — impossible to grep for the failing case — and every senior data engineer has replaced it with ids=["null-input", "empty-str", "unicode-emoji", ...] in a two-hour refactor that pays for itself the first time a single case fails.

The four flavours of parametrization every DE engineer uses.

• Static @pytest.mark.parametrize. The classic — a list of tuples decorating the test function. Best when the test cases are known at collection time and small enough to inline. This covers ~70% of DE parametrized tests.
• @pytest.mark.parametrize(..., indirect=True). Passes the parameter through a fixture instead of directly to the test. Useful when the parameter needs setup — e.g. parametrizing which DB backend a test runs against (Postgres vs DuckDB).
• Fixture-level parametrization (@pytest.fixture(params=...)). Parametrizes the fixture itself; every test that requests the fixture runs once per param. Powerful for "run the whole suite against multiple backends" — but easy to over-use.
• Dynamic pytest_generate_tests. The hook — pytest calls it during collection and you supply the params programmatically. Essential for "run this contract test once per model in the dbt manifest" or "once per table in the schema" — where the axes are discovered at runtime.

Axes senior DEs parametrize over — the canonical five.

• Null / empty / edge / valid / invalid. The base axis for any transform. Every column that can be null must have a null-case test; every string must have empty and unicode-emoji cases; every numeric must have zero, negative, and overflow cases.
• Backend. For code that must work on Postgres, MySQL, Snowflake, DuckDB — parametrize the fixture; run the same test across all backends. Catches dialect-specific SQL bugs.
• Time boundary. For window/session logic — [start-inclusive, end-exclusive, span-crossing-midnight, span-crossing-DST, single-instant]. This axis catches roughly half of all off-by-one bugs.
• Volume. For batch code — [1-row, 100-rows, 10k-rows] cases parametrized over the same transform. The 10k case catches memory-explosion bugs the 100 case doesn't.
• Manifest. For contract tests — parametrize over every model / column / test / source in the dbt manifest, or every table in the information_schema. Amplifies one test into hundreds.

The ids= discipline — the single most important pytest hygiene rule.

• Bad. @pytest.mark.parametrize("x,y", [(1, 2), (3, 4), (5, 6)]) → test IDs test_foo[1-2], test_foo[3-4], test_foo[5-6]. Grep-friendly only if your values are unique and semantic.
• Better. @pytest.mark.parametrize("x,y", [(1, 2), (3, 4)], ids=["simple-case", "boundary-case"]) → readable IDs; the failing case's meaning is obvious from the test name.
• Best. Use a helper that generates IDs from a dataclass — ids=lambda p: f"{p.name}-{p.expected}". Scales to hundreds of cases without hand-writing IDs.
• Never. Do not omit ids= for parametrizations with more than 3 cases. Every failing test that reports test_foo[test_foo0] is a debugging tax.

Common interview probes on pytest parametrize.

• "How do you parametrize a test?" — required answer: @pytest.mark.parametrize with ids=.
• "What does indirect=True do?" — routes the param through a fixture, useful for parameterizing setup.
• "How do you parametrize dynamically at collection time?" — pytest_generate_tests hook.
• "What's the tradeoff between fixture-level params and decorator-level params?" — fixture-level multiplies every test in scope; decorator-level scopes to the specific test.

Worked example — Spark UDF parametrized across null/edge/valid inputs

Detailed explanation. A Spark UDF that normalises phone numbers is a canonical parametrization target. The correctness contract has ~6 axes: normal input, leading whitespace, embedded punctuation, null input, empty string, malformed input. Six tests × one function = six lines of parametrize data. Walk through the pattern.

• Function under test. normalize_phone(raw: str | None) -> str | None.
• Axes. null → null; empty → null; valid → digits-only; punctuation → digits-only; unicode → error; too-short → error.
• ID strategy. Human-readable names describing the axis.
• Assert. Exact-match or exception; no wiggle room.

Question. Write the parametrized test covering all six axes with human-readable IDs.

Input.

Axis	raw input	expected output	id
null	None	None	null-in
empty	""	None	empty-str
valid	"[phone]"	"[phone]"	valid-with-punct
digits	"4155551212"	"[phone]"	digits-only
unicode	"📱4155551212"	ValueError	unicode-emoji
short	"555"	ValueError	too-short

Code.

# tests/unit/test_normalize_phone.py — parametrized Spark UDF test
import pytest
from mypkg.udfs.phone import normalize_phone


@pytest.mark.parametrize("raw,expected", [
    (None,                 None),
    ("",                   None),
    ("[phone]",    "[phone]"),
    ("4155551212",         "[phone]"),
], ids=[
    "null-in",
    "empty-str",
    "valid-with-punct",
    "digits-only",
])
def test_normalize_phone_valid_inputs(raw, expected):
    """Six valid-or-null inputs; exact-match against expected."""
    assert normalize_phone(raw) == expected


@pytest.mark.parametrize("raw", [
    "📱4155551212",
    "555",
    "not-a-phone",
], ids=[
    "unicode-emoji",
    "too-short",
    "totally-garbage",
])
def test_normalize_phone_raises_on_invalid(raw):
    """Three invalid inputs; must raise ValueError."""
    with pytest.raises(ValueError):
        normalize_phone(raw)

Step-by-step explanation.

1. Pytest collects test_normalize_phone_valid_inputs four times — once per row in the parametrize table. Each collection produces a distinct test node with a distinct ID: test_normalize_phone_valid_inputs[null-in], test_normalize_phone_valid_inputs[empty-str], etc.
2. When a test fails, the pytest report shows the ID — so a regression in the punctuation-handling case shows up as FAILED tests/unit/test_normalize_phone.py::test_normalize_phone_valid_inputs[valid-with-punct]. You know exactly which axis broke without running the debugger.
3. Splitting valid-vs-invalid into two parametrize decorators is idiomatic — the assertion style differs (equality vs pytest.raises), so one function per assertion style keeps each test focused.
4. pytest.raises(ValueError) is a context manager that asserts the block raises that exception (or a subclass). Using it inside a parametrized loop tests three invalid inputs with one function.
5. If the UDF later gains a "return None for whitespace-only input" behaviour, adding a row (" ", None, "whitespace-only") takes one line — no new test function, no fixture change. This is the parametrization amplifier at work.

Output.

Test ID	raw	expected	actual	pass/fail
test_normalize_phone_valid_inputs[null-in]	None	None	None	pass
test_normalize_phone_valid_inputs[empty-str]	""	None	None	pass
test_normalize_phone_valid_inputs[valid-with-punct]	"[phone]"	"[phone]"	"[phone]"	pass
test_normalize_phone_valid_inputs[digits-only]	"4155551212"	"[phone]"	"[phone]"	pass
test_normalize_phone_raises_on_invalid[unicode-emoji]	"📱..."	ValueError	ValueError	pass
test_normalize_phone_raises_on_invalid[too-short]	"555"	ValueError	ValueError	pass

Rule of thumb. Every non-trivial transform gets a null/empty/valid/edge/invalid parametrization with ids=. Six lines of table data beats six copy-pasted test functions every time.

Worked example — dbt model parametrized over source-fixture rows

Detailed explanation. A dbt model that computes dim_customer_lifetime_value from raw orders needs to be tested against several source-row patterns: customer with one order, customer with 100 orders, customer with zero orders (should return zero LTV), customer with a returned order (should net out). A single parametrized test can seed the source data and assert the model output per case. Walk through the pattern.

• Source rows. Parametrized dicts describing customer + order data.
• Model. dim_customer_lifetime_value.sql — SUM(revenue) - SUM(returns) per customer.
• Assertion. Materialised model row equals expected LTV.
• Cleanup. Each parametrized case starts with a fresh DB (template clone).

Question. Write the parametrized dbt model test covering four source-data shapes.

Input.

Case	Customer orders	Expected LTV	id
single order	1 × $100	$100	one-hundred
bulk orders	100 × $10	$1000	thousand
no orders	0	$0	zero-orders
returned order	1 × $100 + 1 × -$50	$50	with-return

Code.

# tests/component/test_dim_customer_ltv.py — parametrized dbt model test
import pytest
from dataclasses import dataclass
from mypkg.dbt_helpers import run_dbt_model


@dataclass
class CustomerLTVCase:
    name: str
    orders: list[int]     # cents; negative = return
    expected_ltv_cents: int


CASES = [
    CustomerLTVCase("one-hundred",   [10_000],                                10_000),
    CustomerLTVCase("thousand",      [1_000] * 100,                          100_000),
    CustomerLTVCase("zero-orders",   [],                                          0),
    CustomerLTVCase("with-return",   [10_000, -5_000],                        5_000),
]


@pytest.mark.parametrize(
    "case",
    CASES,
    ids=lambda c: c.name,           # ids from the dataclass field
)
def test_dim_customer_ltv(postgresql, case: CustomerLTVCase):
    """Seed source rows for one customer; run the dbt model; assert LTV."""
    # 1. Seed the source table
    with postgresql.cursor() as cur:
        cur.execute("""
            INSERT INTO public.customers(id, email, name)
            VALUES (%s, %s, %s) RETURNING id
        """, (1, "[e-mail]", "Test User"))
        for i, cents in enumerate(case.orders, start=1):
            cur.execute("""
                INSERT INTO public.orders(id, customer_id, total_cents, status)
                VALUES (%s, %s, %s, %s)
            """, (i, 1, cents, "completed" if cents > 0 else "returned"))
    postgresql.commit()

    # 2. Materialise the dbt model (uses postgresql connection)
    run_dbt_model("dim_customer_lifetime_value", conn=postgresql)

    # 3. Assert the LTV row equals the expected cents
    with postgresql.cursor() as cur:
        cur.execute("""
            SELECT ltv_cents
            FROM   analytics.dim_customer_lifetime_value
            WHERE  customer_id = 1
        """)
        row = cur.fetchone()

    if case.orders:
        assert row is not None
        assert row[0] == case.expected_ltv_cents
    else:
        # No orders means no LTV row (or zero-valued row, depending on the model)
        assert row is None or row[0] == 0

Step-by-step explanation.

1. The CustomerLTVCase dataclass captures every axis of one test case — the name (which drives the ID), the input orders, and the expected LTV. Grouping into a dataclass beats a bare tuple because named fields survive refactoring.
2. ids=lambda c: c.name extracts the ID from the dataclass, giving test IDs like test_dim_customer_ltv[one-hundred] and test_dim_customer_ltv[with-return]. Adding a new case is one dataclass entry.
3. Each parametrized invocation gets its own fresh postgresql (function-scope) DB, so the four cases don't leak data into each other. This is why parametrization works cleanly with per-test DB isolation.
4. run_dbt_model is a thin wrapper around dbt run --select dim_customer_lifetime_value that points at the injected connection — this lets tests exercise real dbt behaviour without a full dbt-core install ceremony.
5. The zero-orders case tests a subtle boundary: does the model emit a zero row or no row? The test asserts both possibilities (is None or == 0) because either is a reasonable design; the test locks in whichever the current model chooses.

Output.

Test ID	Orders seeded	Expected LTV	Actual LTV	pass/fail
test_dim_customer_ltv[one-hundred]	1 × 10,000	10,000	10,000	pass
test_dim_customer_ltv[thousand]	100 × 1,000	100,000	100,000	pass
test_dim_customer_ltv[zero-orders]	0	0 (or no row)	no row	pass
test_dim_customer_ltv[with-return]	10,000, -5,000	5,000	5,000	pass

Rule of thumb. Model dbt tests as parametrized cases with a dataclass per axis. Every dbt model gets at least the four cases: single, bulk, zero, edge-case. The dataclass approach outperforms tuple lists for anything past three fields.

Worked example — schema-contract test parametrized over every table

Detailed explanation. A canonical DE contract test: every table in the public schema must have (a) a primary key, (b) a created_at column, (c) a NOT NULL constraint on the primary key. Writing one test per table is tedious; pytest_generate_tests discovers the tables at collection time and generates one test per discovered table. Walk through the hook.

• Discovery. pytest_generate_tests(metafunc) runs at collection; queries information_schema.tables.
• Generation. metafunc.parametrize supplies the table names as params.
• Assertion. Each generated test validates one table's contract.
• Effect. Adding a new table automatically adds a new test — zero test-file changes.

Question. Implement the pytest_generate_tests hook that generates one contract test per table in public.

Input.

Component	Value
Discovery source	information_schema.tables
Filter	schema = 'public' AND table_type = 'BASE TABLE'
Params per table	(table_name,)
Assertions per test	has PK, has created_at, PK is NOT NULL

Code.

# tests/component/test_schema_contract.py — dynamic parametrization
import pytest
import psycopg2

# ------------------------------------------------------------------
# The pytest_generate_tests hook — discovers tables at collection
# ------------------------------------------------------------------
def pytest_generate_tests(metafunc):
    """Generate one test per table in public schema."""
    if "table_name" not in metafunc.fixturenames:
        return          # this hook only fires for tests requesting `table_name`

    # Query the schema (uses a static connection URL; template DB is up)
    conn = psycopg2.connect(
        "host=/tmp dbname=template_db"      # or use env vars
    )
    with conn.cursor() as cur:
        cur.execute("""
            SELECT table_name
            FROM   information_schema.tables
            WHERE  table_schema = 'public'
              AND  table_type   = 'BASE TABLE'
            ORDER  BY table_name
        """)
        tables = [r[0] for r in cur.fetchall()]
    conn.close()

    metafunc.parametrize(
        "table_name",
        tables,
        ids=tables,             # ID = table name itself
    )


# ------------------------------------------------------------------
# The single test — runs once per discovered table
# ------------------------------------------------------------------
def test_table_has_primary_key(postgresql, table_name):
    """Every public table must have a primary key constraint."""
    with postgresql.cursor() as cur:
        cur.execute("""
            SELECT c.contype
            FROM   pg_constraint c
            JOIN   pg_class      t ON t.oid = c.conrelid
            WHERE  t.relname = %s
              AND  c.contype = 'p'
        """, (table_name,))
        pk = cur.fetchone()

    assert pk is not None, f"table {table_name} has no primary key"


def test_table_has_created_at(postgresql, table_name):
    """Every public table must have a created_at column."""
    with postgresql.cursor() as cur:
        cur.execute("""
            SELECT column_name
            FROM   information_schema.columns
            WHERE  table_schema = 'public'
              AND  table_name   = %s
              AND  column_name  = 'created_at'
        """, (table_name,))
        col = cur.fetchone()

    assert col is not None, f"table {table_name} missing created_at"

Step-by-step explanation.

1. pytest_generate_tests is a hook pytest calls during collection for every test function. Inside the hook, we check whether the current test requests the table_name fixture — if not, we skip (the hook fires for all tests, not just ours).
2. When the hook fires for test_table_has_primary_key, it queries the template DB for the current schema, gets back ['customers', 'orders'] (or whatever exists), and calls metafunc.parametrize("table_name", [...]) — this is exactly like the decorator, but supplied dynamically.
3. ids=tables sets the test IDs to the table names themselves: test_table_has_primary_key[customers], test_table_has_primary_key[orders]. When one fails, you know which table is the culprit.
4. Adding a new table to schema.sql automatically adds a new test to the run — zero test-file changes required. This is the amplifier's leverage.
5. Note that the discovery query targets template_db, not the function-scoped postgresql. The discovery must happen at collection time (before any function fixture is available), so we open a direct connection.

Output.

Test ID	Table	PK found	created_at found	pass/fail
test_table_has_primary_key[customers]	customers	yes	—	pass
test_table_has_primary_key[orders]	orders	yes	—	pass
test_table_has_created_at[customers]	customers	—	no	FAIL
test_table_has_created_at[orders]	orders	—	yes	pass

Rule of thumb. For any DE contract that must hold across every table in a schema, use pytest_generate_tests — never hand-write one test per table. The dynamic generation makes new-table addition automatically covered.

Senior interview question on pytest parametrize discipline

A senior interviewer might ask: "You have a Spark job that computes retention cohorts and a dbt model that computes MRR. You need to cover null/edge/valid axes plus a manifest-driven axis 'every model in the marts folder'. Walk me through the parametrization strategy — decorator vs fixture vs pytest_generate_tests, how you generate readable IDs, and how you keep the parametrize table maintainable."

Solution Using layered parametrize decorators + fixture params + pytest_generate_tests

# 1. Layer A — decorator-level, static, axis parametrize
import pytest
from dataclasses import dataclass


@dataclass
class RetentionCase:
    name: str
    signup_dates: list[str]
    activity_dates: list[list[str]]     # per user
    cohort: str
    expected_retention: float


RETENTION_CASES = [
    RetentionCase("all-active",    ["2026-01-01"] * 3,
                  [["2026-01-08"], ["2026-01-15"], ["2026-01-22"]],
                  "2026-01", 1.0),
    RetentionCase("half-churned",  ["2026-01-01"] * 4,
                  [["2026-01-08"], ["2026-01-08"], [], []],
                  "2026-01", 0.5),
    RetentionCase("zero-cohort",   [], [], "2026-01", 0.0),
]


@pytest.mark.parametrize("case", RETENTION_CASES, ids=lambda c: c.name)
def test_retention_cohort(spark_session, case: RetentionCase):
    from mypkg.retention import cohort_retention
    df = spark_session.createDataFrame(
        [(i, sd, ad) for i, (sd, ad) in enumerate(zip(case.signup_dates, case.activity_dates))],
        schema=["user_id", "signup_date", "activity_dates"],
    )
    result = cohort_retention(df, cohort=case.cohort)
    assert abs(result - case.expected_retention) < 1e-9

# 2. Layer B — fixture-level, params, backend switching
import pytest


@pytest.fixture(params=["postgres", "duckdb"], ids=["pg", "duck"])
def sql_backend(request, postgresql, tmp_path):
    """Yield a connection to either Postgres or DuckDB."""
    if request.param == "postgres":
        yield postgresql
    else:
        import duckdb
        conn = duckdb.connect(str(tmp_path / "test.duckdb"))
        # (schema-load elided)
        yield conn
        conn.close()


def test_mrr_query_matches_across_backends(sql_backend):
    """Same query, both backends, must agree on MRR."""
    from mypkg.mrr import compute_mrr
    result = compute_mrr(sql_backend, month="2026-01")
    assert result >= 0     # basic sanity

# 3. Layer C — pytest_generate_tests, manifest-driven
def pytest_generate_tests(metafunc):
    """Discover every model in the dbt manifest at collection."""
    if "dbt_model" not in metafunc.fixturenames:
        return

    import json
    from pathlib import Path
    manifest = json.loads(Path("target/manifest.json").read_text())
    # Filter to models under models/marts/
    marts_models = [
        node["name"]
        for node in manifest["nodes"].values()
        if node["resource_type"] == "model"
        and node["path"].startswith("marts/")
    ]
    metafunc.parametrize("dbt_model", marts_models, ids=marts_models)


def test_mart_has_row_count_gt_zero(postgresql, dbt_model):
    """Every mart model must materialise to >= 1 row."""
    with postgresql.cursor() as cur:
        cur.execute(f"SELECT COUNT(*) FROM analytics.{dbt_model}")
        row_count = cur.fetchone()[0]
    assert row_count > 0, f"mart {dbt_model} materialised zero rows"

Step-by-step trace.

Layer	Style	Discovery	ID example	Test count
A (decorator)	static list	at import	test_retention_cohort[all-active]	3
A (decorator)	static list	at import	test_retention_cohort[half-churned]	1
A (decorator)	static list	at import	test_retention_cohort[zero-cohort]	1
B (fixture params)	fixture-level	at collection	test_mrr_query_matches_across_backends[pg]	1
B (fixture params)	fixture-level	at collection	test_mrr_query_matches_across_backends[duck]	1
C (generate_tests)	dynamic hook	at collection	test_mart_has_row_count_gt_zero[fct_orders]	N (manifest-driven)

After deployment, the same parametrized suite covers three orthogonal axes with three distinct techniques: (A) static known cases for retention logic, (B) fixture-level cases for cross-backend agreement, (C) manifest-driven cases for every mart model. Adding a new mart is zero test-file changes; adding a new retention edge case is one dataclass row.

Output:

Case	Style	Effort to add new case	Runs
new retention axis	Layer A	1 dataclass row	+1 test
add MySQL backend	Layer B	1 fixture param + branch	+N tests (one per test)
add new mart model	Layer C	zero (manifest picks it up)	+1 test automatically

Why this works — concept by concept:

• Decorator-level parametrize — the classic; best for known, static axes; readable IDs via lambda over dataclass fields.
• Fixture-level parametrize — routes the same params through a fixture; ideal for backend/config axes that need per-param setup; every test that requests the fixture inherits the param multiplication.
• pytest_generate_tests — the escape hatch; discovers axes at collection time from runtime sources (dbt manifest, information_schema, YAML config); the amplifier that lets one test cover every model.
• ids= discipline — the difference between test_foo[0] (hostile) and test_foo[with-return] (grep-friendly). Use lambda c: c.name when working with dataclasses.
• Cost — parametrization overhead is negligible (~µs per case at collection); the dominant cost is the test body itself. The eliminated cost is copy-pasted test functions (typically 10-20 lines each) that drift out of sync. Net: three techniques cover 90% of DE parametrization needs. O(1) code per axis added.

Data validation
Topic — data-validation
Data validation and contract-testing problems

Practice →

Python Topic — defensive-coding Defensive coding and edge-case coverage

Practice →

---

4. Docker-Compose + Testcontainers integration tests

testcontainers-python + a session-scoped DockerCompose fixture is the pattern every senior DE has converged on — hermetic, fast, and CI-safe

The mental model in one line: pytest docker-compose integration testing means declaring a docker-compose.test.yml next to your source, wrapping it in a session-scoped pytest fixture that spins the whole networked stack (Postgres + Kafka + MinIO + whatever) up once per pytest run and tears it down at exit — and if you use testcontainers-python instead of shelling out to docker-compose, you get first-class wait_for health checks, per-container get_service_host accessors, and clean handling of xdist-worker port allocation for free. Every senior data engineer has debugged the "tests pass on my laptop but fail on CI because Postgres wasn't ready yet" flake and every senior data engineer has fixed it with a wait_for on SELECT 1 inside a retry loop.

Two container primitives — pick one and commit.

• testcontainers-python per-service. Instantiate PostgresContainer("postgres:15"), KafkaContainer("confluentinc/cp-kafka:7.6.0"), MinioContainer("minio/minio:latest") in Python. Best for tests that need one or two containers. Handles wait-for-ready, port publishing, and network isolation. Idiomatic in 2026.
• testcontainers.compose.DockerCompose. Wraps a docker-compose.yml file. Best for tests that need the full stack (Airflow scheduler + webserver + Postgres + Redis + Celery worker + Kafka). Handles docker-compose up -d, health-check waits, and docker-compose down -v at teardown.
• pytest-docker-compose plugin. Older wrapper; still maintained. Slightly less flexible than testcontainers-python; retained in legacy suites.
• Raw subprocess.run(['docker-compose', 'up']). The anti-pattern. Loses wait_for, forces you to hand-roll health checks, and breaks under xdist. Only appropriate for one-off manual scripts, never for a test suite.

The three lifecycle patterns for container fixtures.

• Session scope + shared state. Spin the stack up once; every test hits the same DB. Fast (one startup ~10 s amortised over 100 tests) but requires tests to clean up after themselves or use per-test schemas / topics. This is the pattern for most DE integration suites.
• Session scope + per-test isolation. Spin the stack up once; each test creates its own schema (SET search_path) or Kafka topic (test_<uuid>). Fast and hermetic. This is the pattern for Layer-4 tests that must be independent.
• Function scope + full teardown. Spin the stack up per test; tear down completely at teardown. Slow (~10 s per test); use only for tests that need pristine state and can't share containers safely.

The wait-for-ready idiom — non-optional for every container.

• docker-compose up -d returns as soon as the containers are started, not when they are ready. Postgres takes ~2-5 s after start to accept connections; Kafka ~5-10 s; MinIO ~2 s. Racing this window is the #1 source of container-test flakes.
• HTTP health-check pattern. testcontainers.compose.DockerCompose.wait_for("http://localhost:9200/_cluster/health") polls until the endpoint returns 200. Set a 30-second timeout.
• TCP-connect pattern. For services without HTTP (raw Postgres, Kafka broker), poll a psycopg2.connect(...) or KafkaConsumer(bootstrap_servers=...) in a retry loop. Third-party libs like tenacity (@retry(stop_after_delay=30, wait_fixed=0.5)) make this concise.
• Docker-compose healthchecks. Define healthcheck in the yml (test: pg_isready -U test) and use depends_on: postgres: condition: service_healthy — compose itself waits, and the fixture just waits on the top-level service.

Common interview probes on pytest docker-compose.

• "How do you avoid port collisions under xdist?" — either OS-allocated ports (published on the container side) or per-worker compose projects (COMPOSE_PROJECT_NAME=test_${WORKER_ID}).
• "How do you clean up volumes?" — docker-compose down -v at teardown; DockerCompose context manager does this automatically.
• "What's the wait-for-ready story?" — HTTP endpoint poll, TCP connect retry, or compose-native healthcheck + depends_on.condition.
• "When would you pick testcontainers-python over compose?" — small stack (1-2 containers) and per-test isolation is required.

Worked example — pytest-postgresql fixture with a schema template

Detailed explanation. The Layer-3 equivalent of a docker-compose stack: a real Postgres process, but no container. pytest-postgresql bundles a Postgres binary invocation directly — no Docker required — and provides postgresql_proc (session, one binary) plus postgresql (function, one DB clone). This is faster than dockerised Postgres for local dev and works identically on GitHub Actions Ubuntu runners. Walk through the wiring.

• What it starts. A postgres binary running against a temp datadir; no image pull, no Docker daemon.
• How it isolates tests. Function-scoped DB via template clone.
• Where the schema lives. load=[Path('schema.sql')] applied to the template at session start.
• What breaks. Nothing if the runner has postgresql-15 installed; a clear error message if not.

Question. Wire the pytest-postgresql template + a single component test that inserts a row.

Input.

Component	Value
Postgres binary	postgresql-15 on the runner
Data dir	ephemeral temp dir
Schema	tests/schema.sql
Test DB	fresh template clone per test

Code.

# tests/conftest.py — pytest-postgresql fixtures
from pathlib import Path
from pytest_postgresql import factories

TEST_DIR = Path(__file__).parent
SCHEMA_SQL = TEST_DIR / "schema.sql"

postgresql_proc = factories.postgresql_proc(
    port=None,                          # OS-allocated
    unixsocketdir="/tmp",
)

postgresql = factories.postgresql(
    "postgresql_proc",
    dbname="test_db",
    load=[SCHEMA_SQL],
)

# tests/component/test_pg_template.py
def test_schema_loaded(postgresql):
    """Template clone must include the schema.sql tables."""
    with postgresql.cursor() as cur:
        cur.execute("""
            SELECT tablename FROM pg_tables
            WHERE  schemaname = 'public'
            ORDER  BY tablename
        """)
        tables = [r[0] for r in cur.fetchall()]
    assert "orders"    in tables
    assert "customers" in tables


def test_insert_and_select(postgresql):
    """Round-trip an insert; assert the row is visible."""
    with postgresql.cursor() as cur:
        cur.execute("""
            INSERT INTO public.orders(customer_id, total_cents, status)
            VALUES (1, 500, 'pending') RETURNING id
        """)
        oid = cur.fetchone()[0]
    postgresql.commit()

    with postgresql.cursor() as cur:
        cur.execute("SELECT total_cents FROM public.orders WHERE id = %s", (oid,))
        row = cur.fetchone()
    assert row == (500,)

Step-by-step explanation.

1. postgresql_proc at session scope starts one Postgres binary. The port=None argument tells pytest-postgresql to bind to an OS-allocated port; the caller can read the actual port from postgresql_proc.port. This is what makes xdist safe.
2. postgresql at function scope creates a fresh database per test by (a) waiting for postgresql_proc to be up, (b) issuing CREATE DATABASE test_db_<random>, (c) applying schema.sql, (d) opening a psycopg2 connection, (e) at teardown dropping the DB.
3. SCHEMA_SQL is loaded on every clone, so the schema is guaranteed to be present. For faster startup, some teams pre-load the schema onto a template_db and clone from that — pytest-postgresql supports both patterns via template_dbname.
4. test_schema_loaded proves the schema arrived; test_insert_and_select proves the DB is writable and reads back. Both are isolated — the second test starts with an empty orders table even though the first test would have inserted something.
5. Compared to a dockerised Postgres, this pattern skips the Docker daemon entirely — ~500 ms faster per pytest session and works on runners where Docker-in-Docker isn't available (some CI providers).

Output.

Test	Startup	Body	Teardown	Total
test_schema_loaded	~1 s (session, first only) + ~100 ms (clone)	~10 ms	~30 ms	~1.15 s
test_insert_and_select	~100 ms (clone)	~15 ms	~30 ms	~145 ms
(subsequent tests)	~100 ms (clone)	~10-50 ms	~30 ms	~140-180 ms

Rule of thumb. For Layer-3 SQL-emitting tests, prefer pytest-postgresql over a dockerised Postgres — it's faster to start, easier to reason about, and works on runners without Docker access. Save containers for Layer 4.

Worked example — Testcontainers Kafka + a smoke consumer test

Detailed explanation. A test that produces a message to Kafka and consumes it back, using testcontainers-python to spin up a single Kafka container. This is the Layer-4 pattern for CDC / event-driven code where the interaction with Kafka is the thing under test. Walk through the fixture + test.

• Container. KafkaContainer("confluentinc/cp-kafka:7.6.0").
• Fixture scope. Session — one Kafka broker for all Kafka tests in the run.
• Per-test isolation. Each test creates its own topic named test_<uuid>.
• Assertion. Produce a message; consume with a fresh consumer; assert value matches.

Question. Write the Kafka fixture + a smoke test that round-trips a message through a per-test topic.

Input.

Component	Value
Kafka image	confluentinc/cp-kafka:7.6.0
Fixture scope	session
Isolation	per-test topic
Producer / consumer lib	kafka-python

Code.

# tests/conftest.py — Kafka session fixture
import pytest
import uuid
from testcontainers.kafka import KafkaContainer


@pytest.fixture(scope="session")
def kafka_container():
    """Spin up one Kafka broker per pytest run."""
    with KafkaContainer(image="confluentinc/cp-kafka:7.6.0") as k:
        # KafkaContainer.get_bootstrap_server() blocks until ready
        yield k


@pytest.fixture
def kafka_topic(kafka_container):
    """Per-test unique topic; consumers use fresh group-id."""
    topic = f"test_{uuid.uuid4().hex[:8]}"
    yield topic
    # (topic auto-cleaned when broker exits; explicit delete optional)

# tests/integration/test_kafka_roundtrip.py — smoke test
import json
from kafka import KafkaProducer, KafkaConsumer


def test_kafka_produce_and_consume(kafka_container, kafka_topic):
    """Produce one JSON message; consume it back; assert equality."""
    bootstrap = kafka_container.get_bootstrap_server()

    # 1. Produce
    producer = KafkaProducer(
        bootstrap_servers=[bootstrap],
        value_serializer=lambda v: json.dumps(v).encode("utf-8"),
    )
    payload = {"order_id": 42, "status": "placed"}
    producer.send(kafka_topic, payload)
    producer.flush()
    producer.close()

    # 2. Consume from earliest
    consumer = KafkaConsumer(
        kafka_topic,
        bootstrap_servers=[bootstrap],
        auto_offset_reset="earliest",
        group_id=f"grp_{kafka_topic}",
        consumer_timeout_ms=5_000,          # stop after 5 s of no messages
        value_deserializer=lambda v: json.loads(v.decode("utf-8")),
    )
    messages = list(consumer)
    consumer.close()

    # 3. Assert
    assert len(messages) == 1
    assert messages[0].value == payload

Step-by-step explanation.

1. KafkaContainer at session scope pulls the Kafka image (once per CI run — cached by Docker), starts the broker, and blocks on get_bootstrap_server() until the broker accepts client connections. This is the wait-for-ready idiom baked into testcontainers.
2. kafka_topic at function scope generates a per-test topic name using uuid.uuid4().hex[:8] — the 8-char hex is short enough to be readable and long enough to collision-resist across 1000+ concurrent test runs.
3. The producer serialises Python dicts as JSON bytes and sends one message. producer.flush() blocks until the broker acks; producer.close() shuts down cleanly. Both are necessary to guarantee the message is durable before the consumer starts.
4. The consumer uses a per-topic group-id (grp_<topic>) and auto_offset_reset="earliest" — this guarantees a fresh consumer group with no committed offset reads from the start of the topic. consumer_timeout_ms=5_000 ensures the list(consumer) returns even if no more messages arrive.
5. The assertion is pure equality on the deserialised value. Because the topic is per-test, there's no way for a previous test's messages to leak in — the count is deterministic.

Output.

Step	Duration	Detail
Kafka startup (session, first test)	~8 s	image pull cached, broker init
kafka_topic creation	<1 ms	uuid gen only
Producer send + flush	~50 ms	one message
Consumer poll (5s timeout)	~5 s	waits for stall
Test total (first)	~13 s	mostly Kafka startup
Test total (subsequent)	~5.1 s	just consumer stall

Rule of thumb. For Kafka tests, use testcontainers-python's KafkaContainer at session scope; give every test its own uuid-suffixed topic; give every consumer a matching group-id. This is the cheapest correct pattern.

Worked example — Airflow DAG integration test spinning up scheduler + Postgres

Detailed explanation. An Airflow DAG that reads from a source table and writes to a target table needs to be tested end-to-end: metadata DB running, scheduler running, DAG file discovered, task instances actually executing. DockerCompose fixture with an Airflow compose file is the standard pattern. Walk through the yml + fixture + test.

• Compose services. postgres (metadata) + airflow-init + airflow-scheduler + airflow-webserver.
• Fixture scope. Session — one Airflow up per test session; every DAG test triggers via CLI.
• Test isolation. Each test triggers a fresh DAG run with a unique dag_run_id.
• Assertion. After airflow dags trigger, poll airflow tasks state until success/failure.

Question. Write the compose fixture + a DAG test that triggers ingest_daily_orders and asserts success.

Input.

Component	Value
Compose file	docker-compose.test.yml
Services	postgres, airflow-scheduler, airflow-webserver
Trigger	airflow dags trigger ingest_daily_orders --run-id test_
Poll	airflow tasks state ... every 2 s, up to 60 s

Code.

# docker-compose.test.yml — minimal Airflow stack for tests
version: "3.9"

services:
  postgres:
    image: postgres:15
    environment:
      POSTGRES_USER: airflow
      POSTGRES_PASSWORD: airflow
      POSTGRES_DB: airflow
    healthcheck:
      test: ["CMD", "pg_isready", "-U", "airflow"]
      interval: 5s
      timeout: 5s
      retries: 5

  airflow-init:
    image: apache/airflow:2.9.2
    depends_on:
      postgres:
        condition: service_healthy
    environment:
      AIRFLOW__CORE__EXECUTOR: LocalExecutor
      AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow
    entrypoint: /bin/bash
    command: -c "airflow db init && airflow users create --role Admin --username admin --password admin --email a@b.c --firstname a --lastname b"

  airflow-scheduler:
    image: apache/airflow:2.9.2
    depends_on:
      airflow-init:
        condition: service_completed_successfully
    environment:
      AIRFLOW__CORE__EXECUTOR: LocalExecutor
      AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres/airflow
    volumes:
      - ./dags:/opt/airflow/dags
    command: scheduler

# tests/conftest.py — Airflow compose fixture
import pytest
from testcontainers.compose import DockerCompose


@pytest.fixture(scope="session")
def airflow_stack():
    """Spin up Airflow + Postgres via docker-compose."""
    with DockerCompose(
        ".",
        compose_file_name="docker-compose.test.yml",
        pull=True,
    ) as compose:
        # Wait for the scheduler to actually parse the DAGs
        compose.wait_for("http://localhost:8080/health")  # webserver
        yield compose

# tests/integration/test_ingest_dag.py — end-to-end DAG test
import subprocess
import time
import uuid


def _run(cmd: list[str]) -> str:
    """Shell out to airflow via docker exec on the scheduler container."""
    full = ["docker", "exec", "airflow-scheduler"] + cmd
    return subprocess.check_output(full).decode()


def test_ingest_daily_orders_dag_succeeds(airflow_stack):
    """Trigger ingest_daily_orders; poll for success; assert green."""
    run_id = f"test_{uuid.uuid4().hex[:8]}"

    # 1. Trigger
    _run(["airflow", "dags", "trigger", "ingest_daily_orders",
          "--run-id", run_id])

    # 2. Poll state up to 60 s
    deadline = time.time() + 60
    state = "queued"
    while time.time() < deadline and state not in ("success", "failed"):
        time.sleep(2)
        out = _run(["airflow", "dags", "state", "ingest_daily_orders", run_id])
        state = out.strip()

    # 3. Assert
    assert state == "success", f"DAG {run_id} ended in {state}"

Step-by-step explanation.

1. The compose file uses Airflow's official image with LocalExecutor — no Celery, no Redis, minimum surface for a test stack. The metadata DB is a per-test-run Postgres; the scheduler picks up DAGs from a mounted ./dags directory.
2. airflow-init runs airflow db init + creates an admin user, then exits. depends_on: condition: service_completed_successfully on the scheduler ensures the init has finished before the scheduler starts.
3. The airflow_stack fixture wraps DockerCompose — the with block handles docker-compose up -d on entry and docker-compose down -v on exit. pull=True ensures images are pulled once per session.
4. The test triggers a DAG run via airflow dags trigger executed inside the scheduler container (via docker exec). The run-id is uuid-suffixed for per-test isolation.
5. The polling loop calls airflow dags state every 2 seconds up to a 60-second deadline. success or failed breaks the loop; the test assertion pins the required outcome.

Output.

Step	Duration	State
Compose startup (session first)	~30 s	postgres, init, scheduler, webserver
DAG trigger	~200 ms	subprocess exec
Task execution	~5-20 s	depends on DAG
Poll loop	~5-20 s	2 s poll interval
Test total (first)	~55 s	mostly compose startup
Test total (subsequent)	~25 s	just trigger + poll

Rule of thumb. For Airflow DAG integration tests, use DockerCompose at session scope with a compose file that pins Airflow 2.9+ and Postgres 15+. Every test gets a fresh run_id; the scheduler+worker are shared. Session-scope amortises the ~30 s compose startup across all DAG tests.

Senior interview question on pytest docker-compose integration testing

A senior interviewer might ask: "You have a Layer-4 integration suite with 40 tests, each currently spinning up its own docker-compose stack (Postgres + Kafka + MinIO) at ~15 s startup, so the suite takes 10 minutes. You need to bring it under 3 minutes. Walk me through the session-scoped compose fixture, the per-test schema/topic isolation pattern, the wait-for-ready idiom, and how you keep the pattern xdist-safe."

Solution Using session-scoped DockerCompose + per-test schema + wait-for-ready + xdist-safe port allocation

# 1. docker-compose.test.yml — the full stack, one project per xdist worker
version: "3.9"

services:
  postgres:
    image: postgres:15
    environment:
      POSTGRES_USER: test
      POSTGRES_PASSWORD: test
      POSTGRES_DB: test
    ports:
      - "5432"                          # published, port assigned by Docker
    healthcheck:
      test: ["CMD", "pg_isready", "-U", "test"]
      interval: 3s
      retries: 10

  kafka:
    image: confluentinc/cp-kafka:7.6.0
    environment:
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
      KAFKA_LISTENERS:            PLAINTEXT://0.0.0.0:9092
      KAFKA_ZOOKEEPER_CONNECT:    zookeeper:2181
    depends_on: [zookeeper]
    ports:
      - "9092"

  zookeeper:
    image: confluentinc/cp-zookeeper:7.6.0
    environment: { ZOOKEEPER_CLIENT_PORT: 2181 }

  minio:
    image: minio/minio:latest
    command: server /data --console-address ":9001"
    ports:
      - "9000"
      - "9001"
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"]
      interval: 3s
      retries: 10

# 2. tests/conftest.py — session compose fixture with xdist safety
import os
import pytest
import uuid
from testcontainers.compose import DockerCompose


@pytest.fixture(scope="session")
def compose_stack(worker_id):
    """One compose project per xdist worker; session-scoped."""
    project = f"test_{worker_id}"       # 'master' when serial; 'gw0', 'gw1', ...
    os.environ["COMPOSE_PROJECT_NAME"] = project

    with DockerCompose(
        ".",
        compose_file_name="docker-compose.test.yml",
        pull=True,
        wait=True,                      # wait for healthchecks
    ) as compose:
        # Belt-and-braces: also block on our own SELECT 1
        pg_port = int(compose.get_service_port("postgres", 5432))
        _wait_postgres_ready("localhost", pg_port)

        yield compose


def _wait_postgres_ready(host, port, timeout=30):
    """Retry SELECT 1 until success or timeout."""
    import time, psycopg2
    deadline = time.time() + timeout
    while time.time() < deadline:
        try:
            with psycopg2.connect(host=host, port=port, user="test",
                                  password="test", dbname="test") as c:
                with c.cursor() as cur:
                    cur.execute("SELECT 1")
                    if cur.fetchone() == (1,):
                        return
        except Exception:
            time.sleep(0.5)
    raise RuntimeError("postgres never became ready")

# 3. Per-test isolation — new schema per test
@pytest.fixture
def isolated_pg(compose_stack):
    """Yield a connection to a per-test schema in the shared Postgres."""
    import psycopg2
    port = int(compose_stack.get_service_port("postgres", 5432))
    schema = f"t_{uuid.uuid4().hex[:8]}"

    conn = psycopg2.connect(host="localhost", port=port,
                            user="test", password="test", dbname="test")
    with conn.cursor() as cur:
        cur.execute(f'CREATE SCHEMA "{schema}"')
        cur.execute(f'SET search_path TO "{schema}"')
        # (schema DDL for this test)
        cur.execute("""
            CREATE TABLE orders(id BIGSERIAL PRIMARY KEY, cents BIGINT)
        """)
    conn.commit()

    yield conn

    # Teardown — drop the whole schema
    with conn.cursor() as cur:
        cur.execute(f'DROP SCHEMA "{schema}" CASCADE')
    conn.commit()
    conn.close()


@pytest.mark.integration
def test_ingest_writes_orders(isolated_pg):
    """Real Postgres, per-test schema, no leakage."""
    from mypkg.ingest import ingest_batch
    ingest_batch([("A", 100), ("B", 200)], conn=isolated_pg)
    with isolated_pg.cursor() as cur:
        cur.execute("SELECT COUNT(*) FROM orders")
        assert cur.fetchone()[0] == 2

Step-by-step trace.

Step	Before (per-test compose)	After (session + per-schema)
Compose startups	40 × ~15 s = 10 min	1 × ~15 s per worker
Test body	40 × ~1 s = 40 s	40 × ~1 s = 40 s
Schema create/drop	none	40 × ~30 ms = ~1.2 s
xdist worker safety	port-collision risk	COMPOSE_PROJECT_NAME per worker
Total (4-worker xdist)	~4 min	~1 min
Wait-for-ready	ad-hoc sleeps	health checks + SELECT 1 retry

After deployment, the same 40 tests run in ~1 minute on 4-worker xdist, share one compose stack per worker, and every test isolates via a per-test schema. COMPOSE_PROJECT_NAME=test_<worker_id> ensures each xdist worker gets its own container network with distinct published ports — no collision.

Output:

Metric	Before	After
Runtime (serial)	10 min	~2 min
Runtime (4-worker xdist)	~4 min (port collisions)	~1 min (isolated)
Compose startups	40	1 per worker (4 total)
Test isolation	full (own stack)	full (own schema)
Wait-for-ready	absent (flaky)	health check + SELECT 1
Memory pressure	40 × 2 GB stacks	1 × 2 GB per worker (~8 GB)

Why this works — concept by concept:

• Session-scoped DockerCompose fixture — one compose stack per pytest session (per xdist worker). Amortises ~15 s of startup across all Layer-4 tests. The wait=True argument makes testcontainers-python block on the healthcheck-defined ready conditions.
• Per-test schema isolation — each test creates its own Postgres schema (CREATE SCHEMA t_<uuid>) and sets search_path; all its DDL and DML live under that schema. Cleanup is one DROP SCHEMA CASCADE. This gives per-test isolation on top of a shared DB.
• COMPOSE_PROJECT_NAME per worker — worker_id is pytest-xdist's per-worker identifier (gw0, gw1, ...). Setting the project name isolates each worker's compose stack — separate container names, separate networks, no port collision even on the same host.
• Wait-for-ready idiom — compose healthchecks + a belt-and-braces SELECT 1 retry loop. The healthcheck handles ~95% of races; the SELECT 1 retry catches the last 5% (schema not yet migrated, etc).
• Cost — one compose stack per xdist worker (~2 GB RAM), ~15 s startup amortised, ~30 ms per test for schema create/drop. The eliminated cost is 40 × ~15 s = 10 minutes of compose startups. Net O(1) startup per worker versus O(N) startup per test. Volume cleanup is automatic via docker-compose down -v inside the context manager's exit.

ETL
Topic — etl
ETL integration and end-to-end pipeline tests

Practice →

API Topic — api-integration API integration testing patterns

Practice →

---

5. CI wiring — coverage, xdist parallelism, flaky-test discipline

pytest -n auto --cov --cov-fail-under=85 is the two-line CI recipe every DE team eventually converges on

The mental model in one line: pytest for data engineering in CI is the discipline of running the four-layer suite in parallel via pytest-xdist, gating merges on branch-coverage via pytest-cov with a --cov-fail-under threshold, uploading --junit-xml to a flaky-test dashboard, and quarantining known-flaky tests with pytest-rerunfailures plus a nightly retriage job — the combination that turns a 25-minute serial suite into a 7-minute PR gate that catches regressions the day the code is written. Every senior data engineer has inherited a CI that runs pytest serially, has no coverage gate, and ignores flakes; and every senior data engineer has converted that same CI into a fast, gated, parallel suite over the course of a two-week focus block that pays for itself in developer time within the first month.

The three-pillar CI recipe for DE pytest suites.

• Pillar 1 — xdist parallelism. pytest -n auto --dist=loadgroup spawns one worker per CPU; distributes tests to workers by --xdist-group marker for fixture affinity. This is the raw-speed win: an N-core runner runs the suite ~N× faster (minus startup + coordination overhead).
• Pillar 2 — coverage gate. pytest --cov=mypkg --cov-branch --cov-fail-under=85 measures line-and-branch coverage, fails the run if coverage drops below 85%. This is the anti-regression gate — someone adding untested code breaks the build.
• Pillar 3 — flaky discipline. @pytest.mark.flaky(reruns=2) (from pytest-rerunfailures) auto-reruns failing tests up to twice; combined with a quarantine marker (@pytest.mark.flaky_quarantined) and a nightly retriage job, this keeps CI green without hiding real failures.

pytest-cov config — the four settings that matter.

• --cov=mypkg. Measure coverage on the mypkg package. Use one flag per package; don't measure tests/ itself.
• --cov-branch. Include branch coverage — captures whether both sides of every if have been executed. Line coverage alone hides "we tested the happy path but not the exception branch".
• --cov-report=term-missing:skip-covered. Prints uncovered line numbers in the terminal but skips fully-covered files — the report you actually read.
• --cov-fail-under=85. The gate. Below 85% (or whatever threshold you commit to) the run fails. Ratchet upward as the suite matures — start at 70%, aim for 90%.

pytest-xdist distribution modes — pick the one that matches your fixture graph.

• --dist=load (default). Round-robin tests to workers as they become idle. Best for tests with no shared fixtures.
• --dist=loadscope. Group tests by file / class / module. Ensures all tests in a file land on the same worker — useful when a module fixture is expensive.
• --dist=loadgroup. Group tests by the @pytest.mark.xdist_group("name") marker. The most flexible — pin all tests that share a Kafka topic to one worker, for example.
• --dist=each. Run every test on every worker. Only for very specific benchmarking / matrix cases; almost never right.

Flaky-test discipline — the four-step protocol.

• Step 1 — instrument. Add --junit-xml=junit.xml to every CI run; upload the artifact.
• Step 2 — surface. A nightly job parses the junit XMLs from the last N runs and flags tests with pass rate <99%.
• Step 3 — quarantine. Flagged tests get @pytest.mark.flaky(reruns=2) immediately (fixes the visible CI break) and a @pytest.mark.flaky_quarantined marker for tracking.
• Step 4 — retriage. A weekly ticket assigns each quarantined test to an owner; owner has 2 weeks to fix-or-delete. No test stays quarantined forever.

Common interview probes on pytest CI.

• "How do you measure coverage?" — pytest-cov --cov=pkg --cov-branch --cov-fail-under=N.
• "How do you parallelise safely?" — pytest-xdist -n auto --dist=loadgroup with per-worker fixtures.
• "What's your flaky-test policy?" — rerun with pytest-rerunfailures, quarantine, weekly retriage.
• "How do you keep DB fixtures xdist-safe?" — per-worker Postgres process (session-scoped in each worker) or per-worker COMPOSE_PROJECT_NAME.

Worked example — GitHub Actions matrix with coverage upload

Detailed explanation. A canonical GitHub Actions workflow: matrix over Python 3.11 and 3.12, install deps, run pytest with xdist + coverage, upload junit XML + coverage HTML as artifacts, gate on --cov-fail-under=85. Walk through the yaml.

• Matrix. Python 3.11 and 3.12 in parallel.
• Runner. ubuntu-24.04 (has Docker installed).
• Cache. pip cache keyed on requirements hash.
• Coverage upload. GitHub artifact + optional Codecov push.

Question. Write the .github/workflows/tests.yml that runs the four-layer pytest suite with xdist and coverage.

Input.

Parameter	Value
Matrix	Python 3.11, 3.12
xdist workers	auto (CPU count)
Coverage threshold	85%
Junit output	junit.xml per matrix cell
Timeout	15 min

Code.

# .github/workflows/tests.yml
name: tests

on:
  push:
    branches: [main]
  pull_request:

jobs:
  pytest:
    runs-on: ubuntu-24.04
    timeout-minutes: 15
    strategy:
      fail-fast: false
      matrix:
        python-version: ["3.11", "3.12"]

    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: ${{ matrix.python-version }}
          cache: pip

      - name: Install
        run: |
          pip install -e '.[test]'
          pip install pytest-cov pytest-xdist pytest-rerunfailures \
                      pytest-postgresql testcontainers

      - name: Install Postgres
        run: sudo apt-get install -y postgresql-15

      - name: Run pytest
        run: |
          pytest \
            -n auto \
            --dist=loadgroup \
            --cov=mypkg \
            --cov-branch \
            --cov-report=xml \
            --cov-report=term-missing:skip-covered \
            --cov-fail-under=85 \
            --junit-xml=junit-${{ matrix.python-version }}.xml \
            -v \
            tests/

      - name: Upload junit
        if: always()
        uses: actions/upload-artifact@v4
        with:
          name: junit-${{ matrix.python-version }}
          path: junit-${{ matrix.python-version }}.xml

      - name: Upload coverage HTML
        if: always()
        uses: actions/upload-artifact@v4
        with:
          name: coverage-${{ matrix.python-version }}
          path: coverage.xml

Step-by-step explanation.

1. The strategy.matrix runs the workflow twice — once per Python version — in parallel. fail-fast: false lets both cells complete even if one fails; you get coverage data from both.
2. actions/setup-python@v5 with cache: pip uses GitHub's built-in pip cache, shaving ~30 s off cold installs.
3. pytest -n auto uses pytest-xdist to fan out to CPU count workers (ubuntu-24.04 default = 4 vCPUs). --dist=loadgroup respects @pytest.mark.xdist_group markers for fixture affinity.
4. --cov-fail-under=85 fails the step if branch coverage drops below 85%. The step exits non-zero; GitHub marks the check as failed; the PR is blocked from merge.
5. if: always() on the artifact-upload steps ensures junit + coverage are uploaded even when the pytest step failed — you get the data to diagnose the failure. This is a small detail that saves hours of debugging.

Output.

Cell	Duration	Tests run	Coverage	Result
Python 3.11	~5 min	1,290	87%	pass
Python 3.12	~5 min	1,290	87%	pass
Total wall clock	~5 min (parallel cells)	—	—	pass
Artifacts	junit-3.11, junit-3.12, coverage-3.11, coverage-3.12	—	—	—

Rule of thumb. For DE pytest CI, use a Python-version matrix with cache: pip, pytest -n auto, --cov-fail-under=<N>, and if: always() on artifact uploads. This is the minimum viable modern setup.

Worked example — xdist + docker-compose shared across workers

Detailed explanation. The subtle problem: session-scoped fixtures are per-worker in xdist, not per-session. If your compose stack is session-scoped, it starts N times (once per worker). Sometimes that's what you want (isolation); sometimes you want the compose stack shared across all workers (for cost reasons). The pytest-xdist idiom tmp_path_factory + a lockfile solves the "shared across workers" case. Walk through the pattern.

• Problem. @pytest.fixture(scope="session") on compose_stack starts one compose stack per worker.
• Solution. Use a filesystem lock on a shared tmp_path_factory directory to elect one worker to start compose; other workers block until ready.
• When. When compose startup is >30 s and you're xdist-parallelising past 4 workers.

Question. Write the shared-compose fixture using a filesystem lock so N workers use one compose stack.

Input.

Component	Value
Lock file	tmp_path_factory / "compose.lock"
Ready file	tmp_path_factory / "compose.ready"
Compose stack	docker-compose.test.yml
Election	one worker starts; others poll ready file

Code.

# tests/conftest.py — cross-worker shared compose fixture
import os
import time
import pytest
import fcntl
from pathlib import Path
from testcontainers.compose import DockerCompose


@pytest.fixture(scope="session")
def compose_stack(tmp_path_factory, worker_id):
    """One compose stack shared across all xdist workers."""
    # tmp_path_factory.getbasetemp() is shared across workers
    root = tmp_path_factory.getbasetemp().parent
    root.mkdir(parents=True, exist_ok=True)
    lock_path  = root / "compose.lock"
    ready_path = root / "compose.ready"

    if worker_id == "master":
        # Serial pytest — no election needed
        with DockerCompose(".", compose_file_name="docker-compose.test.yml") as c:
            c.wait_for("http://localhost:9001/minio/health/ready")
            yield c
        return

    # Parallel — acquire lock
    with open(lock_path, "w") as lock_fp:
        try:
            fcntl.flock(lock_fp, fcntl.LOCK_EX | fcntl.LOCK_NB)
            # We are the elected worker — start compose
            os.environ["COMPOSE_PROJECT_NAME"] = "test_shared"
            compose = DockerCompose(
                ".", compose_file_name="docker-compose.test.yml"
            ).start()
            _wait_ready(compose)
            ready_path.write_text("ok")
            yield compose
            compose.stop()
            ready_path.unlink(missing_ok=True)

        except BlockingIOError:
            # Some other worker started it; wait for ready file
            deadline = time.time() + 120
            while not ready_path.exists() and time.time() < deadline:
                time.sleep(0.5)
            if not ready_path.exists():
                raise RuntimeError("compose never came ready")

            # Yield a lightweight reference (compose is up)
            yield _AttachedCompose(project_name="test_shared")


def _wait_ready(compose):
    """Poll all healthchecks; block until green."""
    # (implementation elided; use compose.wait_for or per-service polls)
    pass


class _AttachedCompose:
    """Read-only view of a compose stack another worker owns."""
    def __init__(self, project_name):
        self.project_name = project_name
    def get_service_port(self, svc, port):
        import subprocess
        out = subprocess.check_output(
            ["docker", "compose", "-p", self.project_name, "port", svc, str(port)]
        )
        return out.decode().strip().split(":")[-1]

Step-by-step explanation.

1. When pytest runs under xdist, worker_id is gw0, gw1, ... — one per worker. When pytest runs serial, worker_id is master. The fixture branches on this.
2. tmp_path_factory.getbasetemp() returns a per-worker temp dir, but its .parent is the shared base (/tmp/pytest-of-user). Using the shared parent as the lock location lets multiple workers coordinate.
3. fcntl.flock(..., LOCK_EX | LOCK_NB) acquires an exclusive lock non-blockingly. Only one worker succeeds; the others get BlockingIOError and fall into the "wait for ready" branch.
4. The elected worker starts compose, waits for readiness, writes a compose.ready sentinel file, and yields the live compose object. At teardown, it stops compose and cleans the sentinel.
5. Non-elected workers spin-wait on ready_path.exists() for up to 120 s. Once ready, they yield an _AttachedCompose proxy that lets tests query service ports via docker compose port — no actual container control.

Output.

Worker	Elected?	Compose action	Wait
gw0	yes	start compose	writes ready file
gw1	no	attaches	waits ~15 s for ready
gw2	no	attaches	waits ~15 s for ready
gw3	no	attaches	waits ~15 s for ready

Rule of thumb. For very-expensive compose stacks (>30 s startup) and 4+ xdist workers, elect one worker to start compose and have others attach. For cheaper stacks (<15 s) or ≤2 workers, per-worker isolation is simpler and safer.

Worked example — flaky-test dashboard from --junit-xml

Detailed explanation. A tiny Python script that ingests the last N junit XMLs and builds a per-test pass-rate report. Flag anything with pass rate < 99% over the last 20 runs. Walk through the parser + report.

• Input. junit-*.xml files from the last 20 CI runs.
• Parser. xml.etree.ElementTree — each <testcase> element is one test outcome.
• Aggregation. Group by test node id; compute pass rate = passed / total.
• Report. Sorted by pass rate ascending; flag < 99%.

Question. Write the parser + reporter for the flaky-test dashboard.

Input.

Component	Value
Input glob	reports/junit-*.xml
Test key	classname::name
Failure marker	or child
Threshold	99%

Code.

# tools/flaky_dashboard.py
"""Compute per-test pass rates from junit XMLs and flag flakes."""
from __future__ import annotations
import glob
import xml.etree.ElementTree as ET
from collections import defaultdict


def parse_junit(path: str) -> list[tuple[str, bool]]:
    """Return [(test_id, passed?), ...] from one junit XML."""
    root = ET.parse(path).getroot()
    out: list[tuple[str, bool]] = []
    for tc in root.iter("testcase"):
        tid = f"{tc.get('classname', '')}::{tc.get('name', '')}"
        failed = any(child.tag in ("failure", "error") for child in tc)
        out.append((tid, not failed))
    return out


def compute_pass_rates(paths: list[str]) -> dict[str, float]:
    """Aggregate pass rates over all runs."""
    tallies: dict[str, list[bool]] = defaultdict(list)
    for p in paths:
        for tid, passed in parse_junit(p):
            tallies[tid].append(passed)
    return {tid: sum(v) / len(v) for tid, v in tallies.items() if len(v) >= 5}


def flag_flakes(rates: dict[str, float], threshold: float = 0.99) -> list[tuple[str, float]]:
    """Return tests below the pass-rate threshold, worst first."""
    flakes = [(tid, r) for tid, r in rates.items() if r < threshold]
    flakes.sort(key=lambda x: x[1])
    return flakes


def main():
    paths = glob.glob("reports/junit-*.xml")
    rates = compute_pass_rates(paths)
    flakes = flag_flakes(rates)
    if not flakes:
        print(f"OK — all {len(rates)} tests >= 99% pass rate")
        return
    print(f"FLAKY — {len(flakes)} tests below 99% over {len(paths)} runs:")
    print(f"  {'pass_rate':>10}  test_id")
    for tid, rate in flakes[:20]:
        print(f"  {rate * 100:>9.1f}%  {tid}")


if __name__ == "__main__":
    main()

Step-by-step explanation.

1. parse_junit walks the XML and extracts every <testcase>. A test passed iff it has no <failure> or <error> child. Skipped tests (which have <skipped>) are counted as passed — they didn't fail.
2. compute_pass_rates aggregates across all junit files. Tests seen fewer than 5 times are filtered out (insufficient data to call flake vs new). The pass rate is sum(passed) / len(runs).
3. flag_flakes returns tests below the threshold, sorted worst first — the flakiest tests appear at the top of the report and get owner attention first.
4. The main function glob-collects reports/junit-*.xml (uploaded by CI) and prints the top 20 offenders. Wire this into a nightly cron job that pushes the report to Slack or a dashboard.
5. To integrate with pytest-rerunfailures, this report drives which tests get @pytest.mark.flaky(reruns=2) added — a manual review by a triage owner turns "reported flake" into "quarantined test with a tracking ticket."

Output.

FLAKY — 3 tests below 99% over 20 runs:
  pass_rate  test_id
      75.0%  tests.integration.test_kafka::test_kafka_produce_and_consume
      85.0%  tests.integration.test_dag::test_ingest_daily_orders_dag_succeeds
      95.0%  tests.component.test_pg::test_bulk_status_update

Rule of thumb. For any DE test suite past ~200 tests, ship a flaky-test dashboard from --junit-xml. Manual pass-rate tracking doesn't scale; automated flagging + a weekly retriage keeps the suite honest.

Senior interview question on pytest CI and flaky-test discipline

A senior interviewer might ask: "Your team's pytest CI runs serially, has no coverage gate, and has 15 tests that flake ~10% of the time — devs have started -p 'not integration'-skipping the suite on their branches. Walk me through the two-week focus block that gets you to pytest -n auto --cov-fail-under=85 with a quarantine + retriage process, and quantify the CI budget savings."

Solution Using xdist + coverage gate + rerunfailures + quarantine markers + nightly retriage

# 1. pytest.ini — strict-marker gate + coverage + rerunfailures defaults
[pytest]
minversion = 8.0
addopts =
    -ra
    -q
    --strict-markers
    --strict-config
    --cov=mypkg
    --cov-branch
    --cov-report=term-missing:skip-covered
    --cov-report=xml
    --cov-fail-under=85
    --junit-xml=junit.xml

markers =
    unit:              base-layer pure-function tests
    contract:          schema / pydantic contract tests
    component:         Layer-3 pytest-postgresql tests
    integration:       Layer-4 docker-compose tests
    flaky:             tests known to flake; get reruns=2
    flaky_quarantined: assigned to an owner for fix-or-delete

testpaths = tests

# 2. tests/conftest.py — auto-apply reruns for @pytest.mark.flaky
def pytest_collection_modifyitems(config, items):
    """Attach reruns=2 to every test carrying the flaky marker."""
    for item in items:
        if item.get_closest_marker("flaky"):
            item.add_marker("flaky_quarantined")  # for tracking
            # pytest-rerunfailures picks up the reruns param

# 3. .github/workflows/tests.yml — parallel matrix + artifact uploads
name: tests
on: [push, pull_request]

jobs:
  pytest:
    runs-on: ubuntu-24.04
    strategy:
      matrix:
        python: ["3.11", "3.12"]
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: ${{ matrix.python }}, cache: pip }
      - run: pip install -e '.[test]'
      - run: sudo apt-get install -y postgresql-15
      - run: |
          pytest -n auto --dist=loadgroup --reruns 2 tests/
      - if: always()
        uses: actions/upload-artifact@v4
        with:
          name: junit-${{ matrix.python }}
          path: junit.xml

# 4. .github/workflows/flaky-retriage.yml — nightly retriage
name: flaky-retriage
on:
  schedule:
    - cron: "0 6 * * *"      # 06:00 UTC daily

jobs:
  retriage:
    runs-on: ubuntu-24.04
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: "3.12" }
      - name: Download last 20 junit artifacts
        uses: actions/github-script@v7
        with:
          script: |
            const artifacts = await github.rest.actions.listArtifactsForRepo({
              owner: context.repo.owner, repo: context.repo.repo, per_page: 100,
            });
            // (download loop elided)
      - run: python tools/flaky_dashboard.py > flaky_report.txt
      - name: Post to Slack
        run: |
          curl -X POST $SLACK_WEBHOOK -d "text=$(cat flaky_report.txt)"
        env:
          SLACK_WEBHOOK: ${{ secrets.SLACK_WEBHOOK }}

# 5. Example flaky-quarantined test
import pytest

@pytest.mark.flaky
@pytest.mark.integration
def test_kafka_lag_below_threshold(kafka_container):
    """Known flaky under CPU pressure — reruns=2 attached automatically."""
    # ... test body ...
    assert measure_lag() < 100

Step-by-step trace.

Layer	Component	Effect
pytest.ini	strict-markers + coverage gate	fails PRs missing coverage or with unknown markers
conftest.py	@flaky → auto-quarantine marker	tracks quarantined tests without hiding them
tests.yml	xdist + reruns + junit upload	~5 min CI, junit stored per run
flaky-retriage.yml	nightly parse + Slack	flaky tests surfaced daily
Example test	@pytest.mark.flaky	rerun up to 2× on failure
Weekly	manual retriage of Slack digest	owners assigned; fix-or-delete in 2 weeks

After the two-week focus block, CI runtime drops from ~25 minutes serial to ~5 minutes parallel; coverage gate blocks PRs below 85%; flaky tests get reruns automatically and a Slack ping if they persist; the retriage ticket ensures nothing stays flaky forever. Dev unblock: teams can trust pytest -n auto as the pre-push check.

Output:

Metric	Before	After
CI runtime	25 min	5 min
Coverage gate	none	85% branch
Flaky visibility	ad-hoc	nightly Slack digest
Quarantine tracking	manual	@pytest.mark.flaky_quarantined
Devs skipping suite	~30%	~0% (fast enough to always run)
PRs blocked by real bugs	~40% caught	~85% caught

Why this works — concept by concept:

• xdist parallelism — -n auto --dist=loadgroup scales test throughput linearly with CPU count. loadgroup distributes tests grouped by xdist_group marker for fixture affinity, avoiding container-collision.
• Coverage gate — --cov-fail-under=85 is the anti-regression ratchet. PRs adding untested code fail; PRs improving coverage pass. The gate is what makes coverage a leading indicator instead of a lagging one.
• pytest-rerunfailures — @pytest.mark.flaky auto-reruns; --reruns 2 at the CLI applies globally. Buys time to fix flakes without red CI blocking merges. Combined with a quarantine marker for tracking.
• Junit dashboard — --junit-xml per run + a nightly parser + Slack ping = flaky tests get surfaced daily. No test hides in a "we always retry it" limbo.
• Cost — pytest-xdist (one process per CPU), pytest-cov (~3% CPU overhead), pytest-rerunfailures (near-zero), junit XML (~1 KB per test). The eliminated cost is 20 minutes of serial CI per PR + hours of debugging flaky failures + weeks of hidden coverage regressions. Net: O(1) developer time to fix a flake vs O(N) developer time to debug it silently. Two-week focus block, ~1 dev-week of net cost, pays back in ~1 month.

Python
Topic — exception-handling
Exception handling and CI-hardening tests

Practice →

Python
Topic — python
Python testing and CI-workflow problems

Practice →

---

Cheat sheet — pytest recipes for data engineering

• Four-layer test pyramid. Layer 1 unit (pure functions, no I/O, sub-ms per test, hundreds of tests, run on save), Layer 2 contract (dbt schema tests, pydantic validation, ~1 ms per test), Layer 3 component-with-DB (pytest-postgresql template clone, ~100 ms per test), Layer 4 integration-with-compose (testcontainers-python session compose, ~5-30 s per test, run on PR + nightly). The pyramid shape — thousands of Layer 1, dozens of Layer 4 — is what keeps CI under 8 minutes for a 1000+ test data-platform suite.
• Fixture scope decision matrix. Use function scope for anything that mutates during the test (fresh DB row, tmp file, monkeypatched env). Use session scope for anything with >1 s setup (Postgres process, compose stack, dbt manifest). Use module scope for tests-in-one-file that share a large seed. Avoid class scope in modern pytest; avoid autouse=True except for cross-cutting concerns like UTC timezone or logger reset — never for I/O fixtures.
• pytest-postgresql template-clone recipe. postgresql_proc = factories.postgresql_proc(port=None, unixsocketdir="/tmp", postgres_options="-c fsync=off -c synchronous_commit=off -c full_page_writes=off") at session scope; postgresql = factories.postgresql("postgresql_proc", dbname="test_db", load=[SCHEMA_SQL]) at function scope. The port=None is what makes it xdist-safe; the durability flags cut commit latency by ~5×; the template clone is a Postgres-native file-block copy at ~100 ms per test.
• Factory-as-fixture template. @pytest.fixture def make_order(postgresql): def _make(customer_id=1, total_cents=1000, status="pending"): with postgresql.cursor() as cur: cur.execute("INSERT INTO orders(...) VALUES(...) RETURNING id", (...)); return Order(id=..., ...); return _make. Every DE table gets a factory; factories close over the DB fixture; tests call the factory zero or more times per test with defaulted-and-overridable kwargs; cleanup is automatic via the template-clone drop at teardown.
• @pytest.mark.parametrize with ids=. Always supply ids= for parametrizations with more than 3 rows. Prefer readable strings (ids=["null-in", "empty-str", "valid-with-punct"]) or ids=lambda c: c.name when parametrizing over a dataclass. Never accept the default test_foo[0] / test_foo[test_foo0] IDs — they turn every failing test into a debugging tax.
• pytest_generate_tests for dynamic axes. def pytest_generate_tests(metafunc): if "table_name" in metafunc.fixturenames: metafunc.parametrize("table_name", discover_tables(), ids=discover_tables()). Use for manifest-driven axes (every model in the dbt manifest, every table in information_schema, every YAML config in configs/). Amplifies one test function into N tests without hand-writing them; new axis entries are automatically covered.
• testcontainers-python compose fixture. @pytest.fixture(scope="session") def compose_stack(worker_id): os.environ["COMPOSE_PROJECT_NAME"] = f"test_{worker_id}"; with DockerCompose(".", compose_file_name="docker-compose.test.yml", pull=True, wait=True) as compose: compose.wait_for("http://localhost:9001/minio/health/ready"); yield compose. The per-worker project name is what makes xdist safe (isolated networks, no port collision); wait=True blocks on healthchecks; the with statement handles docker-compose down -v at teardown.
• Wait-for-ready idiom. Never trust docker-compose up -d alone — the containers are started, not ready. Use compose.wait_for("http://.../health") for HTTP-endpoint services (MinIO, Airflow webserver, Elasticsearch) and a SELECT 1 retry loop with tenacity for raw Postgres or Kafka. Timeout at 30 s; log clearly on failure ("postgres never became ready"); belt-and-braces both patterns for high-value fixtures.
• pytest-xdist distribution modes. -n auto for CPU-count workers; --dist=load default is round-robin per test; --dist=loadgroup respects @pytest.mark.xdist_group("name") for fixture affinity (pin all Kafka-consumer tests to one worker); --dist=loadscope groups by file/class. Session-scoped fixtures are per-worker, so each worker gets its own Postgres process — this is the correct default; only elect a shared-across-workers stack via a filesystem lock when compose startup >30 s.
• pytest-cov config. --cov=mypkg --cov-branch --cov-report=term-missing:skip-covered --cov-report=xml --cov-fail-under=85. Line-only coverage lies about happy-path bias; branch coverage catches "we tested the if but not the else". term-missing:skip-covered prints only files with gaps; xml report feeds Codecov / SonarQube. Start --cov-fail-under at 70%, ratchet to 85%, aim for 90% on the core packages; never lower the threshold.
• pytest-rerunfailures flaky discipline. @pytest.mark.flaky(reruns=2) on the test (or CLI --reruns 2 globally); combined with a @pytest.mark.flaky_quarantined tracking marker (auto-applied via pytest_collection_modifyitems). A nightly job parses --junit-xml files, computes per-test pass rates, and posts <99% offenders to Slack. Weekly retriage: owner assigned; fix-or-delete deadline; no test stays quarantined forever.
• monkeypatch + tmp_path isolation. For any test touching env vars, cwd, or filesystem writes, use monkeypatch.setenv("KEY", "value") (auto-reverts), monkeypatch.chdir(tmp_path) (auto-reverts), and tmp_path (auto-deletes). Never modify os.environ or os.chdir directly — global-state leakage between tests is the #2 source of "works on my machine" flakes.
• GitHub Actions minimal recipe. strategy.matrix.python: ["3.11", "3.12"] on ubuntu-24.04; actions/setup-python@v5 with cache: pip; install postgresql-15 via apt for pytest-postgresql; pytest -n auto --dist=loadgroup --cov-fail-under=85 --junit-xml=junit.xml; if: always() on artifact uploads for junit + coverage. Wire the Slack notifier separately as a nightly cron job on the retriage workflow.
• Airflow / dbt / Spark specific gotchas. Airflow DAG tests need the airflow-init service to have condition: service_completed_successfully before the scheduler starts. dbt tests need monkeypatch.chdir(project_root) because dbt_project.yml is discovered from cwd. Spark UDF tests can use a session-scoped SparkSession fixture with .master("local[2]") — never local[*] in tests because it stalls other xdist workers. All three: pin the version in requirements-test.txt; version-drift is a silent flaky-test source.

Frequently asked questions

What is pytest in one sentence for a data engineer?

Pytest for data engineering is the Python test framework that unifies four test layers under one runner: unit tests for pure transforms, contract tests for dbt/pydantic schemas, component tests for SQL-emitting code against a real Postgres via pytest-postgresql, and integration tests for end-to-end pipelines spun up via pytest docker-compose or pytest testcontainers. Every plugin senior DE teams reach for — pytest-xdist for parallelism, pytest-cov for coverage, pytest-rerunfailures for flaky discipline, pytest-postgresql for template DBs, testcontainers-python for Kafka/MinIO/Airflow stacks — plugs into pytest's fixture graph without ceremony, and the same pytest command runs on a laptop, in a Codespace, and on GitHub Actions with identical behaviour. It is the load-bearing tool that makes a 1000-test data-platform suite feasible.

Pytest fixtures vs unittest setUp — which do I pick in 2026?

Pick pytest fixtures for every greenfield project and every serious refactor. pytest fixtures compose (fixture A can request fixture B), have explicit scope (function / module / session), support yield-teardown (fixture body reads top-to-bottom like a contextmanager), and are parametrizable (@pytest.fixture(params=[...])). unittest.setUp / tearDown do none of those — they're per-class-method, imperative, and force you into class TestFoo(unittest.TestCase): boilerplate. The only reason to touch unittest in 2026 is if you inherit a legacy suite and the migration cost exceeds the improvement (rare; pytest can run unittest-style tests natively, so you can migrate one file at a time). Every senior DE codebase has converged on pytest fixtures because scope discipline is what makes the difference between a 2-minute test run and a 20-minute one — and only pytest exposes scope as a first-class primitive.

How do I parametrize a pytest test over dbt models?

Use pytest_generate_tests(metafunc) to discover models from the dbt manifest at collection time and inject them as a parameter. Concretely: implement def pytest_generate_tests(metafunc): if "dbt_model" in metafunc.fixturenames: models = [n["name"] for n in json.loads(Path("target/manifest.json").read_text())["nodes"].values() if n["resource_type"]=="model"]; metafunc.parametrize("dbt_model", models, ids=models). Then any test that names dbt_model as a parameter runs once per model with a readable ID like test_mart_has_row_count_gt_zero[fct_orders]. Adding a new dbt model automatically adds a new test — zero test-file changes required. This is the amplifier that turns one contract test into hundreds and lets you enforce invariants like "every mart has ≥1 row" or "every staging model has a _ingested_at column" across the entire warehouse.

What's the difference between docker-compose fixtures and testcontainers-python?

docker-compose fixtures (via testcontainers.compose.DockerCompose or the older pytest-docker-compose plugin) wrap a docker-compose.yml file — best when your test needs the full networked stack (Airflow scheduler + webserver + Postgres + Redis + Celery, or Kafka + Zookeeper + Schema Registry). testcontainers-python per-service (PostgresContainer, KafkaContainer, MinioContainer) instantiates one container at a time in Python code — best when your test needs one or two containers with programmatic setup. Both handle wait-for-ready health checks and clean teardown; both are pytest docker-compose-compatible in the sense that the underlying tech is Docker. Rule of thumb: if your docker-compose.yml has ≥3 services, use DockerCompose; if it's 1-2 services, use per-service testcontainers-python and skip the yaml file. Both patterns are session-scoped; both need per-test isolation via schemas (Postgres) or uuid-suffixed topics (Kafka).

How do I run pytest in parallel with pytest-xdist without breaking Postgres fixtures?

Two rules: (1) make the Postgres fixture per-worker via session scope + OS-allocated port (factories.postgresql_proc(port=None) for pytest-postgresql, or COMPOSE_PROJECT_NAME=test_{worker_id} for docker-compose); (2) make every DB itself per-test via template clone or per-test schema (CREATE SCHEMA t_<uuid>). Concretely, pytest -n auto --dist=loadgroup spawns one worker per CPU; each worker gets its own session-scoped Postgres process on its own port; each test in a worker clones a fresh DB from the template. Under 4-worker xdist on an 8-core machine, a 400-test suite goes from ~2 minutes serial to ~30 seconds — with full isolation, no shared state, no port collisions. The --dist=loadgroup mode respects @pytest.mark.xdist_group("name") markers so all tests sharing a Kafka topic can pin to one worker (fixture affinity). Never try to share a single Postgres process across workers via a filesystem lock unless you've measured that it's actually necessary — per-worker Postgres is the simpler, safer default.

How do I test an Airflow DAG with pytest?

Use a session-scoped DockerCompose fixture with a minimal Airflow compose file — postgres for metadata + airflow-init (runs airflow db init, exits) + airflow-scheduler + optionally airflow-webserver. The compose file uses depends_on: airflow-init: condition: service_completed_successfully so the scheduler waits for init to finish. Test code triggers a DAG run via subprocess.check_output(["docker", "exec", "airflow-scheduler", "airflow", "dags", "trigger", "my_dag", "--run-id", f"test_{uuid.uuid4().hex[:8]}"]), then polls airflow dags state my_dag <run_id> every 2 seconds up to a 60-second deadline. Assert on the final state (success or failed). For unit-testing individual tasks in isolation, use Airflow's Task.test() method (no compose needed) or the @task-decorator's .function() accessor to call the underlying Python directly. For end-to-end DAG behaviour where you need the scheduler + metadata DB, the docker-compose pattern is the standard; ~30-second startup amortised across all DAG tests via session scope keeps the suite fast enough to run on every PR.

Practice on PipeCode

• Drill the Python practice library → for the pytest, fixture-graph, parametrize, and mocking problems senior interviewers love.
• Rehearse on the ETL practice library → for the ingestion-pipeline testing, DB-fixture design, and integration-suite scenarios.
• Sharpen the validation axis with the data-validation practice library → for the contract-test, schema-assertion, and dbt-model-verification patterns.
• Stack the prerequisites against PipeCode's broader 450+ data-engineering catalogue to anchor the four-layer testing decision matrix against real graded inputs.

Lock in pytest-for-DE muscle memory

Docs explain fixtures. PipeCode drills explain the decision — when session-scope makes the difference between a 2-minute and a 20-minute suite, when parametrize with readable ids beats copy-pasted tests, when testcontainers-python earns its place over a shell-out to docker-compose, when the coverage gate + rerunfailures + quarantine protocol turns a flaky suite green. Pipecode.ai is Leetcode for Data Engineering — pattern-first practice tuned for the production testing trade-offs senior data engineers actually face.

Practice Python problems →
Practice ETL problems →