Production Python for Data Engineers

Module 5: Testing Data Pipelines, Not Just Functions

What you'll learn in this module

By the end of this module, you'll be able to:

  • Recognize a test suite that's green but proves almost nothing, and explain exactly why
  • Write unit tests for pure logic, integration tests against a real database, and property-based tests that check an invariant across hundreds of generated inputs
  • Write failure-path tests that check why something failed, not just that it failed
  • Spot the single most common AI mistake in testing: over-mocking until a test can't actually fail

Let's begin.

Let's start with a question

Have you ever seen a test suite that was fully green, right up until the moment something in production broke — something that same test suite was supposedly covering?

That gap — green tests, real bug — is one of the most quietly dangerous situations in software. Not because nobody wrote tests. Because the tests looked like they were doing their job, and everyone trusted that green checkmark right up until it turned out not to mean what everyone assumed it meant.

This module isn't about writing more tests. It's about writing tests that would actually catch the bug you're worried about — and knowing the difference between a test that proves something, and a test that just runs without crashing.

Why this matters in data engineering

Ask yourself something uncomfortable: if your test suite is 100% green right now, do you actually know it would catch a real bug — or do you just know it doesn't currently fail?

Those are different questions. A pipeline can have plenty of tests, all passing, and still ship a bug that swaps two fields, silently corrupts a calculation, or lets a duplicate slip through — if the tests never actually checked the thing that broke.

Data pipelines are especially exposed to this. They touch real databases, real external services, real concurrent access. A test that only exercises a pure function in isolation can be green forever while the actual integration — the part that touches a real database, under real conditions — has never been tested at all.

The core idea: what a test actually proves

The over-mocking trap

Mocking a dependency — replacing a real database or API call with a stand-in for a test — is completely reasonable. The trap isn't mocking the dependency. It's when the assertions themselves stop checking anything real.

Ask yourself: does this assertion check that something happened correctly, or just that something happened at all? assert mock_db.save.called only proves save got called — not with what. Swap two arguments, corrupt a value, pass the wrong type entirely — that assertion still passes, as long as save was invoked once.

Four different jobs, four different kinds of test

  • Unit tests check pure logic in isolation — no database, no network, nothing external. Fast, and good at catching a wrong calculation.
  • Integration tests run against something real — an actual database, in this module's lab. They catch the bugs that only exist in the real interaction, which a mock can quietly hide.
  • Property-based tests don't check one hand-picked example. They ask a tool to generate hundreds of inputs and check that some invariant holds for all of them — catching edge cases nobody thought to write by hand.
  • Failure-path tests deliberately trigger a failure, and check that the specific, correct failure happened — not just that something went wrong.

A real test suite for a real pipeline usually needs more than one of these. Relying on just one — usually unit tests alone — leaves real gaps a green suite will never reveal.

Manual lab: the test suite that's green and wrong

Getting the lab files

Download module-5-materials.zip:

bash
cd ~/courses/pp4de
cp /mnt/c/Users/yourname/Downloads/module-5-materials.zip course-materials/
cd course-materials
unzip module-5-materials.zip

Commit it:

bash
cd ~/courses/pp4de
git add course-materials/module-5
git commit -m "Add Module 5 lab materials"
git push

The scenario

You've inherited a coupon redemption service for an event ticketing platform. Each coupon code should be usable exactly once, by exactly one order — ever. There's already a test suite for it, and it's fully green.

Reproduce the problem first

  1. Go into the starter folder: 
    bash
    cd course-materials/module-5/starter
  2. Run the existing tests: 
    bash
    python3 -m pytest test_discount.py -v
  3. Both tests pass. Read them — notice test_redeem_coupon_rejects_reuse uses a MagicMock for the database connection, and never touches a real database at all.
  4. Now let's find out what a real database says. Run this: 
    bash
    python3 -c "
from db import get_connection, seed_coupon
from discount import redeem_coupon

conn = get_connection(':memory:')
seed_coupon(conn, 'SAVE10', 10.0, '2026-12-31', 20.0)

result1 = redeem_coupon(conn, 'SAVE10', 'order-1', 100.0, '2026-07-01')
print('order-1 redeems SAVE10:', result1)

result2 = redeem_coupon(conn, 'SAVE10', 'order-2', 100.0, '2026-07-01')
print('order-2 redeems the SAME code SAVE10:', result2)
"
  5. Read the output carefully. The same coupon code just got successfully redeemed twice, by two different orders. The test suite is fully green, and this real bug exists anyway.

Sit with that for a second. Every test in this project currently passes. The bug is still there. That gap is the entire subject of this module.

Your task

  1. Find out why the existing tests didn't catch this — read test_discount.py again with that specific question in mind.
  2. Fix the actual bug in discount.py or db.py — the same coupon code must never be redeemable twice, by any order.
  3. Replace the weak test suite with a real one:
    • Unit tests for compute_discounted_total, with no database involved
    • An integration test against a real (in-memory) database, proving the same code can't be redeemed by two different orders
    • A property-based test proving the discounted total is never negative and never exceeds the original order total
    • Explicit failure-path tests for each rejection reason — unknown code, expired, below minimum — each checking the specific exception, not just that some exception happened

A question worth asking before you look at the solution

This time, the honest answer is a little different from earlier modules. The actual bug fix here is small — one column, one constraint. But the amount of test code grows a lot. That's not padding. A pipeline this small having eleven real tests instead of two fake ones isn't excessive — it's what actually proving correctness looks like, once you stop counting "number of tests" and start asking "what does each one actually prove."

Full worked solution

The complete solution lives in course-materials/module-5/solution/:

solution/
├── pyproject.toml
├── src/
│   └── discounts/
│       ├── __init__.py
│       ├── db.py                          # the fix: constraint on the right column
│       └── discount.py                    # the fix: atomic check, same pattern as Module 4
└── tests/
    ├── __init__.py
    ├── test_unit_calculation.py           # pure logic, no database
    ├── test_integration_redemption.py     # real database, catches the real bug
    ├── test_property_based.py             # hypothesis - hundreds of generated inputs
    └── test_failure_paths.py              # each failure, checked specifically

The actual bug, in db.py. Compare these two lines to the starter's version:

python
CREATE TABLE IF NOT EXISTS redemptions (
    code TEXT PRIMARY KEY,
    order_id TEXT NOT NULL,
    redeemed_at TEXT NOT NULL
)

The starter had UNIQUE on order_id. That enforces "one order can only use one coupon, ever" — a real rule, just the wrong one. This version puts the constraint on code, which enforces the actual intended rule: each code can only be used once, by anyone.

The fix in discount.py — the same atomic pattern from Module 4:

python
cursor = conn.execute(
    "INSERT INTO redemptions (code, order_id, redeemed_at) "
    "VALUES (?, ?, ?) ON CONFLICT(code) DO NOTHING",
    (code, order_id, today),
)
if cursor.rowcount == 0:
    raise CouponAlreadyUsedError(f"Coupon {code} has already been used")

The integration test that actually catches it:

python
def test_same_code_cannot_be_used_by_two_different_orders() -> None:
    conn = get_connection(":memory:")
    seed_coupon(conn, "SAVE10", discount_percent=10.0, expires_at="2026-12-31", min_order_total=20.0)

    first_result = redeem_coupon(conn, "SAVE10", "order-1", 100.0, "2026-07-01")
    assert first_result == 90.0

    try:
        redeem_coupon(conn, "SAVE10", "order-2", 100.0, "2026-07-01")
        assert False, "expected CouponAlreadyUsedError for a second, different order"
    except CouponAlreadyUsedError:
        pass

Notice this uses a real get_connection(":memory:") — an actual SQLite database, just an in-memory one for speed. Nothing here is mocked. This is the test that would have caught the starter's bug immediately, the moment it was written.

The property-based test:

python
from hypothesis import given
from hypothesis import strategies as st

from discounts.discount import compute_discounted_total


@given(
    order_total=st.floats(min_value=0, max_value=100_000, allow_nan=False, allow_infinity=False),
    discount_percent=st.floats(min_value=0, max_value=100, allow_nan=False, allow_infinity=False),
)
def test_discounted_total_is_always_between_zero_and_original(
    order_total: float, discount_percent: float
) -> None:
    result = compute_discounted_total(order_total, discount_percent)
    assert 0.0 <= result <= round(order_total, 2)

hypothesis generates a wide range of order_total and discount_percent values — not just the two or three you'd think to write by hand — and checks the invariant holds for every one of them.

The failure-path tests, each checking a specific exception, not a generic one:

python
def test_unknown_code_raises_invalid_coupon_error() -> None:
    conn = get_connection(":memory:")
    with pytest.raises(InvalidCouponError):
        redeem_coupon(conn, "DOES-NOT-EXIST", "order-1", 100.0, "2026-07-01")

Also worth noticing: test_order_exactly_at_minimum_is_accepted — a boundary test, checking that an order exactly at the minimum is accepted, not rejected. >= and > read almost identically at a glance, and only a test at the exact boundary value would ever catch the difference.

Verify it, step by step

  1. Create a virtual environment and install: 
    bash
    cd course-materials/module-5/solution
python3 -m venv .venv
source .venv/bin/activate
pip install -e .
pip install mypy pytest hypothesis
  2. Run mypy --strict: 
    bash
    python -m mypy --strict src/discounts/ tests/
    Expected: Success: no issues found.
  3. Run the full test suite: 
    bash
    python -m pytest tests/ -v
    You should see 11 tests, all passing.
  4. Run the exact same manual check that exposed the starter's bug — this time against the fixed code: 
    bash
    python3 -c "
from discounts.db import get_connection, seed_coupon
from discounts.discount import redeem_coupon, CouponAlreadyUsedError

conn = get_connection(':memory:')
seed_coupon(conn, 'SAVE10', 10.0, '2026-12-31', 20.0)

result1 = redeem_coupon(conn, 'SAVE10', 'order-1', 100.0, '2026-07-01')
print('order-1:', result1)

try:
    result2 = redeem_coupon(conn, 'SAVE10', 'order-2', 100.0, '2026-07-01')
    print('BUG STILL PRESENT:', result2)
except CouponAlreadyUsedError as e:
    print('Correctly rejected:', e)
"

If step 3 showed 11 passing tests, and step 4 showed the second redemption correctly rejected, you've verified the actual thing this module set out to teach.

AI-assisted round

The task

Ask your assistant to add a new feature and its tests:

Output / Note

"Add a cancel_redemption function that reverses a coupon redemption — removing it from the redemptions table so the code becomes usable again, for cases like a cancelled order. Write tests for it, and run them before telling me you're done."

The known failure pattern to watch for

This is the exact pattern this whole module is built around: an assistant writing tests that over-mock until the test can't actually fail.

Here's what that tends to look like:

diff
+ def test_cancel_redemption():
+     mock_conn = MagicMock()
+     cancel_redemption(mock_conn, "SAVE10")
+     assert mock_conn.execute.called

Read this assertion closely. It checks that conn.execute was called — once, for any SQL statement, with any arguments. It would pass even if cancel_redemption executed completely the wrong query, or deleted every row in the table instead of one, or didn't actually delete anything related to "SAVE10" at all. The test cannot fail in any way that would actually matter.

This is dangerous specifically because it doesn't look lazy. It looks like a real test — it has a mock, it has an assertion, it runs and passes. The problem is invisible unless you specifically ask: what would have to go wrong in this function for this test to actually fail?

The guardrail

If you catch this, write it down:

Output / Note

Module 5 guardrail: For any AI-generated test with a mock, explicitly ask: what would have to break in the real code for this test to fail? If the honest answer is "almost nothing," the test isn't testing the thing it claims to. Prefer a real (in-memory) database over a mock wherever the test is meant to verify actual data correctness — mock the dependency only when the point of the test is genuinely about isolation, not about verifying real behavior.

If your assistant wrote a real integration test against the actual database without being told to — log that too.

Common mistakes

  • assert mock.method.called as the entire assertion. Proves a call happened. Proves nothing about whether it happened correctly.
  • Testing only the happy path. A pipeline's failure paths are often where the real risk lives — an untested failure path is exactly where a silent data-corrupting bug likes to hide.
  • One example instead of a property. Hand-picking two or three test values feels thorough. It rarely is — property-based testing exists specifically because humans are bad at guessing which input will break something.
  • Mocking so much of a test that it can't distinguish a working implementation from a broken one. If you could replace the real function with pass and the test would still pass, the test isn't testing anything.

Capstone tie-in

Step 1: Open your capstone repo and start the infrastructure

Confirm the title bar says pp4de [WSL: Ubuntu].

bash
docker compose start
docker compose ps

Step 2: Build a layered test suite for your capstone

In VS Code's Explorer:

  1. If you don't have one yet, right-click the top-level pp4de folder → New Folder → name it tests.
  2. Add at least one test from each category, applied to your actual capstone code so far:
    • A unit test for something in your Module 2 validation logic (IngestionRecord) — no database, no API, pure logic
    • An integration test against your real Postgres container, proving your Module 4 upsert genuinely doesn't duplicate rows — you can adapt the same pattern from this module's lab
    • A failure-path test that feeds an intentionally malformed record through your pipeline and checks it's rejected with the specific error you expect, not just "something happened"

Step 3: Install pytest and hypothesis into your capstone venv

Open the integrated terminal, confirm your venv is active, and run:

bash
pip install pytest hypothesis

Same reminder as every module — these are dev tools, not runtime dependencies of your package. Add them to your own notes if you're tracking what's installed where, but they don't belong in pyproject.toml's dependencies list.

Step 4: Run your tests — the one mandatory terminal step

bash
python -m pytest tests/ -v

Confirm everything passes. If your integration test against Postgres fails, check that docker compose start actually succeeded and the container is healthy before assuming the test itself is wrong.

Step 5: Commit and push through VS Code's Source Control panel

  1. Click the Source Control icon in the sidebar.
  2. Confirm .venv is not listed under Changes.
  3. Stage, commit with a message like Module 5: unit, integration, and failure-path tests, and click Commit.
  4. Click Sync Changes to push to GitHub.
  5. Confirm the changes appear on GitHub.

Check row 5 off your SPEC.md checklist. Four to go.

Before you close for the day:

bash
docker compose stop

Interview drill

Every question below follows the same pattern. First, the question. Then, what the interviewer wants to hear. Then, a junior engineer's answer. Then, a strong senior answer.

Recall

Question: "What's the actual difference between a unit test, an integration test, and a property-based test? When would you reach for each one?"

What the interviewer wants to hear: clear, distinct definitions, plus a real sense of when each is the right tool — not just recitation.

Junior answer: "Unit tests test one function. Integration tests test more than one thing together. Property-based tests use a library to generate test cases automatically."

(Technically not wrong, but shallow — no sense of when each one is actually the right choice, or what each one is good at catching that the others aren't.)

Senior answer: "A unit test isolates pure logic — no database, no network — and it's fast and precise for catching a wrong calculation. An integration test runs against something real, like an actual database, and it catches bugs that only exist in the real interaction — exactly the kind a mock can hide, like an actual bad uniqueness constraint. A property-based test doesn't check one example at all — it generates a wide range of inputs and checks an invariant holds across all of them, which is good for catching edge cases nobody thought to write by hand, like a boundary value or an unusual combination of inputs. I reach for unit tests first, for fast feedback on logic. I reach for integration tests specifically where the real dependency's behavior matters to correctness, not just the code around it. I reach for property-based tests when there's a genuine invariant — something that should always be true regardless of the specific input — rather than just a fixed set of expected outputs."

Debugging

Question: shown as a real snippet and incident story, with no explanation yet:

python
def test_process_watch_event():
    mock_db = Mock()
    mock_enricher = Mock()
    mock_enricher.enrich.return_value = {"genre": "drama", "duration_sec": 3600}

    event = {"user_id": "u123", "content_id": "c456", "watched_sec": 1800}
    process_watch_event(event, db=mock_db, enricher=mock_enricher)

    assert mock_db.save.called

"This test is green. Last week, process_watch_event shipped a change that silently swapped user_id and content_id when saving to the database — every watch event since then has been attributed to the wrong user. This test was already in the suite before and after that change, and never caught it. Why not, and what would?"

What the interviewer wants to hear: precisely identifying that the assertion checks "was called," not "was called correctly" — and proposing the actual fix, not just a vague "add more tests."

Junior answer: "Maybe the test needs to run more than once, or check more of the database calls."

(Doesn't identify the actual problem — the assertion itself is checking the wrong thing, regardless of how many times it runs.)

Senior answer: "The assertion only checks mock_db.save.called — that save was invoked at all, not what it was invoked with. A completely broken call — wrong argument order, swapped values, missing fields — still makes this pass, as long as save gets called once. That's exactly what happened here: user_id and content_id got swapped, save still got called, and the test had no way to notice, because it was never actually looking at the values. The fix is checking the real call arguments — mock_db.save.assert_called_once_with(user_id='u123', content_id='c456', ...) — or inspecting mock_db.save.call_args directly. Mocking the dependency, the database, is fine. The problem is that the assertion was mocked away too, in effect — it stopped checking real behavior. I'd also point out this exact bug — a field-order swap — is the kind of thing a property-based or schema-validated test would catch structurally, rather than relying on someone remembering to assert the right field by hand."

AI-review

Question: shown this diff, with no explanation yet:

diff
+ def test_cancel_redemption():
+     mock_conn = MagicMock()
+     cancel_redemption(mock_conn, "SAVE10")
+     assert mock_conn.execute.called

What the interviewer wants to hear: recognizing this as the exact over-mocking pattern from earlier in the module, and being able to articulate precisely what this test fails to prove.

Junior answer: "It calls the function and checks the database was used — that's a reasonable smoke test."

(Accepts a test that "does something" as sufficient, without asking what specific behavior it's meant to verify.)

Senior answer: "This test can't actually fail in any way that matters. mock_conn.execute.called is true if execute was called once, for any SQL, with any arguments — it would pass even if cancel_redemption deleted the wrong row, deleted every row, or ran a completely unrelated query. Nothing here checks that the right thing happened to the right coupon code. I'd ask: what would actually have to break in cancel_redemption for this test to fail? Right now, the honest answer is 'almost nothing.' I'd tell the assistant to either use a real in-memory database — the same pattern already used everywhere else in this project — and check the row is actually gone afterward, or, if a mock is genuinely necessary, assert on the specific SQL and parameters passed to execute, not just that it was called."

Judgment

Interviewer: "You're reviewing a pull request for a service that processes leaderboard events for a live gaming platform. The PR adds a new scoring rule and includes three new tests, all passing. All three tests mock the database entirely. Your teammate says the mocks make the tests fast and isolated, which is exactly what unit tests should be. Do you approve the PR?"

Candidate (asking first): "Before I answer — what's actually being tested in those three cases? Is the new scoring rule itself pure logic, or does it involve reading or writing real leaderboard state?"

Interviewer: "Good question. The scoring rule reads a player's current streak from the database, and the new logic changes how bonus points get calculated based on that streak."

Candidate (first answer): "Then I wouldn't block the PR outright, but I wouldn't approve it as-is either. If the scoring calculation itself — given a streak value — is pure logic, mocked unit tests are completely reasonable for that part, and my teammate's right that it keeps them fast. But the part where the streak actually gets read from the database is a different concern, and mocking that away means nothing here actually proves the real read behaves correctly."

Interviewer (pushback): "But isn't testing the database read itself someone else's responsibility — like, that's just calling the DB layer, which presumably has its own tests?"

Candidate (round 1): "That's a fair point in isolation, but the actual risk here isn't 'does the database layer work in general' — the existing DB layer is presumably already tested elsewhere. The risk is specifically 'does this new scoring rule correctly interact with a real streak value coming from a real read.' That's a new integration point, even if both pieces individually have their own tests. I'd want at least one test that exercises the real path — new scoring rule, real database, real streak value — because that's the one place a mismatch between the two could hide, and none of these three tests would catch it."

Interviewer (second pushback): "This is a live gaming platform — leaderboard updates happen constantly, at real scale. Wouldn't adding a real database call to every test slow the suite down too much to be practical?"

Candidate (round 2): "I wouldn't propose converting all three tests to hit a real database — that probably is overkill, and my teammate's speed concern is legitimate. I'd propose keeping the three fast, mocked tests for the pure calculation logic, and adding exactly one additional integration test — a real in-memory or test database, one realistic streak value, checking the whole path end to end. That's a small, targeted addition, not a wholesale rewrite of the existing tests, and it specifically covers the one gap the current three can't."

Interviewer (final challenge): "Convince me one extra test is actually worth blocking a merge over, on a platform this size."

Candidate (final defense): "It's worth it because of exactly what kind of bug this gap would let through. If the scoring rule reads the streak value incorrectly — wrong field, wrong type, off-by-one — every one of these three mocked tests would stay green, because none of them touch a real streak value at all. That's not a hypothetical; it's the same shape of bug as a real incident this course covers elsewhere — a field swap that a fully mocked test suite genuinely didn't catch, because it was never checking real values in the first place. One targeted integration test is a small cost. Shipping a live scoring bug to a real leaderboard, discovered by players instead of a test suite, is not."

Model strong answer — a single answer, given upfront:

Output / Note

"I'd ask first whether the new scoring rule is pure calculation, or whether it involves reading real state from the database — in this case, a player's streak. Assuming it reads a real streak value, I wouldn't approve the PR as-is, but I also wouldn't ask for a wholesale rewrite. I'd keep the three fast, mocked tests for the pure calculation logic — that part's a reasonable use of mocking, and my teammate's right that it keeps things fast. But I'd add exactly one additional integration test, against a real in-memory or test database, with a realistic streak value, exercising the whole path end to end. The reason: none of the three existing tests can catch a bug where the scoring rule misreads the real streak value — wrong field, wrong type, off-by-one — because none of them touch a real value at all. That's a small, targeted addition, not a performance problem for the suite, and it's specifically aimed at the one gap a fully mocked test suite can't see."