#!/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())