Capstone Phase 5 to 7: Storage, Reporting, and the Test Suite

By the end of this lesson you can store the validated batch so a second run is safe, produce the report a reviewer actually receives, and cover the whole pipeline with tests in all four categories including a regression.

These three phases turn a working script into something somebody else could run. Storage can preserve yesterday's shortlist; reporting brings results to a person who will not run the code; tests make tomorrow's change safer to review.

Phase 6 is also where the project either keeps its promises or quietly stops. The whole point of phases 3 and 4 was that verdicts carry reasons and figures carry denominators. A report that prints five rows and a total, dropping the reasons and the exclusions, throws away the work.

The tests are what let you finish. Without them the last three phases are a slow, anxious sequence of manual checks, repeated every time anything changes.

Storage. A small normalised schema: an agencies table seeded from the CSV fixture, and a notices table with the notice id as primary key and a foreign key to the agency. Constraints carry real weight here. NOT NULL on the agency code and the foreign key give you a second, independent detection of the unrecognised agency your normalisation already reported, which is a good property rather than duplicated work.

The load is an upsert inside one transaction, so a second run refreshes rather than duplicates, and a failure partway through rolls back. Schema changes go in numbered forward-only migration files. Values supplied by records and queries bind as parameters; the fixed schema seed literals below are illustrative constants.

Reporting. Two outputs.

The CSV is what a reviewer opens in a spreadsheet: written with csv.DictWriter so a title containing a comma round-trips, named columns, one row per shortlisted notice. The expected result is stored as a fixture, so the test compares against a file rather than against a description.

The Markdown summary is what a reviewer reads first, and every figure states what it was computed from: records read, records normalised, records excluded and why, records matched, and, for any average, its denominator. A summary that cannot be reconciled to the CSV is a defect.

Tests, in the four categories from Module 5, over this pipeline:

• Success: the fixture batch produces exactly the expected shortlist.
• Boundary: a record worth exactly the minimum qualifies; one unit below does not.
• Malformed: each defect in the malformed fixture produces an error naming its field, and both the clean-plus-errors count and per-input identity checks hold.
• Regression: the exact input from a defect you diagnosed and repaired.

Two properties the whole suite must have. It runs with no network access or project-specific credentials and configuration variables, which makes the fixture tests reproducible. And it does not edit an existing assertion merely to make a change pass; a failing test is reporting something.

Give every assertion a message naming the behaviour it protects. When one fails months from now, that sentence is the entire diagnosis.

import csv
import io
import sqlite3

connection = sqlite3.connect(":memory:")
connection.execute("PRAGMA foreign_keys = ON")
connection.executescript(
    """
    CREATE TABLE agencies (agency_code TEXT PRIMARY KEY, name TEXT NOT NULL);
    CREATE TABLE notices (
        notice_id TEXT PRIMARY KEY,
        agency_code TEXT NOT NULL REFERENCES agencies(agency_code),
        title TEXT NOT NULL,
        amount INTEGER NOT NULL
    );
    INSERT INTO agencies VALUES ('GSA', 'General Services Administration');
    INSERT INTO agencies VALUES ('EPA', 'Environmental Protection Agency');
    """
)

UPSERT = (
    "INSERT INTO notices (notice_id, agency_code, title, amount) VALUES (?, ?, ?, ?)"
    " ON CONFLICT(notice_id) DO UPDATE SET"
    " agency_code = excluded.agency_code, title = excluded.title, amount = excluded.amount"
)


def load(rows):
    with connection:
        connection.executemany(UPSERT, rows)
    return len(rows)


def shortlist(minimum):
    return connection.execute(
        "SELECT n.notice_id, n.title, a.name, n.amount FROM notices n"
        " JOIN agencies a ON a.agency_code = n.agency_code"
        " WHERE n.amount >= ? ORDER BY n.amount DESC",
        (minimum,),
    ).fetchall()


def to_csv(rows):
    out = io.StringIO()
    writer = csv.DictWriter(out, fieldnames=["notice_id", "title", "agency_name", "amount"])
    writer.writeheader()
    for notice_id, title, agency_name, amount in rows:
        writer.writerow(
            {"notice_id": notice_id, "title": title, "agency_name": agency_name, "amount": amount}
        )
    return out.getvalue()


BATCH = [
    ("SYN-0001", "GSA", "Grounds Maintenance, Building 7", 310000),
    ("SYN-0011", "EPA", "Groundwater Monitoring Wells", 640000),
    ("SYN-0006", "GSA", "Snow Removal", 42000),
]

print(load(BATCH), load(BATCH))
print(connection.execute("SELECT COUNT(*) FROM notices").fetchone()[0])
print(to_csv(shortlist(100000)).splitlines()[2])
print(len(shortlist(100000)), "of", connection.execute("SELECT COUNT(*) FROM notices").fetchone()[0])

The load runs twice on purpose. The printed count shows no duplicate rows after a second run; the upsert statement also updates existing values. The full phase 5 test must assert both the stable row count and refreshed values, rather than treating this printout as proof of both.

shortlist joins to agencies so the report carries the readable name rather than the code, and the join is an inner one, which is correct here: a notice whose agency code is unrecognised was already rejected during normalisation, and the foreign key would have refused it anyway.

to_csv uses DictWriter, so the title containing a comma is quoted correctly on the way out.

Predict all four printed lines.

3 3
3
SYN-0001,"Grounds Maintenance, Building 7",General Services Administration,310000
2 of 3

Change to_csv to build each line with ",".join(str(value) for value in row) instead of using DictWriter. Predict the third printed line, and what a consumer sees.

SYN-0001,Grounds Maintenance, Building 7,General Services Administration,310000

An assistant was asked for the summary section of the report. It produced this.

def summary(shortlisted, all_records, errors):
    average = sum(r["amount"] for r in shortlisted) / len(shortlisted)
    return (
        f"# Opportunity Review\n\n"
        f"- {len(shortlisted)} opportunities matched\n"
        f"- average value {average:,.0f}\n"
        f"- {len(shortlisted) / len(all_records):.0%} of the pipeline\n"
    )


print(summary([], [], []))

def summary(shortlisted, normalised, errors):
    lines = [
        "# Opportunity Review",
        "",
        f"- {len(normalised)} records normalised, {len(errors)} validation errors",
        f"- {len(shortlisted)} of {len(normalised)} met the worked amount minimum",
    ]
    amounts = [r["amount"] for r in shortlisted if r.get("amount") is not None]
    if amounts:
        lines.append(f"- average value for amount-minimum rows {sum(amounts) / len(amounts):,.0f} over {len(amounts)} records")
    else:
        lines.append("- no amount-minimum rows with a published value")
    lines.extend(f"- validation error: {error}" for error in errors)
    return "\n".join(lines)

Four assertions, one per category. One of them fails. Read the message, repair the defect, and confirm the others still hold.

Loading this exercise…

Optional, and it completes phases 5 to 7. Build the real schema with its migration file, load the twelve-record fixture batch, generate the CSV, and compare it against capstone/fixtures/expected-shortlist.csv row for row. Then write the four-category suite over the real pipeline, seed one defect of your own by changing a single comparison, confirm which test fails, repair it, and keep the test as your regression case. Record which defect you seeded and which test caught it; the retrospective in phase 11 asks for it.

Sign in to track your progress on this exercise.

1. Why does an upsert matter more than a plain insert for the capstone's load?
2. Why store the expected CSV as a fixture rather than describing it in a test?
3. Name the two properties the whole suite must have.
4. What must a summary figure always state?

Sign in to track your progress on this exercise.

Store with constraints and an upsert inside a transaction, report with the CSV module and a summary whose every figure names its base, and cover the pipeline with success, boundary, malformed, and regression cases that run offline. Next: reading the API contract behind the fixtures, the read-only interface, one reviewed change, and the documents that hand the project to somebody else.