stackchain/troupe-standup
2
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
aeca104283f257e4fc3c87cde9899340bda9ca60
troupe-standup/tools/selftest_sweep.py
T
volt c56e3be726 served_drift_check (volt): a deploy is a claim about a revision, and nothing graded it 
The leathermint lane re-proves the same served-bytes fact by hand every tick --
reply 45 lists six paths, my 18:25Z block lists six more, each with sizes typed
out. `landing_check.py` grades the FORGE, whose raw path is a convention
(`<base>/branch/<branch>/<path>`); a static host has none, so its own question
went ungraded: are the bytes on the host the bytes this revision says?

New instrument `tools/served_drift_check.py` -- one row per tracked file at
--rev, graded against GET `<base>/<path>`; verdicts OK / DIFFERS / MISSING /
HTTP_ERROR / UNREACHABLE / READ_FAILED / SKIPPED_LARGE, the same vocabulary as
landing_check so no reader has to learn a third table. DIFFERS carries both
lengths and both digests (the direction of the staleness is the useful part);
UNREACHABLE is kept apart from MISSING so an outage cannot read as a clean
negative; rc 2 when it cannot grade at all (not a git work tree, no --base-url,
a scheme it cannot read, a revision that tracks nothing).

Self-proving: `--selftest` is 22 cells over a fixture git repo and a real
http.server on 127.0.0.1:0 -- mirror-is-OK, stale-copy-is-DIFFERS with both
digests, 404-is-MISSING, 500-is-HTTP_ERROR, refused-port-is-UNREACHABLE,
HEAD-vs-HEAD~1 grades a different blob, --prefix/--exclude/--max-bytes, the
three rc-2 preconditions, and a --json round trip. One defect found in my own
first draft by running it: `return 1 if bad else 0` on a list made every
green selftest exit 1; the cell count was 22/22 with rc 1 until it was
`bad[0]`.

Live measurement, this host, this commit's parent: 23 of 23 tracked files of
/var/www/leathermint OK against https://leathermint.alexanderwhitestone.com,
0 drift, rc 0 -- the first machine-checkable form of what the leathermint
blocks assert in prose.

Both sweep axes get their row (read-refusal: --root {nope}; malformed:
not-a-repo -> rc 2 naming "not a git work tree"), which is what keeps the new
tool from being silently ungraded; docs/TOOLS.md re-rendered in the next
commit, since tool_map reads the tree at HEAD.
2026-10-20 19:28:32 +00:00

827 lines
39 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""selftest_sweep.py -- every instrument proves itself, in one command, and the
number it advertises is the number it prints.
WHY THIS EXISTS
`tools/append_only_audit.py` carried, on its own usage line, the words
"synthetic repo, N cells" beside a selftest that printed a different N.
The control plane found it and flagged it twice; it stood until 580a41c because
nothing in this archive ever compared a tool's self-advertised count against the
count the tool prints. `doc_census_check.py` now checks that class -- but its
subject is the census blocks (AMBIGUOUS / MATCHES / NOT_PRINTED / ...) and it
drives every tool against a revision, so a full run is minutes, not seconds.
The selftest cell count is a cheaper and more basic claim: it is the number a
reader uses to decide whether a tool is worth trusting, and the number a
check-in quotes as proof. Eight tools carry one. One was wrong for days. This
is the gate for that sentence, and it runs in seconds because a selftest is a
unit test, not a census.
WHAT IT MEASURES -- one row per tracked tool, one verdict per row
NO_SELFTEST_FLAG the source never mentions "--selftest"; the tool is not run
(a tool without the flag may do real work on `--selftest`).
TIMEOUT the selftest did not finish inside --timeout seconds.
RUN_FAILED non-zero exit and no count line: the run died, or failed
before it could report.
NO_COUNT_LINE exit 0 but nothing matching
`selftest: <n> cells|fixtures, <n> passed, <n> failed`.
SELFTEST_FAILED a count line exists and reports failed > 0 or passed < total.
SPLIT two different values of one unit on the tool's own
`--selftest` usage line(s) -- the four-hundred-and-seventy-one
over-count defect, at
selftest scale.
COUNT_MISMATCH exactly one advertised value, and it is not the printed one
(value or unit). This is the append_only_audit defect.
NO_CLAIM prints a count, advertises none. ADVISORY, not a fault: an
absent claim cannot be false.
OK printed a count, every advertised value agrees with it.
FATAL = every class above except NO_CLAIM. Exit 1 if any row is fatal, else 0.
THE READ-REFUSAL AXIS, ADDED 2026-10-17
The same command asks every tool the second question this archive kept
getting wrong: what exit code does it give an input it cannot read? The
contract is three codes and no others -- rc 0 ran clean, rc 1 ran and graded
a fault, rc 2 could not grade. The axis runs each tool against a path that
does not exist (and, for the four file readers, against bytes that are not
UTF-8) and grades the answer OK / RC_NOT_2 / TRACEBACK / SILENT; the last
three are fatal. So is NO_PROBE (below): a tool the tables never name is a
tool this axis never graded, and an ungraded tool may not be reported clean.
rc 3 is retired: the two queue readers used it for "cannot
read the queue", so one meaning carried two codes and a stale index (rc 1)
could be confused with a queue that was never read.
THE MALFORMED-INPUT CLASS, ADDED 2026-10-18
A missing path and non-UTF-8 bytes are the easy two. The class the axis left
unmeasured was the one in between: an input that DECODES but cannot be made
sense of -- a `.json` that is not JSON, a valid `{}` where an audit file
should be, a directory that is not a git repository, a queue file with no
records. Those are the inputs where a tool is most tempted to answer "0" and
return 0, which reads exactly like a clean pass (`doc_census_check`'s
`--tools` bug, `76043b7`, was this class). The axis runs one such probe per
tool and grades each against ITS OWN stated answer, because "must exit 2" is
not a universal truth: most tools must refuse at rc 2, while `check_lint` grades
a junk log MISSING (rc 0 in report mode, rc 1 under --fail-on-fault),
`checkin_check` grades a heading-less check-in as an advisory (rc 0, stated
in its docstring) and `queue_claim_check` reports 0 opening records (rc 0,
its --fail-on-stale is the gate). The axis asserts the message too, so a
silent 0 cannot pass as a refusal.
WHAT COUNTS AS A CLAIM, PRECISELY
Only the module docstring (read through `ast.get_docstring`, so a fixture
string in the BODY is not a claim -- `doc_census_check.py` contains the literal
"selftest: N cells, N passed, N failed" inside a test, and reading the file as
text would call that a claim), and only on a line that also contains the token
`--selftest`. Prose elsewhere in the docstring is not a claim: the same
doc_census_check docstring quotes a thirteen-cell claim as the defect the tool
itself catches, and a scanner that flagged its own bug report would be wrong.
DISCOVERY, AND WHY UNTRACKED FILES ARE PRINTED
Tracked paths only, from `git ls-files` in --repo -- the lesson doc_census_check
paid for: an untracked tool is invisible to a revision-scoped census, and this
gate inherits that blindness rather than pretending to fix it. Every `tools/*.py`
that IS present but untracked is therefore listed as SKIPPED_UNTRACKED with its
name, never silently dropped. With --tools-dir, or with no git, discovery is a
plain glob of that directory and the row says so.
READ-ONLY: the only write path is `--json PATH`. It does not edit a tool, and it
does not decide that a claim SHOULD exist -- NO_CLAIM is advisory by design.
NO_PROBE is not: an ungraded tool cannot be reported clean (see the read axis).
USAGE:
python3 selftest_sweep.py --repo /root/troupe-standup
python3 selftest_sweep.py --tools-dir /tmp/fake --json /tmp/sweep.json
python3 selftest_sweep.py --selftest # 30 cells, parser, sweep and both input axes
"""
from __future__ import annotations
import argparse
import ast
import glob
import json
import os
import re
import subprocess
import sys
import tempfile
UNITS = ("cells", "fixtures")
# `selftest: 18 cells, 18 passed, 0 failed`
# `self-test: 19 fixtures, 19 passed, 0 failed`
# `selftest: 14 fixtures (10 claim-check, 4 unit), 14 passed, 0 failed`
PRINTED_RE = re.compile(
r"(?:selftest|self-test)\s*:\s*([\d,]+)\s+(cells|fixtures)\b"
r"[^\n]*?\b([\d,]+)\s+passed\b[^\n]*?\b([\d,]+)\s+failed\b"
)
CLAIM_RE = re.compile(r"(?<![\w.])([\d,]+)\s+(cells|fixtures)\b")
SELFTEST_FLAG_RE = re.compile(r"""["']--selftest["']""")
FATAL = (
"SELFTEST_FAILED",
"RUN_FAILED",
"TIMEOUT",
"NO_COUNT_LINE",
"NO_SELFTEST_FLAG",
"COUNT_MISMATCH",
"SPLIT",
"UNREADABLE",
)
# --------------------------------------------------------------- read axis
# THE SECOND GATE, ADDED 2026-10-17: the exit code a tool gives an input it
# cannot read. The repo-wide contract is three codes and no others --
# rc 0 ran, nothing to report
# rc 1 ran, and GRADED something as faulty (STALE, TAUTOLOGY, MISMATCH, ...)
# rc 2 COULD NOT GRADE: usage error, precondition unmet, input unreadable
# Before this axis existed the third class was spread over rc 1 (check_lint,
# census_row_delta, both append-only tools as a raw traceback, this sweep) and
# rc 3 (the two queue readers), so "1" meant both "bad verdict" and "no
# verdict" and an operator could not tell them apart. The axis re-measures the
# class on every tool, from the outside, against the real CLI.
READ_AXIS_UNIT = 2
# NO_PROBE is fatal, not advisory (2026-10-18): a tool the tables never name is
# a tool neither axis graded, and the pre-fix sweep reported a 15th tool with no
# probe entry as "15 tools, 14 OK, 0 fatal ... rc 0" -- two NO_PROBE rows and a
# green gate. An ungraded tool may not be reported clean; the tables are the
# gate's only cover, so a gap in them has to fail like a fault in a tool.
FATAL_READ = ("TRACEBACK", "RC_NOT_2", "SILENT", "NO_TOOL", "RC_WRONG",
"UNEXPLAINED", "NO_PROBE")
NO_PROBE = "NO_PROBE" # fatal: a tool with no probe defined
# {nope} = a path that does not exist; {bin} = a file that is not UTF-8;
# {bindir} = a directory holding one non-UTF-8 *.md (added 2026-10-18 with
# checkin_check.py, whose input is a directory of UTF-8 files -- a {bin} file
# would only have proven it refuses a non-directory).
READ_PROBES = (
("append_only_audit.py", ["--repo", "{nope}"]),
("append_only_owners.py", ["{nope}/audit.json"]),
("census_row_delta.py", ["{nope}/a.json", "{nope}/b.json"]),
("check_lint.py", ["--log", "{nope}/log.txt"]),
("checkin_check.py", ["--dir", "{nope}", "--roster", "volt"]),
("citation_hash_class.py", ["--repo", "{nope}"]),
("citation_resolve_check.py", ["--repo", "{nope}"]),
("doc_command_check.py", ["--repo", "{nope}"]),
("doc_census_check.py", ["--repo", "{nope}", "--no-run"]),
("gate_blame.py", ["--repo", "{nope}"]),
("landing_check.py", ["--repo", "{nope}"]),
("queue_claim_check.py", ["--queue", "{nope}/q.md"]),
("queue_index.py", ["--queue", "{nope}/q.md", "--check", "{nope}/i.md"]),
("queue_append_audit.py", ["--repo", "{nope}"]),
("reply_check.py", ["--dir", "{nope}"]),
("selftest_sweep.py", ["--repo", "{nope}"]),
("served_drift_check.py", ["--root", "{nope}", "--base-url", "file:///nonexistent"]),
("tool_map.py", ["--repo", "{nope}"]),
("verify.py", ["--repo", "{nope}"])
)
NOT_UTF8_PROBES = (
("append_only_owners.py", ["{bin}"]),
("census_row_delta.py", ["{bin}", "{bin}"]),
("checkin_check.py", ["--dir", "{bindir}", "--roster", "volt"]),
("queue_claim_check.py", ["--queue", "{bin}"]),
("queue_index.py", ["--queue", "{bin}", "--check", "{bin}"]),
)
# THE THIRD CLASS (2026-10-18): input that decodes as UTF-8 but cannot be made
# sense of. Each entry carries the answer the TOOL's own contract states --
# (tool, label, argv, expected rc, a substring the output must contain) -- so
# the axis measures the documented behaviour instead of imposing one. Most
# tools must refuse at rc 2; the ones that grade a malformed input are listed
# with the rc and the word their grade prints, which is what makes a silent
# rc 0 (the doc_census_check --tools bug class) fail here.
# The counts are deliberately NOT written here: a number in a comment has no
# reader, which is the defect class this file's own docstring names, and the
# two counts that used to sit on these lines were both stale ("eleven",
# "Fourteen") before gate_blame.py's row was added on 2026-10-20.
# Fixtures: {notrepo} = a directory with a file and no .git; {badjson} = a
# UTF-8 file that is not JSON; {emptyjson} = "{}" (valid JSON, wrong shape);
# {junklog} = a log with no CHECK: line; {junkqueue} = 3 prose lines, no
# records; {junkindex} = a file with no pinned parse; {badcheckins} = a
# directory holding one heading-less check-in.
MALFORMED_PROBES = (
("append_only_audit.py", "not-a-repo", ["--repo", "{notrepo}"],
2, "cannot read repo"),
("append_only_owners.py", "malformed-json", ["{badjson}"],
2, "cannot read audit"),
("append_only_owners.py", "wrong-shape", ["{emptyjson}"],
2, "no 'rows' list"),
("census_row_delta.py", "malformed-json", ["{badjson}", "{badjson}"],
2, "cannot read census"),
("census_row_delta.py", "wrong-shape", ["{emptyjson}", "{emptyjson}"],
2, "no 'reports' list"),
("check_lint.py", "junk-log", ["--log", "{junklog}"],
0, "MISSING"),
("check_lint.py", "junk-log-gated", ["--log", "{junklog}", "--fail-on-fault"],
1, "MISSING"),
("checkin_check.py", "no-headings", ["--dir", "{badcheckins}", "--roster", "volt"],
0, "advisory"),
("citation_hash_class.py", "not-a-repo", ["--repo", "{notrepo}"],
2, "not a git"),
("citation_resolve_check.py", "not-a-repo", ["--repo", "{notrepo}"],
2, "not a git"),
("doc_census_check.py", "not-a-repo", ["--repo", "{notrepo}", "--no-run"],
2, "no tools found"),
("doc_command_check.py", "not-a-repo", ["--repo", "{notrepo}"],
2, "cannot read repo"),
# gate_blame.py's input is a git work tree: a directory that is not one is the
# malformed shape, and it must name that rather than pass a window of nothing.
("gate_blame.py", "not-a-repo", ["--repo", "{notrepo}"],
2, "not a git work tree"),
("landing_check.py", "not-a-repo", ["--repo", "{notrepo}"],
2, "not a git work tree"),
("queue_claim_check.py", "junk-queue", ["--queue", "{junkqueue}"],
0, "0 opening records"),
("queue_index.py", "junk-index", ["--queue", "{junkqueue}", "--check", "{junkindex}"],
2, "UNVERIFIABLE"),
("queue_append_audit.py", "not-a-repo", ["--repo", "{notrepo}"],
2, "not a git repository"),
# reply_check.py's input is a directory of reply files: a file where that
# directory belongs is the malformed shape (a missing one is the read axis).
("reply_check.py", "not-a-dir", ["--dir", "{bin}"],
2, "not a directory"),
# served_drift_check.py's input is a git work tree that a host is supposed to
# be serving: a directory that is not one must be named, not graded as empty.
("served_drift_check.py", "not-a-repo", ["--root", "{notrepo}", "--base-url", "file:///nonexistent"],
2, "not a git work tree"),
("selftest_sweep.py", "not-a-repo", ["--repo", "{notrepo}"],
2, "no tools discovered"),
("tool_map.py", "not-a-repo", ["--repo", "{notrepo}"],
2, "cannot read repo"),
# verify.py's own two legs are the repo it is asked to grade; a directory
# that is not a git repository is the malformed input for both of them.
("verify.py", "not-a-repo", ["--repo", "{notrepo}"],
2, "cannot read repo"),
)
def _int(text: str) -> int:
return int(text.replace(",", ""))
def parse_printed(text: str):
"""(count, unit, passed, failed) from the first count line, or None."""
m = PRINTED_RE.search(text or "")
if not m:
return None
return _int(m.group(1)), m.group(2), _int(m.group(3)), _int(m.group(4))
def usage_claims(doc: str):
"""[(value, unit, line)] -- claims on lines that name --selftest."""
out = []
for line in (doc or "").splitlines():
if "--selftest" not in line:
continue
for m in CLAIM_RE.finditer(line):
out.append((_int(m.group(1)), m.group(2), line.strip()))
return out
def module_docstring(path: str):
"""(docstring, source). The docstring is AST-read so body fixtures are not claims."""
try:
with open(path, "r", encoding="utf-8", errors="replace") as fh:
src = fh.read()
except OSError as exc:
return None, f"<unreadable: {exc}>"
try:
tree = ast.parse(src)
except SyntaxError:
return None, src
return (ast.get_docstring(tree) or ""), src
# ---------------------------------------------------------------- discovery
def discover(repo: str, tools_dir: str | None):
"""(rows, source, untracked).
rows: [(display_name, path)] -- path is absolute, display_name is tools/x.py.
"""
if tools_dir:
d = os.path.abspath(tools_dir)
paths = sorted(p for p in glob.glob(os.path.join(d, "*.py"))
if os.path.isfile(p))
return ([(os.path.basename(p), p) for p in paths],
f"glob of --tools-dir {d} (no git consulted)", [])
d = os.path.join(os.path.abspath(repo), "tools")
have_git = subprocess.run(["git", "-C", os.path.abspath(repo), "rev-parse",
"--git-dir"], capture_output=True).returncode == 0
if not have_git:
paths = sorted(p for p in glob.glob(os.path.join(d, "*.py"))
if os.path.isfile(p))
return ([(os.path.basename(p), p) for p in paths],
f"glob of {d} (not a git tree)", [])
tracked = subprocess.run(
["git", "-C", os.path.abspath(repo), "ls-files", "--", "tools"],
capture_output=True, text=True).stdout.splitlines()
named = [t for t in tracked
if t.startswith("tools/") and t.endswith(".py") and "/" not in t[6:]]
rows = [(os.path.basename(t), os.path.join(os.path.abspath(repo), t))
for t in sorted(named)]
untracked = subprocess.run(
["git", "-C", os.path.abspath(repo), "ls-files", "--others",
"--exclude-standard", "--", "tools"],
capture_output=True, text=True).stdout.splitlines()
stray = [t for t in untracked
if t.startswith("tools/") and t.endswith(".py") and "/" not in t[6:]]
rev = subprocess.run(["git", "-C", os.path.abspath(repo), "rev-parse",
"--short", "HEAD"], capture_output=True, text=True).stdout.strip()
return rows, f"git ls-files at {rev or 'HEAD'} ({len(rows)} tracked in tools/)", sorted(stray)
# ------------------------------------------------------------------- judging
def judge(doc, out: str, rc: int, timed_out: bool):
"""One verdict dict from a completed run. Pure: no I/O, no clock."""
printed = parse_printed(out)
claims = usage_claims(doc or "")
rec = {
"printed": ({"count": printed[0], "unit": printed[1],
"passed": printed[2], "failed": printed[3]} if printed else None),
"claims": [{"value": v, "unit": u, "line": ln} for v, u, ln in claims],
"verdict": None,
"detail": "",
}
if timed_out:
rec["verdict"] = "TIMEOUT"
rec["detail"] = "selftest did not finish"
return rec
if printed is None:
if rc != 0:
rec["verdict"] = "RUN_FAILED"
tail = [l for l in (out or "").splitlines() if l.strip()][-1:] or [""]
rec["detail"] = f"rc={rc}; last line: {tail[0][:160]}"
else:
rec["verdict"] = "NO_COUNT_LINE"
rec["detail"] = "rc=0 but no '<n> cells|fixtures, <n> passed, <n> failed' line"
return rec
count, unit, passed, failed = printed
if failed > 0 or passed < count:
rec["verdict"] = "SELFTEST_FAILED"
rec["detail"] = f"{passed}/{count} passed, {failed} failed"
return rec
same_unit = [(v, ln) for v, u, ln in claims if u == unit]
values = sorted({v for v, _ in same_unit})
if len(values) > 1:
rec["verdict"] = "SPLIT"
rec["detail"] = ("advertises " + " and ".join(f"{v} {unit}" for v in values)
+ f"; prints {count}")
return rec
if not claims:
rec["verdict"] = "NO_CLAIM"
rec["detail"] = f"prints {count} {unit}; advertises none (advisory)"
return rec
if values and values[0] == count:
rec["verdict"] = "OK"
rec["detail"] = f"{count} {unit}"
return rec
rec["verdict"] = "COUNT_MISMATCH"
adv = ", ".join(f"{v} {u}" for v, u, _ in claims)
rec["detail"] = f"advertises {adv}; prints {count} {unit}"
return rec
def sweep(rows, repo: str, timeout: int, verbose: bool):
results = []
for name, path in rows:
doc, src = module_docstring(path)
rec = {"tool": name, "path": path}
if doc is None and not SELFTEST_FLAG_RE.search(src or ""):
rec.update({"verdict": "UNREADABLE", "detail": "source unreadable",
"printed": None, "claims": []})
results.append(rec)
continue
if not SELFTEST_FLAG_RE.search(src or ""):
rec.update({"verdict": "NO_SELFTEST_FLAG", "printed": None,
"claims": [{"value": v, "unit": u, "line": ln}
for v, u, ln in usage_claims(doc or "")],
"detail": "no \"--selftest\" in source; not run"})
results.append(rec)
continue
timed_out = False
rc, out = 0, ""
try:
proc = subprocess.run([sys.executable, path, "--selftest"],
cwd=repo, capture_output=True, text=True,
errors="replace", timeout=timeout)
rc, out = proc.returncode, (proc.stdout or "") + (proc.stderr or "")
except subprocess.TimeoutExpired:
timed_out = True
except OSError as exc:
rc, out = 127, f"<exec failed: {exc}>"
verdict = judge(doc, out, rc, timed_out)
rec.update(verdict)
rec["rc"] = rc
results.append(rec)
if verbose:
print(f" . {name}: {rec['verdict']} rc={rc}")
return results
# --------------------------------------------------------------------- report
def report(results, source, untracked, quiet=False):
fatal = [r for r in results if r["verdict"] in FATAL]
if not quiet:
print(f"discovery: {source}")
for name in untracked:
print(f" SKIPPED_UNTRACKED {name} (present in tools/, not in the tree)")
width = max([len(r["tool"]) for r in results] + [4])
print()
print(f"{'tool':<{width}} {'verdict':<16} {'claim':>12} {'printed':>12} detail")
for r in results:
claims = r.get("claims") or []
claim_s = (", ".join(f"{c['value']} {c['unit']}" for c in claims)
if claims else "-")
p = r.get("printed")
printed_s = f"{p['count']} {p['unit']}" if p else "-"
print(f"{r['tool']:<{width}} {r['verdict']:<16} {claim_s:>12} "
f"{printed_s:>12} {r['detail']}")
print()
ok = sum(1 for r in results if r["verdict"] == "OK")
print(f"sweep: {len(results)} tools, {ok} OK, {len(fatal)} fatal"
+ (f" ({', '.join(sorted({r['verdict'] for r in fatal}))})" if fatal else ""))
return 1 if fatal else 0
# ------------------------------------------------------- read-refusal axis
def _first_line(text: str) -> str:
for line in (text or "").splitlines():
if line.strip():
return line.strip()[:96]
return ""
def judge_read(rc, text):
"""One probe's verdict. Three classes, in this order on purpose: a
traceback is a defect whatever its exit code, and a silent rc 2 is a
refusal an operator cannot read."""
if "Traceback (most recent call last)" in (text or ""):
return {"verdict": "TRACEBACK", "rc": rc, "detail": _first_line(text)}
if rc != READ_AXIS_UNIT:
return {"verdict": "RC_NOT_2", "rc": rc,
"detail": _first_line(text) or f"rc {rc}, no output"}
if not (text or "").strip():
return {"verdict": "SILENT", "rc": rc, "detail": "rc 2 with no message"}
return {"verdict": "OK", "rc": rc, "detail": _first_line(text)}
def run_read_probe(path: str, argv, timeout: int):
"""(rc, combined output) for one probe, run in a throwaway cwd."""
with tempfile.TemporaryDirectory() as td:
binpath = os.path.join(td, "not-utf8.bin")
with open(binpath, "wb") as fh:
fh.write(b"\xff\xfe\x00binary\x00\n")
bindir = os.path.join(td, "not-utf8-dir")
os.makedirs(bindir, exist_ok=True)
with open(os.path.join(bindir, "2026-01-01-volt.md"), "wb") as fh:
fh.write(b"\xff\xfe\x00binary\x00\n")
nope = os.path.join(td, "there-is-no-such-path")
# the malformed class: each of these DECODES, none of it parses
notrepo = os.path.join(td, "not-a-repo")
os.makedirs(notrepo, exist_ok=True)
with open(os.path.join(notrepo, "README.txt"), "w") as fh:
fh.write("files, but no .git here\n")
paths = {"badjson": os.path.join(td, "bad.json"),
"emptyjson": os.path.join(td, "empty.json"),
"junklog": os.path.join(td, "log.txt"),
"junkqueue": os.path.join(td, "q.md"),
"junkindex": os.path.join(td, "i.md")}
bodies = {"badjson": "not json at all\n",
"emptyjson": "{}\n",
"junklog": "garbage line\nnot a lint entry\n",
"junkqueue": "this is not a queue\nno headings here at all\nQ\n",
"junkindex": "not an index either\n"}
for key, body in bodies.items():
with open(paths[key], "w", encoding="utf-8") as fh:
fh.write(body)
badcheckins = os.path.join(td, "checkins")
os.makedirs(badcheckins, exist_ok=True)
with open(os.path.join(badcheckins, "2026-01-01-volt.md"), "w",
encoding="utf-8") as fh:
fh.write("no heading, no template\n")
repl = {"{nope}": nope, "{bin}": binpath, "{bindir}": bindir,
"{notrepo}": notrepo, "{badcheckins}": badcheckins}
repl.update({"{" + k + "}": v for k, v in paths.items()})
real = []
for a in argv:
for key, val in repl.items():
a = a.replace(key, val)
real.append(a)
try:
p = subprocess.run([sys.executable, path] + real, cwd=td,
capture_output=True, text=True, errors="replace",
timeout=timeout)
except subprocess.TimeoutExpired:
return None, "<timeout>"
except OSError as exc:
return 127, f"<exec failed: {exc}>"
return p.returncode, (p.stdout or "") + (p.stderr or "")
def judge_malformed(rc, text, expect_rc, need):
"""One malformed probe's verdict, against the answer its tool states.
Graded in this order: a traceback is a defect whatever the code; a code
that is not the documented one is the defect this class exists to catch
(the silent rc 0); a refusal with no message cannot be acted on; and a
right code with the wrong verdict word is not the documented answer."""
if "Traceback (most recent call last)" in (text or ""):
return {"verdict": "TRACEBACK", "rc": rc, "detail": _first_line(text)}
if rc != expect_rc:
return {"verdict": "RC_WRONG", "rc": rc,
"detail": "rc %s, contract says %s: %s"
% (rc, expect_rc, _first_line(text) or "no output")}
if not (text or "").strip():
return {"verdict": "SILENT", "rc": rc, "detail": "rc %s with no message" % rc}
if need and need not in (text or ""):
return {"verdict": "UNEXPLAINED", "rc": rc,
"detail": "wanted %r, got: %s" % (need, _first_line(text))}
return {"verdict": "OK", "rc": rc, "detail": _first_line(text)}
def axis_faults(axis):
"""The rows that make the axis red. NO_PROBE is one of them: a tool with no
probe entry has not been graded, and 'not graded' is not 'clean'."""
return [r for r in axis if r["verdict"] in FATAL_READ]
def read_axis(rows, timeout: int):
"""One row per (tool, probe). Tools the sweep discovered are probed by path;
a probe naming a tool that was not discovered is NO_TOOL (fatal -- the table
and the probes have drifted apart), and a discovered tool the tables never
name is NO_PROBE on each axis (also fatal: the tool is uncovered)."""
paths = dict(rows)
out = []
for probe, specs in (("missing", READ_PROBES), ("not-utf8", NOT_UTF8_PROBES)):
for name, argv in specs:
rec: dict = {"tool": name, "probe": probe}
if name not in paths:
rec.update({"verdict": "NO_TOOL", "rc": None,
"detail": "probe names a tool the sweep did not discover"})
else:
rc, text = run_read_probe(paths[name], argv, timeout)
rec.update(judge_read(rc, text))
out.append(rec)
for name, label, argv, expect_rc, need in MALFORMED_PROBES:
rec = {"tool": name, "probe": label, "expect_rc": expect_rc}
if name not in paths:
rec.update({"verdict": "NO_TOOL", "rc": None,
"detail": "probe names a tool the sweep did not discover"})
else:
rc, text = run_read_probe(paths[name], argv, timeout)
rec.update(judge_malformed(rc, text, expect_rc, need))
out.append(rec)
probed = {n for n, _ in READ_PROBES}
malformed = {t for t, *_ in MALFORMED_PROBES}
for name, _p in rows:
if name not in probed:
out.append({"tool": name, "probe": "missing", "verdict": NO_PROBE,
"rc": None,
"detail": "no read probe defined -- ungraded, fatal"})
if name not in malformed:
out.append({"tool": name, "probe": "malformed", "verdict": NO_PROBE,
"rc": None,
"detail": "no malformed probe defined -- ungraded, fatal"})
return out
def report_read(axis, quiet=False):
fatal = axis_faults(axis)
hard = [r for r in axis if r["probe"] in ("missing", "not-utf8")]
soft = [r for r in axis if r["probe"] not in ("missing", "not-utf8")]
if not quiet:
print()
print(f"read-refusal axis: an unreadable input exits {READ_AXIS_UNIT}, "
"never 1 (a graded fault)")
width = max([len(r["tool"]) for r in axis] + [4])
for r in axis:
rc = "-" if r["rc"] is None else str(r["rc"])
print(f"{r['tool']:<{width}} {r['probe']:<15} rc={rc:<4} "
f"{r['verdict']:<11} {r['detail']}")
for label, group in (("read-refusal", hard), ("malformed-input", soft)):
ok = sum(1 for r in group if r["verdict"] == "OK")
bad = [r for r in group if r["verdict"] in FATAL_READ]
print(f"{label}: {len(group)} probes, {ok} ok, {len(bad)} fault"
+ (f" ({', '.join(sorted({r['verdict'] for r in bad}))})" if bad else ""))
return 1 if fatal else 0
# ------------------------------------------------------------------- selftest
def selftest():
bad = []
def cell(name, ok, detail=""):
print(f"{'PASS' if ok else 'FAIL'} {name}"
+ (f" [{detail}]" if (detail and not ok) else ""))
if not ok:
bad.append(name)
cell("c1 parse plain cells line",
parse_printed("selftest: 18 cells, 18 passed, 0 failed") == (18, "cells", 18, 0))
cell("c2 parse fixtures with parenthetical",
parse_printed("selftest: 14 fixtures (10 claim-check, 4 unit), 14 passed, 0 failed")
== (14, "fixtures", 14, 0))
cell("c3 parse hyphenated self-test",
parse_printed("self-test: 19 fixtures, 19 passed, 0 failed")
== (19, "fixtures", 19, 0))
cell("c4 parse thousands separator",
parse_printed("selftest: 1,204 cells, 1,204 passed, 0 failed")
== (1204, "cells", 1204, 0))
cell("c5 no count line -> None",
parse_printed("all good, everything fine") is None)
doc = ("USAGE:\n python3 t.py --selftest # synthetic repo, 18 cells\n")
cell("c6 claim on the --selftest usage line",
usage_claims(doc) and usage_claims(doc)[0][:2] == (18, "cells"))
prose = (" returned a docstring claim of `13 cells` against a selftest that\n"
" prints `18`\n")
cell("c7 prose beside the word selftest is not a claim",
usage_claims(prose) == [])
cell("c8 claim with a unit the tool never prints is still a claim",
usage_claims(" --selftest # 7 fixtures\n")[0][:2] == (7, "fixtures"))
j = judge(doc, "selftest: 13 cells, 13 passed, 0 failed", 0, False)
cell("c9 advertised 18 vs printed 13 -> COUNT_MISMATCH",
j["verdict"] == "COUNT_MISMATCH", j["verdict"])
j = judge(doc, "selftest: 18 cells, 18 passed, 0 failed", 0, False)
cell("c10 advertised 18 vs printed 18 -> OK", j["verdict"] == "OK", j["verdict"])
j = judge(" --selftest # 7 fixtures\n", "selftest: 7 cells, 7 passed, 0 failed", 0, False)
cell("c11 unit mismatch is a mismatch, not a pass",
j["verdict"] == "COUNT_MISMATCH", j["verdict"])
j = judge(" --selftest # 18 cells\n --selftest # 13 cells\n",
"selftest: 18 cells, 18 passed, 0 failed", 0, False)
cell("c12 two advertised values -> SPLIT", j["verdict"] == "SPLIT", j["verdict"])
j = judge("", "selftest: 8 cells, 7 passed, 1 failed", 0, False)
cell("c13 failed>0 is fatal even at rc=0",
j["verdict"] == "SELFTEST_FAILED", j["verdict"])
j = judge("", "Traceback (most recent call last):", 1, False)
cell("c14 rc!=0 with no count line -> RUN_FAILED", j["verdict"] == "RUN_FAILED", j["verdict"])
j = judge("", "", 0, False)
cell("c15 rc=0 with no count line -> NO_COUNT_LINE", j["verdict"] == "NO_COUNT_LINE", j["verdict"])
# c16/c17: end to end through main(), on a directory of fake instruments.
good = ('#!/usr/bin/env python3\n"""fake_good.py -- a fake.\n\nUSAGE:\n'
' python3 fake_good.py --selftest # 7 cells\n"""\n'
'import sys\nif "--selftest" in sys.argv:\n'
' print("selftest: 7 cells, 7 passed, 0 failed")\n sys.exit(0)\n')
bad_tool = ('#!/usr/bin/env python3\n"""fake_bad.py -- a fake.\n\nUSAGE:\n'
' python3 fake_bad.py --selftest # 13 cells\n"""\n'
'import sys\nif "--selftest" in sys.argv:\n'
' print("selftest: 18 cells, 18 passed, 0 failed")\n sys.exit(0)\n')
flagless = ('#!/usr/bin/env python3\n"""fake_flagless.py -- no selftest."""\n'
'print("did some work")\n')
with tempfile.TemporaryDirectory() as td:
for nm, body in (("fake_good.py", good), ("fake_bad.py", bad_tool),
("fake_flagless.py", flagless)):
with open(os.path.join(td, nm), "w") as fh:
fh.write(body)
rc_bad = main(["--tools-dir", td, "--quiet"])
rows, source, untracked = discover(td, td)
res = sweep(rows, td, 30, False)
verdicts = {r["tool"]: r["verdict"] for r in res}
cell("c16 end to end: a wrong claim is fatal",
rc_bad == 1 and verdicts.get("fake_bad.py") == "COUNT_MISMATCH",
f"rc={rc_bad} {verdicts}")
cell("c17 end to end: correct claim OK, flagless caught, exit 1",
verdicts.get("fake_good.py") == "OK"
and verdicts.get("fake_flagless.py") == "NO_SELFTEST_FLAG",
str(verdicts))
jd = os.path.join(td, "sweep.json")
main(["--tools-dir", td, "--quiet", "--json", jd])
try:
with open(jd) as fh:
blob = json.load(fh)
ok_json = (isinstance(blob, dict) and len(blob.get("tools", [])) == 3
and blob.get("fatal") == 2)
except (OSError, ValueError):
ok_json = False
cell("c18 --json carries every row and the fatal count", ok_json)
# c19-c22: the read-refusal axis's judgement, one cell per class. The axis
# is the gate on "an input I cannot read must not exit 1"; these cells are
# the gate on the gate.
jr = judge_read(2, "cannot read census /nope/a.json: [Errno 2]")
cell("c19 a refusal at rc 2 with a message -> OK", jr["verdict"] == "OK", jr["verdict"])
jr = judge_read(1, "MISSING: the check has a fault")
cell("c20 a graded fault's rc 1 on an unreadable input -> RC_NOT_2",
jr["verdict"] == "RC_NOT_2", jr["verdict"])
jr = judge_read(2, "Traceback (most recent call last):\n File x")
cell("c21 a traceback is fatal even when it exits 2",
jr["verdict"] == "TRACEBACK", jr["verdict"])
jr = judge_read(2, "")
cell("c22 rc 2 with no message -> SILENT", jr["verdict"] == "SILENT", jr["verdict"])
# c23-c28: the malformed class's judgement, and its coverage. A decodable
# input has no single right answer -- each tool's contract decides -- so
# these cells pin the judgement itself: the documented rc passes, a wrong
# code fails, and a right code with no explanation fails too.
jm = judge_malformed(0, "0 opening records; 0 closed; 0 open.", 0, "0 opening records")
cell("c23 a documented rc 0 grade on malformed input -> OK",
jm["verdict"] == "OK", jm["verdict"])
jm = judge_malformed(0, "no tools found at HEAD", 2, "no tools found")
cell("c24 the silent-zero: rc 0 where the contract says 2 -> RC_WRONG",
jm["verdict"] == "RC_WRONG", jm["verdict"])
jm = judge_malformed(2, " ", 2, "")
cell("c25 a refusal with no message -> SILENT", jm["verdict"] == "SILENT", jm["verdict"])
jm = judge_malformed(2, "something went wrong", 2, "cannot read")
cell("c26 the right code with the wrong reason -> UNEXPLAINED",
jm["verdict"] == "UNEXPLAINED", jm["verdict"])
jm = judge_malformed(2, "Traceback (most recent call last):", 2, "")
cell("c27 a traceback is fatal even at the documented code",
jm["verdict"] == "TRACEBACK", jm["verdict"])
missing_probed = {t for t, _ in READ_PROBES}
malformed_probed = {t for t, *_ in MALFORMED_PROBES}
cell("c28 every tool with a missing-probe also has a malformed probe",
missing_probed == malformed_probed,
str(sorted(missing_probed ^ malformed_probed)))
# c29/c30: the NO_PROBE hole, measured from the other side -- a tool the
# tables never name must come out uncovered on BOTH axes and must make the
# axis red. Before this cell existed, read_axis emitted two advisory rows and
# a 15-tool directory graded "14 OK, 0 fatal, rc 0".
alone = read_axis([("zzz_unprobed.py", "/nonexistent/zzz_unprobed.py")], 5)
mine = [r for r in alone if r["tool"] == "zzz_unprobed.py"]
cell("c29 an unprobed tool is NO_PROBE on both axes and is a fault",
sorted(r["probe"] for r in mine) == ["malformed", "missing"]
and len(axis_faults(mine)) == 2,
str([(r["probe"], r["verdict"]) for r in mine]))
cell("c30 every FATAL_READ verdict is red and OK is not",
all(axis_faults([{"verdict": v}]) for v in FATAL_READ)
and not axis_faults([{"verdict": "OK"}])
and NO_PROBE in FATAL_READ,
str(FATAL_READ))
print(f"selftest: 30 cells, {30 - len(bad)} passed, {len(bad)} failed")
return 1 if bad else 0
# ----------------------------------------------------------------------- main
def main(argv=None):
ap = argparse.ArgumentParser(
description="Run every tool's --selftest and check its advertised cell count.")
ap.add_argument("--repo", default="/root/troupe-standup")
ap.add_argument("--tools-dir", default=None,
help="sweep this directory instead of the repo's tracked tools/")
ap.add_argument("--timeout", type=int, default=120, metavar="S")
ap.add_argument("--json", default=None, metavar="PATH")
ap.add_argument("--quiet", action="store_true")
ap.add_argument("--verbose", action="store_true")
ap.add_argument("--selftest", action="store_true")
a = ap.parse_args(argv)
if a.selftest:
return selftest()
rows, source, untracked = discover(a.repo, a.tools_dir)
if not rows:
# A sweep that discovered nothing has verified nothing. It is not a
# clean run (rc 0) and not a fatal verdict (rc 1): it is rc 2, the
# repo-wide code for "could not grade" -- the same contract the
# read-refusal axis below measures on every other tool.
print("cannot read: no tools discovered in %s (%s)"
% (a.tools_dir or os.path.join(a.repo, "tools"), source),
file=sys.stderr)
return 2
results = sweep(rows, a.repo if not a.tools_dir else os.path.abspath(a.tools_dir),
a.timeout, a.verbose)
rc = report(results, source, untracked, a.quiet)
axis = read_axis(rows, a.timeout)
rrc = report_read(axis, a.quiet)
if a.json:
fatal = [r for r in results if r["verdict"] in FATAL]
with open(a.json, "w") as fh:
json.dump({"repo": os.path.abspath(a.repo),
"tools_dir": a.tools_dir,
"discovery": source,
"untracked": untracked,
"tools": results,
"fatal": len(fatal),
"ok": sum(1 for r in results if r["verdict"] == "OK"),
"read_axis": axis,
"read_fatal": sum(1 for r in axis
if r["verdict"] in FATAL_READ)}, fh, indent=1)
return 1 if (rc or rrc) else 0
if __name__ == "__main__":
sys.exit(main())
Reference in new issue View Git Blame Copy Permalink