07 — smoke_pipeline.py: end-to-end verification

Duration: 10 min Prerequisites: chapters 02–06 (you know what each module does).

Why a smoke test

When you change any of the 5 modules (ingest.py, normalize.py, db.py, categorize.py, dashboards.py), how do you know nothing broke?

Three options:

1. Run both apps and click around. Slow, flaky, requires Ollama or an API key.
2. A full unit-test suite. Worth doing for production — but heavy.
3. A smoke test: one script, no LLM, no Streamlit, runs in ~1 second, asserts known invariants.

smoke_pipeline.py is option 3. It’s the fast feedback loop during library refactoring.

The full script

csv-llm-shared/smoke_pipeline.py

"""End-to-end smoke test for csv-llm-shared.
Runs in ~1 second. No LLM, no Streamlit. Asserts known invariants.

Usage:
    python csv-llm-shared/smoke_pipeline.py

Exits with code 0 on success, non-zero on failure.
"""

import sys
from pathlib import Path

HERE = Path(__file__).parent
sys.path.insert(0, str(HERE))

import ingest
import normalize
import db

CSV_PATH = HERE.parent / "csv-llm-ollama" / "data" / "data1-anonymized.csv"
if not CSV_PATH.exists():
    CSV_PATH = HERE / "data" / "data1-anonymized.csv"


def check(name: str, actual, expected):
    if actual == expected:
        print(f"  OK  {name}: {actual}")
        return True
    print(f"  FAIL  {name}: expected {expected}, got {actual}")
    return False


def main() -> int:
    print(f"Loading CSV: {CSV_PATH}")
    df = ingest.read_csv(CSV_PATH)
    print(f"  ingested {len(df)} rows, columns: {df.columns.tolist()}")

    print("\nNormalizing...")
    nf = normalize.normalize(df, profile="data1")
    print(f"  normalized {len(nf)} rows, columns: {nf.columns.tolist()}")

    print("\nLoading SQLite...")
    sqlite_path = HERE / ".cache" / "smoke.sqlite"
    if sqlite_path.exists():
        sqlite_path.unlink()
    conn = db.open_db(str(sqlite_path))
    inserted = db.load_dataframe(conn, nf, profile="data1")
    print(f"  inserted {inserted} rows")

    print("\nRunning safe queries...")
    failures = 0

    # 1. Total row count
    r = db.safe_query(conn, "SELECT COUNT(*) AS n FROM transactions WHERE profile='data1'")
    failures += not check("row count", int(r.iloc[0]["n"]), 734)

    # 2. Expense vs credit count
    r = db.safe_query(conn,
        "SELECT COUNT(*) AS n FROM transactions WHERE profile='data1' AND amount > 0")
    failures += not check("expense count", int(r.iloc[0]["n"]), 675)

    r = db.safe_query(conn,
        "SELECT COUNT(*) AS n FROM transactions WHERE profile='data1' AND amount < 0")
    failures += not check("credit count", int(r.iloc[0]["n"]), 59)

    # 3. Total spend (rounded to 2 decimals)
    r = db.safe_query(conn,
        "SELECT ROUND(SUM(amount), 2) AS t FROM transactions WHERE profile='data1' AND amount > 0")
    failures += not check("total spend", float(r.iloc[0]["t"]), 118667.69)

    # 4. Top category
    r = db.safe_query(conn, """
        SELECT category, ROUND(SUM(amount), 2) AS t
        FROM transactions
        WHERE profile='data1' AND amount > 0
        GROUP BY category
        ORDER BY SUM(amount) DESC
        LIMIT 1
    """)
    failures += not check("top category name",   r.iloc[0]["category"], "Housing")
    failures += not check("top category amount", float(r.iloc[0]["t"]), 30895.29)

    # 5. Coastal Spoon Cafe count + total
    r = db.safe_query(conn, """
        SELECT COUNT(*) AS n, ROUND(SUM(amount), 2) AS t
        FROM transactions
        WHERE profile='data1' AND description='Coastal Spoon Cafe' AND amount > 0
    """)
    failures += not check("Coastal Spoon count", int(r.iloc[0]["n"]),   19)
    failures += not check("Coastal Spoon total", float(r.iloc[0]["t"]), 1221.34)

    # 6. safe_query rejects mutation
    try:
        db.safe_query(conn, "DELETE FROM transactions")
        print("  FAIL  safe_query did not reject DELETE")
        failures += 1
    except ValueError:
        print("  OK  safe_query rejected DELETE")

    conn.close()
    print()
    if failures == 0:
        print("All checks passed.")
        return 0
    print(f"{failures} check(s) failed.")
    return 1


if __name__ == "__main__":
    sys.exit(main())

What it asserts

#	Invariant	Expected
1	Row count	734
2	Expense count	675
3	Credit count	59
4	Total spend	$118,667.69
5	Top category	Housing, $30,895.29
6	Coastal Spoon Cafe Q3 ground truth	19 visits, $1,221.34
7	safe_query rejects DELETE	raises ValueError

If any of these breaks, you know immediately which module to look at:

• Wrong row count → ingest.py or normalize.py (dropped too many rows).
• Wrong total → _parse_amount regression.
• safe_query lets DELETE through → security regression in db.py.

Running it

python csv-llm-shared/smoke_pipeline.py

Expected output:

Loading CSV: csv-llm-ollama/data/data1-anonymized.csv
  ingested 734 rows, columns: ['date', 'card', 'description', 'category', 'debit', 'credit']

Normalizing...
  normalized 734 rows, columns: ['profile', 'date', 'card', 'description', 'category', 'amount']

Loading SQLite...
  inserted 734 rows

Running safe queries...
  OK  row count: 734
  OK  expense count: 675
  OK  credit count: 59
  OK  total spend: 118667.69
  OK  top category name: Housing
  OK  top category amount: 30895.29
  OK  Coastal Spoon count: 19
  OK  Coastal Spoon total: 1221.34
  OK  safe_query rejected DELETE

All checks passed.

Runtime: ~0.8 seconds on a typical laptop.

Wiring it into CI

A GitHub Actions workflow that catches regressions on every push:

.github/workflows/smoke.yml

name: smoke
on: [push, pull_request]
jobs:
  smoke:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: "3.11" }
      - run: pip install -r csv-llm-shared/requirements.txt
      - run: python csv-llm-shared/smoke_pipeline.py

3 minutes for the full job (most of it is pip install). 1 second of actual test.

What it doesn’t check

• The LLM agent loop (covered by csv-llm-ollama/smoke_chat.py separately).
• Plotly figure rendering (visual — would need snapshot tests).
• Streamlit UI (manual or Playwright).
• Performance (would need a separate benchmark script).

A smoke test is not a comprehensive test suite. It’s a fast canary that catches “you broke the basics”.

Adding new invariants

When you add a new module or feature, add 1–3 assertions here:

• New CSV format support? Add check("rbc csv row count", ...) after running ingest on a sample RBC file.
• New category rule? Assert a known merchant gets the expected category.
• New SQL pattern? Add the expected result.

The script grows linearly with the surface area of the library.

Takeaways

• ~1 second per run, no LLM, no UI — the fastest feedback loop.
• 7 invariants cover row count, totals, top category, ground-truth merchant, security guard.
• Wire it into CI; treat any failure as a release blocker.
• Add a new check every time you add a feature.

Next: Extending for a new CSV format →