249 lines
11 KiB
Python
249 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""q57_mutation.py -- the committed negative control for check.py.
|
|
|
|
Exit code IS the verdict: 0 = the gate behaves the way its own docstring says on
|
|
a mutated entry, 1 = it does not. Nothing here is asserted; every cell is a run.
|
|
|
|
python3 /root/troupe-standup/wizard-work/null/q57_mutation.py
|
|
|
|
WHY A FIXTURE, AND NOT THE LIVE TREE
|
|
`python3 check.py` reads whatever is on disk, so running it in the shared
|
|
checkout grades the worktree and calls the result the commit -- Timmy's own
|
|
first pass on this artifact made exactly that mistake (reply 06). This
|
|
harness never touches the live tree. It takes a `git clone --bare` snapshot
|
|
of it, clones THAT, and mutates only the clone. origin/main inside the
|
|
fixture is the frozen bare repo, so a concurrent push by another wizard
|
|
cannot change what the test is measuring.
|
|
|
|
WHAT IT MEASURES -- seven cases, expectations computed here, never hardcoded
|
|
case 0 clean clone -> rc 0, ALL CHECKS GREEN
|
|
case 1 one byte inside a work-file bullet -> rc 1, names that line
|
|
case 2 the operator line -> rc 1, names that line
|
|
case 3 the base-commit line -> rc 1, names that line
|
|
case 4 one line appended -> rc 1, names the line past the end
|
|
case 5 final newline becomes CRLF -> rc 1, "a line terminator differs",
|
|
and NO line number is invented
|
|
case 6 the published gate moved on -> rc 1, names check.py, while the
|
|
entry check still PASSES
|
|
|
|
Case 5 was GREEN before this artifact. Both sides of the old comparison ran
|
|
under universal-newline translation (subprocess text=True and open(...,
|
|
encoding=...)), so the CRLF never reached the comparison: a doctored entry
|
|
read "byte-identical". Transcript of that run: ../q57-binding/transcript-before.txt.
|
|
Case 6 was impossible before: nothing compared the gate on disk with the gate
|
|
on main, so an entry could be certified by a private build.
|
|
|
|
Expected line numbers are computed two independent ways and both must agree:
|
|
by locating the mutated text in the graded bytes, and by a first-difference
|
|
scan over `graded.split(b"\\n")`. Neither reads the gate's own message.
|
|
|
|
SET NULL_Q57_KEEP=1 to keep the fixture directory for inspection.
|
|
"""
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
|
|
REPO = os.environ.get("NULL_REPO", "/root/troupe-standup")
|
|
LOGREL = "wizard-work/null/wizard.log.md"
|
|
GATEREL = "wizard-work/null/check.py"
|
|
PASSED, FAILED = [], []
|
|
|
|
|
|
def sh(args):
|
|
return subprocess.run(args, capture_output=True)
|
|
|
|
|
|
def git(root, *args):
|
|
return sh(["git", "-C", root] + list(args))
|
|
|
|
|
|
def check(name, ok, detail=""):
|
|
ok = bool(ok)
|
|
(PASSED if ok else FAILED).append(name)
|
|
print(("PASS " if ok else "FAIL ") + name + (" -> " + detail if detail else ""))
|
|
return ok
|
|
|
|
|
|
def line_of(lines, needle):
|
|
"""1-based number of the line containing `needle`, or 0 if it is absent."""
|
|
for i, l in enumerate(lines):
|
|
if needle in l:
|
|
return i + 1
|
|
return 0
|
|
|
|
|
|
def first_diff_scan(want, got):
|
|
"""Independent first-difference scan over raw bytes split on b'\\n'."""
|
|
w, g = want.split(b"\n"), got.split(b"\n")
|
|
for i in range(max(len(w), len(g))):
|
|
a = w[i] if i < len(w) else None
|
|
b = g[i] if i < len(g) else None
|
|
if a != b:
|
|
return i + 1
|
|
return None
|
|
|
|
|
|
def main():
|
|
root = tempfile.mkdtemp(prefix="q57-mutation-")
|
|
bare = os.path.join(root, "origin.git")
|
|
fx = os.path.join(root, "fixture")
|
|
logp = os.path.join(fx, LOGREL)
|
|
gatep = os.path.join(fx, GATEREL)
|
|
print("fixture root: %s" % root)
|
|
|
|
r = sh(["git", "clone", "--bare", "--quiet", REPO, bare])
|
|
if r.returncode != 0:
|
|
print("FAIL cannot snapshot the repo: " + r.stderr.decode("utf-8", "replace")[:200])
|
|
return 1
|
|
r = sh(["git", "clone", "--quiet", bare, fx])
|
|
if r.returncode != 0:
|
|
print("FAIL cannot clone the fixture: " + r.stderr.decode("utf-8", "replace")[:200])
|
|
return 1
|
|
|
|
def run_gate():
|
|
p = subprocess.run([sys.executable, gatep], capture_output=True, text=True)
|
|
return p.returncode, (p.stdout or "") + (p.stderr or "")
|
|
|
|
graded = git(fx, "show", "origin/main:" + LOGREL).stdout
|
|
text = graded.decode("utf-8")
|
|
lines = text.splitlines()
|
|
if not graded or not lines:
|
|
print("FAIL the fixture has no graded entry to mutate")
|
|
return 1
|
|
|
|
# ---- case 0: the fixture is clean and the gate is green on it ---------
|
|
disk = open(logp, "rb").read()
|
|
pre = check("fixture precondition: disk entry == the blob the gate grades",
|
|
disk == graded, "%d bytes" % len(graded))
|
|
rc, out = run_gate()
|
|
check("case 0 clean clone exits 0", rc == 0 and "ALL CHECKS GREEN" in out,
|
|
"rc %d, last line %r" % (rc, out.strip().splitlines()[-1] if out.strip() else ""))
|
|
if not pre:
|
|
print("\nFAILED: the fixture is not clean; no mutation result is meaningful")
|
|
return 1
|
|
|
|
def reset_log():
|
|
with open(logp, "wb") as fh:
|
|
fh.write(graded)
|
|
|
|
def case(label, mutated, expect_line, needles=()):
|
|
"""Write `mutated`, run the gate, and require it to name `expect_line`."""
|
|
with open(logp, "wb") as fh:
|
|
fh.write(mutated)
|
|
rc, out = run_gate()
|
|
named = re.search(r"line (\d+) differs", out)
|
|
ok = (rc == 1
|
|
and "ALL CHECKS GREEN" not in out
|
|
and expect_line is not None
|
|
and named is not None
|
|
and int(named.group(1)) == expect_line
|
|
and all(n in out for n in needles))
|
|
check(label, ok,
|
|
"rc %d, gate named line %s, expected %s"
|
|
% (rc, named.group(1) if named else "nothing", expect_line))
|
|
if not ok:
|
|
print(" gate output:\n" + "\n".join(" " + l for l in out.strip().splitlines()))
|
|
return ok
|
|
|
|
# ---- case 1: one byte inside a data line -----------------------------
|
|
needle1 = b"arguments, not the clock"
|
|
i1 = line_of(lines, needle1.decode())
|
|
m1 = list(graded.split(b"\n"))
|
|
m1[i1 - 1] = m1[i1 - 1].replace(b"arguments", b"argumentz")
|
|
mutated1 = b"\n".join(m1)
|
|
check("case 1 expectation cross-check: needle line == byte-scan line",
|
|
i1 > 0 and i1 == first_diff_scan(graded, mutated1), "line %d" % i1)
|
|
case("case 1 one byte inside a work-file bullet is named by line",
|
|
mutated1, i1, needles=("graded=", "disk="))
|
|
|
|
# ---- case 2: the operator line ---------------------------------------
|
|
reset_log()
|
|
needle2 = b"- operator: "
|
|
i2 = line_of(lines, needle2.decode())
|
|
m2 = list(graded.split(b"\n"))
|
|
m2[i2 - 1] = m2[i2 - 1].replace(b"null", b"nobody-at-all")
|
|
mutated2 = b"\n".join(m2)
|
|
check("case 2 expectation cross-check: needle line == byte-scan line",
|
|
i2 > 0 and i2 == first_diff_scan(graded, mutated2), "line %d" % i2)
|
|
case("case 2 a doctored operator line is named by line", mutated2, i2)
|
|
|
|
# ---- case 3: the base-commit line ------------------------------------
|
|
reset_log()
|
|
needle3 = b"- base commit: "
|
|
i3 = line_of(lines, needle3.decode())
|
|
m3 = list(graded.split(b"\n"))
|
|
m3[i3 - 1] = re.sub(rb"(?<=- base commit: )[0-9a-f]+", b"deadbeef", m3[i3 - 1])
|
|
mutated3 = b"\n".join(m3)
|
|
check("case 3 expectation cross-check: needle line == byte-scan line",
|
|
i3 > 0 and i3 == first_diff_scan(graded, mutated3), "line %d" % i3)
|
|
case("case 3 a doctored base commit is named by line", mutated3, i3)
|
|
|
|
# ---- case 4: one line appended ---------------------------------------
|
|
reset_log()
|
|
mutated4 = graded + b"- extra: a line no writer produced\n"
|
|
i4 = len(lines) + 1
|
|
check("case 4 expectation cross-check: appended line == byte-scan line",
|
|
i4 == first_diff_scan(graded, mutated4), "line %d" % i4)
|
|
case("case 4 an appended line is named, and the graded side reads <no such line>",
|
|
mutated4, i4, needles=("<no such line>",))
|
|
|
|
# ---- case 5: a line terminator, and no line to name -------------------
|
|
reset_log()
|
|
mutated5 = graded[:-1] + b"\r\n"
|
|
with open(logp, "wb") as fh:
|
|
fh.write(mutated5)
|
|
rc, out = run_gate()
|
|
check("case 5 a CRLF mutation is a failure, not a PASS",
|
|
rc == 1 and "ALL CHECKS GREEN" not in out, "rc %d" % rc)
|
|
check("case 5 every line is equal, so no line number is invented",
|
|
"line terminator" in out and re.search(r"line \d+ differs", out) is None,
|
|
"reported as a terminator")
|
|
check("case 5 the byte count is printed, so the 1-byte delta is visible",
|
|
"%d bytes" % len(graded) in out and "%d bytes" % len(mutated5) in out)
|
|
|
|
# ---- case 6: the published gate has moved on --------------------------
|
|
reset_log()
|
|
m = re.search(r"^- base commit: ([0-9a-f]{40})$", text, re.M)
|
|
if not m:
|
|
check("case 6 the entry names a 40-hex base commit", False, "no base commit line")
|
|
else:
|
|
named = m.group(1)
|
|
with open(gatep, "ab") as fh:
|
|
fh.write(b"\n# fixture: a later commit on main changed this file\n")
|
|
git(fx, "-c", "user.name=null", "-c", "user.email=null@agents.invalid",
|
|
"commit", "-aqm", "fixture: publish a gate that moved on")
|
|
push = git(fx, "push", "--quiet", "origin", "HEAD:main")
|
|
check("case 6 fixture: the frozen origin advanced past the named commit",
|
|
push.returncode == 0,
|
|
"origin/main is now %s" % git(fx, "rev-parse", "HEAD").stdout.decode().strip()[:12])
|
|
# Put the GRADED gate back on disk: disk == named commit's blob (so the
|
|
# named-commit check still passes) but != the published blob.
|
|
with open(gatep, "wb") as fh:
|
|
fh.write(git(fx, "show", "%s:%s" % (named, GATEREL)).stdout)
|
|
rc, out = run_gate()
|
|
check("case 6 a gate that main does not carry is refused",
|
|
rc == 1 and "FAIL the gate on disk is the gate published on the remote main" in out,
|
|
"rc %d" % rc)
|
|
check("case 6 isolation: the entry itself is still byte-identical on main",
|
|
"FAIL entry bytes are the blob on the remote main" not in out
|
|
and "PASS entry bytes are the blob on the remote main" in out,
|
|
"only the published-gate cell moved")
|
|
|
|
print("\n%d passed, %d failed" % (len(PASSED), len(FAILED)))
|
|
if os.environ.get("NULL_Q57_KEEP") != "1":
|
|
shutil.rmtree(root, ignore_errors=True)
|
|
else:
|
|
print("fixture kept at %s" % root)
|
|
if FAILED:
|
|
print("FAILED: " + ", ".join(FAILED))
|
|
return 1
|
|
print("MUTATION CONTROL GREEN")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|