Testing, Debugging, Logging & Operability

A test is a small piece of code that checks your code. You give it an input and the answer you expect.

A pipeline runs again and again. Tomorrow it reads new data. Next month a teammate changes it.

A test tells you right away when the answer changes.

Here is the shop from the earlier chapters. The function paid_total adds up the amounts of paid orders. The test uses assert.

The word assert means: this must be true. If it is false, Python stops with an AssertionError and shows your message.

orders = [
    {"order_id": 101, "customer_id": 1, "status": "paid",
     "amount": 40.0, "order_date": "2026-01-03"},
    {"order_id": 102, "customer_id": 1, "status": "paid",
     "amount": 25.0, "order_date": "2026-01-10"},
    {"order_id": 103, "customer_id": 2, "status": "refunded",
     "amount": 60.0, "order_date": "2026-01-11"},
    {"order_id": 104, "customer_id": 3, "status": "paid",
     "amount": 15.5, "order_date": "2026-01-15"},
    {"order_id": 105, "customer_id": 3, "status": "pending",
     "amount": None, "order_date": "2026-01-20"},
    {"order_id": 106, "customer_id": 4, "status": "paid",
     "amount": 80.0, "order_date": "2026-01-21"},
]

def paid_total(orders):
    total = 0
    for order in orders:
        if order["status"] == "paid":
            total += order["amount"]
    return total

def test_paid_total():
    total = paid_total(orders)
    assert total == 160.5, f"expected 160.5, got {total}"

test_paid_total()
print("test passed")

Now a teammate wants to “simplify” the filter. They write != "pending" instead of == "paid". Run the same test again:

def paid_total(orders):
    total = 0
    for order in orders:
        if order["status"] != "pending":
            total += order["amount"]
    return total

test_paid_total()

Without the test, the job runs with no error and the report is wrong.

What you will learn

1. How to test a function, including odd inputs.
2. How to run tests with a tool called pytest, and how to fake a clock or an API.
3. How to find a bug: read an error message, and stop a program to look inside.
4. How to write counts to a log, and keep secrets out of it.
5. How to check that row counts add up.

A unit test checks one small function. The easiest function to test is a pure one.

A pure function gives the same output for the same input. It reads no file, calls no API and asks no clock.

paid_total is pure. You can test it with a few rows in a list. No file or database is needed.

Write one test for each rule you care about. Give each test a name that says the rule.

The tests use the first version of paid_total, with == "paid". It is in transforms.py.

The six shop orders are in shop.py. The test file is test_unit.py:

from shop import orders
from transforms import paid_total

def test_paid_total_of_shop():
    assert paid_total(orders) == 160.5

def test_paid_total_of_two_rows():
    assert paid_total(orders[:2]) == 65.0

def test_refunded_and_pending_not_counted():
    assert paid_total([orders[2], orders[4]]) == 0

Read each test as a rule:

• The whole shop has 160.5 in paid orders.
• Orders 101 and 102 give 40.0 + 25.0 = 65.0.
• Order 103 is refunded and order 105 is pending. Neither adds anything, so the total is 0.

You ran pytest in the validation chapter. Now three new tools: parametrize, -v, and reading a failure.

parametrize runs one test many times, once for each set of values you list. You write the test once and add one line for each case.

The line that starts with @ is a decorator. It sits above a function and changes how it runs. Here the list holds three pairs, so pytest runs the test three times:

import pytest
from shop import orders
from transforms import paid_total

def test_paid_total_of_shop():
    assert paid_total(orders) == 160.5

def test_empty_input_gives_zero():
    assert paid_total([]) == 0

@pytest.mark.parametrize(
    "status, expected",
    [("paid", 40.0), ("refunded", 0), ("pending", 0)],
)
def test_only_paid_counts(status, expected):
    order = dict(orders[0], status=status)
    assert paid_total([order]) == expected

Run pytest -v. The -v means verbose: it shows one line for each test. These are the lines, without the header:

pytest -v

Now make a test fail on purpose. In the parametrize list, change ("refunded", 0) to ("refunded", 60.0). Run pytest -q (quiet: short output):

pytest -q

An edge case is an unusual input. It is where code breaks first.

You met the habit in the first SQL chapters. Ask what happens with a missing value, with nothing, and with the same row twice.

Ask the same three questions about a Python function. We test split_orders, the quarantine step from the first Python chapter. It sends each order to a good list or a rejected list.

It uses check_order, which raises an error when a rule breaks. This version also rejects an order_id that it has already seen:

REQUIRED = ["order_id", "customer_id", "status", "amount", "order_date"]

def check_order(order):
    missing = [field for field in REQUIRED if field not in order]
    if missing:
        raise ValueError(f"order {order.get('order_id')}: missing {missing}")
    if order["status"] == "paid" and order["amount"] is None:
        raise ValueError(f"order {order['order_id']}: paid order has no amount")

def split_orders(rows):
    good = []
    rejected = []
    seen = set()
    for row in rows:
        try:
            check_order(row)
            if row["order_id"] in seen:
                raise ValueError(f"order {row['order_id']}: duplicate order_id")
            seen.add(row["order_id"])
            good.append(row)
        except ValueError as error:
            rejected.append({"order_id": row.get("order_id"), "reason": str(error)})
    return good, rejected

Each rule gets one test. A rule with no test is a gap. This table lists the rules and their tests:

Here are the four tests. Order 107 is the new bad order: it is paid and has no amount. The file is test_edge.py:

from shop import orders
from s3 import split_orders

def test_pending_order_may_have_no_amount():
    good, rejected = split_orders([orders[4]])
    assert len(good) == 1 and rejected == []

def test_paid_order_without_amount_is_rejected():
    bad = dict(orders[0], order_id=107, amount=None)
    good, rejected = split_orders([bad])
    assert good == []
    assert rejected == [
        {"order_id": 107, "reason": "order 107: paid order has no amount"}
    ]

def test_empty_input_gives_two_empty_lists():
    assert split_orders([]) == ([], [])

def test_duplicate_order_id_is_rejected():
    good, rejected = split_orders([orders[0], orders[0]])
    assert len(good) == 1
    assert rejected[0]["reason"] == "order 101: duplicate order_id"

Optional: read this when many tests need the same input. A fixture is a function that builds that input for a test.

You mark it with @pytest.fixture. A test asks for it by writing its name as an argument. pytest calls the fixture first and passes in the result:

import pytest
from shop import orders
from transforms import paid_total

@pytest.fixture
def shop_orders():
    return [dict(order) for order in orders]

def test_change_a_copy(shop_orders):
    shop_orders[0]["status"] = "refunded"
    assert paid_total(shop_orders) == 120.5

def test_next_test_gets_a_fresh_copy(shop_orders):
    assert paid_total(shop_orders) == 160.5

The copy matters. Here the tests use the shared orders list directly, with no copy:

def test_change_shared_list():
    orders[0]["status"] = "refunded"
    assert paid_total(orders) == 120.5

def test_next_test():
    assert paid_total(orders) == 160.5

Some code depends on something outside your control: the clock or an API. A real clock changes every day. A real API can be slow or down.

A test should give the same result every time. So you replace the outside thing with a fake.

In the APIs chapter you wrote fake functions. Do the same here.

Passing the date in is the simplest fake. An argument is a value you give to a function. Here the caller gives today’s date as an argument:

from datetime import date

def is_late(order, today):
    age = (today - date.fromisoformat(order["order_date"])).days
    return order["status"] == "pending" and age > 7

print(is_late(orders[4], date(2026, 1, 25)))
print(is_late(orders[4], date(2026, 1, 27)))
print(is_late(orders[4], date(2026, 1, 28)))
print(is_late(orders[0], date(2026, 1, 30)))

The two middle lines test the boundary: day 7 and day 8. A boundary is where the answer changes. The test sets the date, so it never depends on today.

For an API, pass in the function that makes the call. In a test, pass a plain fake function. It returns fixed rows, or it raises an error:

def paid_ids(fetch):
    rows = fetch("/orders")
    return [row["order_id"] for row in rows if row["status"] == "paid"]

def fake_fetch(path):
    return orders[:3]

def slow_fetch(path):
    raise TimeoutError("api slow")

print(paid_ids(fake_fetch))
try:
    paid_ids(slow_fetch)
except TimeoutError as error:
    print("caught:", error)

Python also has unittest.mock, a ready-made tool for fakes. The plain fakes above are enough to start.

Do not fake your own pure functions. Fake only what is outside your code: the clock, files, APIs and databases.

Optional: read this when a function returns many values. A golden output is a saved, correct result for a small input. The test runs the code and compares the whole result with the saved one.

One comparison replaces many asserts. The function totals the paid amounts for each day. This is the saved file expected_daily.json:

{
  "2026-01-03": 40.0,
  "2026-01-10": 25.0,
  "2026-01-15": 15.5,
  "2026-01-21": 80.0
}

The test loads the file and compares it with the result of the function:

import json

def daily_paid(orders):
    totals = {}
    for order in orders:
        if order["status"] == "paid":
            day = order["order_date"]
            totals[day] = totals.get(day, 0) + order["amount"]
    return totals

with open("expected_daily.json") as file:
    expected = json.load(file)

print(daily_paid(orders) == expected)

Now a teammate cuts the date to the month, with order_date[:7]. The comparison fails. To see where, list the entries that are missing, extra or different:

def daily_paid(orders):
    totals = {}
    for order in orders:
        if order["status"] == "paid":
            day = order["order_date"][:7]
            totals[day] = totals.get(day, 0) + order["amount"]
    return totals

actual = daily_paid(orders)
print(actual == expected)
print("missing:", sorted(set(expected) - set(actual)))
print("extra:", sorted(set(actual) - set(expected)))
print("different:", sorted(d for d in expected if d in actual and expected[d] != actual[d]))

Keep the input small, like these six orders. A big golden file is hard to read, and people stop checking why it changed.

When Python stops with an error, it prints a traceback. A traceback lists the function calls that led to the error.

Here a loader reads lines from a file. Order 105 has an empty amount at the end of its line:

def parse_row(line):
    order_id, status, amount = line.split(",")
    return {
        "order_id": int(order_id),
        "status": status,
        "amount": float(amount),
    }

def load(lines):
    return [parse_row(line) for line in lines]

lines = ["101,paid,40.0", "105,pending,"]
print(load(lines))

The first line says “most recent call last”. So the call that failed is at the bottom. Read it from the bottom:

1. The last line is the error: ValueError, could not convert string to float: ''. An empty text cannot become a number.
2. The lines above it show where the error happened: inside parse_row, at float(amount).
3. The lines above that show who called it: load called parse_row, and the main script called load.

The fix is not in float. The row is the problem: order 105 has no amount. Check the row before you convert it, or reject the row with a reason, as in the first Python chapter.

You have two simple tools to see a value inside running code.

print() shows a value and keeps going. breakpoint() stops the program, so you can look around.

Use print when you know which value to look at. It is quick, but you must edit the code and run it again.

Use breakpoint() when you do not know what to look at. It opens the Python debugger, called pdb. It shows a prompt, (Pdb), where you type commands.

Put the call in parse_row, after the split line, and run the script. Type the commands after (Pdb):

def parse_row(line):
    order_id, status, amount = line.split(",")
    breakpoint()
    return {
        "order_id": int(order_id),
        "status": status,
        "amount": float(amount),
    }

for line in ["101,paid,40.0", "105,pending,"]:
    print(parse_row(line))

Optional: the validation chapter showed that read = kept + rejected. Here you use that check to find which step lost rows.

Every row that comes in must be either kept or rejected. None may vanish. This uses split_orders from the edge-case section, with the two bad orders 107 and 108:

bad_paid = dict(orders[0], order_id=107, amount=None)
incomplete = {"order_id": 108, "status": "paid"}
raw = orders + [bad_paid, incomplete]
good, rejected = split_orders(raw)

def reconcile(read, kept, rejected):
    if read != kept + rejected:
        raise ValueError(
            f"read {read}, but kept {kept} + rejected {rejected} = {kept + rejected}"
        )
    print(f"balanced: read {read} = kept {kept} + rejected {rejected}")

reconcile(len(raw), len(good), len(rejected))

Now see a leak. Two steps drop rows, and neither records what it dropped. Count after each step to see which one lost rows:

with_date = [row for row in raw if "order_date" in row]
with_amount = [row for row in with_date if row["amount"] is not None]

print("read:", len(raw))
print("after date check:", len(with_date))
print("after amount check:", len(with_amount))

The same check works between two systems. Say the source sent 8 orders.

The target table holds 5, and the rejected list holds 2. Sets help you find the missing id:

source_ids = {101, 102, 103, 104, 105, 106, 107, 108}
target_ids = {101, 102, 103, 105, 106}
rejected_ids = {107, 108}

print("source:", len(source_ids))
print("target + rejected:", len(target_ids) + len(rejected_ids))
print("missing:", sorted(source_ids - target_ids - rejected_ids))
print("extra:", sorted(target_ids - source_ids))

Compare the counts first, and look at single rows second. Counts show which step lost the rows. Then you only read the rows around that step.

A log is a record of what your program did while it ran. When a job fails at night, the log is all you have.

print() is too simple for this. It has no level, no time and no way to turn it off. Python’s logging module fixes this.

Every log line has a level. The level says how serious the line is. You set one level, and Python hides everything below it.

A logger is the object you call to write log lines. logging.getLogger(__name__) makes one named after the current file. When you run a file directly, that name is __main__. Set the level to INFO and send four lines:

import logging
import sys

logging.basicConfig(
    level=logging.INFO,
    format="%(levelname)s %(name)s %(message)s",
    stream=sys.stdout,
)
log = logging.getLogger(__name__)

log.debug("checking order 101")
log.info("start run")
log.warning("order 107 rejected")
log.error("could not write the output file")
print("done")

basicConfig sets up logging once, at the start of the program. It has three settings here:

• level=logging.INFO shows INFO and everything above it.
• format is the layout of a line. %(levelname)s is the level, %(name)s is the logger name, and %(message)s is your text.
• stream=sys.stdout sends the lines to the normal output, so they show here. By default Python uses the error stream.

If you do not set a level, Python shows WARNING and above. Run the same lines without level=logging.INFO, and only the warning appears:

import logging
import sys

logging.basicConfig(
    format="%(levelname)s %(name)s %(message)s",
    stream=sys.stdout,
)
log = logging.getLogger(__name__)

log.info("start run")
log.warning("order 107 rejected")

The most useful log line for a pipeline is the counts. It says how many rows were read, kept and rejected.

Write the counts as name=value pairs. A person can read them fast, and a tool can search them.

Log only the first 5 rejected rows, with id and reason. Then say how many there were in all. This uses the INFO logger from the first logging example:

bad_paid = dict(orders[0], order_id=107, amount=None)
incomplete = {"order_id": 108, "status": "paid"}
raw = orders + [bad_paid, incomplete]
good, rejected = split_orders(raw)

log.info("read=%d kept=%d rejected=%d", len(raw), len(good), len(rejected))
for item in rejected[:5]:
    log.warning("rejected order_id=%s reason=%s", item["order_id"], item["reason"])
if len(rejected) > 5:
    log.warning("showing 5 of %d rejected rows", len(rejected))

The log has the id and the reason, but not the whole row. A whole row can hold personal data. The redact section shows how to handle that.

The INFO line answers three questions: how big was the input, how much was kept, and did the numbers balance. Compare it with yesterday’s line. A sudden jump in rejected rows is then easy to see.

Optional: the validation chapter showed manifests. A run manifest is a small summary of one run. It is the receipt: what came in and what went out.

It holds a run id and the counts. It also holds the answer to the reconcile check. Here it uses the setup lines of the earlier sections:

import json

bad_paid = dict(orders[0], order_id=107, amount=None)
incomplete = {"order_id": 108, "status": "paid"}
raw = orders + [bad_paid, incomplete]
good, rejected = split_orders(raw)

manifest = {
    "run_id": "shop-2026-01-22",
    "read": len(raw),
    "kept": len(good),
    "rejected": len(rejected),
    "balanced": len(raw) == len(good) + len(rejected),
}
print(json.dumps(manifest))

A test can read the manifest too. If balanced is false, the run is not safe to trust.

Optional: read this after the counts section. A metric is a number you track on every run. A job can finish with no error and still be unhealthy.

One good metric is the share of rows that were rejected. Set a limit, and warn when a run goes over it.

The if raw else 0.0 part guards empty input. Without it, a run with no rows would divide by zero:

bad_paid = dict(orders[0], order_id=107, amount=None)
incomplete = {"order_id": 108, "status": "paid"}
raw = orders + [bad_paid, incomplete]
good, rejected = split_orders(raw)

share = len(rejected) / len(raw) if raw else 0.0
print(f"rejected_share={share:.2f}")
if share > 0.10:
    print("ALERT: more than 10% rejected")

Pick the limit from what is normal for the source. A share that is usually 1% and jumps to 25% means something changed in an earlier step.

Logs are copied, searched and kept for a long time. Many people can read them. So a log must never hold a secret or personal data.

A secret is a password, a token or a key. Personal data is anything about a person, such as a name or an email address.

To redact a record means to hide the sensitive values and keep the keys. The function redact returns a copy of a record with each sensitive value replaced by ***.

This customer row is new data for this section. It has two personal fields and one secret:

customer = {
    "customer_id": 1,
    "name": "Asha",
    "email": "asha@example.com",
    "api_token": "tok_9f2c71",
}

SENSITIVE = {"name", "email", "api_token"}

def redact(record):
    return {
        key: "***" if key in SENSITIVE else value
        for key, value in record.items()
    }

log.info("loaded customer %s", redact(customer))
print(customer["email"] == "asha@example.com")

A list of sensitive names misses any new column. An allow-list is safer: a list of fields you may log, with everything else hidden. Review the list when the data changes.

Interviewers often ask what you test in a pipeline. Here is a short answer:

You learned to test a function, read a traceback, log counts and hide secrets. Each exercise opens in the Python workspace with its own data.

Before you run each exercise, write down one thing you expect to go wrong.