284 lines
11 KiB
Python
284 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""q153_verify.py -- grade the Q153 repair: a beam test that graded the host.
|
|
|
|
The claim this gate exists to keep true:
|
|
|
|
wizard-work/beam/test_next_greater.py resolves `next_greater` from its own
|
|
directory, grades the COMMITTED bytes, and the lane's CHECK now reds when a
|
|
beam test binds a host copy -- the defect that landed at `096ba9b`.
|
|
|
|
Six legs, each an independent reading, and the last three are behavioural
|
|
rather than textual:
|
|
|
|
tool tools/import_resolve_check.py over the work tree reads rc 0 and
|
|
names no OUTSIDE_REPO row (the repo authority agrees with the fix).
|
|
binding the test file's source carries a `__file__` binding and an
|
|
assertion over the imported module's path, and carries NO absolute
|
|
constant in a `sys.path` call (parsed with `ast`, so a comment or
|
|
docstring that quotes the old line is not read as a binding).
|
|
sibling the test and its module copied together into a directory outside
|
|
the tree run there at rc 0 -- the file is self-contained.
|
|
poison the same copy with `next_greater_right` poisoned reads rc 1: the
|
|
cell grades the bytes beside it, which is the whole repair.
|
|
mutant the pre-repair binding (the line committed at `096ba9b`, put back
|
|
into a copy of the repaired file) reads rc 1 and names the host
|
|
path -- the gate can see the defect it is meant to catch.
|
|
lane wizard-work/beam/check_bindings.py --selftest reads 0 failures,
|
|
and its live reading of this lane is rc 0.
|
|
|
|
Exit: 0 every leg passed, 1 a leg faulted, 2 could not grade (a file or the
|
|
interpreter is missing). Nothing here writes to the tracked tree; every copy
|
|
lives under `--scratch`.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import ast
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
|
|
DEFAULT_REPO = os.path.dirname(os.path.dirname(os.path.dirname(
|
|
os.path.abspath(__file__))))
|
|
TEST_REL = os.path.join("wizard-work", "beam", "test_next_greater.py")
|
|
MODULE_REL = os.path.join("wizard-work", "beam", "next_greater.py")
|
|
TOOL_REL = os.path.join("tools", "import_resolve_check.py")
|
|
GUARD_REL = os.path.join("wizard-work", "beam", "check_bindings.py")
|
|
|
|
# The defect this gate must still be able to build: the binding line as it
|
|
# landed at 096ba9b. The mutant leg puts it back into a copy of the repaired
|
|
# file, so the gate proves it can see the class rather than asserting it can.
|
|
HEAD_BINDING = 'sys.path.insert(0, "/root/.hermes/wizard-outputs/beam")\n'
|
|
|
|
|
|
class CannotGrade(Exception):
|
|
pass
|
|
|
|
|
|
def _read(path):
|
|
try:
|
|
with open(path, "rb") as fh:
|
|
return fh.read()
|
|
except OSError as exc:
|
|
raise CannotGrade("cannot read %s: %s" % (path, exc))
|
|
|
|
|
|
def _sha(raw):
|
|
return hashlib.sha256(raw).hexdigest()
|
|
|
|
|
|
def _run(args, cwd, timeout=180):
|
|
env = {"PATH": "/usr/bin:/bin", "HOME": os.environ.get("HOME", "/root")}
|
|
try:
|
|
return subprocess.run(args, cwd=cwd, capture_output=True, text=True,
|
|
timeout=timeout, env=env)
|
|
except (OSError, subprocess.TimeoutExpired) as exc:
|
|
raise CannotGrade("could not run %s: %s" % (args[0], exc))
|
|
|
|
|
|
def _last(proc):
|
|
lines = (proc.stdout + proc.stderr).strip().splitlines()
|
|
return lines[-1][:200] if lines else ""
|
|
|
|
|
|
# ------------------------------------------------------------------------ legs
|
|
|
|
def leg_tool(repo, scratch):
|
|
proc = _run([sys.executable, TOOL_REL, "--repo", repo, "--rev", "WORKTREE"],
|
|
repo)
|
|
if proc.returncode == 2:
|
|
raise CannotGrade("import_resolve_check could not grade the tree")
|
|
rows = [l for l in proc.stdout.splitlines() if l.startswith("FAULT")]
|
|
if proc.returncode != 0 or rows:
|
|
return False, "import_resolve_check rc %d: %s" % (
|
|
proc.returncode, rows[0] if rows else _last(proc))
|
|
return True, _last(proc)
|
|
|
|
|
|
def leg_binding(repo, scratch):
|
|
raw = _read(os.path.join(repo, TEST_REL))
|
|
text = raw.decode("utf-8")
|
|
try:
|
|
tree = ast.parse(text, filename=TEST_REL)
|
|
except SyntaxError as exc:
|
|
return False, "does not parse: line %s: %s" % (exc.lineno, exc.msg)
|
|
absolute = []
|
|
for node in ast.walk(tree):
|
|
if not isinstance(node, ast.Call):
|
|
continue
|
|
f = node.func
|
|
if not (isinstance(f, ast.Attribute) and f.attr in
|
|
("insert", "append", "extend")):
|
|
continue
|
|
inner = f.value
|
|
if not (isinstance(inner, ast.Attribute) and inner.attr == "path"):
|
|
continue
|
|
for arg in node.args:
|
|
if isinstance(arg, ast.Constant) and isinstance(arg.value, str) \
|
|
and os.path.isabs(arg.value):
|
|
absolute.append((node.lineno, arg.value))
|
|
if absolute:
|
|
return False, "absolute sys.path constant at line %d: %s" % absolute[0]
|
|
uses_file = any(isinstance(n, ast.Name) and n.id == "__file__"
|
|
for n in ast.walk(tree))
|
|
asserts_path = False
|
|
for node in tree.body: # module level only
|
|
if isinstance(node, ast.Assert):
|
|
dumped = ast.dump(node.test)
|
|
if "next_greater.py" in dumped and "Div" in dumped:
|
|
asserts_path = True
|
|
if not uses_file:
|
|
return False, "no __file__-based resolution in the file"
|
|
if not asserts_path:
|
|
return False, ("no module-level assert that the imported module is "
|
|
"HERE / next_greater.py")
|
|
return True, "sha256 %s, __file__ binding asserted at import" % _sha(raw)
|
|
|
|
|
|
def _copy_pair(repo, dest):
|
|
os.makedirs(dest, exist_ok=True)
|
|
shutil.copy2(os.path.join(repo, TEST_REL), os.path.join(dest, "test_next_greater.py"))
|
|
shutil.copy2(os.path.join(repo, MODULE_REL), os.path.join(dest, "next_greater.py"))
|
|
return dest
|
|
|
|
|
|
def leg_sibling(repo, scratch):
|
|
dest = _copy_pair(repo, os.path.join(scratch, "sibling"))
|
|
proc = _run([sys.executable, "test_next_greater.py"], dest)
|
|
if proc.returncode != 0:
|
|
return False, "rc %d outside the tree: %s" % (proc.returncode, _last(proc))
|
|
return True, "runs in a directory outside the repo, rc 0 (%s)" % _last(proc)
|
|
|
|
|
|
def leg_poison(repo, scratch):
|
|
dest = _copy_pair(repo, os.path.join(scratch, "poison"))
|
|
mod = os.path.join(dest, "next_greater.py")
|
|
src = _read(mod).decode("utf-8")
|
|
marker = 'def next_greater_right(nums: List[int]) -> List[int]:\n'
|
|
if marker not in src:
|
|
raise CannotGrade("next_greater.py no longer defines "
|
|
"next_greater_right with that signature")
|
|
poisoned = src.replace(
|
|
marker + ' """For each i, index of first j>i with nums[j] > nums[i], '
|
|
'else -1."""',
|
|
marker + " return [-2] * len(nums) # POISON", 1)
|
|
if poisoned == src:
|
|
raise CannotGrade("the poison target moved: the docstring line is gone")
|
|
with open(mod, "w") as fh:
|
|
fh.write(poisoned)
|
|
proc = _run([sys.executable, "test_next_greater.py"], dest)
|
|
if proc.returncode != 1:
|
|
return False, ("a poisoned sibling module still reads rc %d -- the cell "
|
|
"is not grading the bytes beside it" % proc.returncode)
|
|
return True, "a poisoned sibling module flips the cell to rc 1"
|
|
|
|
|
|
def leg_mutant(repo, scratch):
|
|
dest = os.path.join(scratch, "mutant")
|
|
os.makedirs(dest, exist_ok=True)
|
|
src = _read(os.path.join(repo, TEST_REL)).decode("utf-8")
|
|
lines = src.splitlines(True)
|
|
# Put the pre-repair binding back: replace the __file__ insert with the
|
|
# host path, keeping the assert (which is what makes it fail loudly here).
|
|
out, swapped = [], False
|
|
for line in lines:
|
|
if not swapped and line.startswith("sys.path.insert(0, str(HERE))"):
|
|
out.append(HEAD_BINDING)
|
|
swapped = True
|
|
else:
|
|
out.append(line)
|
|
if not swapped:
|
|
raise CannotGrade("the repaired insert line is absent -- cannot build "
|
|
"the pre-repair mutant")
|
|
with open(os.path.join(dest, "test_next_greater.py"), "w") as fh:
|
|
fh.write("".join(out))
|
|
shutil.copy2(os.path.join(repo, MODULE_REL), os.path.join(dest, "next_greater.py"))
|
|
proc = _run([sys.executable, "test_next_greater.py"], dest)
|
|
text = proc.stdout + proc.stderr
|
|
if proc.returncode == 0:
|
|
return False, "the pre-repair binding still reads rc 0"
|
|
if "wizard-outputs" not in text and "MODULE_PATH" not in text \
|
|
and "AssertionError" not in text:
|
|
return False, "rc %d but nothing names the host binding: %s" % (
|
|
proc.returncode, _last(proc))
|
|
note = _last(proc)
|
|
if "AssertionError" in note:
|
|
note = "assert fired: %s" % note[:160]
|
|
return True, "pre-repair binding reads rc %d (%s)" % (proc.returncode, note)
|
|
|
|
|
|
def leg_lane(repo, scratch):
|
|
guard = os.path.join(repo, GUARD_REL)
|
|
if not os.path.isfile(guard):
|
|
raise CannotGrade("%s is missing" % GUARD_REL)
|
|
st = _run([sys.executable, guard, "--selftest"], os.path.dirname(guard))
|
|
if st.returncode != 0:
|
|
return False, "guard selftest rc %d: %s" % (st.returncode, _last(st))
|
|
live = _run([sys.executable, guard], os.path.dirname(guard))
|
|
if live.returncode != 0:
|
|
return False, "the lane's live binding reading is rc %d" % live.returncode
|
|
cells = [l for l in st.stdout.splitlines() if l.startswith("selftest:")]
|
|
return True, "%s; %s" % (cells[0] if cells else "selftest rc 0", _last(live))
|
|
|
|
|
|
LEGS = [
|
|
("tool", leg_tool),
|
|
("binding", leg_binding),
|
|
("sibling", leg_sibling),
|
|
("poison", leg_poison),
|
|
("mutant", leg_mutant),
|
|
("lane", leg_lane),
|
|
]
|
|
|
|
|
|
def main(argv=None):
|
|
ap = argparse.ArgumentParser(description=(__doc__ or "").splitlines()[0])
|
|
ap.add_argument("--repo", default=DEFAULT_REPO)
|
|
ap.add_argument("--scratch", default=None,
|
|
help="where the throwaway copies go (default: a temp dir)")
|
|
ap.add_argument("--json", metavar="PATH")
|
|
a = ap.parse_args(argv)
|
|
repo = os.path.abspath(a.repo)
|
|
if not os.path.isdir(os.path.join(repo, "tools")):
|
|
print("cannot grade: %s does not look like the repo root" % repo,
|
|
file=sys.stderr)
|
|
return 2
|
|
rows, faults, ungraded, rc = [], 0, 0, 0
|
|
scratch = a.scratch or tempfile.mkdtemp(prefix="q153-")
|
|
if os.path.isdir(scratch):
|
|
shutil.rmtree(scratch)
|
|
os.makedirs(scratch)
|
|
try:
|
|
for name, leg in LEGS:
|
|
try:
|
|
ok, detail = leg(repo, scratch)
|
|
except CannotGrade as exc:
|
|
print("UNGRADABLE %-8s %s" % (name, exc))
|
|
rows.append({"leg": name, "ok": None, "detail": str(exc)})
|
|
ungraded += 1
|
|
rc = 2
|
|
continue
|
|
print("%s %-8s %s" % ("PASS" if ok else "FAIL", name, detail))
|
|
rows.append({"leg": name, "ok": ok, "detail": detail})
|
|
if not ok:
|
|
faults += 1
|
|
finally:
|
|
shutil.rmtree(scratch, ignore_errors=True)
|
|
if rc == 0:
|
|
rc = 1 if faults else 0
|
|
print("q153: %d leg(s), %d passed, %d fault(s), %d ungradable"
|
|
% (len(LEGS), len(LEGS) - faults - ungraded, faults, ungraded))
|
|
if a.json:
|
|
with open(a.json, "w") as fh:
|
|
json.dump({"repo": repo, "rows": rows, "rc": rc}, fh, indent=1)
|
|
return rc
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|