# tests/test_prevent_runner.py
"""runner.py — the pure dispatcher. Mechanics proven with TEMP echo detectors (hermetic): selection,
repo-scope single invocation, contract-JSON parsing, and the block policy ladder (error→BLOCK;
warning+confirmed→BLOCK; else WARN; degraded→COVERAGE-INCOMPLETE, non-blocking)."""
import importlib.util, json, os, textwrap

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))

def _load(name, path):
    spec = importlib.util.spec_from_file_location(name, path)
    mod = importlib.util.module_from_spec(spec); spec.loader.exec_module(mod); return mod

runner = _load("sg_runner", os.path.join(ROOT, "prevent", "runner.py"))

def _echo_detector(tmp_path, did, scope, level, precise, *, with_green=True, unresolved=False):
    """Create a temp detector that prints one canned contract finding. Absolute exec path."""
    ddir = tmp_path / "domains" / "sec" / "detectors" / did
    ddir.mkdir(parents=True)
    if with_green:
        (ddir / "cells").mkdir()
        (ddir / "cells" / "x_safe.json").write_text("{}")
    script = ddir / f"{did}.py"
    unres = "['/x']" if unresolved else "[]"
    script.write_text(textwrap.dedent(f"""
        import json, sys
        f = {{"ruleId":"{did}-1","level":"{level}","class":"S11","message":"m","file":sys.argv[1] if len(sys.argv)>1 else "?","line":0,"symbol":"{did}-sym"}}
        print(json.dumps({{"detector":"{did}","status":("degraded" if {unres} else "ok"),
            "findings":[f],"coverage":{{"scanned":[f["file"]],"unresolved":{unres}}}}}))
    """))
    manifest = {"id": did, "exec": ["python3", str(script)], "scope": scope,
                "scope_globs": (["**/*.ts"] if scope == "per-file" else []),
                "trigger_globs": (["**/*.lock"] if scope == "repo" else []),
                "needs_context": "none", "triggers": ["pre-commit"],
                "precision": ("precise" if precise else "imprecise"), "class": "S11"}
    (ddir / "detector.json").write_text(json.dumps(manifest))
    return ddir

def test_precise_error_blocks(tmp_path):
    _echo_detector(tmp_path, "p1", "per-file", "error", precise=True)
    reg = _load("reg6a", os.path.join(ROOT, "prevent", "registry.py"))
    dets, _ = reg.load(str(tmp_path))
    rep = runner.run(["a.ts"], "pre-commit", dets, str(tmp_path))
    assert rep["exit_code"] == 1 and len(rep["blocking"]) == 1

def test_imprecise_warning_does_not_block(tmp_path):
    _echo_detector(tmp_path, "w1", "per-file", "warning", precise=False)
    reg = _load("reg6b", os.path.join(ROOT, "prevent", "registry.py"))
    dets, _ = reg.load(str(tmp_path))
    rep = runner.run(["a.ts"], "pre-commit", dets, str(tmp_path))
    assert rep["exit_code"] == 0 and rep["warnings"] and not rep["blocking"]

def test_ratcheted_warning_blocks(tmp_path):
    _echo_detector(tmp_path, "w2", "per-file", "warning", precise=False)
    reg = _load("reg6c", os.path.join(ROOT, "prevent", "registry.py"))
    dets, _ = reg.load(str(tmp_path))
    confirmed = [{"class": "S11", "file": "a.ts", "symbol": "w2-sym"}]
    rep = runner.run(["a.ts"], "pre-commit", dets, str(tmp_path), confirmed)
    assert rep["exit_code"] == 1 and len(rep["blocking"]) == 1

def test_precise_without_green_cell_demoted_to_warn(tmp_path):
    # admission gate: precision=precise but NO green cell → NOT block-authorized → error demoted to warn
    _echo_detector(tmp_path, "p2", "per-file", "error", precise=True, with_green=False)
    reg = _load("reg6d", os.path.join(ROOT, "prevent", "registry.py"))
    dets, _ = reg.load(str(tmp_path))
    rep = runner.run(["a.ts"], "pre-commit", dets, str(tmp_path))
    assert rep["exit_code"] == 0 and rep["warnings"] and not rep["blocking"]

def test_repo_scope_runs_once_with_repo_root(tmp_path):
    _echo_detector(tmp_path, "r1", "repo", "warning", precise=False)
    reg = _load("reg6e", os.path.join(ROOT, "prevent", "registry.py"))
    dets, _ = reg.load(str(tmp_path))
    rep = runner.run(["deps.lock", "deps.lock"], "pre-commit", dets, str(tmp_path))  # listed twice
    assert len(rep["warnings"]) == 1  # repo detector runs ONCE, not per changed file
    assert rep["warnings"][0]["file"] == str(tmp_path)  # argv = repo root

def test_degraded_detector_surfaces_coverage_incomplete(tmp_path):
    _echo_detector(tmp_path, "d1", "per-file", "warning", precise=False, unresolved=True)
    reg = _load("reg6f", os.path.join(ROOT, "prevent", "registry.py"))
    dets, _ = reg.load(str(tmp_path))
    rep = runner.run(["a.ts"], "pre-commit", dets, str(tmp_path))
    assert rep["incomplete"] and rep["exit_code"] == 0  # incomplete is surfaced but NEVER blocks

def test_ratchet_respects_class_when_set(tmp_path):
    # F2: a confirmed entry with a DIFFERENT class must NOT ratchet a same-suffix+symbol finding to BLOCK
    _echo_detector(tmp_path, "w3", "per-file", "warning", precise=False)  # emits class S11
    reg = _load("reg6g", os.path.join(ROOT, "prevent", "registry.py"))
    dets, _ = reg.load(str(tmp_path))
    wrong = [{"class": "S99", "file": "a.ts", "symbol": "w3-sym"}]  # class mismatch
    rep = runner.run(["a.ts"], "pre-commit", dets, str(tmp_path), wrong)
    assert rep["exit_code"] == 0 and not rep["blocking"] and rep["warnings"], f"class mismatch must NOT block; got {rep}"
    classless = [{"file": "a.ts", "symbol": "w3-sym"}]  # empty class = backward-compat wildcard → blocks
    rep2 = runner.run(["a.ts"], "pre-commit", dets, str(tmp_path), classless)
    assert rep2["exit_code"] == 1 and rep2["blocking"], f"classless seed must still block (compat); got {rep2}"

def test_degraded_status_with_empty_unresolved_still_surfaces(tmp_path):
    # F3: a detector emitting status=degraded with EMPTY unresolved must still surface as COVERAGE-INCOMPLETE
    ddir = tmp_path / "domains" / "sec" / "detectors" / "g1"
    ddir.mkdir(parents=True)
    script = ddir / "g1.py"
    script.write_text("import json\nprint(json.dumps({'detector':'g1','status':'degraded',"
                      "'findings':[],'coverage':{'scanned':[],'unresolved':[]}}))\n")
    (ddir / "detector.json").write_text(json.dumps({"id": "g1", "exec": ["python3", str(script)],
        "scope": "per-file", "scope_globs": ["**/*.ts"], "trigger_globs": [], "needs_context": "none",
        "triggers": ["pre-commit"], "precision": "imprecise", "class": "S11"}))
    reg = _load("reg6h", os.path.join(ROOT, "prevent", "registry.py"))
    dets, _ = reg.load(str(tmp_path))
    rep = runner.run(["a.ts"], "pre-commit", dets, str(tmp_path))
    assert rep["incomplete"] and rep["exit_code"] == 0, f"degraded+empty-unresolved must surface; got {rep}"

def test_skipped_manifest_seeds_incomplete_and_ran_count(tmp_path):
    # F4 + F7: malformed manifests passed in surface as incomplete; ran=0 when nothing applied
    skipped = [(str(tmp_path / "domains" / "x" / "detectors" / "broken" / "detector.json"), "boom")]
    rep = runner.run(["a.md"], "pre-commit", [], str(tmp_path), (), skipped)
    assert rep["ran"] == 0, f"no detectors → ran 0; got {rep}"
    assert any("broken" in i for i in rep["incomplete"]), f"skipped manifest must surface; got {rep}"
    assert rep["exit_code"] == 0  # reduced coverage never blocks

def test_exec_with_interpreter_flag_resolves_script_not_flag(tmp_path):
    # F5: exec=[python3, -O, script] must SKIP the flag and resolve the script. A broken loop would treat
    # "-O" as the script → rebase/run it → subprocess error → no finding + COVERAGE-INCOMPLETE. So a clean
    # finding with NO incomplete proves the flag was skipped and the real script ran.
    ddir = tmp_path / "domains" / "sec" / "detectors" / "o1"
    ddir.mkdir(parents=True)
    script = ddir / "o1.py"
    script.write_text("import json, sys\n"
                      "print(json.dumps({'detector':'o1','status':'ok',"
                      "'findings':[{'ruleId':'o1-1','level':'warning','class':'S11','message':'m',"
                      "'file':sys.argv[1] if len(sys.argv)>1 else '?','line':0,'symbol':'o1-sym'}],"
                      "'coverage':{'scanned':[],'unresolved':[]}}))\n")
    (ddir / "detector.json").write_text(json.dumps({"id": "o1", "exec": ["python3", "-O", str(script)],
        "scope": "per-file", "scope_globs": ["**/*.ts"], "trigger_globs": [], "needs_context": "none",
        "triggers": ["pre-commit"], "precision": "imprecise", "class": "S11"}))
    reg = _load("reg6i", os.path.join(ROOT, "prevent", "registry.py"))
    dets, _ = reg.load(str(tmp_path))
    rep = runner.run(["a.ts"], "pre-commit", dets, str(tmp_path))
    assert rep["warnings"] and rep["warnings"][0]["symbol"] == "o1-sym", f"flag-skip exec must run; got {rep}"
    assert not rep["incomplete"], f"flag mis-resolved would error→incomplete; got {rep}"

def _sleep_detector(root, did, declared_timeout, sleep_s):
    """A per-file detector that sleeps `sleep_s` then emits one ok warning. Declares its own manifest timeout."""
    ddir = root / "domains" / "sec" / "detectors" / did
    ddir.mkdir(parents=True)
    script = ddir / f"{did}.py"
    script.write_text("import json, sys, time\n"
                      f"time.sleep({sleep_s})\n"
                      "print(json.dumps({'detector':%r,'status':'ok',"
                      "'findings':[{'ruleId':%r,'level':'warning','class':'S11','message':'m',"
                      "'file':sys.argv[1] if len(sys.argv)>1 else '?','line':0,'symbol':%r}],"
                      "'coverage':{'scanned':[],'unresolved':[]}}))\n" % (did, f"{did}-1", f"{did}-sym"))
    (ddir / "detector.json").write_text(json.dumps({"id": did, "exec": ["python3", str(script)],
        "scope": "per-file", "scope_globs": ["**/*.ts"], "trigger_globs": [], "needs_context": "none",
        "triggers": ["pre-commit"], "timeout": declared_timeout, "precision": "imprecise", "class": "S11"}))

def test_per_detector_timeout_honored_from_manifest(tmp_path):
    # The manifest "timeout" must reach _run_detector (deps needs >10s for network-bound pnpm audit). Lower the
    # DEFAULT to 1s so the field's effect is observable WITHOUT a >10s sleep:
    #  A) declares timeout=10, sleeps 2s → completes ONLY if the larger DECLARED budget is honored (2s>1s default).
    #  B) declares timeout=1,  sleeps 2s → exceeds its OWN budget → fail-open COVERAGE-INCOMPLETE, never blocks.
    orig = runner.TIMEOUT
    runner.TIMEOUT = 1
    try:
        _sleep_detector(tmp_path, "slow_ok", declared_timeout=10, sleep_s=2)
        reg = _load("reg6j", os.path.join(ROOT, "prevent", "registry.py"))
        dets, _ = reg.load(str(tmp_path))
        rep = runner.run(["a.ts"], "pre-commit", dets, str(tmp_path))
        assert rep["warnings"] and rep["warnings"][0]["symbol"] == "slow_ok-sym", \
            f"declared timeout=10 must let a 2s detector complete despite the 1s default; got {rep}"
        assert not rep["incomplete"], f"a completed detector must NOT be COVERAGE-INCOMPLETE; got {rep}"
    finally:
        runner.TIMEOUT = orig  # never leak the lowered default into other tests
    sub = tmp_path / "b"; sub.mkdir()
    _sleep_detector(sub, "slow_to", declared_timeout=1, sleep_s=2)  # default now back to 10 → declared=1 must win
    reg2 = _load("reg6k", os.path.join(ROOT, "prevent", "registry.py"))
    dets2, _ = reg2.load(str(sub))
    rep2 = runner.run(["a.ts"], "pre-commit", dets2, str(sub))
    assert rep2["incomplete"] and not rep2["warnings"] and rep2["exit_code"] == 0, \
        f"declared timeout=1 vs 2s sleep must COVERAGE-INCOMPLETE, never block (proves declared<default is read); got {rep2}"

def _add_trigger_filter(ddir, run, args):
    """Write a trigger_filter that returns a fixed {run, args} (ignores stdin) and wire it into the manifest."""
    fs = ddir / "filter.py"
    fs.write_text("import json\nprint(json.dumps(%r))\n" % {"run": run, "args": args})
    mf = ddir / "detector.json"
    m = json.loads(mf.read_text()); m["trigger_filter"] = ["python3", str(fs)]; mf.write_text(json.dumps(m))

def test_trigger_filter_run_false_skips_repo_detector(tmp_path):
    # REGRESSION (the user's exports-only case at the runner layer): a repo detector whose trigger_filter returns
    # run:false must NOT run — no finding, no block, no incomplete. ran==0 proves the detector never executed.
    ddir = _echo_detector(tmp_path, "r2", "repo", "error", precise=True)  # would BLOCK if it ran
    _add_trigger_filter(ddir, run=False, args=[])
    reg = _load("reg6l", os.path.join(ROOT, "prevent", "registry.py"))
    dets, _ = reg.load(str(tmp_path))
    rep = runner.run(["deps.lock"], "pre-commit", dets, str(tmp_path))
    assert rep["ran"] == 0 and not rep["blocking"] and not rep["warnings"] and not rep["incomplete"], \
        f"trigger_filter run:false must skip the detector entirely; got {rep}"

def test_trigger_filter_run_true_passes_args_to_exec(tmp_path):
    # run:true + args → the args must reach the detector exec (before the repo_root file arg). The detector echoes
    # its argv into the finding message so the test can assert the scope args arrived.
    ddir = tmp_path / "domains" / "sec" / "detectors" / "r3"
    ddir.mkdir(parents=True)
    (ddir / "cells").mkdir(); (ddir / "cells" / "x_safe.json").write_text("{}")
    script = ddir / "r3.py"
    script.write_text("import json, sys\n"
                      "print(json.dumps({'detector':'r3','status':'ok',"
                      "'findings':[{'ruleId':'r3-1','level':'warning','class':'S11','message':' '.join(sys.argv[1:]),"
                      "'file':'f','line':0,'symbol':'r3-sym'}],"
                      "'coverage':{'scanned':[],'unresolved':[]}}))\n")
    (ddir / "detector.json").write_text(json.dumps({"id": "r3", "exec": ["python3", str(script)],
        "scope": "repo", "scope_globs": [], "trigger_globs": ["**/*.lock"], "needs_context": "none",
        "triggers": ["pre-commit"], "precision": "imprecise", "class": "S11"}))
    _add_trigger_filter(ddir, run=True, args=["--commit-scope", "--scope", "packages/x"])
    reg = _load("reg6m", os.path.join(ROOT, "prevent", "registry.py"))
    dets, _ = reg.load(str(tmp_path))
    rep = runner.run(["deps.lock"], "pre-commit", dets, str(tmp_path))
    assert rep["warnings"], f"run:true must run the detector; got {rep}"
    msg = rep["warnings"][0]["message"]
    assert "--commit-scope" in msg and "--scope packages/x" in msg, f"scope args must reach exec; got {msg!r}"

def test_trigger_filter_crash_fails_open_runs_unscoped(tmp_path):
    # no-false-clean: a broken trigger_filter must NOT silently skip a precise detector — it runs UNSCOPED (blocks).
    ddir = _echo_detector(tmp_path, "r4", "repo", "error", precise=True)
    fs = ddir / "filter.py"; fs.write_text("import sys\nsys.exit(3)\n")  # crashes, emits no JSON
    mf = ddir / "detector.json"; m = json.loads(mf.read_text())
    m["trigger_filter"] = ["python3", str(fs)]; mf.write_text(json.dumps(m))
    reg = _load("reg6n", os.path.join(ROOT, "prevent", "registry.py"))
    dets, _ = reg.load(str(tmp_path))
    rep = runner.run(["deps.lock"], "pre-commit", dets, str(tmp_path))
    assert rep["ran"] == 1 and rep["blocking"], f"broken filter must fail-OPEN (run unscoped → block); got {rep}"