241 lines
8.3 KiB
Python
241 lines
8.3 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
mutate_teeth.py -- does spec_schedkit.py actually bite?
|
|
|
|
A green spec proves nothing on its own: a spec that asserts nothing is green
|
|
forever. This harness mutates schedkit.py and requires the spec to go red.
|
|
|
|
Three parts:
|
|
control the unmutated module must PASS (else the harness is broken)
|
|
baseline the pre-fix module (schedkit_v1_quadratic.py) must FAIL
|
|
mutants 12 single-substitution bugs, each of which the spec must catch
|
|
|
|
Each run happens in its own scratch directory with a copy of the module and a
|
|
copy of the spec, so the working tree is never touched.
|
|
|
|
Exit 0 == control passed, baseline failed, every mutant was caught.
|
|
Exit 1 == something survived; the line says what.
|
|
|
|
Usage: python3 mutate_teeth.py [--budget SECONDS]
|
|
"""
|
|
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
SPEC = os.path.join(HERE, "spec_schedkit.py")
|
|
MODULE = os.path.join(HERE, "schedkit.py")
|
|
BASELINE = os.path.join(HERE, "schedkit_v1_quadratic.py")
|
|
|
|
# (label, old, new). `old` must occur exactly once in schedkit.py.
|
|
MUTANTS = [
|
|
(
|
|
"plan: waves not sorted (the pre-fix behaviour)",
|
|
" return tuple(tuple(sorted(w)) for w in waves)",
|
|
" return tuple(tuple(w) for w in waves)",
|
|
),
|
|
(
|
|
"plan: wave depth uses max of deps, not 1+max",
|
|
" wave_of[t] = 1 + max((wave_of[d] for d in g.deps(t)), default=-1)",
|
|
" wave_of[t] = max((wave_of[d] for d in g.deps(t)), default=-1)",
|
|
),
|
|
(
|
|
"find_cycles: cycles not canonically rotated",
|
|
"lambda c: found.add(_canon(tuple(ids[comp[i]] for i in c))),",
|
|
"lambda c: found.add(tuple(ids[comp[i]] for i in c)),",
|
|
),
|
|
(
|
|
"find_cycles: self-loops dropped",
|
|
" if v in adj[v]:\n found.add((ids[v],))",
|
|
" if False:\n found.add((ids[v],))",
|
|
),
|
|
(
|
|
"find_cycles: singleton SCCs never checked for self-loops",
|
|
" if len(comp) == 1:",
|
|
" if False:",
|
|
),
|
|
(
|
|
"CIRCUIT: nodes never unblocked",
|
|
" if closed:\n _unblock(v, blocked, blocked_map)",
|
|
" if False:\n _unblock(v, blocked, blocked_map)",
|
|
),
|
|
(
|
|
"CIRCUIT: floor ignored (removed vertices stay visible)",
|
|
" if w < floor or w not in nodes:",
|
|
" if w not in nodes:",
|
|
),
|
|
(
|
|
"topological_order: heap replaced by list pop",
|
|
" t = heapq.heappop(ready)",
|
|
" t = ready.pop()",
|
|
),
|
|
(
|
|
"plan_json: n_waves off by one",
|
|
' "n_waves": len(waves),',
|
|
' "n_waves": len(waves) + 1,',
|
|
),
|
|
(
|
|
"iter_plan: drops the last task of every wave",
|
|
" for w in waves:\n for t in w:\n yield t",
|
|
" for w in waves:\n for t in w[:-1]:\n yield t",
|
|
),
|
|
(
|
|
"relabel: dependency ids not remapped",
|
|
" sorted(mapping.get(d, d) for d in self._deps[t]),",
|
|
" sorted(self._deps[t]),",
|
|
),
|
|
(
|
|
"PERF: Johnson's source-vertex skip removed (ring becomes O(k^2))",
|
|
" if indeg[s]: # a self-loop counts: it is an incoming edge",
|
|
" if True:",
|
|
),
|
|
]
|
|
|
|
|
|
def run_spec(workdir: str, budget: float):
|
|
"""Run the spec in `workdir`. Returns (rc|None, output, seconds)."""
|
|
t0 = time.monotonic()
|
|
try:
|
|
p = subprocess.run(
|
|
[sys.executable, "spec_schedkit.py"],
|
|
cwd=workdir,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
timeout=budget,
|
|
text=True,
|
|
)
|
|
return p.returncode, p.stdout, time.monotonic() - t0
|
|
except subprocess.TimeoutExpired as exc:
|
|
out = exc.stdout or ""
|
|
if isinstance(out, bytes):
|
|
out = out.decode("utf-8", "replace")
|
|
return None, out, time.monotonic() - t0
|
|
|
|
|
|
def first_failure(out: str) -> str:
|
|
lines = out.splitlines()
|
|
for i, ln in enumerate(lines):
|
|
if ln.startswith("FAILURES ("):
|
|
for nxt in lines[i + 1 : i + 4]:
|
|
if nxt.strip().startswith("- "):
|
|
return nxt.strip()[2:]
|
|
break
|
|
for ln in lines:
|
|
if ln.startswith("FAILURES"):
|
|
return ln
|
|
return "(no failure line)"
|
|
|
|
|
|
def summary(out: str) -> str:
|
|
for ln in reversed(out.splitlines()):
|
|
if " checks, " in ln and " failed" in ln:
|
|
return ln.strip()
|
|
return "(no summary line)"
|
|
|
|
|
|
def main() -> int:
|
|
budget = 300.0
|
|
if "--budget" in sys.argv:
|
|
budget = float(sys.argv[sys.argv.index("--budget") + 1])
|
|
|
|
for path in (SPEC, MODULE):
|
|
if not os.path.exists(path):
|
|
print("missing %s" % path)
|
|
return 1
|
|
|
|
print("spec : %s" % SPEC)
|
|
print("module : %s" % MODULE)
|
|
print("budget : %.0fs per run\n" % budget)
|
|
|
|
failures = []
|
|
|
|
# ------------------------------------------------------------- control
|
|
tmp = tempfile.mkdtemp(prefix="teeth-control-")
|
|
try:
|
|
shutil.copy(MODULE, os.path.join(tmp, "schedkit.py"))
|
|
shutil.copy(SPEC, os.path.join(tmp, "spec_schedkit.py"))
|
|
rc, out, dt = run_spec(tmp, budget)
|
|
print("CONTROL unmutated module rc=%s %5.1fs %s" % (rc, dt, summary(out)))
|
|
if rc != 0:
|
|
failures.append("control: the unmutated module failed the spec")
|
|
print(first_failure(out))
|
|
finally:
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
|
|
# ------------------------------------------------------------ baseline
|
|
if os.path.exists(BASELINE):
|
|
tmp = tempfile.mkdtemp(prefix="teeth-baseline-")
|
|
try:
|
|
shutil.copy(BASELINE, os.path.join(tmp, "schedkit.py"))
|
|
shutil.copy(SPEC, os.path.join(tmp, "spec_schedkit.py"))
|
|
rc, out, dt = run_spec(tmp, min(budget, 120.0))
|
|
caught = rc is None or rc != 0
|
|
verdict = "did not finish in %.0fs" % dt if rc is None else "rc=%d" % rc
|
|
print(
|
|
"BASELINE pre-fix module (v1) %s %5.1fs %s"
|
|
% ("CAUGHT " if caught else "SURVIVED", dt, verdict)
|
|
)
|
|
if not caught:
|
|
failures.append("baseline: the pre-fix module passed the spec")
|
|
finally:
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
|
|
# ------------------------------------------------------------- mutants
|
|
caught_n = 0
|
|
for i, (label, old, new) in enumerate(MUTANTS, 1):
|
|
src = open(MODULE).read()
|
|
if src.count(old) != 1:
|
|
failures.append("mutant %d anchor not unique (%d): %s" % (i, src.count(old), label))
|
|
print("M%-2d ANCHOR %s" % (i, label))
|
|
continue
|
|
tmp = tempfile.mkdtemp(prefix="teeth-m%02d-" % i)
|
|
try:
|
|
mutated = src.replace(old, new)
|
|
assert mutated != src
|
|
with open(os.path.join(tmp, "schedkit.py"), "w") as fh:
|
|
fh.write(mutated)
|
|
shutil.copy(SPEC, os.path.join(tmp, "spec_schedkit.py"))
|
|
rc, out, dt = run_spec(tmp, budget)
|
|
if rc is None:
|
|
caught_n += 1
|
|
print(
|
|
"M%-2d CAUGHT %-55s %5.0fs wall-clock gate (no exit)"
|
|
% (i, label, dt)
|
|
)
|
|
elif rc != 0:
|
|
caught_n += 1
|
|
print(
|
|
"M%-2d CAUGHT %-55s %5.1fs %s"
|
|
% (i, label, dt, first_failure(out))
|
|
)
|
|
else:
|
|
failures.append("mutant %d SURVIVED: %s" % (i, label))
|
|
print("M%-2d SURVIVED %s" % (i, label))
|
|
finally:
|
|
shutil.rmtree(tmp, ignore_errors=True)
|
|
|
|
print(
|
|
"\n%d/%d mutants caught, baseline %s, control %s"
|
|
% (
|
|
caught_n,
|
|
len(MUTANTS),
|
|
"caught" if os.path.exists(BASELINE) else "absent",
|
|
"passes" if not any(f.startswith("control") for f in failures) else "FAILS",
|
|
)
|
|
)
|
|
if failures:
|
|
print("\nTEETH FAILED (%d):" % len(failures))
|
|
for f in failures:
|
|
print(" - %s" % f)
|
|
return 1
|
|
print("TEETH OK: the spec is not a decoration")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|