q63: q63_prove.py -- the -O proof harness, re-runnable: six old/-O rc 0 (89cda97) vs six new/-O rc 1, on deliberately broken subjects
This commit is contained in:
spark committed
1 parent
5853fe4faa
commit
2a484e4873
1 file changed
+179
@@ -0,0 +1,179 @@
|
||||
"""q63_prove.py -- Q63 proof harness: for each of the six stripped-assert
|
||||
verifiers, run
|
||||
|
||||
(old committed blob, normal) (old committed blob, -O)
|
||||
(new worktree file, normal) (new worktree file, -O)
|
||||
|
||||
against a DELIBERATELY BROKEN input, and print every rc. The class-1 defect is
|
||||
old/-O returning 0 (PASS) on a broken subject; the fix is rc != 0 there.
|
||||
Measured 2026-10-24: RED at `89cda97` (six old/-O runs rc 0), GREEN at `c033059`
|
||||
(six new/-O runs rc 1), rc 0 from this harness only when all six are fixed.
|
||||
|
||||
Isolation: each case runs in its own scratch dir. Where a verifier hardcodes an
|
||||
absolute input path, the demonstration copy differs from the committed file in
|
||||
that one literal only (input path / subject cwd).
|
||||
|
||||
Usage: python3 wizard-work/volt/q63_prove.py [repo] [old-rev]
|
||||
defaults: repo /root/troupe-standup, old-rev c033059's parent 89cda97
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
REPO = sys.argv[1] if len(sys.argv) > 1 else "/root/troupe-standup"
|
||||
OLDREV = sys.argv[2] if len(sys.argv) > 2 else "89cda97"
|
||||
WORK = os.path.join(os.environ.get("TMPDIR", "/tmp"), "q63_proof")
|
||||
PY = sys.executable
|
||||
|
||||
|
||||
def sh(cmd, cwd=None):
|
||||
p = subprocess.run(cmd, capture_output=True, text=True, cwd=cwd)
|
||||
return p.returncode, (p.stdout + p.stderr).strip()
|
||||
|
||||
|
||||
def case(name, old_src, new_src, fixture, runs):
|
||||
"""fixture(dir) builds the broken input; runs is a list of argv lists."""
|
||||
d = os.path.join(WORK, name)
|
||||
shutil.rmtree(d, ignore_errors=True)
|
||||
os.makedirs(d)
|
||||
fixture(d)
|
||||
out = []
|
||||
for label, src in (("old", old_src), ("new", new_src)):
|
||||
path = os.path.join(d, f"v_{label}.py")
|
||||
with open(path, "w") as f:
|
||||
f.write(src)
|
||||
for mode, flag in (("normal", []), ("-O", ["-O"])):
|
||||
for argv in runs:
|
||||
rc, txt = sh([PY, *flag, path, *argv], cwd=d)
|
||||
last = txt.splitlines()[-1] if txt.splitlines() else ""
|
||||
out.append((label, mode, " ".join(argv), rc, last[:90]))
|
||||
return out
|
||||
|
||||
|
||||
def broken_json(d, fname, obj):
|
||||
with open(os.path.join(d, fname), "w") as f:
|
||||
json.dump(obj, f)
|
||||
|
||||
|
||||
results = {}
|
||||
|
||||
# 1. beam -- subject module shadowed by a topo_sort that mangles None deps
|
||||
old = subprocess.run(["git", "-C", REPO, "show", f"{OLDREV}:wizard-work/beam/verify_toposort_none.py"],
|
||||
capture_output=True, text=True).stdout
|
||||
new = open(f"{REPO}/wizard-work/beam/verify_toposort_none.py").read()
|
||||
|
||||
|
||||
def fx_beam(d):
|
||||
topo = subprocess.run(["git", "-C", REPO, "show", f"{OLDREV}:wizard-work/beam/toposort.py"],
|
||||
capture_output=True, text=True).stdout
|
||||
with open(os.path.join(d, "toposort.py"), "w") as f:
|
||||
f.write(topo + "\n\n# --- Q63 broken-subject shim: mangle None-dependency results ---\n"
|
||||
"_real = topo_sort\n\n\nclass _R:\n sorted = None\n cycle = None\n\n\n"
|
||||
"def topo_sort(deps):\n return _R()\n")
|
||||
|
||||
|
||||
results["beam_toposort_none"] = case("beam", old, new, fx_beam, [[]])
|
||||
|
||||
# 2. ledger -- CSV whose header is wrong and which has 3 rows
|
||||
old = subprocess.run(["git", "-C", REPO, "show", f"{OLDREV}:wizard-work/ledger/verify_arxiv_csv.py"],
|
||||
capture_output=True, text=True).stdout
|
||||
new = open(f"{REPO}/wizard-work/ledger/verify_arxiv_csv.py").read()
|
||||
|
||||
|
||||
def fx_csv(d):
|
||||
# correct header, 3 well-formed rows: only the row-count check can object,
|
||||
# so the -O run reaches the success print instead of crashing on a KeyError.
|
||||
rows = ["arxiv_id,title,authors,published,abstract"]
|
||||
for i in range(3):
|
||||
rows.append(f"2610.{10000+i},t{i},a{i},2026-10-19,abstract{i}")
|
||||
with open(os.path.join(d, "b.csv"), "w") as f:
|
||||
f.write("\n".join(rows) + "\n")
|
||||
|
||||
|
||||
results["ledger_verify_arxiv_csv"] = case("ledger", old, new, fx_csv, [["b.csv"]])
|
||||
|
||||
# 3. spark/verify_report.py -- summary counts wrong
|
||||
old = subprocess.run(["git", "-C", REPO, "show", f"{OLDREV}:wizard-work/spark/verify_report.py"],
|
||||
capture_output=True, text=True).stdout
|
||||
new = open(f"{REPO}/wizard-work/spark/verify_report.py").read()
|
||||
OLD_PATH = "/root/.hermes/wizard-outputs/spark/validation_report.json"
|
||||
|
||||
|
||||
def fx_report(d):
|
||||
broken_json(d, "r.json", {"summary": {"valid_count": 5, "invalid_count": 0}, "test_passed": False})
|
||||
|
||||
|
||||
def repoint(src, path, fname):
|
||||
assert path in src, path
|
||||
return src.replace(path, fname)
|
||||
|
||||
|
||||
results["spark_verify_report"] = case(
|
||||
"report", repoint(old, OLD_PATH, "r.json"), repoint(new, OLD_PATH, "r.json"), fx_report, [[]])
|
||||
|
||||
# 4. spark/verify_result.py -- wrong title
|
||||
old = subprocess.run(["git", "-C", REPO, "show", f"{OLDREV}:wizard-work/spark/verify_result.py"],
|
||||
capture_output=True, text=True).stdout
|
||||
new = open(f"{REPO}/wizard-work/spark/verify_result.py").read()
|
||||
OLD_RES = "/root/.hermes/wizard-outputs/spark/scrape_result.json"
|
||||
|
||||
|
||||
def fx_result(d):
|
||||
broken_json(d, "s.json", {"title": "Wrong Domain", "status": "success"})
|
||||
|
||||
|
||||
results["spark_verify_result"] = case(
|
||||
"result", repoint(old, OLD_RES, "s.json"), repoint(new, OLD_RES, "s.json"), fx_result, [[]])
|
||||
|
||||
# 5. spark/verify_scratch_analysis.py -- total_files 7 against 7 listed files
|
||||
old = subprocess.run(["git", "-C", REPO, "show", f"{OLDREV}:wizard-work/spark/verify_scratch_analysis.py"],
|
||||
capture_output=True, text=True).stdout
|
||||
new = open(f"{REPO}/wizard-work/spark/verify_scratch_analysis.py").read()
|
||||
OLD_SCR = "/root/.hermes/wizard-outputs/spark/scratch_analysis.json"
|
||||
|
||||
|
||||
def fx_scratch(d):
|
||||
files = [{"name": f"f{i}.py", "path": f"/x/f{i}.py", "bytes": 1, "extension": ".py"} for i in range(7)]
|
||||
broken_json(d, "a.json", {"directory": "/x", "total_files": 7, "total_bytes": 7,
|
||||
"file_types": {".py": 7}, "files": files})
|
||||
|
||||
|
||||
results["spark_verify_scratch_analysis"] = case(
|
||||
"scratch", repoint(old, OLD_SCR, "a.json"), repoint(new, OLD_SCR, "a.json"), fx_scratch, [[]])
|
||||
|
||||
# 6. spark/verify_dir_analyzer.py -- subject prints JSON with 0 files
|
||||
old = subprocess.run(["git", "-C", REPO, "show", f"{OLDREV}:wizard-work/spark/verify_dir_analyzer.py"],
|
||||
capture_output=True, text=True).stdout
|
||||
new = open(f"{REPO}/wizard-work/spark/verify_dir_analyzer.py").read()
|
||||
SPARK_CWD = "cwd='/root/troupe-standup/wizard-work/spark'"
|
||||
|
||||
|
||||
def fx_dir(d):
|
||||
with open(os.path.join(d, "dir_analyzer.py"), "w") as f:
|
||||
f.write('import json, sys\nif "--json" in sys.argv:\n'
|
||||
' print(json.dumps({"total_files": 0, "total_size_bytes": 0}))\n'
|
||||
'else:\n print("nothing here")\n')
|
||||
|
||||
|
||||
results["spark_verify_dir_analyzer"] = case(
|
||||
"dir", old.replace(SPARK_CWD, "cwd=__import__('os').path.dirname(__file__)"),
|
||||
new.replace(SPARK_CWD, "cwd=__import__('os').path.dirname(__file__)"), fx_dir, [[]])
|
||||
|
||||
print(f"{'case':32s} {'who':4s} {'mode':7s} {'argv':10s} rc last line")
|
||||
bad = 0
|
||||
for name, rows in results.items():
|
||||
for label, mode, argv, rc, last in rows:
|
||||
print(f"{name:32s} {label:4s} {mode:7s} {argv:10s} {rc:<3d} {last}")
|
||||
old_o = [r[3] for r in rows if r[0] == "old" and r[1] == "-O"][0]
|
||||
new_o = [r[3] for r in rows if r[0] == "new" and r[1] == "-O"][0]
|
||||
new_n = [r[3] for r in rows if r[0] == "new" and r[1] == "normal"][0]
|
||||
if old_o == 0 and new_o != 0:
|
||||
verdict = "FIXED (old/-O falsely PASSed, new/-O fails)"
|
||||
else:
|
||||
verdict = f"UNEXPECTED (old/-O rc {old_o}, new/-O rc {new_o}, new/normal rc {new_n})"
|
||||
bad += 1
|
||||
print(f" -> {name}: {verdict}")
|
||||
print(f"\ncases where the defect is not demonstrated+fixed: {bad}")
|
||||
sys.exit(1 if bad else 0)
|
||||
Reference in new issue
Block a user