volt
6ee8fcee5a
Six tools answered an input they could not read with a code that means
something else: check_lint and census_row_delta exited 1 (the code that
means "the grading found a fault"), the two append-only tools exited 1 out
of a raw traceback, selftest_sweep exited 1 with "0 tools, nothing
discovered", and the two queue readers exited 3 -- one meaning, three
codes. Measured before the change: 11 of 15 broken-input probes were
mis-coded (2 tracebacks, 9 wrong code).
The contract is now three codes and no others: rc 0 ran clean, rc 1 ran
and graded a fault, rc 2 could not grade (usage, precondition, unreadable
input). rc 3 is retired.
And the class is gated, not merely fixed: selftest_sweep.py gains a
read-refusal axis -- 15 probes (11 missing-input, 4 non-UTF-8) run
against the real CLI in a throwaway cwd, graded OK / RC_NOT_2 / TRACEBACK
/ SILENT, fatal on any of the last three. On HEAD's tools the axis reads
11 faults and exits 1; on this tree it reads 15 ok and exits 0.
Also: doc_census_check.py advertises its own count (33 cells) so the sweep
has no advisory row left (11 tools, 11 OK); .mailmap folds volt's five
commit spellings (18+7+3+1+1 = 30 commits) to Volt <volt@agents.invalid>,
the identity 016d26e lost to the repo's Timmy config.
Selftests: 18/15/33/37/54/16/22, 0 failed. Sweep 11 tools, 11 OK, 0 fatal.
queue_index --check rc 0 (49 sections, 45 ids, 9 nested). landing_check
346/346.
596 lines
27 KiB
Python
596 lines
27 KiB
Python
#!/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. 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.
|
|
|
|
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.
|
|
|
|
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 # 22 cells, parser, sweep and read axis
|
|
"""
|
|
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
|
|
FATAL_READ = ("TRACEBACK", "RC_NOT_2", "SILENT", "NO_TOOL")
|
|
NO_PROBE = "NO_PROBE" # advisory: a tool with no probe defined
|
|
|
|
# {nope} = a path that does not exist; {bin} = a file that is not UTF-8.
|
|
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"]),
|
|
("citation_hash_class.py", ["--repo", "{nope}"]),
|
|
("citation_resolve_check.py", ["--repo", "{nope}"]),
|
|
("doc_census_check.py", ["--repo", "{nope}", "--no-run"]),
|
|
("landing_check.py", ["--repo", "{nope}"]),
|
|
("queue_claim_check.py", ["--queue", "{nope}/q.md"]),
|
|
("queue_index.py", ["--queue", "{nope}/q.md", "--check", "{nope}/i.md"]),
|
|
("selftest_sweep.py", ["--repo", "{nope}"]),
|
|
)
|
|
NOT_UTF8_PROBES = (
|
|
("append_only_owners.py", ["{bin}"]),
|
|
("census_row_delta.py", ["{bin}", "{bin}"]),
|
|
("queue_claim_check.py", ["--queue", "{bin}"]),
|
|
("queue_index.py", ["--queue", "{bin}", "--check", "{bin}"]),
|
|
)
|
|
|
|
|
|
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")
|
|
nope = os.path.join(td, "there-is-no-such-path")
|
|
real = [a.replace("{nope}", nope).replace("{bin}", binpath) for a in argv]
|
|
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 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)."""
|
|
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)
|
|
probed = {n for n, _ in READ_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 (advisory)"})
|
|
return out
|
|
|
|
|
|
def report_read(axis, quiet=False):
|
|
fatal = [r for r in axis if r["verdict"] in FATAL_READ]
|
|
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']:<9} rc={rc:<4} "
|
|
f"{r['verdict']:<10} {r['detail']}")
|
|
ok = sum(1 for r in axis if r["verdict"] == "OK")
|
|
print(f"read-refusal: {len(axis)} probes, {ok} ok, {len(fatal)} fault"
|
|
+ (f" ({', '.join(sorted({r['verdict'] for r in fatal}))})" if fatal 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"])
|
|
|
|
print(f"selftest: 22 cells, {22 - 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())
|