#!/usr/bin/env python3 """Re-runs every check behind the post "When one page of a statement will not open". Two kinds of row live in dataset.json: git the quote must still appear in Evidilo's source at origin/main, read with `git show` and never from the working tree; lab twelve statements are generated again, one page of rows is removed from each, and the product's own balance check (packages/shared/src/chain.ts at origin/main, run by Node) is run over the rows that are left. The tally must equal the value in dataset.json. Exit 0 when every row holds, 1 otherwise. Needs git, Node 22.7 or later, and the repository. python3 check.py every row python3 check.py --recheck the same, but the lab is skipped if it runs past a minute python3 check.py --write rewrites the lab values in dataset.json and all of dataset.csv """ import csv import html import json import os import re import subprocess import sys import tempfile import unicodedata from datetime import date, timedelta from random import Random HERE = os.path.dirname(os.path.abspath(__file__)) DATASET = os.path.join(HERE, "dataset.json") CSV = os.path.join(HERE, "dataset.csv") REPO = os.environ.get("EVIDILO_REPO", os.path.normpath(os.path.join(HERE, "../../../../.."))) REF = os.environ.get("EVIDILO_REF", "origin/main") LAB_BUDGET_S = 60 SPACE = re.compile(r"\s+") QUOTES = str.maketrans({"‘": "'", "’": "'", "“": '"', "”": '"', " ": " "}) def normalise(text): text = html.unescape(text) text = unicodedata.normalize("NFKC", text).translate(QUOTES) return SPACE.sub(" ", text).strip() def normalise_code(text): # A comment sentence wraps over several `//` or ` *` lines; joining them lets one quote match. return normalise(re.sub(r"(?m)^\s*(?://|\*)\s?", " ", text)) def git_show(path): result = subprocess.run( ["git", "-C", REPO, "show", "%s:%s" % (REF, path)], capture_output=True, text=True ) if result.returncode != 0: raise RuntimeError("git show %s:%s failed: %s" % (REF, path, result.stderr.strip())) return result.stdout # --- the lab --------------------------------------------------------------------------------- OPENING = 250000 # 2,500.00 USD in cents AMOUNT_POOL = [-12500, -4300, -990, 125000, 2399, 75002, -22000, 890, 45000, -1500] PAGE_SIZES = [5, 6, 7, 8] * 3 PAGES = 3 RUNNER = """ import { readFileSync } from "node:fs"; import { checkBalanceChain } from "./chain.ts"; import { money } from "./money.ts"; const usd = (cents) => (cents === null ? null : money(BigInt(cents), "USD")); const cases = JSON.parse(readFileSync(process.argv[2], "utf8")); const verdicts = cases.map((c) => { const v = checkBalanceChain({ openingBalance: usd(c.opening), closingBalance: usd(c.closing), rows: c.rows.map(([amount, balance]) => ({ amount: usd(amount), runningBalance: usd(balance) })), }); return v.status === "balanced" ? { status: v.status, break: null, firstBrokenRow: null } : { status: v.status, break: v.break, firstBrokenRow: v.firstBrokenRow }; }); process.stdout.write(JSON.stringify(verdicts)); """ def business_days(start, count): days, day = [], start while len(days) < count: if day.weekday() < 5: days.append(day) day += timedelta(days=1) return days def build_statement(rows_per_page, rng): days = business_days(date(2026, 1, 2), rows_per_page * PAGES) rows, balance = [], OPENING for day in days: amount = rng.choice(AMOUNT_POOL) if balance + amount < 0: amount = abs(amount) + 1000 balance += amount rows.append({"date": day, "amount": amount, "balance": balance}) return rows def zero_net_middle_page(rows, rows_per_page, rng): left, right = rng.randint(1, 40) * 1000, rng.randint(1, 40) * 1000 pattern = ([left, -left, right, -right] + [0] * rows_per_page)[:rows_per_page] for row, amount in zip(rows[rows_per_page : 2 * rows_per_page], pattern): row["amount"] = amount balance = OPENING for row in rows: balance += row["amount"] row["balance"] = balance return rows def layout(name, prepare=None, opening=OPENING, drop_last_page=False, row_balances=True): """One case per statement: the middle page removed, or the last one when it holds the closing.""" rng = Random(20261026) cases = [] for rows_per_page in PAGE_SIZES: rows = build_statement(rows_per_page, rng) if prepare is not None: rows = prepare(rows, rows_per_page, rng) if name == "control": kept, closing = rows, rows[-1]["balance"] elif drop_last_page: kept, closing = rows[: 2 * rows_per_page], None else: kept, closing = rows[:rows_per_page] + rows[2 * rows_per_page :], rows[-1]["balance"] cases.append( { "layout": name, "opening": opening, "closing": closing, "rows": [[r["amount"], r["balance"] if row_balances else None] for r in kept], "firstAfterGap": rows_per_page + 1, "kept": len(kept), "gapDays": (rows[2 * rows_per_page]["date"] - rows[2 * rows_per_page - 1]["date"]).days, } ) return cases def run_chain(cases, timeout): """Runs the product's checkBalanceChain, exported from origin/main, over every case.""" with tempfile.TemporaryDirectory() as tmp: for name in ("chain", "money", "errors"): source = git_show("packages/shared/src/%s.ts" % name) # Node resolves the source's `./x.js` specifiers only when they name the .ts file. source = re.sub(r'from "(\./[a-z-]+)\.js"', r'from "\1.ts"', source) with open(os.path.join(tmp, name + ".ts"), "w") as handle: handle.write(source) with open(os.path.join(tmp, "run.mjs"), "w") as handle: handle.write(RUNNER) with open(os.path.join(tmp, "cases.json"), "w") as handle: json.dump(cases, handle) result = subprocess.run( ["node", "--experimental-transform-types", "--no-warnings", "run.mjs", "cases.json"], cwd=tmp, capture_output=True, text=True, timeout=timeout, ) if result.returncode != 0: raise RuntimeError("node failed: %s" % result.stderr.strip()) return json.loads(result.stdout) def run_lab(timeout): layouts = { "C0": layout("control"), "L1": layout("row-balances"), "L2": layout("no-row-balances", row_balances=False), "L3": layout("no-opening", opening=None), "L4": layout("closing-page-gone", drop_last_page=True), "L5": layout("zero-net-page", prepare=zero_net_middle_page), } flat = [case for cases in layouts.values() for case in cases] verdicts = iter(run_chain(flat, timeout)) for cases in layouts.values(): for case in cases: case["verdict"] = next(verdicts) def count(key, test): cases = layouts[key] return "%d of %d" % (sum(1 for c in cases if test(c)), len(cases)) status = lambda wanted: (lambda c: c["verdict"]["status"] == wanted) broke = lambda wanted: (lambda c: c["verdict"]["break"] == wanted) no_row = lambda c: c["verdict"]["firstBrokenRow"] is None named_after_gap = [c for c in layouts["L1"] if c["verdict"]["firstBrokenRow"] == c["firstAfterGap"]] days = sorted(c["gapDays"] for c in named_after_gap) return { "C0.verdict": "balanced in " + count("C0", status("balanced")), "L1.verdict": "needs_review in " + count("L1", status("needs_review")), "L1.break": "running_balance in " + count("L1", broke("running_balance")), "L1.named": "the first row after the gap in " + count("L1", lambda c: c["verdict"]["firstBrokenRow"] == c["firstAfterGap"]), "L1.days": "%d to %d" % (days[0], days[-1]) if days else "none", "L2.verdict": "needs_review in " + count("L2", status("needs_review")), "L2.break": "closing in " + count("L2", broke("closing")), "L2.named": "the last kept row in " + count("L2", lambda c: c["verdict"]["firstBrokenRow"] == c["kept"]), "L3.break": "opening_missing in " + count("L3", broke("opening_missing")), "L3.named": "no row in " + count("L3", no_row), "L4.break": "closing_missing in " + count("L4", broke("closing_missing")), "L4.named": "no row in " + count("L4", no_row), "L5.verdict": "balanced in " + count("L5", status("balanced")), } # --- reporting ------------------------------------------------------------------------------- def write_csv(rows): fields = ("option", "property", "value", "quote", "url", "method") with open(CSV, "w", newline="") as handle: writer = csv.writer(handle, quoting=csv.QUOTE_ALL, lineterminator="\n") writer.writerow(fields) for row in rows: writer.writerow([row.get(field, "") for field in fields]) def main(argv): recheck, write = "--recheck" in argv, "--write" in argv with open(DATASET) as handle: dataset = json.load(handle) failures, report = [], [] measured = None try: measured = run_lab(LAB_BUDGET_S if recheck else None) except subprocess.TimeoutExpired: report.append("skip lab: over the %d s budget of --recheck" % LAB_BUDGET_S) except RuntimeError as error: failures.append("lab did not run: %s" % error) sources = {} for row in dataset["results"]: if row["source"] == "git": path = row["url"].split(":", 2)[2] if path not in sources: try: sources[path] = normalise_code(git_show(path)) except RuntimeError as error: sources[path] = None failures.append(str(error)) if sources[path] is None: continue found = normalise(row["quote"]) in sources[path] report.append("%s git %s | %s" % ("ok " if found else "GONE ", path, row["property"])) if not found: failures.append("quote gone from %s: %s" % (path, row["quote"])) elif row["source"] == "lab": if measured is None: continue got = measured.get(row["key"]) if write: row["value"] = got same = row["value"] == got report.append("%s lab %s | %s = %s" % ("ok " if same else "DIFF ", row["key"], row["option"], got)) if not same: failures.append("%s: dataset says %r, the lab measured %r" % (row["key"], row["value"], got)) else: failures.append("row without a known source: %s" % row.get("property")) if write: with open(DATASET, "w") as handle: json.dump(dataset, handle, indent=2, ensure_ascii=False) handle.write("\n") write_csv(dataset["results"]) report.append("wrote dataset.json and dataset.csv") print("\n".join(report)) if failures: print("\n%d failure(s):" % len(failures)) for line in failures: print(" - " + line) return 1 print("\nall %d rows hold" % len(dataset["results"])) return 0 if __name__ == "__main__": sys.exit(main(sys.argv[1:]))