An integration test for the whole pipeline

Description

Every test you wrote since module 2 tests an isolated piece: extract_orders() alone, validate_orders() alone, transform_or_quarantine() alone. Even test_retry.py's tests (module 4), which do exercise several pieces together, never go through cli.py — they call internal package functions directly, exactly how you came in through the kitchen window before the front door existed (lesson 2). This lesson writes tests/test_pipeline_integration.py, this entire guide's first test file calling main() — the front door — with real arguments, and confirms the complete result: extraction, validation, the production engine lesson 4 decided, retries, and load, all together, with assert.

Connection to the module. This lesson builds half of this module's second pillar: an integration test confirming the complete package, invoked exactly as a real scheduler would invoke it, produces the correct result. Lesson 6 adds the other half — the idempotency test — over the same file.

An analogy: the crash test of the complete car, not the airbag alone

A car manufacturer tests the airbag on its own: detonates it on an isolated test bench, measures how many milliseconds it takes to inflate, confirms the pressure is right. It's a real, necessary test — but it doesn't confirm the airbag, installed inside the complete car, connects to the right sensor, receives the signal at the right moment, and doesn't interfere with the seatbelt. Only a crash test of the entire car, assembled, exactly as it's going to leave the factory, confirms that.

test_retry.py's tests (module 4) are the isolated airbag test: they confirm transform_or_quarantine() correctly translates ValueError into DataQualityError, with nothing else from the pipeline involved. This lesson is the complete car's crash test: it invokes main() with a real --date, lets the run pass through every piece — config.py, logging_config.py, the production engine, retry.py, load.py — in the exact same order a real scheduler would, and confirms, with assert, that the result in the warehouse is correct. Neither test replaces the other — module 4 still needs its five deterministic tests over the isolated mechanism; this lesson adds confirmation that, assembled, the complete set also works.

Worked example: main(argv) as the test's entry point

Remember, from lesson 2, why main() takes argv: list[str] | None = None as a parameter instead of reading sys.argv directly: it makes it possible to invoke the complete command, with controlled arguments, with no new terminal process launched. This lesson is, precisely, the moment that design decision pays off. Create tests/test_pipeline_integration.py:

# tests/test_pipeline_integration.py
"""End-to-end tests: cli.main() -> run_pipeline() -> the full extract/validate/
transform(Polars)/load(resilient) chain, over a real (tmp, throwaway) SQLite
warehouse -- never a single function mocked in isolation, the whole wiring.

These tests never sleep (no retry failure is exercised here -- test_retry.py
already covers that mechanism in isolation) and never touch data/warehouse.db,
Kiosko's real warehouse file: db_path always points inside pytest's tmp_path,
a fresh throwaway directory per test.
"""

import sqlite3

from kiosko_pipeline.cli import main


def _warehouse_counts(db_path) -> tuple[int, float]:
    con = sqlite3.connect(db_path)
    row_count, revenue = con.execute("SELECT COUNT(*), ROUND(SUM(revenue), 2) FROM fact_orders").fetchone()
    con.close()
    return row_count, revenue


def test_cli_run_for_a_known_date_loads_the_right_row_count_and_revenue(tmp_path, data_dir):
    db_path = tmp_path / "warehouse.db"

    exit_code = main(["--date", "2026-08-05", "--data-dir", str(data_dir), "--db-path", str(db_path)])

    assert exit_code == 0
    assert _warehouse_counts(db_path) == (2, 9.55)

Stop at every new piece. main(["--date", "2026-08-05", "--data-dir", str(data_dir), "--db-path", str(db_path)]) — exactly the same list of strings argparse would get from sys.argv if you typed that command in a real terminal, passed directly to the function. There's no subprocess.run(["kiosko-pipeline", ...]), no new process, no interpreter startup cost — it's a normal Python function call, as fast as any other test in this suite, running exactly the same code the installed command would run.

tmp_path is a pytest fixture you didn't define — it comes included in pytest's base install, available in any test with no import at all — a pathlib.Path object pointing at a temporary folder, fresh and empty, created automatically before each test and deleted automatically afterward. db_path = tmp_path / "warehouse.db" builds a path inside that throwaway folder — this test's warehouse never touches data/warehouse.db, the real file you'd use on a genuine run. data_dir, on the other hand, is this package's own fixture, the same one you already know from module 2's conftest.py — it points at the real data/ folder, with the known week's seven fixed CSVs — this test's source data is real and shared across the whole suite, but the destination (db_path) is throwaway and exclusive to each test. It's the same discipline you already saw with sqlite3.connect(":memory:") in test_retry.py, now applied with a real file instead of an in-memory connection, because run_pipeline() needs a file path, not an already-open connection.

_warehouse_counts(db_path) is a helper function — not a test, doesn't start with test_ — opening the resulting warehouse and returning (row_count, revenue) in a single call, the exact same "helper function that isn't a test" pattern you already saw with _row() in module 2. The final assert compares against (2, 9.55) — 2026-08-05's exact numbers you already know by heart since module 3.

Run this first test:

uv run pytest tests/test_pipeline_integration.py -v

What to expect:

============================= test session starts ==============================
platform darwin -- Python 3.12.3, pytest-9.1.1, pluggy-1.6.0
rootdir: /path/to/kiosko_pipeline
configfile: pyproject.toml
collecting ... collected 1 item

tests/test_pipeline_integration.py::test_cli_run_for_a_known_date_loads_the_right_row_count_and_revenue PASSED [100%]

============================== 1 passed in 0.02s ===============================

Worked example: the complete week, one day at a time — like a real scheduler

The first test confirms one day. Add one confirming all seven, invoking the command seven times, once per partition — exactly how a real scheduler would do it, triggering a new run of the command every day, not a single run over a list of dates:

def test_cli_run_for_every_known_day_matches_foundations_numbers(tmp_path, data_dir):
    # One partition per call, exactly how a daily-batch scheduler would invoke
    # this command -- same db_path across all seven calls, one connection's
    # worth of history accumulating one partition at a time.
    db_path = tmp_path / "warehouse.db"
    known_days_and_counts = [
        ("2026-08-03", 8), ("2026-08-04", 6), ("2026-08-05", 2), ("2026-08-06", 5),
        ("2026-08-07", 7), ("2026-08-08", 9), ("2026-08-09", 3),
    ]

    for day, _ in known_days_and_counts:
        exit_code = main(["--date", day, "--data-dir", str(data_dir), "--db-path", str(db_path)])
        assert exit_code == 0

    row_count, revenue = _warehouse_counts(db_path)
    assert row_count == sum(count for _, count in known_days_and_counts) == 40
    assert revenue == 106.15

Notice the detail making this test genuinely different from any run you already know: db_path is the same file across all seven calls, but every main() call is a completely independent invocation of the command — with its own run_id (derived from a single-day list each time, so kiosko-2026-08-03-2026-08-03, not kiosko-2026-08-03-2026-08-09), its own SQLite connection opened and closed. load_to_warehouse() (unchanged since module 1) is still what makes this work: every call deletes only its own day's partition before inserting, so accumulating seven independent calls over the same file produces exactly the same result as a single run over all seven days together — 106.15, the same revenue as always.

Worked example: quarantine, inside the complete path

The third test confirms something lesson 4 already demonstrated by hand: that a partition with an orphaned record gets quarantined with no crash of the command, run through main() end to end, not just calling transform_or_quarantine_polars() in isolation the way test_retry.py's tests do:

def test_cli_run_for_the_orphan_store_date_quarantines_without_crashing(tmp_path, data_dir):
    # orders_2026-08-11.csv (fixture data_dir, module 2) has exactly one row,
    # from store S04 -- never added to DIM_STORE. This is the same partition
    # test_transform.py and test_retry.py already proved raises DataQualityError
    # in isolation; this test proves the WHOLE cli.main() -> run_pipeline() ->
    # transform_or_quarantine_polars() chain survives it: exit code 0, nothing
    # loaded, no Traceback (see pipeline.py's total_loaded > 0 guard, added in
    # this module after this exact scenario raised sqlite3.OperationalError).
    db_path = tmp_path / "warehouse.db"

    exit_code = main(["--date", "2026-08-11", "--data-dir", str(data_dir), "--db-path", str(db_path)])

    assert exit_code == 0

This test is deliberately simple — a single assert, over the exit code — and that simplicity is the point: it isn't confirming any number (test_data_quality_error_is_never_retried_by_the_load_policy, in module 4, already tests the quarantine mechanism in detail), it's confirming the complete path doesn't crash. Before lesson 4's fix, this assert would have failed — not because exit_code was different from 0, but because main() would never have returned anything at all: sqlite3.OperationalError would have propagated with nobody catching it, and pytest would have reported the test as ERROR, not FAILED, the specific signal that something blows up before the test itself can even evaluate its assert.

Run all three together:

uv run pytest tests/test_pipeline_integration.py -v

What to expect (partial collection — lesson 6 adds two more tests to this same file):

============================= test session starts ==============================
platform darwin -- Python 3.12.3, pytest-9.1.1, pluggy-1.6.0
rootdir: /path/to/kiosko_pipeline
configfile: pyproject.toml
collecting ... collected 3 items

tests/test_pipeline_integration.py::test_cli_run_for_a_known_date_loads_the_right_row_count_and_revenue PASSED [ 33%]
tests/test_pipeline_integration.py::test_cli_run_for_every_known_day_matches_foundations_numbers PASSED [ 66%]
tests/test_pipeline_integration.py::test_cli_run_for_the_orphan_store_date_quarantines_without_crashing PASSED [100%]

======================== 3 passed, 2 deselected in 0.03s =========================

Diagram: what each test layer in this guide proves

flowchart TD
    subgraph Unitario["Unit -- one function, isolated"]
        A["test_extract.py, test_quality.py,\ntest_transform.py (module 2)"]
        B["test_retry.py (module 4) --\ntransform_or_quarantine() alone,\nload_to_warehouse_resilient() alone"]
        C["test_transform_duckdb.py,\ntest_transform_polars.py\n(modules 5-6) -- one engine at a time"]
    end
    subgraph Integracion["Integration -- the complete assembly"]
        D["test_pipeline_integration.py\n(this module) -- complete cli.main(),\nevery call goes through EVERY piece"]
    end
    A --> D
    B --> D
    C --> D
    D --> E["Confidence that the pieces,\nalready tested separately,\nwork TOGETHER"]

Neither layer replaces the other. A failing unit test tells you, with surgical precision, exactly which function broke — the integration test, if it failed, would only tell you "something, somewhere in the complete chain" didn't work as expected, with no exact line pointed at. That's why this guide keeps both layers: the unit tests' diagnostic speed (thirty-five of them, running in fractions of a second), and the complete-assembly confidence only an integration test can give.

Going deeper: why call main() directly, and not subprocess.run()

There's an alternative to main(["--date", ...]) you might have thought of: launching the installed command as a real process, with subprocess.run(["kiosko-pipeline", "--date", "2026-08-05", ...], capture_output=True), and checking the captured output. It's a valid technique, and some test suites prefer it when they genuinely need to test the command exactly as an end user would install it — including [project.scripts]'s resolution in pyproject.toml. This lesson picks main(argv) instead, for two concrete reasons. First, speed: launching a new Python process — with its own interpreter startup, its own loading of every dependency — costs orders of magnitude more than a function call inside the same process already running pytest; with five tests in this file, the difference is marginal, but with a suite of hundreds of integration tests, it becomes real. Second, and more important: main(argv) gives you direct access to the function's return value — exit_code, a normal int — with no subprocess exit code to parse nor stdout/stderr to capture separately. The coverage you lose by not using subprocess.run() — confirming [project.scripts] is correctly configured, that the installed command genuinely exists — is exactly what you already confirmed by hand, run, in lessons 2, 3, and 4 of this module; there's no need for an automated test to repeat that confirmation every time, because it isn't the kind of thing that breaks silently between one run and the next.

Common mistakes

Pointing db_path at data/warehouse.db, the real file, instead of tmp_path. What happens: someone writes a test calling main(["--date", "2026-08-05"]) with no --db-path passed, letting config.py fall back to its default (data/warehouse.db). The test might pass — the numbers are correct — but every run of the test suite leaves the real development warehouse with test data mixed in with whatever real data was there before. Why it happens: it's shorter to write main([...]) without the configuration flags, and "the test passes" hides the side effect. How to spot it: if data/warehouse.db changes size or content after running uv run pytest, with you never having run any kiosko-pipeline by hand yourself, some test is touching the real file. How to fix it: every test in this file passes --db-path explicitly, pointing inside tmp_path — the same isolation discipline you already saw with sqlite3.connect(":memory:") in test_retry.py, adapted to a case where run_pipeline() genuinely needs a real file path (even if temporary), not an already-open connection.

Confusing a failing test (FAILED) with one blowing up before evaluating its assert (ERROR). What happens: someone sees a test from this file reported as ERROR instead of FAILED, and assumes it's the same kind of problem — an assert that didn't hold — when actually pytest uses ERROR specifically to signal an exception interrupted the test before its body finished running, never even reaching any assert. Why it happens: both look like "the test didn't pass" in a quick summary, and it's easy to miss the distinction if you don't read pytest's complete output. How to spot it: pytest -v's output clearly distinguishes the two cases — FAILED shows the exact assert line that didn't hold and the values compared; ERROR shows a complete Traceback from an exception unrelated to any assert. How to fix it: an ERROR in an integration test like this lesson's is, almost always, the signal of a real bug in the code under test — exactly what happened with sqlite3.OperationalError before lesson 4's fix — not an expectation to adjust in the test itself. Treat it as a clue to review the package's code, not the test.

Writing an integration test that replaces, instead of complementing, an existing unit test. What happens: someone, satisfied that test_cli_run_for_the_orphan_store_date_quarantines_without_crashing covers S04's case, deletes or stops maintaining module 4's test_data_quality_error_is_never_retried_by_the_load_policy, reasoning "it's already covered anyway." Why it happens: the two tests confirm related behaviors, and it's easy to see redundancy where there are actually two different levels of confidence. How to spot it: ask yourself what exact line of code a failing test would point at — this lesson's integration test, if it failed, wouldn't tell you whether the problem is in transform_or_quarantine_polars(), in pipeline.py, or in cli.py; module 4's unit test would, precisely, because it tests a single isolated piece. How to fix it: keep both — this lesson's diagram shows it precisely: unit tests and integration tests aren't interchangeable, they're two different questions about the same system.

Exercises

Exercise 1 — Add a test confirming the production engine, not just the result. This lesson's three tests confirm what the pipeline loads, but none explicitly confirms the engine used is Polars. Write a new test running main() over a known date and confirming, with the pipeline_run_started event or any other available evidence, that the real engine is "polars" — hint: check what constant cli.py exposes.

from kiosko_pipeline.cli import PRODUCTION_ENGINE


def test_cli_uses_polars_as_the_production_engine():
    assert PRODUCTION_ENGINE == "polars"

Exercise 2 — Break the integration on purpose, and confirm the test catches it. Temporarily change, in pipeline.py, the if engine not in ("dict", "polars") validation so it only accepts ("dict",) (remove "polars" from the tuple). Run test_cli_run_for_a_known_date_loads_the_right_row_count_and_revenue and observe what pytest reports. Undo the change when done.

Exercise 3 — Explain, in your own words, why data_dir is real but db_path is temporary in these tests. In 2-3 sentences, explain why this lesson's tests read from the project's real data/ folder (through the data_dir fixture) but never write to the real data/warehouse.db (they always use tmp_path).

Summary and next step

In this lesson you built the start of tests/test_pipeline_integration.py, with three tests invoking main() — the package's front door, built in lesson 2 — end to end: one confirms a single known day's result, one confirms the complete week invoked as a real scheduler would (one call per day, same warehouse accumulating), and one confirms the partition with the orphaned record gets quarantined with no crash of the command, exercising along the way the fix you discovered in lesson 4. None of the three replaces the thirty-five inherited unit tests — they complement them, with a different layer of confidence: that the pieces, already tested separately, work together.

Before moving on you should be able to: explain the difference between a unit test and an integration test with a concrete example from this guide; explain why tmp_path (a pytest fixture, not this package's) is the piece making it possible to isolate every test from the real warehouse; and explain why this lesson chose calling main(argv) instead of launching the command with subprocess.run().

This lesson's three tests confirm the pipeline works. None yet confirms the most important guarantee of all: that running it twice over the same date produces the exact same result, with nothing duplicated. Lesson 6 adds that test — this guide's central idempotency, finally verified with an assert, not by running a command twice and comparing the output by eye.

Resources

• pytest — official documentation for the tmp_path fixture, included with no extra import needed. docs.pytest.org/en/stable/how-to/tmp_path.html.
• pytest — official documentation on the difference between FAILED and ERROR in a run's report. docs.pytest.org/en/stable/how-to/output.html.
• Python — official documentation for subprocess.run(), the alternative this lesson evaluates and doesn't choose in "going deeper." docs.python.org/3/library/subprocess.html#subprocess.run.
• Martin Fowler — "Integration Test," the reference definition of what distinguishes an integration test from a unit test. martinfowler.com/bliki/IntegrationTest.html.