"""test_gate_emit.py — deterministic: gate.py --emit contract serialization + report byte-identity.
No LLM: build_emit_dict / build_report are pure over synthetic groups/oracle. Path-load per convention."""
import importlib.util, os

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
GOLDEN = os.path.join(ROOT, "tests", "golden", "gate_report_single_file.md")


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


gate = _load("sg_gate", os.path.join(ROOT, "orchestrator", "gate.py"))

# synthetic, deterministic inputs (no LLM).
# TARGET is RELATIVE on purpose: build_report/build_emit_dict render paths via os.path.relpath(target),
# which for an ABSOLUTE target climbs `../` from the runtime cwd (filesystem-depth-dependent) — that baked a
# non-portable golden that passed only from the capture cwd. A relative input is cwd-independent
# (relpath('abs/target.ts') == 'abs/target.ts' from any dir), so the golden is hermetic across checkouts/CI.
GROUPS = [
    {"title": "reset token single-use not enforced", "sev": "high", "rolls": {0}},
    {"title": "missing rate limit on reset", "sev": "low", "rolls": {0}},
]
ORACLE_SILENT = [("abs/target.ts", "(oracle silent)")]
TARGET = "abs/target.ts"


def test_emit_dict_is_contract_shape_with_additive_keys():
    d = gate.build_emit_dict(TARGET, "baseline", ["S1", "S2"], 1, GROUPS, ORACLE_SILENT, {}, False)
    assert d["detector"] == "baseline"
    assert d["status"] == "ok"                       # no unresolved → ok
    assert set(d["coverage"]) == {"scanned", "unresolved"}
    f0 = d["findings"][0]
    # frozen contract keys present
    for kpresent in ("ruleId", "level", "class", "message", "file", "line", "symbol"):
        assert kpresent in f0
    assert f0["level"] == "error"                    # high → error
    assert f0["class"] == "S1,S2"                    # covers joined
    assert f0["message"] == "reset token single-use not enforced"
    # additive keys
    assert f0["sev"] == "high" and f0["rolls"] == 1 and f0["of"] == 1
    assert d["findings"][1]["level"] == "note"       # low → note


def test_emit_dict_degrades_on_dropped_dep():
    d = gate.build_emit_dict(TARGET, "baseline", ["S1"], 1, [], ORACLE_SILENT,
                             {"/abs/dep.ts": "over max-scope"}, False)
    assert d["status"] == "degraded"                 # non-empty unresolved → degraded (no-false-clean)
    assert any("dep.ts" in u for u in d["coverage"]["unresolved"])


def test_over_budget_fanin_drops_real_and_emit_is_degraded(tmp_path):
    # BINDING budget invariant (no-false-clean, spec band2-resolver-generalization §"Budget invariant"):
    # a REAL over-budget fan-in MUST drop imports in collect_deps AND surface status='degraded' E2E — never a
    # clean SILENT. Unlike test_emit_dict_degrades_on_dropped_dep (hand-built dropped dict), this proves the
    # resolver ACTUALLY drops under real over-budget and the emit path carries it through to coverage.unresolved.
    import json
    resolver = _load("sg_resolver", os.path.join(ROOT, "orchestrator", "resolver.py"))

    d = str(tmp_path)
    open(os.path.join(d, "pnpm-workspace.yaml"), "w").write("packages:\n  - 'apps/*'\n")
    app = os.path.join(d, "apps/api/src"); os.makedirs(app, exist_ok=True)
    open(os.path.join(d, "apps/api/package.json"), "w").write(
        json.dumps({"name": "@zync/api", "exports": {".": "./src/index.ts"}}))
    # 3 first-party helper defs + an importer that CALLS all 3. Names are keyword-free (doThing*) on purpose:
    # the pull is shape-driven (every first-party called import), NOT name-gated — CRITICAL is ordering-only.
    for i in range(3):
        open(os.path.join(app, f"helper{i}.ts"), "w").write(f"export function doThing{i}(a){{ return a }}\n")
    importer = os.path.join(app, "route.ts")
    open(importer, "w").write(
        "".join(f"import {{ doThing{i} }} from './helper{i}'\n" for i in range(3)) +
        "doThing0(1); doThing1(2); doThing2(3)\n")

    aliases = resolver.build_workspace_aliases(d)
    # budget = 2 < 3 first-party called imports -> the resolver MUST drop at least one (no silent skip)
    deps, dropped = resolver.collect_deps(importer, aliases, 1, 2)
    assert len(deps) == 2, f"budget cap not enforced; deps={deps}"
    assert dropped, "over-budget fan-in MUST populate dropped (no silent skip)"
    assert "doThing" in " ".join(dropped.values()), f"dropped reason must name the dropped import; got {dropped}"

    emit = gate.build_emit_dict(importer, "baseline", ["S1"], 3, [], [], dropped, False)
    assert emit["status"] == "degraded", f"dropped deps MUST degrade the emit (no-false-clean); got {emit}"
    assert any("max-scope" in u for u in emit["coverage"]["unresolved"]), \
        f"the dropped import must appear in coverage.unresolved; got {emit['coverage']}"


def test_report_is_byte_identical_to_golden():
    rep = gate.build_report(TARGET, [], {}, ORACLE_SILENT, GROUPS, 1, "sonnet", "medium", False, False)
    assert rep == open(GOLDEN, encoding="utf-8").read()


# ---------------------------------------------------------------------------
# finding-citation spine: build_emit_dict resolves the REAL line from the snippet (NOT the LLM's hint,
# which is bundle-offset-wrong in bundle mode) and emits a dedicated `code` field (NOT stuffed into symbol).
# ---------------------------------------------------------------------------
def test_emit_resolves_real_line_from_snippet_not_hint(tmp_path):
    target = os.path.join(str(tmp_path), "refund.ts")
    open(target, "w").write("line one\nline two\nawait db.refund(orderId)\nline four\n")
    groups = [{"title": "Missing ownership check", "sev": "high", "rolls": {0, 1, 2},
               "snippet": "await db.refund(orderId)", "line_hint": 99}]
    d = gate.build_emit_dict(target, "baseline", ["S5"], 3, groups, [], {}, False)
    f = d["findings"][0]
    assert f["line"] == 3, f"must resolve from snippet (line 3), not hint 99; got {f['line']}"
    assert f["code"] == "await db.refund(orderId)"
    assert f["symbol"] == "", "snippet must NOT be stuffed into symbol"
    assert f["anchor"] == "resolved"


def test_emit_unanchored_when_snippet_absent_falls_back_to_hint(tmp_path):
    # no snippet -> finding KEPT, line falls back to the hint, marked unverified (no-false-clean: never drop).
    target = os.path.join(str(tmp_path), "refund.ts")
    open(target, "w").write("a\nb\nc\n")
    groups = [{"title": "Missing idempotency guard", "sev": "high", "rolls": {0},
               "snippet": "", "line_hint": 42}]
    f = gate.build_emit_dict(target, "baseline", ["S5"], 3, groups, [], {}, False)["findings"][0]
    assert f["line"] == 42 and f["code"] == "" and f["anchor"] == "unverified"


def test_emit_unanchored_when_snippet_not_found(tmp_path):
    # snippet present but NOT in the file (paraphrased / from a dep section) -> fall back to hint, unverified.
    target = os.path.join(str(tmp_path), "refund.ts")
    open(target, "w").write("a\nb\nc\n")
    groups = [{"title": "X", "sev": "low", "rolls": {0}, "snippet": "not in this file", "line_hint": 2}]
    f = gate.build_emit_dict(target, "baseline", ["S5"], 3, groups, [], {}, False)["findings"][0]
    assert f["line"] == 2 and f["anchor"] == "unverified"


def test_anchor_survives_union_and_merge_end_to_end(tmp_path):
    # THE production-path test: rolls -> union_rolls -> _fold (semantic_merge) -> build_emit_dict. _fold
    # rebuilds the group dict by explicit fields; if it drops snippet/line_hint the emit silently re-emits
    # line=0 even though parse_findings captured the anchor. merge_groups runs by DEFAULT, so a direct
    # build_emit_dict test alone would pass while production stays non-navigable. This guards that seam.
    sm = _load("sg_merge", os.path.join(ROOT, "orchestrator", "semantic_merge.py"))
    target = os.path.join(str(tmp_path), "vuln.ts")
    open(target, "w").write("x\ny\nawait resetUserPassword(db, userId, hash)\nz\n")
    roll = (
        "1. **Reset token never invalidated — replay within TTL**\n"
        "   - **Location:** `vuln.ts:91-92`\n"
        "   - **Code:**\n"
        "     ```ts\n"
        "     await resetUserPassword(db, userId, hash)\n"
        "     ```\n"
        "   - **Severity:** critical\n")
    groups = gate.union_rolls([roll, roll, roll])
    assert groups[0].get("snippet") == "await resetUserPassword(db, userId, hash)"
    folded = sm._fold(groups)  # drive the fold directly: it is the seam that drops unknown keys
    assert folded["snippet"] == "await resetUserPassword(db, userId, hash)", "_fold dropped the snippet"
    assert folded["line_hint"] == 91, "_fold dropped the line_hint"
    f = gate.build_emit_dict(target, "baseline", ["S1"], 3, [folded], [], {}, False)["findings"][0]
    assert f["line"] == 3 and f["anchor"] == "resolved" and f["code"].startswith("await resetUserPassword")


def test_real_capture_patched_prompt_resolves_in_corpus_file():
    # end-to-end on REAL patched-prompt output (/tmp/sg_fmt_roll_B.txt) against the REAL corpus file:
    # the captured Code snippets must resolve (anchor=resolved) in the actual vuln.ts. Proves the anchor
    # works on real model output, not just synthetic input. Dev-artifact gated -> clean skip off-host.
    cap = "/tmp/sg_fmt_roll_B.txt"
    corpus = os.path.join(ROOT, "domains/security/corpus/S1-reset-token-reuse/vuln.ts")
    if not (os.path.exists(cap) and os.path.exists(corpus)):
        return
    rows = gate.parse_findings(open(cap, encoding="utf-8").read())
    anchored = [r for r in rows if r[3]]  # rows with a non-empty snippet
    assert anchored, "patched-prompt capture should carry Code snippets"
    resolved = sum(1 for _, _, _, snip, hint in anchored if gate._resolve_line(corpus, snip, hint)[1])
    assert resolved >= 1, "at least one captured snippet must resolve in the real corpus file"