Files
troupe-standup/wizard-work/volt/q143-check-sweep/check_sweep.py
T

517 lines
22 KiB
Python

#!/usr/bin/env python3
"""check_sweep.py -- every CHECK entry point in this tree, run, and graded
against a baseline that states the reading already reviewed.
WHY THIS EXISTS
The gate pins eleven legs and one lane's closed artifacts; nothing in this
tree runs the 87 `CHECK` entry points the archive carries. So a closed item's
evidence is green only where someone happened to name it that tick, and the
class the queue keeps filing -- a repair that stops existing while its gate
keeps printing what it prints (Q127, Q130, Q142) -- has no repo-wide
instrument at all. This is the cheap half of that instrument: run every entry
point at the tree, record its exit code, and compare the codes with the codes
a reader signed off on.
WHAT IT GRADES -- two directions, because both are real
* a path the baseline names that is gone -> FAULT (the evidence left)
* an entry point the baseline does not name -> FAULT (unreviewed)
* a code that moved 0 -> 1 (or 1 -> 2, or -> 124) -> FAULT (a repair was
reverted, or a new defect landed, and the instrument says which path)
* a code that moved 1 -> 0 -> FAULT (the repair is
real; re-take the baseline so the improvement is signed off, not silent)
Equality is the contract, so neither direction can pass by not looking.
WHAT IT DOES NOT GRADE
Whether a green entry point measures the right thing. An entry point that
prints PASS and exits 0 unconditionally is green here and always will be;
only the tool's own --selftest can say otherwise, and that is the sweep's
other leg, not this one. Nor does it grade content: two different failures
with the same exit code read the same, which is why the baseline notes carry
the first fault line each path printed when the reading was taken.
EXIT CODES
0 the sweep and the baseline agree, 1 they do not (each disagreement named),
2 the baseline could not be read or the sweep could not run.
USAGE
python3 check_sweep.py --repo . --check
python3 check_sweep.py --repo . --only CHECK-2026-11-01-posts-census --check
python3 check_sweep.py --repo . --list
python3 check_sweep.py --repo . --take wizard-work/volt/q143-check-sweep/check_entry_baseline.tsv
python3 check_sweep.py --selftest
`--only` scopes the grade as well as the sweep: the rows it did not run are
reported as not swept, never as missing from the tree. Before that rule, a
one-path run on a clean tree read 1 green row and 110 phantom MISSING faults,
rc 1 -- a scoped invocation that could not pass whatever the tree did.
"""
import argparse
import concurrent.futures
import contextlib
import io
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
HERE = os.path.dirname(os.path.abspath(__file__))
DEFAULT_BASELINE = os.path.join(HERE, "check_entry_baseline.tsv")
# A CHECK entry point is a tracked file whose basename is `CHECK` or begins
# `CHECK-`/`CHECK.`. `CHECKER`, `CHECKS.md` and `CHECKSUMS` are deliberately
# not matched: the archive carries all three kinds of near-miss and discovery
# that guesses is discovery that invents paths.
ENTRY_RE = re.compile(r"^CHECK$|^CHECK[-.]")
HEADER = "# path\trc\tnote -- the reading taken at the rev named below, one row per CHECK entry point"
def repo_root_from_here():
return os.path.dirname(os.path.dirname(os.path.dirname(HERE)))
def tracked(repo):
"""Tracked paths, or a tree walk for a fixture with no .git."""
if os.path.isdir(os.path.join(repo, ".git")):
try:
out = subprocess.run(["git", "-C", repo, "ls-files"],
capture_output=True, text=True, timeout=120)
if out.returncode == 0:
return out.stdout.splitlines()
except (OSError, subprocess.SubprocessError):
pass
found = []
for base, dirs, files in os.walk(repo):
dirs[:] = sorted(d for d in dirs if d not in (".git", "__pycache__"))
for name in sorted(files):
full = os.path.join(base, name)
found.append(os.path.relpath(full, repo))
return found
def entry_points(repo):
return sorted(p for p in tracked(repo)
if ENTRY_RE.match(os.path.basename(p)))
def interpreter(path):
"""The interpreter the file's own first line names; else by extension."""
try:
with open(path, "rb") as fh:
first = fh.read(200).split(b"\n")[0].decode("utf-8", "replace")
except OSError:
return ["bash"]
if first.startswith("#!"):
words = first[2:].split()
if words and words[0].endswith("/env") and len(words) > 1:
return [shutil.which(words[1]) or words[1]]
if words:
return [shutil.which(os.path.basename(words[0])) or words[0]]
if path.endswith(".py"):
return [sys.executable]
return ["bash"]
FAULT_HINT = re.compile(r"^(FAIL|FAULT|STALE|VERDICT: FAIL|Traceback|NotImplemented|cannot|Error)")
def run_entry(repo, rel, timeout):
full = os.path.join(repo, rel)
argv = interpreter(full) + [full]
try:
r = subprocess.run(argv, capture_output=True, text=True, timeout=timeout,
cwd=repo, stdin=subprocess.DEVNULL)
text = (r.stdout or "") + (r.stderr or "")
rc = r.returncode
except subprocess.TimeoutExpired:
return 124, "TIMEOUT after %ss" % timeout
except OSError as exc:
return 125, "SPAWN %s" % exc
lines = [l.strip() for l in text.splitlines() if l.strip()]
note = ""
for line in lines:
if FAULT_HINT.match(line):
note = line
break
if not note and lines:
note = lines[-1]
return rc, note[:160]
def classify(rc):
if rc == 0:
return "green"
if rc == 1:
return "red"
if rc == 2:
return "ungradable"
if rc == 124:
return "timeout"
return "crashed(rc=%d)" % rc
def sweep(repo, timeout, only=None, jobs=4):
"""Every entry point, run `jobs` at a time; the RESULT is sorted by path so a
parallel sweep and a serial one read the same."""
paths = [rel for rel in entry_points(repo) if not only or only in rel]
if jobs < 2 or len(paths) < 2:
rows = []
for rel in paths:
rc, note = run_entry(repo, rel, timeout)
rows.append({"path": rel, "rc": rc, "state": classify(rc), "note": note})
return rows
rows = []
with concurrent.futures.ThreadPoolExecutor(max_workers=jobs) as pool:
futures = {pool.submit(run_entry, repo, rel, timeout): rel for rel in paths}
for fut in concurrent.futures.as_completed(futures):
rel = futures[fut]
rc, note = fut.result()
rows.append({"path": rel, "rc": rc, "state": classify(rc), "note": note})
return sorted(rows, key=lambda r: r["path"])
def read_baseline(path):
"""-> (rev, [(path, rc, note)]); raises ValueError on a malformed file."""
if not os.path.isfile(path):
raise ValueError("no baseline at %s" % path)
rev, rows = None, []
with open(path, "r", encoding="utf-8") as fh:
for n, raw in enumerate(fh, 1):
line = raw.rstrip("\n")
if not line.strip():
continue
if line.startswith("#"):
m = re.search(r"rev (\S+)", line)
if m:
rev = m.group(1)
continue
parts = line.split("\t")
if len(parts) < 2:
raise ValueError("line %d has no tab: %r" % (n, line))
path, rc = parts[0], parts[1].strip()
if not rc.isdigit():
raise ValueError("line %d rc is not an integer: %r" % (n, rc))
rows.append((path, int(rc), parts[2] if len(parts) > 2 else ""))
return rev, rows
def grade(rows, base_rows):
"""Two directions: paths, then codes. -> (faults, tolerances)."""
faults = []
live = {r["path"]: r for r in rows}
base = {p: (rc, note) for p, rc, note in base_rows}
for path in sorted(set(base) - set(live)):
faults.append("MISSING %s -- baseline rc %d, no such entry point in the tree"
% (path, base[path][0]))
for path in sorted(set(live) - set(base)):
faults.append("NEW %s -- rc %d, not in the baseline (%s)"
% (path, live[path]["rc"], live[path]["state"]))
for path in sorted(set(live) & set(base)):
was, note = base[path]
now = live[path]["rc"]
if was == now:
continue
if was == 0:
faults.append("REVERT %s -- rc moved 0 -> %d (%s): %s"
% (path, now, classify(now), live[path]["note"]))
elif now == 0:
faults.append("IMPROVED %s -- rc moved %d -> 0, re-take the baseline"
% (path, was))
else:
faults.append("MOVED %s -- rc moved %d -> %d (%s): %s"
% (path, was, now, classify(now), live[path]["note"]))
return faults, len(set(live) & set(base))
def take(repo, path, timeout, jobs=4):
rows = sweep(repo, timeout, jobs=jobs)
rev = "worktree"
if os.path.isdir(os.path.join(repo, ".git")):
try:
r = subprocess.run(["git", "-C", repo, "rev-parse", "--short", "HEAD"],
capture_output=True, text=True, timeout=60)
if r.returncode == 0:
rev = r.stdout.strip()
except (OSError, subprocess.SubprocessError):
pass
with open(path, "w", encoding="utf-8") as fh:
fh.write(HEADER + "\n")
fh.write("# rev %s -- every code below was measured, not read, and a move\n"
"# in either direction is a fault until this file is re-taken\n" % rev)
for row in rows:
note = " ".join(row["note"].replace("\t", " ").split())
fh.write("%s\t%d\t%s\n" % (row["path"], row["rc"], note))
return rows, rev
# --------------------------------------------------------------------------
# selftest -- fixture trees, every leg graded by the code it returns
# --------------------------------------------------------------------------
FIXTURE = {
"CHECK-green": "#!/bin/bash\necho \"VERDICT: PASS\"\nexit 0\n",
"CHECK-red": "#!/bin/bash\necho \"FAIL: two rows disagree\"\nexit 1\n",
"sub/CHECK-ungradable": "#!/bin/bash\necho \"cannot grade: no input\"\nexit 2\n",
"sub/CHECK-slow": "#!/bin/bash\nsleep 30\nexit 0\n",
"sub/CHECK-py": "#!/usr/bin/env python3\nprint('verdict: pass')\nraise SystemExit(0)\n",
"sub/CHECKER": "#!/bin/bash\nexit 1\n",
"notes.md": "# not an entry point\n",
}
def write_fixture(root, extra=None, drop=None):
for rel, body in FIXTURE.items():
if drop and rel in drop:
continue
full = os.path.join(root, rel)
os.makedirs(os.path.dirname(full), exist_ok=True)
with open(full, "w", encoding="utf-8") as fh:
fh.write(body)
for rel, body in (extra or {}).items():
full = os.path.join(root, rel)
os.makedirs(os.path.dirname(full), exist_ok=True)
with open(full, "w", encoding="utf-8") as fh:
fh.write(body)
class Cells(object):
def __init__(self):
self.passed, self.failed = 0, []
def check(self, name, cond, detail=None):
if cond:
self.passed += 1
print(" PASS %s" % name)
else:
self.failed.append(name)
print(" FAIL %s %s" % (name, "" if detail is None else detail))
def selftest():
cells = Cells()
tmp = tempfile.mkdtemp(prefix="check-sweep-")
try:
# --- discovery -----------------------------------------------------
tree = os.path.join(tmp, "tree")
write_fixture(tree)
found = entry_points(tree)
cells.check("discover-finds-five",
len(found) == 5, found)
cells.check("discover-skips-near-miss",
"sub/CHECKER" not in found and "notes.md" not in found, found)
cells.check("discover-sorted", found == sorted(found), found)
# --- classification: every code the runner can produce --------------
rows = {r["path"]: r for r in sweep(tree, timeout=3)}
cells.check("run-green-rc0", rows["CHECK-green"]["rc"] == 0, rows["CHECK-green"])
cells.check("run-red-rc1", rows["CHECK-red"]["rc"] == 1, rows["CHECK-red"])
cells.check("run-ungradable-rc2",
rows["sub/CHECK-ungradable"]["rc"] == 2, rows["sub/CHECK-ungradable"])
cells.check("run-timeout-rc124", rows["sub/CHECK-slow"]["rc"] == 124,
rows["sub/CHECK-slow"])
cells.check("run-python-shebang", rows["sub/CHECK-py"]["rc"] == 0,
rows["sub/CHECK-py"])
cells.check("run-note-names-fault",
rows["CHECK-red"]["note"].startswith("FAIL:"), rows["CHECK-red"])
# --- the clean control ---------------------------------------------
base = os.path.join(tmp, "base.tsv")
take(tree, base, timeout=3)
rev, base_rows = read_baseline(base)
cells.check("take-writes-all-rows", len(base_rows) == 5, len(base_rows))
faults, graded = grade(sweep(tree, timeout=3), base_rows)
cells.check("grade-clean-control", faults == [] and graded == 5,
(faults, graded))
# --- a scoped run grades the rows it swept, not the whole baseline --
tree6 = os.path.join(tmp, "tree6")
write_fixture(tree6)
base6 = os.path.join(tmp, "base6.tsv")
take(tree6, base6, timeout=3)
buf = io.StringIO()
with contextlib.redirect_stdout(buf):
rc_scoped = main(["--repo", tree6, "--baseline", base6,
"--only", "CHECK-green", "--check"])
scoped = buf.getvalue()
cells.check("only-scoped-is-green", rc_scoped == 0, scoped)
cells.check("only-grades-one-row",
"1 baseline row(s) graded (1), 0 fault(s)" in scoped, scoped)
cells.check("only-names-the-unswept",
"4 row(s) not swept (--only)" in scoped, scoped)
# --- four mutants, each must be caught by name ----------------------
(faults, _) = grade(sweep(tree, timeout=3),
base_rows + [("CHECK-ghost", 0, "fixture")])
cells.check("grade-missing-path", any(f.startswith("MISSING CHECK-ghost") for f in faults),
faults)
extra = dict(FIXTURE)
extra["CHECK-extra"] = "#!/bin/bash\nexit 0\n"
tree2 = os.path.join(tmp, "tree2")
write_fixture(tree2, extra={"CHECK-extra": "#!/bin/bash\nexit 0\n"})
(faults, _) = grade(sweep(tree2, timeout=3), base_rows)
cells.check("grade-new-path", any(f.startswith("NEW CHECK-extra") for f in faults),
faults)
cells.check("grade-new-path-count",
sum(1 for f in faults if f.startswith("NEW")) == 1, faults)
tree3 = os.path.join(tmp, "tree3")
write_fixture(tree3)
with open(os.path.join(tree3, "CHECK-green"), "w", encoding="utf-8") as fh:
fh.write("#!/bin/bash\necho 'FAIL: reverted'\nexit 1\n")
(faults, _) = grade(sweep(tree3, timeout=3), base_rows)
cells.check("grade-revert-0-to-1",
any(f.startswith("REVERT CHECK-green") for f in faults), faults)
tree4 = os.path.join(tmp, "tree4")
write_fixture(tree4)
with open(os.path.join(tree4, "CHECK-red"), "w", encoding="utf-8") as fh:
fh.write("#!/bin/bash\nexit 0\n")
(faults, _) = grade(sweep(tree4, timeout=3), base_rows)
cells.check("grade-improved-1-to-0",
any(f.startswith("IMPROVED CHECK-red") for f in faults), faults)
tree5 = os.path.join(tmp, "tree5")
write_fixture(tree5, drop=["CHECK-red"])
(faults, _) = grade(sweep(tree5, timeout=3), base_rows)
cells.check("grade-dropped-file-is-missing",
any(f.startswith("MISSING CHECK-red") for f in faults), faults)
# --- baseline unreadable is rc 2, not a fault ------------------------
cells.check("baseline-absent-raises",
_raises(lambda: read_baseline(os.path.join(tmp, "nope.tsv"))), "")
bad = os.path.join(tmp, "bad.tsv")
with open(bad, "w", encoding="utf-8") as fh:
fh.write("CHECK-green 0 no-tab-here\n")
cells.check("baseline-malformed-raises", _raises(lambda: read_baseline(bad)), "")
# --- determinism: two takes are byte-identical ------------------------
base2 = os.path.join(tmp, "base2.tsv")
take(tree, base2, timeout=3)
with open(base, "rb") as fh:
first = fh.read()
with open(base2, "rb") as fh:
second = fh.read()
cells.check("take-deterministic", first == second, "")
finally:
shutil.rmtree(tmp, ignore_errors=True)
total = cells.passed + len(cells.failed)
print("selftest: %d case(s), %d failed" % (total, len(cells.failed)))
return 0 if not cells.failed else 1
def _raises(fn):
try:
fn()
except ValueError:
return True
return False
def main(argv=None):
ap = argparse.ArgumentParser(
prog="check_sweep.py",
description="run every CHECK entry point in this tree and grade its exit "
"code against a baseline that states the reading already reviewed")
ap.add_argument("--repo", default=repo_root_from_here(),
help="the tree to sweep (default: the repo holding this file)")
ap.add_argument("--baseline", default=DEFAULT_BASELINE,
help="the baseline TSV to grade against")
ap.add_argument("--timeout", type=int, default=120,
help="seconds one entry point may run before it reads rc 124")
ap.add_argument("--only", default=None,
help="restrict the sweep to paths containing this substring")
ap.add_argument("--jobs", type=int, default=4,
help="entry points to run at a time; the result is sorted, so "
"the reading does not depend on it")
ap.add_argument("--check", action="store_true",
help="sweep and grade against the baseline")
ap.add_argument("--paths", action="store_true",
help="grade the entry-point set against the baseline without running any")
ap.add_argument("--list", action="store_true",
help="print the entry points discovery finds and exit")
ap.add_argument("--take", metavar="PATH",
help="measure the sweep and write the baseline to PATH")
ap.add_argument("--json", metavar="PATH", default=None,
help="write the measured rows as JSON to PATH")
ap.add_argument("--selftest", action="store_true",
help="run the fixture corpus and exit 0 only if every cell passes")
args = ap.parse_args(argv)
if args.selftest:
return selftest()
if args.list:
for rel in entry_points(args.repo):
print(rel)
return 0
if args.take:
rows, rev = take(args.repo, args.take, args.timeout, jobs=args.jobs)
green = sum(1 for r in rows if r["rc"] == 0)
print("baseline: %d entry point(s), %d green, %d not, rev %s"
% (len(rows), green, len(rows) - green, rev))
return 0
if args.paths:
try:
rev, base_rows = read_baseline(args.baseline)
except ValueError as exc:
print("BASELINE UNREADABLE: %s" % exc)
return 2
live = set(entry_points(args.repo))
base = {p for p, _, _ in base_rows}
faults = ["MISSING %s -- baseline names it, the tree does not have it" % p
for p in sorted(base - live)]
faults += ["NEW %s -- in the tree, not in the baseline" % p
for p in sorted(live - base)]
for line in faults:
print("FAIL: " + line)
print("paths: %d entry point(s), %d baseline path(s), %d fault(s)"
% (len(live), len(base), len(faults)))
return 0 if not faults else 1
if not args.check:
ap.error("one of --check, --paths, --list, --take, --selftest is required")
try:
rev, base_rows = read_baseline(args.baseline)
except ValueError as exc:
print("BASELINE UNREADABLE: %s" % exc)
return 2
try:
rows = sweep(args.repo, args.timeout, only=args.only, jobs=args.jobs)
except Exception as exc: # noqa: BLE001 -- a sweep that cannot run is rc 2
print("SWEEP COULD NOT RUN: %r" % (exc,))
return 2
if args.json:
with open(args.json, "w", encoding="utf-8") as fh:
json.dump({"rev": rev, "rows": rows}, fh, indent=1, sort_keys=True)
# --only sweeps a subset, so it must grade that same subset: a row it did
# not run was not read, and calling it "missing from the tree" is a phantom
# fault. Measured before this line existed: `--only <one path> --check`
# graded 1 row green and 110 unswept rows as MISSING, rc 1, on a clean tree.
base_in_scope = [r for r in base_rows if not args.only or args.only in r[0]]
for row in rows:
print(" %-11s rc=%-3d %s" % (row["state"], row["rc"], row["path"]))
faults, graded = grade(rows, base_in_scope)
for line in faults:
print("FAIL: " + line)
green = sum(1 for r in rows if r["rc"] == 0)
scope_note = ("" if not args.only else
", %d row(s) not swept (--only)"
% (len(base_rows) - len(base_in_scope)))
print("check-sweep: %d entry point(s), %d green, %d baseline row(s) graded (%d), "
"%d fault(s)%s" % (len(rows), green, len(base_in_scope), graded,
len(faults), scope_note))
return 0 if not faults else 1
if __name__ == "__main__":
sys.exit(main())