"""Shadow validation regression for dry-run executor (task-2634).

규칙:
- read-only · pure function 호출만 (merge/push/PR/cron/branch-protection/admin-override 호출 0)
- frozen fixture 만 입력 (tests/fixtures/dryrun_shadow/) · live GitHub/git 의존 0
- dryrun_route 함수 본체 수정 0 (validation only)

검증 범위 (spec system_dryrun_executor_shadow_validation_spec_260523.md §7):
- 16 fixture (6 PASS + 4 HOLD + 4 CHAIR_REQUIRED + 2 UNKNOWN) 입력→출력 byte-equal
- 안전 불변식: actually_executed=False · executor_action WOULD_* 접두사 100%
- verdict 4값(PASS/HOLD/CHAIR_REQUIRED/UNKNOWN) 전부 커버
- idempotency: 동일 입력 2회 호출 → byte-identical
- HOLD sub-action 매핑 정합 (CI_PENDING/EVIDENCE_STALE/MEDIUM/THREAD)
- CHAIR_REQUIRED chair_triggers 정합 + UNKNOWN reason=INSUFFICIENT_EVIDENCE
- PASS auto_merge_10_conditions 전부 True + requires_chair_activation=True

dev6 담당: 페룬
단일소스 스펙: memory/specs/system_dryrun_executor_shadow_validation_spec_260523.md
"""

import json
import sys
from pathlib import Path

import pytest

_WORKTREE_ROOT = Path(__file__).resolve().parents[2]
if str(_WORKTREE_ROOT) not in sys.path:
    sys.path.insert(0, str(_WORKTREE_ROOT))

from utils import merge_ready_dryrun_executor as E
from utils import merge_ready_states as S

FIXTURE_ROOT = Path(__file__).resolve().parents[1] / "fixtures" / "dryrun_shadow"

# spec §3 정본 — 16 시나리오 (이름·verdict 매핑은 spec frozen)
PASS_FIXTURES = [
    "shadow_pass_pr131_l3_classifier",
    "shadow_pass_pr132_l1l2_credential_passthrough",
    "shadow_pass_pr133_l4_wiring_mixed_remediation",
    "shadow_pass_pr134_e2e_production_code_zero",
    "shadow_pass_pr135_loop_boundary_resolved",
    "shadow_pass_pr136_dryrun_only",
]
HOLD_FIXTURES = [
    "shadow_hold_ci_pending",
    "shadow_hold_gemini_evidence_stale",
    "shadow_hold_gemini_medium_within_expected",
    "shadow_hold_unresolved_medium_thread",
]
CHAIR_FIXTURES = [
    "shadow_chair_blocking_secret",
    "shadow_chair_admin_override_required",
    "shadow_chair_replacement_pr_runner_modified",
    "shadow_chair_loop_boundary_critical_repetition",
]
UNKNOWN_FIXTURES = [
    "shadow_unknown_evidence_missing",
    "shadow_unknown_lifecycle_incident_normal_miss",
]
ALL_FIXTURES = PASS_FIXTURES + HOLD_FIXTURES + CHAIR_FIXTURES + UNKNOWN_FIXTURES

# HOLD sub-action 매핑 단언 (spec §3)
HOLD_SUB_ACTION = {
    "shadow_hold_ci_pending": ("CI_PENDING", E.ACTION_WOULD_WAIT_RECHECK),
    "shadow_hold_gemini_evidence_stale": (
        "GEMINI_EVIDENCE_STALE", E.ACTION_WOULD_OWNER_GEMINI_REVIEW_NUDGE,
    ),
    "shadow_hold_gemini_medium_within_expected": (
        "GEMINI_MEDIUM_WITHIN_EXPECTED",
        E.ACTION_WOULD_AUTO_FIX_REGRESS_PUSH_RESOLVE_RECHECK,
    ),
    "shadow_hold_unresolved_medium_thread": (
        "UNRESOLVED_MEDIUM_THREAD", E.ACTION_WOULD_RESOLVE_THREAD_RECHECK,
    ),
}


# ─────────────────────────────────────────────────────────────────────────────
# 헬퍼
# ─────────────────────────────────────────────────────────────────────────────

def _load(name):
    base = FIXTURE_ROOT / name
    evidence = json.loads((base / "evidence.json").read_text(encoding="utf-8"))
    expected = json.loads((base / "expected.json").read_text(encoding="utf-8"))
    return evidence, expected


def _route(evidence):
    return E.dryrun_route(evidence["classifier_result"], evidence["pr_identity"])


# ─────────────────────────────────────────────────────────────────────────────
# 1) fixture 골격 — 16 시나리오 전부 존재 + 3 파일 (evidence/expected/PROVENANCE)
# ─────────────────────────────────────────────────────────────────────────────

def test_all_16_fixtures_present():
    assert len(ALL_FIXTURES) == 16
    dirs = sorted(p.name for p in FIXTURE_ROOT.iterdir() if p.is_dir())
    expected = sorted(ALL_FIXTURES)
    assert dirs == expected, f"shadow fixture set mismatch: {dirs}"


@pytest.mark.parametrize("name", ALL_FIXTURES)
def test_fixture_triplet_present(name):
    base = FIXTURE_ROOT / name
    for fn in ("evidence.json", "expected.json", "PROVENANCE.md"):
        assert (base / fn).exists(), f"{name}/{fn} missing"


# ─────────────────────────────────────────────────────────────────────────────
# 2) byte-equal 단언 — dryrun_route(evidence) == expected.json (16 fixture 전부)
# ─────────────────────────────────────────────────────────────────────────────

@pytest.mark.parametrize("name", ALL_FIXTURES)
def test_route_matches_expected_byte_equal(name):
    ev, exp = _load(name)
    art = _route(ev)
    # semantic equality
    assert art == exp, f"{name}: route output != expected.json"
    # strict byte-equal — canonicalized json.dumps (sort_keys, ensure_ascii=False) 일치
    canon_art = json.dumps(art, ensure_ascii=False, sort_keys=True)
    canon_exp = json.dumps(exp, ensure_ascii=False, sort_keys=True)
    assert canon_art == canon_exp, (
        f"{name}: byte-equal mismatch on canonicalized json (sort_keys=True)"
    )


# ─────────────────────────────────────────────────────────────────────────────
# 3) 안전 불변식 — actually_executed=False · WOULD_* 접두사 (100% 단언)
# ─────────────────────────────────────────────────────────────────────────────

@pytest.mark.parametrize("name", ALL_FIXTURES)
def test_invariant_actually_executed_false(name):
    ev, _ = _load(name)
    art = _route(ev)
    assert art["actually_executed"] is False
    assert isinstance(art["actually_executed"], bool)


@pytest.mark.parametrize("name", ALL_FIXTURES)
def test_invariant_executor_action_would_prefix(name):
    ev, _ = _load(name)
    art = _route(ev)
    action = art["executor_action"]
    assert isinstance(action, str)
    assert action.startswith("WOULD_"), f"{name}: action={action!r}"
    assert action in E.ALL_EXECUTOR_ACTIONS


@pytest.mark.parametrize("name", ALL_FIXTURES)
def test_invariant_expected_fixture_satisfies_safety(name):
    """expected.json 자체가 안전 불변식 만족 — fixture 자체가 안전 doctrine 100%."""
    _, exp = _load(name)
    assert exp["actually_executed"] is False
    assert exp["executor_action"].startswith("WOULD_")
    assert exp["schema"] in (
        E.SCHEMA_AUTO_MERGE_CANDIDATE,
        E.SCHEMA_REMEDIATION_REQUIRED,
        E.SCHEMA_HOLD_FOR_CHAIR,
    )


# ─────────────────────────────────────────────────────────────────────────────
# 4) idempotency — 동일 입력 2회 호출 byte-identical
# ─────────────────────────────────────────────────────────────────────────────

@pytest.mark.parametrize("name", ALL_FIXTURES)
def test_idempotent_byte_identical(name):
    ev, _ = _load(name)
    a1 = _route(ev)
    a2 = _route(ev)
    s1 = json.dumps(a1, ensure_ascii=False, sort_keys=False)
    s2 = json.dumps(a2, ensure_ascii=False, sort_keys=False)
    assert s1 == s2, f"{name}: non-deterministic output"
    assert a1 == a2


# ─────────────────────────────────────────────────────────────────────────────
# 5) verdict 분류별 정합성
# ─────────────────────────────────────────────────────────────────────────────

@pytest.mark.parametrize("name", PASS_FIXTURES)
def test_pass_routes_to_auto_merge_candidate(name):
    ev, _ = _load(name)
    art = _route(ev)
    assert art["schema"] == E.SCHEMA_AUTO_MERGE_CANDIDATE
    assert art["verdict"] == S.PASS
    assert art["executor_action"] == E.ACTION_WOULD_MERGE
    assert art["requires_chair_activation"] is True
    # auto_merge 10조건 전부 True
    conds = art["auto_merge_10_conditions"]
    assert set(conds.keys()) == set(S.AUTO_MERGE_10_CONDITION_KEYS), name
    assert all(conds.values()), f"{name}: PASS but 10조건 미충족 — {conds}"
    # post_merge_smoke_plan + callback_lifecycle_artifact_plan 보존
    assert art["post_merge_smoke_plan"]["defined"] is True
    assert art["callback_lifecycle_artifact_plan"]["would_generate"] is True


@pytest.mark.parametrize("name", HOLD_FIXTURES)
def test_hold_routes_to_remediation_required(name):
    ev, _ = _load(name)
    art = _route(ev)
    assert art["schema"] == E.SCHEMA_REMEDIATION_REQUIRED
    assert art["verdict"] == S.HOLD
    assert art["executor_action"] == E.ACTION_WOULD_AUTO_REMEDIATE
    assert art["loop_boundary_guard"] == E.LOOP_BOUNDARY_GUARD_TEXT
    assert art["next_recheck"] == E.NEXT_RECHECK_ACTION
    # auto_remediable 최소 1개
    assert len(art["auto_remediable"]) >= 1
    # remediation_plan 각 item · action 매핑 정합
    for entry in art["remediation_plan"]:
        assert entry["action"] in E.ALL_SUB_ACTIONS, f"{name}: {entry}"


@pytest.mark.parametrize("name", list(HOLD_SUB_ACTION.keys()))
def test_hold_sub_action_mapping_exact(name):
    """spec §3 sub-action 매핑 정본 — 각 HOLD fixture 의 1차 auto_remediable item/action."""
    ev, _ = _load(name)
    art = _route(ev)
    expected_item, expected_action = HOLD_SUB_ACTION[name]
    assert expected_item in art["auto_remediable"], f"{name}: missing {expected_item}"
    plan_for_item = [p for p in art["remediation_plan"] if p["item"] == expected_item]
    assert len(plan_for_item) == 1, f"{name}: remediation_plan missing {expected_item}"
    assert plan_for_item[0]["action"] == expected_action


@pytest.mark.parametrize("name", CHAIR_FIXTURES)
def test_chair_required_routes_to_hold_for_chair(name):
    ev, _ = _load(name)
    art = _route(ev)
    assert art["schema"] == E.SCHEMA_HOLD_FOR_CHAIR
    assert art["verdict"] == S.CHAIR_REQUIRED
    assert art["executor_action"] == E.ACTION_WOULD_ESCALATE_CHAIR
    assert "reason" not in art, f"{name}: CHAIR_REQUIRED 에 reason 필드 부재여야 함"
    assert len(art["chair_triggers"]) >= 1, f"{name}: chair_triggers empty"


def test_chair_blocking_secret_signals_credential_expansion():
    ev, _ = _load("shadow_chair_blocking_secret")
    art = _route(ev)
    assert S.CHAIR_CREDENTIAL_PERMISSION_EXPANSION in art["chair_triggers"]
    # 평문 시크릿 미포함 (redacted only) — credential 3-tier doctrine
    raw = json.dumps(art, ensure_ascii=False)
    # 각 ghp_ 출현 위치 직후 16자 이내에 redacted 표식이 반드시 와야 함.
    # 좁은 윈도우로 다른 필드 redacted 가 통과시키지 못하도록 인접성 강제 (round-4 security-medium 대응).
    # 정상 평문 토큰이면 ghp_ 직후 영숫자가 연속 — redacted 표식이 16자 내 없을 수 있음.
    idx = 0
    while True:
        pos = raw.find("ghp_", idx)
        if pos == -1:
            break
        # ghp_ 토큰 위치 직후 16자 안에 redacted 표식 (좁은 윈도우)
        nearby = raw[pos:pos + 16]
        assert "redacted" in nearby.lower() or "<redacted" in nearby, (
            f"평문 ghp_ 노출 의심 (16자 윈도우 내 redacted 없음): nearby={nearby!r}"
        )
        idx = pos + 4
    assert "-----BEGIN" not in raw


def test_chair_admin_override_signals_override_trigger():
    ev, _ = _load("shadow_chair_admin_override_required")
    art = _route(ev)
    assert S.CHAIR_CRITICAL7 in art["chair_triggers"]
    assert S.CHAIR_ADMIN_OVERRIDE in art["chair_triggers"]
    assert "C7_OVERRIDE" in art["report_envelope"]["evidence_refs"]


def test_chair_replacement_pr_runner_modified_signals_forbidden_path():
    ev, _ = _load("shadow_chair_replacement_pr_runner_modified")
    art = _route(ev)
    assert S.CHAIR_CRITICAL7 in art["chair_triggers"]
    assert S.CHAIR_OUT_OF_EXPECTED_FILES in art["chair_triggers"]
    assert "C7_FORBIDDEN_PATH" in art["report_envelope"]["evidence_refs"]


def test_chair_loop_boundary_signals_auto_remediation_boundary():
    ev, _ = _load("shadow_chair_loop_boundary_critical_repetition")
    art = _route(ev)
    assert S.CHAIR_AUTO_REMEDIATION_LOOP_BOUNDARY in art["chair_triggers"]


@pytest.mark.parametrize("name", UNKNOWN_FIXTURES)
def test_unknown_routes_to_regather(name):
    ev, _ = _load(name)
    art = _route(ev)
    assert art["schema"] == E.SCHEMA_HOLD_FOR_CHAIR
    assert art["verdict"] == S.UNKNOWN
    assert art["executor_action"] == E.ACTION_WOULD_REGATHER
    assert art["reason"] == E.REASON_INSUFFICIENT_EVIDENCE
    # UNKNOWN 은 회장 escalate 아님 — chair_triggers 빈 리스트
    assert art["chair_triggers"] == []


# ─────────────────────────────────────────────────────────────────────────────
# 6) pr_identity 결정적 보존 (4-tuple + ts_kst)
# ─────────────────────────────────────────────────────────────────────────────

@pytest.mark.parametrize("name", ALL_FIXTURES)
def test_pr_identity_passthrough(name):
    ev, _ = _load(name)
    art = _route(ev)
    pi = ev["pr_identity"]
    assert art["pr"] == int(pi.get("pr") or 0)
    assert art["head_sha"] == (pi.get("head_sha") or "")
    assert art["task_id"] == (pi.get("task_id") or "")
    assert art["branch"] == (pi.get("branch") or "")
    assert art["ts_kst"] == (pi.get("ts_kst") or "")


# ─────────────────────────────────────────────────────────────────────────────
# 7) ANCHOR-2 종합 — verdict 4값 routing 매핑 전부 커버 (6+4+4+2)
# ─────────────────────────────────────────────────────────────────────────────

def test_anchor2_all_four_verdicts_covered_by_16_fixtures():
    verdicts_seen = set()
    actions_seen = set()
    schemas_seen = set()
    for name in ALL_FIXTURES:
        ev, _ = _load(name)
        art = _route(ev)
        verdicts_seen.add(art["verdict"])
        actions_seen.add(art["executor_action"])
        schemas_seen.add(art["schema"])
    assert verdicts_seen == {S.PASS, S.HOLD, S.CHAIR_REQUIRED, S.UNKNOWN}
    assert actions_seen == {
        E.ACTION_WOULD_MERGE, E.ACTION_WOULD_AUTO_REMEDIATE,
        E.ACTION_WOULD_ESCALATE_CHAIR, E.ACTION_WOULD_REGATHER,
    }
    assert schemas_seen == {
        E.SCHEMA_AUTO_MERGE_CANDIDATE, E.SCHEMA_REMEDIATION_REQUIRED,
        E.SCHEMA_HOLD_FOR_CHAIR,
    }


def test_shadow_fixture_count_breakdown_6_4_4_2():
    """spec §3 정본 분포 단언 — 6 PASS + 4 HOLD + 4 CHAIR_REQUIRED + 2 UNKNOWN."""
    counts = {S.PASS: 0, S.HOLD: 0, S.CHAIR_REQUIRED: 0, S.UNKNOWN: 0}
    for name in ALL_FIXTURES:
        ev, _ = _load(name)
        art = _route(ev)
        counts[art["verdict"]] += 1
    assert counts == {S.PASS: 6, S.HOLD: 4, S.CHAIR_REQUIRED: 4, S.UNKNOWN: 2}


# ─────────────────────────────────────────────────────────────────────────────
# 8) JSON serialization round-trip — artifact 직렬화 안전성
# ─────────────────────────────────────────────────────────────────────────────

@pytest.mark.parametrize("name", ALL_FIXTURES)
def test_artifact_json_round_trip(name):
    ev, _ = _load(name)
    art = _route(ev)
    s = json.dumps(art, ensure_ascii=False)
    assert json.loads(s) == art


# ─────────────────────────────────────────────────────────────────────────────
# 9) live 의존 0 — fixture 만 로드, dryrun_route 만 호출 (정적 안전망)
# ─────────────────────────────────────────────────────────────────────────────

def test_no_live_imports_in_this_module():
    """본 테스트 모듈이 GitHub/git/subprocess/network 의존 import 를 가지지 않음.

    AST 기반 정확 검사 — alias·괄호 다중 라인·from-import 전부 견고 처리 (Gemini medium round-3 대응)."""
    import ast

    forbidden_top = {"subprocess", "requests", "urllib", "httpx", "socket"}
    forbidden_sub = "utils.replacement_pr_runner"

    tree = ast.parse(Path(__file__).read_text(encoding="utf-8"))
    for node in ast.walk(tree):
        if isinstance(node, ast.Import):
            for alias in node.names:
                name = alias.name
                assert name.split(".")[0] not in forbidden_top, (
                    f"forbidden top-level import: {name}"
                )
                assert forbidden_sub not in name, (
                    f"forbidden dotted import: {name}"
                )
        elif isinstance(node, ast.ImportFrom):
            mod = node.module or ""
            assert mod.split(".")[0] not in forbidden_top, (
                f"forbidden from-import top-level: {mod}"
            )
            assert forbidden_sub not in mod, (
                f"forbidden from-import dotted: {mod}"
            )
            for alias in node.names:
                full = f"{mod}.{alias.name}" if mod else alias.name
                assert forbidden_sub not in full, (
                    f"forbidden from-import member: {full}"
                )
        elif isinstance(node, ast.Call):
            # __import__('subprocess') / importlib.import_module('subprocess')
            # 동적 import 도 차단 (round-4 security-medium 대응).
            fn = node.func
            fn_name = None
            if isinstance(fn, ast.Name):
                fn_name = fn.id
            elif isinstance(fn, ast.Attribute):
                fn_name = fn.attr
            if fn_name in ("__import__", "import_module") and node.args:
                first = node.args[0]
                if isinstance(first, ast.Constant) and isinstance(first.value, str):
                    target = first.value
                    head = target.split(".")[0]
                    assert head not in forbidden_top, (
                        f"forbidden dynamic import: {target}"
                    )
                    assert forbidden_sub not in target, (
                        f"forbidden dynamic dotted import: {target}"
                    )