Tests and Data Quality Monitoring

Introduction

How do you know your analysis is right? Looking at a chart is not enough. Software engineers use tests, small programs that check code behaves correctly. Data work needs two kinds: tests for your code, and checks on your data.

Unit tests for code

A unit test runs a small function on a tiny input where you know the answer. Suppose we write a function that converts a pair of dates into loan days:

import pandas as pd

def loan_days(loaned_on, returned_on):
    return (returned_on - loaned_on).dt.days

A test with simple, hand-checkable values:

def test_loan_days():
    loaned = pd.to_datetime(pd.Series(["2024-01-01", "2024-02-10"]))
    returned = pd.to_datetime(pd.Series(["2024-01-11", "2024-02-10"]))
    result = loan_days(loaned, returned)
    assert result.tolist() == [10, 0]

Ten days from 1 to 11 January, and zero for same-day returns. Run tests with pytest: put the function in a file named test_something.py and run pytest. The assert fails the test if the statement is false.

Test the edges, not just the easy case:

• A missing return date should give a missing result, not an error.
• Empty input.
• A single row.
• Unusual values, such as a leap day.

Testing an aggregation

def test_loans_by_genre():
    df = pd.DataFrame({"id": [1, 2, 3], "genre": ["Crime", "Crime", "Food"]})
    counts = df.groupby("genre")["id"].count()
    assert counts["Crime"] == 2
    assert counts["Food"] == 1

Also test invariants: facts that should always hold, such as the by-city counts adding up to the total loans. The 18 + 11 + 1 = 30 check in the library data is exactly this.

Data quality checks

Code tests use made-up data. Data checks run on the real data each time it loads, because new data can break old assumptions:

def check_loans(df):
    problems = []
    if not df["id"].is_unique:
        problems.append("duplicate loan ids")
    if df["loaned_on"].isna().any():
        problems.append("missing loan dates")
    if (df["returned_on"] < df["loaned_on"]).any():
        problems.append("return before loan")
    if len(df) == 0:
        problems.append("no rows")
    return problems

problems = check_loans(df)
print(problems if problems else "all checks passed")

all checks passed

Decide, for each check, whether a failure should stop the pipeline (for serious problems) or only warn. Report all the problems at once, so people can fix them together.

Monitoring over time

If the pipeline runs regularly, watch the results as well as the code:

• Row counts: did today's load shrink or double unexpectedly?
• Missing values: is the share of empty values rising?
• Ranges and distributions: are loan lengths suddenly very different?
• Freshness: is the newest date recent, or has the feed stopped?
• Key totals: compare with last period and with other sources.

Drift means the data changes in character over time: a new category appears, a column changes meaning, or a system starts recording differently. Compare each load to a baseline, and investigate big changes before trusting the results.

Keep a record

• Save check results with timestamps.
• Alert the owner when checks fail.
• Fix the cause, add a test for it and note it in a log.

Try it yourself

Write three unit tests for functions in your pipeline, including one edge case. Add a data check that fails if the number of rows falls by more than half compared with a saved previous count.

Summary

• Unit tests run small functions on tiny hand-checked inputs, including edge cases and invariants.
• Data checks run on every load to catch duplicates, missing values and impossible values.
• Monitor row counts, missing values, ranges and freshness over time, and investigate drift.