|
| 1 | +#!/usr/bin/env python3 |
| 2 | +"""Standalone TDD ledger backed by SQLite. |
| 3 | +
|
| 4 | +Drop this file into any repo and run `python3 tdd_ledger.py --help`. |
| 5 | +The ledger enforces a red->green workflow by recording failing tests first. |
| 6 | +""" |
| 7 | + |
| 8 | +from __future__ import annotations |
| 9 | + |
| 10 | +import argparse |
| 11 | +import os |
| 12 | +import sqlite3 |
| 13 | +import subprocess |
| 14 | +import sys |
| 15 | +from typing import Iterable, Optional |
| 16 | + |
| 17 | + |
| 18 | +def parse_args() -> argparse.Namespace: |
| 19 | + parser = argparse.ArgumentParser(description="TDD ledger backed by SQLite.") |
| 20 | + parser.add_argument( |
| 21 | + "--db", |
| 22 | + default=os.environ.get("TDD_LEDGER_DB", ".tdd_ledger.sqlite"), |
| 23 | + help="Path to sqlite db (default: .tdd_ledger.sqlite or TDD_LEDGER_DB).", |
| 24 | + ) |
| 25 | + sub = parser.add_subparsers(dest="command", required=True) |
| 26 | + |
| 27 | + sub.add_parser("init", help="Initialize the ledger database.") |
| 28 | + |
| 29 | + red = sub.add_parser("red", help="Record a failing test.") |
| 30 | + red.add_argument("test", help="Test name or identifier.") |
| 31 | + red.add_argument("--note", default=None, help="Optional note.") |
| 32 | + red.add_argument("--run", nargs=argparse.REMAINDER, help="Command to run (expects failure).") |
| 33 | + |
| 34 | + green = sub.add_parser("green", help="Record a passing test.") |
| 35 | + green.add_argument("test", help="Test name or identifier.") |
| 36 | + green.add_argument("--note", default=None, help="Optional note.") |
| 37 | + green.add_argument("--run", nargs=argparse.REMAINDER, help="Command to run (expects success).") |
| 38 | + |
| 39 | + sub.add_parser("require-green", help="Exit non-zero if any tests are still red.") |
| 40 | + sub.add_parser("status", help="Print status summary.") |
| 41 | + |
| 42 | + return parser.parse_args() |
| 43 | + |
| 44 | + |
| 45 | +def connect(db_path: str) -> sqlite3.Connection: |
| 46 | + conn = sqlite3.connect(db_path) |
| 47 | + conn.execute("PRAGMA journal_mode = WAL;") |
| 48 | + conn.execute("PRAGMA foreign_keys = ON;") |
| 49 | + return conn |
| 50 | + |
| 51 | + |
| 52 | +def init_db(conn: sqlite3.Connection) -> None: |
| 53 | + conn.execute( |
| 54 | + """ |
| 55 | + CREATE TABLE IF NOT EXISTS ledger ( |
| 56 | + test_name TEXT PRIMARY KEY, |
| 57 | + status TEXT NOT NULL CHECK(status IN ('red','green')), |
| 58 | + note TEXT, |
| 59 | + last_command TEXT, |
| 60 | + last_exit_code INTEGER, |
| 61 | + created_at TEXT NOT NULL DEFAULT (datetime('now')), |
| 62 | + updated_at TEXT NOT NULL DEFAULT (datetime('now')) |
| 63 | + ); |
| 64 | + """ |
| 65 | + ) |
| 66 | + conn.execute( |
| 67 | + """ |
| 68 | + CREATE TABLE IF NOT EXISTS events ( |
| 69 | + id INTEGER PRIMARY KEY AUTOINCREMENT, |
| 70 | + test_name TEXT NOT NULL, |
| 71 | + status TEXT NOT NULL CHECK(status IN ('red','green')), |
| 72 | + note TEXT, |
| 73 | + command TEXT, |
| 74 | + exit_code INTEGER, |
| 75 | + created_at TEXT NOT NULL DEFAULT (datetime('now')) |
| 76 | + ); |
| 77 | + """ |
| 78 | + ) |
| 79 | + conn.commit() |
| 80 | + |
| 81 | + |
| 82 | +def normalize_run_args(run_args: Optional[Iterable[str]]) -> Optional[list[str]]: |
| 83 | + if not run_args: |
| 84 | + return None |
| 85 | + args = list(run_args) |
| 86 | + if args and args[0] == "--": |
| 87 | + args = args[1:] |
| 88 | + if not args: |
| 89 | + return None |
| 90 | + return args |
| 91 | + |
| 92 | + |
| 93 | +def run_command(args: list[str]) -> int: |
| 94 | + proc = subprocess.run(args, check=False) |
| 95 | + return proc.returncode |
| 96 | + |
| 97 | + |
| 98 | +def upsert_ledger( |
| 99 | + conn: sqlite3.Connection, |
| 100 | + test: str, |
| 101 | + status: str, |
| 102 | + note: Optional[str], |
| 103 | + command: Optional[str], |
| 104 | + exit_code: Optional[int], |
| 105 | +) -> None: |
| 106 | + conn.execute( |
| 107 | + """ |
| 108 | + INSERT INTO ledger (test_name, status, note, last_command, last_exit_code) |
| 109 | + VALUES (?, ?, ?, ?, ?) |
| 110 | + ON CONFLICT(test_name) DO UPDATE SET |
| 111 | + status=excluded.status, |
| 112 | + note=excluded.note, |
| 113 | + last_command=excluded.last_command, |
| 114 | + last_exit_code=excluded.last_exit_code, |
| 115 | + updated_at=datetime('now'); |
| 116 | + """, |
| 117 | + (test, status, note, command, exit_code), |
| 118 | + ) |
| 119 | + conn.execute( |
| 120 | + """ |
| 121 | + INSERT INTO events (test_name, status, note, command, exit_code) |
| 122 | + VALUES (?, ?, ?, ?, ?); |
| 123 | + """, |
| 124 | + (test, status, note, command, exit_code), |
| 125 | + ) |
| 126 | + conn.commit() |
| 127 | + |
| 128 | + |
| 129 | +def cmd_red(conn: sqlite3.Connection, args: argparse.Namespace) -> int: |
| 130 | + run_args = normalize_run_args(args.run) |
| 131 | + exit_code: Optional[int] = None |
| 132 | + cmd_str: Optional[str] = None |
| 133 | + if run_args: |
| 134 | + cmd_str = " ".join(run_args) |
| 135 | + exit_code = run_command(run_args) |
| 136 | + if exit_code == 0: |
| 137 | + print("expected failing test, but command exited 0", file=sys.stderr) |
| 138 | + return 2 |
| 139 | + upsert_ledger(conn, args.test, "red", args.note, cmd_str, exit_code) |
| 140 | + return 0 |
| 141 | + |
| 142 | + |
| 143 | +def cmd_green(conn: sqlite3.Connection, args: argparse.Namespace) -> int: |
| 144 | + run_args = normalize_run_args(args.run) |
| 145 | + exit_code: Optional[int] = None |
| 146 | + cmd_str: Optional[str] = None |
| 147 | + if run_args: |
| 148 | + cmd_str = " ".join(run_args) |
| 149 | + exit_code = run_command(run_args) |
| 150 | + if exit_code != 0: |
| 151 | + print("expected passing test, but command failed", file=sys.stderr) |
| 152 | + return 2 |
| 153 | + upsert_ledger(conn, args.test, "green", args.note, cmd_str, exit_code) |
| 154 | + return 0 |
| 155 | + |
| 156 | + |
| 157 | +def cmd_require_green(conn: sqlite3.Connection) -> int: |
| 158 | + rows = conn.execute( |
| 159 | + "SELECT test_name FROM ledger WHERE status = 'red' ORDER BY test_name;" |
| 160 | + ).fetchall() |
| 161 | + if rows: |
| 162 | + print("red tests remain:", file=sys.stderr) |
| 163 | + for (name,) in rows: |
| 164 | + print(f"- {name}", file=sys.stderr) |
| 165 | + return 1 |
| 166 | + return 0 |
| 167 | + |
| 168 | + |
| 169 | +def cmd_status(conn: sqlite3.Connection) -> int: |
| 170 | + rows = conn.execute( |
| 171 | + "SELECT status, COUNT(*) FROM ledger GROUP BY status ORDER BY status;" |
| 172 | + ).fetchall() |
| 173 | + counts = {status: count for status, count in rows} |
| 174 | + red = counts.get("red", 0) |
| 175 | + green = counts.get("green", 0) |
| 176 | + print(f"red={red} green={green}") |
| 177 | + return 0 |
| 178 | + |
| 179 | + |
| 180 | +def main() -> int: |
| 181 | + args = parse_args() |
| 182 | + conn = connect(args.db) |
| 183 | + init_db(conn) |
| 184 | + |
| 185 | + if args.command == "init": |
| 186 | + return 0 |
| 187 | + if args.command == "red": |
| 188 | + return cmd_red(conn, args) |
| 189 | + if args.command == "green": |
| 190 | + return cmd_green(conn, args) |
| 191 | + if args.command == "require-green": |
| 192 | + return cmd_require_green(conn) |
| 193 | + if args.command == "status": |
| 194 | + return cmd_status(conn) |
| 195 | + |
| 196 | + print("unknown command", file=sys.stderr) |
| 197 | + return 2 |
| 198 | + |
| 199 | + |
| 200 | +if __name__ == "__main__": |
| 201 | + raise SystemExit(main()) |
0 commit comments