volt
52784d4799
q65_no_verdict.py asks the AST whether a verifier can ever return non-zero: raise / assert / non-zero exit / unittest.main. 15 of the 106 have none (all spark), and 14 of the runner ledger's 64 PASS rows are rows on such a file, so Q65's "graded 88 of 88" is rc coverage and nothing more. Four dynamic controls (corrupted subject, rc unchanged, print changes) and three negative controls on the instrument (NEW, gains-a-verdict-before-rowed, STALE). Wired as CHECK's fourth leg. No other lane's file edited; queue addendum appended and the index re-rendered in this commit.
327 lines
14 KiB
Python
327 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""q65_no_verdict.py -- does a committed verifier have a failure path at all?
|
|
|
|
Q65's census grades each committed `wizard-work/**/verify_*.py` by running it and
|
|
recording its rc. A verifier whose source has no failure path returns rc 0 for
|
|
every input, so its row is a constant: it can never be a FAIL, and `graded 88 of
|
|
88` counts it as graded. Q65 cycle 4 found one such file by hand
|
|
(`ledger/verify_dates.py`: prints six lines and falls off the end of main());
|
|
this instrument is the static question asked of all of them.
|
|
|
|
WHAT COUNTS AS A VERDICT PATH (ast, so docstrings and comments are invisible --
|
|
which is the defect being measured: prose about failing is not a failure path):
|
|
|
|
raise <anything> a Raise node
|
|
assert <expr> an Assert node
|
|
sys.exit(<nonzero>) / exit(<nonzero>) / raise SystemExit(<nonzero>)
|
|
unittest.main() / pytest.main() the runner exits (pytest.main: see WEAK)
|
|
subprocess.run(..., check=True) raises on a bad rc -- WEAK, see below
|
|
|
|
WEAK-SUBPROCESS is its own class and not a verdict: a `check=True` grades the
|
|
child process, not the subject. `ledger/verify_dates.py` carries exactly one
|
|
(its `git log` call) and scores WEAK here while its comparison loop prints and
|
|
returns -- so WEAK is printed per-file for a human to read and is never a fault
|
|
on its own.
|
|
|
|
WHAT THIS CANNOT SEE, stated so the numbers are not read as stronger than they
|
|
are: a `raise` inside a swallowed `try/except` counts as a verdict path (false
|
|
negative for the printer class); a comparison whose result is printed and
|
|
`sys.exit(0 if ok else 1)`-ed counts correctly; a file that never runs to
|
|
completion is the runner's question (its row is TIMEOUT, not mine).
|
|
|
|
Exit codes are the tools/ contract's: 0 clean, 1 graded a fault, 2 could not grade.
|
|
--check ledger vs tree at --rev; NEW no-verdict file or a STALE ledger row = rc 1
|
|
--emit rewrite the ledger from the tree
|
|
--json the full classification, for re-derivation
|
|
"""
|
|
import argparse
|
|
import ast
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import warnings
|
|
|
|
# several committed verifiers carry a "\u" in a docstring; that is a SyntaxWarning
|
|
# on parse and not this instrument's business.
|
|
warnings.filterwarnings("ignore", category=SyntaxWarning)
|
|
|
|
DEFAULT_LEDGER = os.path.join(os.path.dirname(os.path.abspath(__file__)),
|
|
"q65_no_verdict.tsv")
|
|
RUNNER_LEDGER = os.path.join(os.path.dirname(os.path.abspath(__file__)),
|
|
"q65_runner_baseline.tsv")
|
|
EXIT_NAMES = {"exit", "_exit", "quit"}
|
|
WEAK_KW = {"check_call", "check_output"}
|
|
|
|
|
|
def git(repo, *args):
|
|
p = subprocess.run(["git", "-C", repo, *args], capture_output=True, text=True)
|
|
if p.returncode != 0:
|
|
raise RuntimeError((p.stderr or p.stdout).strip())
|
|
return p.stdout
|
|
|
|
|
|
def _nonzero(arg):
|
|
"""Is this exit() argument a non-zero code? Unknown expressions count as
|
|
non-zero: an exit(<expr>) is a deliberate status, and the defect class is the
|
|
file with no status at all."""
|
|
if arg is None:
|
|
return False
|
|
if isinstance(arg, ast.Constant):
|
|
if arg.value is None or arg.value == 0 or arg.value is False:
|
|
return False
|
|
return True
|
|
return True
|
|
|
|
|
|
def classify(src):
|
|
"""Return (klass, tokens) for one verifier's source text."""
|
|
t = {"raise": 0, "assert": 0, "exit": 0, "runner_main": 0, "weak": 0,
|
|
"exit_zero": 0}
|
|
try:
|
|
tree = ast.parse(src)
|
|
except SyntaxError as e:
|
|
return "UNPARSEABLE", dict(t, error=str(e))
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.Raise):
|
|
t["raise"] += 1
|
|
elif isinstance(node, ast.Assert):
|
|
t["assert"] += 1
|
|
elif isinstance(node, ast.Call):
|
|
f = node.func
|
|
name = (f.attr if isinstance(f, ast.Attribute)
|
|
else f.id if isinstance(f, ast.Name) else "")
|
|
if name in EXIT_NAMES:
|
|
arg = node.args[0] if node.args else None
|
|
if _nonzero(arg):
|
|
t["exit"] += 1
|
|
else:
|
|
t["exit_zero"] += 1 # sys.exit() / exit(0) is rc 0
|
|
elif name in WEAK_KW:
|
|
t["weak"] += 1
|
|
elif name == "run":
|
|
for kw in node.keywords:
|
|
if kw.arg == "check" and isinstance(kw.value, ast.Constant) \
|
|
and kw.value.value:
|
|
t["weak"] += 1
|
|
elif name == "main" and isinstance(f, ast.Attribute) and \
|
|
isinstance(f.value, ast.Name) and f.value.id == "unittest":
|
|
t["runner_main"] += 1
|
|
elif name == "main" and isinstance(f, ast.Attribute) and \
|
|
isinstance(f.value, ast.Name) and f.value.id == "pytest":
|
|
t["weak"] += 1 # pytest.main returns an int, does not exit
|
|
strong = t["raise"] + t["assert"] + t["exit"] + t["runner_main"]
|
|
if strong and not t["weak"]:
|
|
return "has-verdict", t
|
|
if strong:
|
|
return "has-verdict+weak", t
|
|
if t["weak"]:
|
|
return "weak-subprocess", t
|
|
return "no-verdict", t
|
|
|
|
|
|
def verifiers(repo, rev):
|
|
out = git(repo, "ls-tree", "-r", "--name-only", rev).split()
|
|
return sorted(p for p in out
|
|
if p.startswith("wizard-work/") and "/verify_" in p
|
|
and p.endswith(".py"))
|
|
|
|
|
|
def census(repo, rev):
|
|
rows = []
|
|
for p in verifiers(repo, rev):
|
|
src = git(repo, "show", f"{rev}:{p}")
|
|
klass, t = classify(src)
|
|
rows.append({"path": p, "class": klass, "tokens": t,
|
|
"loc": src.count("\n") + 1,
|
|
"blob": git(repo, "rev-parse", f"{rev}:{p}").strip()[:12]})
|
|
return rows
|
|
|
|
|
|
def read_ledger(path):
|
|
led = {}
|
|
if not os.path.isfile(path):
|
|
return led
|
|
for ln in open(path, encoding="utf-8"):
|
|
if not ln.strip() or ln.lstrip().startswith("#"):
|
|
continue
|
|
parts = ln.rstrip("\n").split("\t")
|
|
if len(parts) >= 2:
|
|
led[parts[1]] = parts
|
|
return led
|
|
|
|
|
|
def read_runner(path):
|
|
"""path -> verdict, for rows whose rc agrees with the verdict."""
|
|
rows = {}
|
|
if not os.path.isfile(path):
|
|
return rows
|
|
for ln in open(path, encoding="utf-8"):
|
|
if not ln.strip() or ln.lstrip().startswith("#"):
|
|
continue
|
|
parts = ln.rstrip("\n").split("\t")
|
|
if len(parts) >= 4:
|
|
rows[parts[1]] = (parts[2], parts[3])
|
|
return rows
|
|
|
|
|
|
def emit(rows, path, rev):
|
|
bad = [r for r in rows if r["class"] in ("no-verdict", "weak-subprocess",
|
|
"UNPARSEABLE")]
|
|
with open(path, "w", encoding="utf-8") as f:
|
|
f.write("# Q65 no-verdict census -- one row per committed verifier with no\n")
|
|
f.write("# failure path of its own (or only a subprocess check). Written by\n")
|
|
f.write("# q65_no_verdict.py --emit; a row states the finding, it is not an\n")
|
|
f.write("# exemption: --check faults on a no-verdict file this file does not\n")
|
|
f.write("# name, and on a row whose file has since gained a verdict path.\n")
|
|
f.write(f"# rev {rev} committed verifiers {len(rows)} rows {len(bad)}\n")
|
|
f.write("# class\tpath\tloc\tblob\ttokens\tevidence\n")
|
|
for r in bad:
|
|
tok = ",".join(f"{k}={v}" for k, v in sorted(r["tokens"].items()) if v)
|
|
f.write("\t".join([r["class"], r["path"], str(r["loc"]), r["blob"],
|
|
tok or "no-failure-token", ""]) + "\n")
|
|
return bad
|
|
|
|
|
|
def check(rows, led, runner):
|
|
nov = [r for r in rows if r["class"] == "no-verdict"]
|
|
weak = [r for r in rows if r["class"] == "weak-subprocess"]
|
|
known = {p: v for p, v in led.items() if v[0] == "no-verdict"}
|
|
new = [r for r in nov if r["path"] not in known]
|
|
stale = [p for p in known if p not in {r["path"] for r in nov}]
|
|
# a PASS row on a file with no verdict path is a constant, not a grade
|
|
vacuous = [r for r in nov
|
|
if runner.get(r["path"], ("", ""))[0] in ("PASS",)]
|
|
return nov, weak, new, stale, vacuous
|
|
|
|
|
|
def report(repo, rev, led, runner, verbose=False):
|
|
rows = census(repo, rev)
|
|
nov, weak, new, stale, vacuous = check(rows, led, runner)
|
|
lanes = {}
|
|
for r in nov:
|
|
lane = r["path"].split("/")[1]
|
|
lanes[lane] = lanes.get(lane, 0) + 1
|
|
print(f"q65_no_verdict -- rev {rev}")
|
|
print(" committed verifiers : %d" % len(rows))
|
|
print(" no failure path : %d (%s)"
|
|
% (len(nov), " ".join(f"{k} {v}" for k, v in sorted(lanes.items(),
|
|
key=lambda kv: -kv[1]))
|
|
or "none"))
|
|
for r in nov:
|
|
tok = ",".join(f"{k}={v}" for k, v in sorted(r["tokens"].items()) if v) or "-"
|
|
print(f" NO-VERDICT {r['path']:64s} loc {r['loc']:3d} {tok}")
|
|
print(" weak only (subprocess) : %d" % len(weak))
|
|
for r in weak:
|
|
print(f" WEAK {r['path']}")
|
|
print(" runner PASS rows that cannot fail: %d of %d"
|
|
% (len(vacuous), sum(1 for v in runner.values() if v[0] == "PASS")))
|
|
if verbose:
|
|
for r in vacuous:
|
|
print(f" VACUOUS {r['path']} "
|
|
f"(runner row {runner[r['path']][0]} rc {runner[r['path']][1]})")
|
|
for r in new:
|
|
print(f" NEW {r['path']} (no-verdict, not in {os.path.basename(DEFAULT_LEDGER)})")
|
|
for p in stale:
|
|
print(f" STALE {p} (ledger row, file now has a verdict path or is gone)")
|
|
faults = len(new) + len(stale)
|
|
print("q65_no_verdict: %d fault(s)" % faults)
|
|
return faults
|
|
|
|
|
|
FIXTURES = [
|
|
("printer", "print('all good')\nopen('/tmp/x','w').write('ok')\n",
|
|
"no-verdict"),
|
|
("bare exit", "import sys\nprint('fine')\nsys.exit()\n", "no-verdict"),
|
|
("exit 0", "import sys\nsys.exit(0)\n", "no-verdict"),
|
|
("assert only", "assert 1 + 1 == 2\n", "has-verdict"),
|
|
("sys.exit(nonzero)", "import sys\nsys.exit(1)\n", "has-verdict"),
|
|
("raise SystemExit", "raise SystemExit(2)\n", "has-verdict"),
|
|
("subprocess only", "import subprocess\nsubprocess.run(['true'], check=True)\n",
|
|
"weak-subprocess"),
|
|
("unittest.main", "import unittest\nclass T(unittest.TestCase):\n"
|
|
" def test_a(self):\n self.assertTrue(True)\n"
|
|
"if __name__ == '__main__':\n unittest.main()\n",
|
|
"has-verdict"),
|
|
("assert + swallowed", "try:\n assert False\nexcept Exception:\n pass\n",
|
|
"has-verdict"),
|
|
]
|
|
|
|
|
|
def selftest():
|
|
bad = 0
|
|
for name, src, want in FIXTURES:
|
|
got, t = classify(src)
|
|
ok = got == want
|
|
bad += 0 if ok else 1
|
|
print(f" {'ok ' if ok else 'FAIL'} {name:22s} -> {got} (want {want})")
|
|
# check(): a NEW no-verdict file faults, a STALE row faults, a match is clean
|
|
rows = [{"path": "wizard-work/spark/verify_a.py", "class": "no-verdict",
|
|
"tokens": {}, "loc": 1, "blob": "x"},
|
|
{"path": "wizard-work/spark/verify_b.py", "class": "has-verdict",
|
|
"tokens": {}, "loc": 1, "blob": "y"}]
|
|
led = {"wizard-work/spark/verify_a.py": ["no-verdict", "wizard-work/spark/verify_a.py"]}
|
|
nov, weak, new, stale, vac = check(rows, led, {})
|
|
ok = (len(nov) == 1 and len(new) == 0 and len(stale) == 0)
|
|
print(f" {'ok ' if ok else 'FAIL'} ledger match -> new {len(new)} stale {len(stale)}")
|
|
bad += 0 if ok else 1
|
|
rows[0]["class"] = "has-verdict"
|
|
nov, weak, new, stale, vac = check(rows, led, {})
|
|
ok = len(stale) == 1
|
|
print(f" {'ok ' if ok else 'FAIL'} fixed file is STALE -> stale {len(stale)}")
|
|
bad += 0 if ok else 1
|
|
rows[0]["class"] = "no-verdict"
|
|
nov, weak, new, stale, vac = check(rows, {}, {})
|
|
ok = len(new) == 1
|
|
print(f" {'ok ' if ok else 'FAIL'} unrowed file is NEW -> new {len(new)}")
|
|
bad += 0 if ok else 1
|
|
vac_rows = check(rows, led, {"wizard-work/spark/verify_a.py": ("PASS", "0")})
|
|
ok = len(vac_rows[4]) == 1
|
|
print(f" {'ok ' if ok else 'FAIL'} PASS row is vacuous -> vacuous {len(vac_rows[4])}")
|
|
bad += 0 if ok else 1
|
|
print(f"q65_no_verdict selftest: {'PASS' if not bad else 'FAIL'} "
|
|
f"({len(FIXTURES) + 4 - bad}/{len(FIXTURES) + 4})")
|
|
return bad
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--repo", default=".")
|
|
ap.add_argument("--rev", default="HEAD")
|
|
ap.add_argument("--ledger", default=DEFAULT_LEDGER)
|
|
ap.add_argument("--runner-ledger", default=RUNNER_LEDGER)
|
|
ap.add_argument("--check", action="store_true")
|
|
ap.add_argument("--emit", action="store_true")
|
|
ap.add_argument("--json")
|
|
ap.add_argument("--verbose", action="store_true")
|
|
ap.add_argument("--selftest", action="store_true")
|
|
a = ap.parse_args()
|
|
|
|
if a.selftest:
|
|
return 0 if selftest() == 0 else 1
|
|
try:
|
|
git(a.repo, "rev-parse", a.rev)
|
|
except (RuntimeError, FileNotFoundError) as e:
|
|
print(f"q65_no_verdict: cannot grade -- {e}", file=sys.stderr)
|
|
return 2
|
|
try:
|
|
if a.emit:
|
|
rows = census(a.repo, a.rev)
|
|
bad = emit(rows, a.ledger, a.rev)
|
|
print(f"q65_no_verdict: {len(bad)} row(s) -> {a.ledger} "
|
|
f"({len(rows)} committed verifiers at {a.rev})")
|
|
return 0
|
|
faults = report(a.repo, a.rev, read_ledger(a.ledger),
|
|
read_runner(a.runner_ledger), a.verbose)
|
|
if a.json:
|
|
with open(a.json, "w", encoding="utf-8") as f:
|
|
json.dump(census(a.repo, a.rev), f, indent=1)
|
|
return 0 if faults == 0 else 1
|
|
except (RuntimeError, FileNotFoundError) as e:
|
|
print(f"q65_no_verdict: cannot grade -- {e}", file=sys.stderr)
|
|
return 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|