"""Regression tests for merge_ready_dryrun_executor (task-2633).

규칙:
- 순수 함수 호출만 (merge 실행 0 · GitHub write 0 · live git/subprocess 0 · network 0)
- frozen fixture 만 입력 (tests/fixtures/merge_ready_dryrun/) · live workspace 의존 0
- pytest 로 실행 가능

검증 범위:
- verdict 4값(PASS/HOLD/CHAIR_REQUIRED/UNKNOWN) → 3종 routing artifact 매핑 전부 커버
- 안전 불변식: actually_executed=false · executor_action WOULD_* 접두사 · 3종 schema id
- 결정적 idempotency: 동일 입력 2회 → byte-identical 출력
- auto_remediable → sub-action 매핑 (CI_PENDING / GEMINI_MEDIUM_WITHIN_EXPECTED)
- UNKNOWN → reason=INSUFFICIENT_EVIDENCE 표식
- 정적 검증: 모듈 코드에 merge/push/PR/subprocess/network/open(write) 호출 0

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

import json
import re
import sys
from pathlib import Path

import pytest

# worktree root → sys.path (안전망)
_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
from utils.merge_ready_classifier import classify_merge_ready

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

ALL_FIXTURES = [
    "pass_routing",
    "hold_routing_ci_pending",
    "hold_routing_gemini_medium",
    "chair_routing_critical7",
    "chair_routing_credential",
    "unknown_routing",
    "idempotent_routing",
]


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

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 골격 존재
# ─────────────────────────────────────────────────────────────────────────────

def test_all_routing_fixtures_present():
    dirs = sorted(p.name for p in FIXTURE_ROOT.iterdir() if p.is_dir())
    assert dirs == sorted(ALL_FIXTURES), f"fixture set mismatch: {dirs}"
    for name in ALL_FIXTURES:
        base = FIXTURE_ROOT / name
        for fn in ("evidence.json", "expected.json", "PROVENANCE.md"):
            assert (base / fn).exists(), f"{name}/{fn} missing"


# ─────────────────────────────────────────────────────────────────────────────
# 2) verdict → routing 매핑 전부 커버
# ─────────────────────────────────────────────────────────────────────────────

def test_pass_routes_to_auto_merge_candidate():
    ev, exp = _load("pass_routing")
    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
    assert art == exp


def test_hold_ci_pending_routes_to_remediation_required():
    ev, exp = _load("hold_routing_ci_pending")
    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 {"item": "CI_PENDING", "action": E.ACTION_WOULD_WAIT_RECHECK} in art["remediation_plan"]
    assert art == exp


def test_hold_gemini_medium_routes_to_remediation_required():
    ev, exp = _load("hold_routing_gemini_medium")
    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 {
        "item": "GEMINI_MEDIUM_WITHIN_EXPECTED",
        "action": E.ACTION_WOULD_AUTO_FIX_REGRESS_PUSH_RESOLVE_RECHECK,
    } in art["remediation_plan"]
    assert art["loop_boundary_guard"] == E.LOOP_BOUNDARY_GUARD_TEXT
    assert art["next_recheck"] == E.NEXT_RECHECK_ACTION
    assert art == exp


def test_chair_critical7_routes_to_hold_for_chair():
    ev, exp = _load("chair_routing_critical7")
    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 S.CHAIR_CRITICAL7 in art["chair_triggers"]
    assert "C7_FORBIDDEN_PATH" in art["report_envelope"]["evidence_refs"]
    assert "reason" not in art  # CHAIR_REQUIRED 는 reason 필드 없음
    assert art == exp


def test_chair_credential_routes_to_hold_for_chair():
    ev, exp = _load("chair_routing_credential")
    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 S.CHAIR_CREDENTIAL_PERMISSION_EXPANSION in art["chair_triggers"]
    # 평문 비밀 미포함 (credential 3계층 doctrine)
    raw = json.dumps(art, ensure_ascii=False)
    assert "ghp-PAT-placeholder" not in raw or "redacted" in raw
    assert art == exp


def test_unknown_routes_to_hold_for_chair_regather():
    ev, exp = _load("unknown_routing")
    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
    # chair_triggers 는 classifier UNKNOWN 시 빈 리스트 — 회장 escalate 아님
    assert art["chair_triggers"] == []
    assert art == exp


# ─────────────────────────────────────────────────────────────────────────────
# 3) 안전 불변식 (전체 fixture 일괄)
# ─────────────────────────────────────────────────────────────────────────────

@pytest.mark.parametrize("name", ALL_FIXTURES)
def test_invariant_actually_executed_is_false(name):
    ev, _ = _load(name)
    art = _route(ev)
    assert art["actually_executed"] is False
    assert isinstance(art["actually_executed"], bool)  # 0/None 등 falsy 가짜 통과 차단


@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}: executor_action={action!r}"
    assert action in E.ALL_EXECUTOR_ACTIONS


@pytest.mark.parametrize("name", ALL_FIXTURES)
def test_invariant_schema_is_one_of_three(name):
    ev, _ = _load(name)
    art = _route(ev)
    assert art["schema"] in (
        E.SCHEMA_AUTO_MERGE_CANDIDATE,
        E.SCHEMA_REMEDIATION_REQUIRED,
        E.SCHEMA_HOLD_FOR_CHAIR,
    )


@pytest.mark.parametrize("name", ALL_FIXTURES)
def test_invariant_pr_identity_passthrough(name):
    """pr_identity 4튜플(pr,head_sha,task_id,branch)이 결정적으로 보존."""
    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 "")


# ─────────────────────────────────────────────────────────────────────────────
# 4) idempotency — 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) sub-action 매핑 enum 완전성
# ─────────────────────────────────────────────────────────────────────────────

def test_auto_remediable_action_map_covers_all_enum():
    for item in S.AUTO_REMEDIABLE_ORDER:
        assert item in E.AUTO_REMEDIABLE_ACTION_MAP, f"missing action map: {item}"
        action = E.AUTO_REMEDIABLE_ACTION_MAP[item]
        assert action.startswith("WOULD_"), f"{item}: action={action!r}"


# ─────────────────────────────────────────────────────────────────────────────
# 6) UNKNOWN 라우팅 — 미인지 verdict 도 안전하게 REGATHER 로 fallback
# ─────────────────────────────────────────────────────────────────────────────

def test_unrecognized_verdict_falls_back_to_regather():
    """미인지 verdict (e.g. None, 'INVALID_VERDICT_STRING') → UNKNOWN/REGATHER 와 동일 처리.
    추정 금지 + 안전 불변식 유지."""
    fake_result = {"verdict": "WAT", "chair_triggers": [], "blocking_reasons": []}
    art = E.dryrun_route(fake_result, {"pr": 1, "head_sha": "x", "task_id": "t", "branch": "b"})
    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
    assert art["actually_executed"] is False


def test_none_pr_identity_does_not_crash():
    """pr_identity=None None-guard (classifier 입력 견고성 doctrine 동형)."""
    fake_result = {"verdict": S.PASS, "auto_merge_10_conditions": {}}
    art = E.dryrun_route(fake_result, None)
    assert art["pr"] == 0
    assert art["head_sha"] == ""
    assert art["actually_executed"] is False


def test_none_classifier_result_falls_back_to_regather():
    art = E.dryrun_route(None, {"pr": 7, "head_sha": "h", "task_id": "t", "branch": "b"})
    assert art["schema"] == E.SCHEMA_HOLD_FOR_CHAIR
    assert art["executor_action"] == E.ACTION_WOULD_REGATHER


# ─────────────────────────────────────────────────────────────────────────────
# 7) classifier integration — classifier 출력을 직접 executor 에 흘려 결정성 검증
# ─────────────────────────────────────────────────────────────────────────────

def test_classifier_output_consumed_unmodified():
    """executor 는 classifier output 을 read-only 로 소비해야 한다 (변형 금지)."""
    ev = json.loads((Path(__file__).resolve().parents[1] / "fixtures" / "merge_ready"
                     / "pass_all_green" / "evidence.json").read_text(encoding="utf-8"))
    declared = (ev.get("scope") or {}).get("expected_files") or []
    result = classify_merge_ready(ev, anu_keys=S.DEFAULT_ANU_KEYS, expected_files=declared)
    snapshot = json.dumps(result, ensure_ascii=False, sort_keys=True)
    _ = E.dryrun_route(result, {"pr": 1, "head_sha": "a", "task_id": "t", "branch": "b",
                                 "ts_kst": "2026-05-22T00:00:00+09:00"})
    after = json.dumps(result, ensure_ascii=False, sort_keys=True)
    assert snapshot == after, "executor must not mutate classifier_result"


# ─────────────────────────────────────────────────────────────────────────────
# 8) 정적 검증 — 모듈 코드에 merge/push/PR/subprocess/network/외부 호출 부재
# ─────────────────────────────────────────────────────────────────────────────

def _module_source():
    path = Path(E.__file__)
    return path.read_text(encoding="utf-8")


def _strip_strings_and_comments(src):
    """docstring/주석을 제거한 코드만 반환 — 정적 검증의 거짓 양성(narrative 문구 일치) 방지.
    완벽한 파서는 아니지만 forbidden 코드 패턴 탐지 목적에 충분히 안전한 휴리스틱."""
    # triple-quoted strings (""" / ''')
    code = re.sub(r'"""[\s\S]*?"""', '""', src)
    code = re.sub(r"'''[\s\S]*?'''", "''", code)
    # single-line strings (보수적 — escape 무시 OK)
    code = re.sub(r'"(?:\\.|[^"\\])*"', '""', code)
    code = re.sub(r"'(?:\\.|[^'\\])*'", "''", code)
    # comments
    code = re.sub(r"#[^\n]*", "", code)
    return code


def test_module_has_no_subprocess_or_network_or_write_io():
    """안전 불변식 정적 검증: merge/push/PR/cron/subprocess/network/file write 코드 경로 부재.

    검사 대상 패턴(반드시 부재 — narrative 문구 제외한 실 코드 토큰):
    - subprocess / os.system / shlex.run
    - urllib / requests / httpx / socket
    - open(..., 'w'/'a') / Path.write_text / write_bytes
    - branch protection bypass / admin override execute
    """
    code = _strip_strings_and_comments(_module_source())
    forbidden_imports = [
        r"\bimport\s+subprocess\b",
        r"\bfrom\s+subprocess\b",
        r"\bimport\s+urllib\b", r"\bfrom\s+urllib\b",
        r"\bimport\s+requests\b", r"\bfrom\s+requests\b",
        r"\bimport\s+httpx\b", r"\bfrom\s+httpx\b",
        r"\bimport\s+socket\b", r"\bfrom\s+socket\b",
        r"\bimport\s+os\b", r"\bfrom\s+os\b",
        r"\bimport\s+shutil\b", r"\bfrom\s+shutil\b",
    ]
    forbidden_calls = [
        r"\bsubprocess\.",
        r"\bos\.system\s*\(", r"\bos\.popen\s*\(",
        r"\bshutil\.",
        r"\burllib\.", r"\brequests\.", r"\bhttpx\.", r"\bsocket\.",
        r"\.write_text\s*\(", r"\.write_bytes\s*\(",
        r"\bopen\s*\(",
    ]
    for pat in forbidden_imports + forbidden_calls:
        assert not re.search(pat, code), f"forbidden code pattern: {pat}"


def test_module_imports_only_states():
    """utils.merge_ready_dryrun_executor 의 import 는 utils.merge_ready_states 만 허용.
    anu_v3/런타임/외부 의존 0 (decoupled doctrine)."""
    src = _module_source()
    import_lines = [
        ln for ln in src.splitlines()
        if ln.strip().startswith(("import ", "from "))
    ]
    for ln in import_lines:
        assert "utils.merge_ready_states" in ln, f"forbidden import: {ln!r}"


def test_module_does_not_call_classifier_or_modify_it():
    """executor 는 classifier 함수/모듈을 호출/import 하지 않는다 (입력 소비만 — Anchor 4).
    docstring/주석의 narrative 언급은 허용 (코드 의존성 부재만 검사)."""
    code = _strip_strings_and_comments(_module_source())
    assert "classify_merge_ready" not in code, "classifier 함수 호출 발견"
    assert "merge_ready_classifier" not in code, "classifier 모듈 import/호출 발견"


# ─────────────────────────────────────────────────────────────────────────────
# 9) JSON serialization 안전성 — artifact 가 json.dumps 가능해야 함
# ─────────────────────────────────────────────────────────────────────────────

@pytest.mark.parametrize("name", ALL_FIXTURES)
def test_artifact_is_json_serializable(name):
    ev, _ = _load(name)
    art = _route(ev)
    # ensure round-trip
    s = json.dumps(art, ensure_ascii=False)
    art2 = json.loads(s)
    assert art2 == art


# ─────────────────────────────────────────────────────────────────────────────
# 10) ANCHOR-2 spec 준수: verdict 4값 매핑 전부 검증 — sanity 종합
# ─────────────────────────────────────────────────────────────────────────────

def test_anchor2_all_four_verdicts_covered():
    """ANCHOR-2: verdict 4값(PASS/HOLD/CHAIR_REQUIRED/UNKNOWN) routing 매핑 전부 커버."""
    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,
    }