#!/usr/bin/env python3 """Self-mock detector. Runs both gates and writes the run's ``qa/`` sidecar. Gate A is static: it scans changed TEST files for SUT-stub signals. Gate B is mutation-based (``okstra_ctl.mutation_probe``) and runs over the ``--changed-file`` set — the stage's WHOLE diff, because each mutation adapter picks its own production sources out of it. ``overall`` is FAIL when either gate fails; ``unsupported(...)`` from gate B is not a failure, it means that gate could not run here at all. Signals come from the SSOT in ``okstra_ctl.self_mock_signals`` — never redefine a pattern here. Each file is matched as ONE whole-file string rather than line by line, because some signals span lines (java ``injectmocks-spy`` is two annotations on separate lines); the reported line number is derived from the match offset. Writes a ``qa/`` sidecar JSON and prints ``QA-RESULT: PASS|FAIL`` as its last stdout line. The exit code follows ``overall``, not gate A alone: 0 = PASS, 1 = FAIL from EITHER gate. A mutation failure with a clean static scan still exits 1, because the verifier records this exit code as the command's outcome. Alongside the hits the sidecar records every ``--test-file`` it received, split into ``scannedFiles`` and ``skippedFiles``: hits alone cannot distinguish "scanned the changed test files and found nothing" from "scanned an empty or wrong input" — both are an empty hit list. Reporting the skipped ones too is what keeps the gate from blocking a run whose changed test file the detector legitimately cannot read. ``--waivers`` is the false-positive escape hatch. A regex gate produces some false accusations, and with no channel for them one wedges the stage forever — the sidecar is detector-written and hand-editing it is a contract violation. A waived hit moves out of ``staticDetect.hits`` into ``staticDetect.waived`` and stops counting toward the verdict. The detector matches only; it never judges whether the waiver is legitimate. ``validate-run.py::_validate_selfmock`` does that, by requiring a ``reason`` and a user ``acknowledgedBy`` on every waived entry — so an entry lacking either is carried through to that gate rather than dropped here, and an agent cannot clear its own finding by writing a waiver. The argument itself is recorded as ``staticDetect.waiverSource`` for the same reason: the gate pins it to the task's own ``qa/self-mock-waivers.json``, so redirecting ``--waivers`` at a self-authored file blocks instead of passing. """ from __future__ import annotations import argparse import io import json import sys import tokenize from datetime import datetime, timezone from pathlib import Path # scripts/ (repo) and python/ (installed under ~/.okstra/lib) are not packages; # insert whichever exists so okstra_ctl is importable directly. _VALIDATORS_DIR = Path(__file__).resolve().parent for _ssot_dir in (_VALIDATORS_DIR.parent / "scripts", _VALIDATORS_DIR.parent / "python"): if _ssot_dir.is_dir() and str(_ssot_dir) not in sys.path: sys.path.insert(0, str(_ssot_dir)) from okstra_ctl.mutation_probe import probe_changed_files from okstra_ctl.self_mock_signals import ( EXT_TO_LANG, SIGNALS, partition_waived_entries, selfmock_path_key, ) def overall_verdict(static_status: str, mutation_status: str) -> str: """Fold both gates into the single verdict `validate-run.py` reads. Either gate failing fails the stage. `unsupported(...)` is not a failure — it means that gate could not run at all (no adapter for the language, no tool installed), and blocking on it would wedge every repo that has no mutation tooling. The reason stays in the sidecar for audit either way. """ return "FAIL" if "FAIL" in (static_status, mutation_status) else "PASS" def is_scannable(path: Path) -> bool: """True when the scan can actually read this path. A file whose extension has no signal set (Go, Ruby, a JSON fixture) and one a stage deleted are skipped rather than treated as failures. Both are still reported — as ``skippedFiles`` — because the gate needs to know the detector RECEIVED them: its trigger for a changed test file is extension- and existence-agnostic, so demanding those appear among the scanned ones would block every such run. """ return EXT_TO_LANG.get(path.suffix) is not None and path.is_file() def scannable_files(paths: list[Path]) -> list[Path]: """Return the subset a scan can actually read, in the caller's own order.""" return [p for p in paths if is_scannable(p)] def _mask_python_non_code(text: str) -> str: """Blank Python strings and comments without moving any match positions.""" chars = list(text) line_offsets = [0] for line in text.splitlines(keepends=True): line_offsets.append(line_offsets[-1] + len(line)) try: tokens = tokenize.generate_tokens(io.StringIO(text).readline) for token in tokens: if token.type not in {tokenize.STRING, tokenize.COMMENT}: continue start = line_offsets[token.start[0] - 1] + token.start[1] end = line_offsets[token.end[0] - 1] + token.end[1] for index in range(start, end): if chars[index] not in "\r\n": chars[index] = " " except (tokenize.TokenError, IndentationError, SyntaxError): # Broken source is scanned conservatively so a syntax error cannot hide a hit. return text return "".join(chars) def scan_files(paths: list[Path]) -> list[dict]: """Return one hit dict ``{file, line, signal}`` per signal match.""" hits: list[dict] = [] for p in scannable_files(paths): lang = EXT_TO_LANG[p.suffix] text = p.read_text(encoding="utf-8", errors="replace") scan_text = _mask_python_non_code(text) if lang == "python" else text file_hits: list[dict] = [] for sig in SIGNALS.get(lang, []): for m in sig.pattern.finditer(scan_text): line = scan_text[: m.start()].count("\n") + 1 file_hits.append({"file": str(p), "line": line, "signal": sig.name}) hits.extend(sorted(file_hits, key=lambda h: (h["line"], h["signal"]))) return hits def load_waivers(path: Path | None) -> list[dict]: """Read the user-acknowledged false-positive entries; `[]` when there are none. An absent file is the normal case — the flag is authored unconditionally in `prompts/profiles/_implementation-verifier.md` while most stages never need a waiver. An unreadable or malformed one raises instead: silently falling back to "no waivers" would leave the operator re-reading an unchanged FAIL with no hint that their acknowledgement never parsed. """ if path is None or not path.is_file(): return [] try: data = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise ValueError(f"waiver file unreadable at {path}: {exc}") from exc if not isinstance(data, list) or any(not isinstance(e, dict) for e in data): raise ValueError( f"waiver file at {path} must be a JSON array of " "{file, line, signal, reason, acknowledgedBy} objects" ) return data def partition_waived( hits: list[dict], waivers: list[dict] ) -> tuple[list[dict], list[dict]]: """Gate A's half of the shared waiver matching: keyed on `signal`.""" return partition_waived_entries(hits, waivers, "signal") def main(argv=None) -> int: ap = argparse.ArgumentParser( description="Detect self-mocked tests (SUT stubbed by its own test)." ) ap.add_argument("--test-file", action="append", default=[], dest="test_files") # Gate B needs the WHOLE changed set, not the test files gate A scans: each # mutation adapter selects its own production sources out of it. Passing only # the test files leaves every adapter with nothing to mutate, which reports a # vacuous PASS while gate B is silently dead. ap.add_argument("--changed-file", action="append", default=[], dest="changed_files") ap.add_argument("--sidecar", required=True) ap.add_argument("--stage-name", default=None) ap.add_argument("--waivers", default=None) ap.add_argument("--diff", default=None) ap.add_argument("--worktree", default=None) args = ap.parse_args(argv) try: waivers = load_waivers(Path(args.waivers) if args.waivers else None) except ValueError as exc: ap.error(str(exc)) # Paths are recorded as the caller spelled them (never resolved): the gate in # validate-run.py compares these against the report's repo-relative §5.7.3 # rows, and both come from `git diff --name-only` in the worktree cwd. received = [Path(f) for f in args.test_files] # Gate B's own record of its input, the counterpart to gate A's # scannedFiles/skippedFiles: without it a run that forgot `--changed-file` # produces a sidecar indistinguishable from one where gate B had nothing to # flag. Spelled exactly as passed, like the gate A lists. changed = [Path(f) for f in args.changed_files] scanned = [p for p in received if is_scannable(p)] skipped = [p for p in received if not is_scannable(p)] hits, waived = partition_waived(scan_files(scanned), waivers) static_status = "FAIL" if hits else "PASS" # The SAME waiver list feeds both gates: one user-managed file, and a single # `waiverSource` for the gate to pin. Gate A reads its `signal` entries, gate # B its `mutant` ones. mutation = probe_changed_files( changed, Path(args.diff) if args.diff else None, Path(args.worktree) if args.worktree else None, waivers, ) # Recorded for the same reason `staticDetect.waiverSource` is: the gate pins # it to the task's own file, so redirecting `--waivers` at something the run # wrote itself blocks instead of passing. mutation = {**mutation, "waiverSource": args.waivers} status = overall_verdict(static_status, str(mutation["status"])) sidecar = { "stageName": args.stage_name, "overall": status, "ranAt": datetime.now(timezone.utc).isoformat(), "scannedFiles": [str(p) for p in scanned], "skippedFiles": [str(p) for p in skipped], "changedFiles": [str(p) for p in changed], # `waiverSource` is the `--waivers` argument exactly as passed. The gate # pins it to the task's own `qa/self-mock-waivers.json`, which is what # stops a run from reading its acknowledgement out of a file it wrote # itself somewhere the task bundle never records. "staticDetect": { "status": static_status, "hits": hits, "waived": waived, "waiverSource": args.waivers, }, "mutation": mutation, } sidecar_path = Path(args.sidecar) sidecar_path.parent.mkdir(parents=True, exist_ok=True) sidecar_path.write_text(json.dumps(sidecar, indent=2), encoding="utf-8") for h in hits: print(f"SELF-MOCK {h['file']}:{h['line']} {h['signal']}") for s in mutation["survived"]: print(f"MUTANT-SURVIVED {s['file']}:{s['line']} {s['mutant']} ({s['status']})") print(f"QA-RESULT: {status}") return 1 if status == "FAIL" else 0 if __name__ == "__main__": raise SystemExit(main())