"""tests/regression/test_ci_gemini_watcher_runner_2718.py — task-2718 회귀 테스트 스위트.

회장 verbatim 매핑 (시나리오 1~13):
  1.  test_merge_ready_candidate          — FRESH + escalated 0 + CI SUCCESS → MERGE_READY_CANDIDATE
  2.  test_loop_boundary_on_escalation    — FRESH + escalated ≥1 → LOOP_BOUNDARY
  3.  test_hold_stale_head                — actual_head != expected_head → HOLD_STALE_HEAD
  4.  test_hold_scope_unclean             — diff_paths 에 scope 외 경로 → HOLD_SCOPE_UNCLEAN
  5.  test_ci_failed_non_remediable       — CI FAILURE non-remediable → CI_FAILED_NON_REMEDIABLE
  6.  test_allow_owner_trigger            — STALE + is_owner + dedupe False → ALLOW_OWNER_TRIGGER
  7.  test_external_trigger_required_without_owner — STALE + owner None → GEMINI_EXTERNAL_TRIGGER_REQUIRED
  8.  test_self_key_blocked               — STALE + self_key=True → BLOCKED_BY_CAPABILITY / SELF_GEMINI_TRIGGER_BLOCKED
  9.  test_blocked_by_capability_no_gh_runner — gh_runner=None → BLOCKED_BY_CAPABILITY
  10. test_auto_gemini_triage_called      — FRESH 경로에서 AutoGeminiTriage.triage_batch mock 주입 후 호출 검증
  11. test_owner_trigger_decision_validate_called — STALE+owner 경로에서 validate_decision spy 검증
  12. test_github_write_zero              — 모든 주요 경로에서 github_writes == 0
  13. test_idempotent_dedupe              — 동일 head 2회 호출, 2회차 dedupe → OWNER_TRIGGER_DEDUPED
"""

from __future__ import annotations

import unittest.mock as mock
from typing import Any, Callable

from anu_v2.auto_gemini_triage import AutoGeminiTriage
from anu_v2.ci_gemini_watcher_runner import (
    ALL_TERMINALS,
    TERMINAL_BLOCKED_BY_CAPABILITY,
    TERMINAL_CI_FAILED_NON_REMEDIABLE,
    TERMINAL_GEMINI_EXTERNAL_TRIGGER_REQUIRED,
    TERMINAL_HOLD_SCOPE_UNCLEAN,
    TERMINAL_HOLD_STALE_HEAD,
    TERMINAL_LOOP_BOUNDARY,
    TERMINAL_MERGE_READY_CANDIDATE,
    TRIGGER_ALLOW_OWNER,
    TRIGGER_EXTERNAL_REQUIRED,
    TRIGGER_NONE,
    TRIGGER_OWNER_DEDUPED,
    TRIGGER_SELF_BLOCKED,
    WatchResult,
    run_watch_cycle,
)
from anu_v2.owner_trigger_decision import validate_decision

# ──────────────────────────────────────────────────────────────────────────────
# SHA fixture 헬퍼
# ──────────────────────────────────────────────────────────────────────────────
HEAD_A = "a" * 40   # expected_head (= actual head for FRESH/CLEAN tests)
HEAD_B = "b" * 40   # alternate head (= stale actual head)

# 기본 expected_files
DEFAULT_FILES = ["anu_v2/ci_gemini_watcher_runner.py", "anu_v2/auto_gemini_triage.py"]


# ──────────────────────────────────────────────────────────────────────────────
# gh_runner 빌더 헬퍼
# ──────────────────────────────────────────────────────────────────────────────

def make_gh_runner(
    *,
    actual_head: str = HEAD_A,
    diff_paths: list[str] | None = None,
    ci_state: str = "SUCCESS",
    ci_remediable: bool = True,
    reviews: list[dict] | None = None,
    findings: list[dict] | None = None,
) -> Callable[..., Any]:
    """op 분기 dispatch 하는 gh_runner closure 를 반환한다."""
    _diff_paths = diff_paths if diff_paths is not None else list(DEFAULT_FILES)
    _reviews = reviews if reviews is not None else []
    _findings = findings if findings is not None else []

    def gh_runner(op: str, **params: Any) -> Any:
        if op == "actual_head":
            return actual_head
        if op == "diff_paths":
            return _diff_paths
        if op == "ci_rollup":
            return {"state": ci_state, "remediable": ci_remediable}
        if op == "reviews":
            # freshness checker 가 "GET" method 로 호출함
            return _reviews
        if op == "findings":
            return _findings
        raise ValueError(f"unknown op: {op!r}")

    return gh_runner


def make_gemini_review(commit_id: str) -> dict:
    """FRESH fixture: gemini-code-assist[bot] review with given commit_id."""
    return {
        "user": {"login": "gemini-code-assist[bot]"},
        "commit_id": commit_id,
        "state": "COMMENTED",
        "body": "Looks good.",
    }


def make_finding(
    *,
    rule_id: str = "no-op",
    severity: str = "low",
    category: str = "style",
    path: str = "anu_v2/ci_gemini_watcher_runner.py",
    body: str = "",
    title: str = "",
) -> dict:
    """finding fixture 헬퍼."""
    return {
        "rule_id": rule_id,
        "severity": severity,
        "category": category,
        "path": path,
        "body": body,
        "title": title,
    }


# ──────────────────────────────────────────────────────────────────────────────
# 시나리오 1
# ──────────────────────────────────────────────────────────────────────────────

def test_merge_ready_candidate():
    """FRESH review + escalated 0 + CI SUCCESS + scope clean → MERGE_READY_CANDIDATE."""
    reviews = [make_gemini_review(HEAD_A)]
    # style finding 은 dismiss_style_only 로 분류 → escalated 0
    findings = [make_finding(severity="low", category="style", path=DEFAULT_FILES[0])]

    runner = make_gh_runner(
        actual_head=HEAD_A,
        diff_paths=list(DEFAULT_FILES),
        ci_state="SUCCESS",
        reviews=reviews,
        findings=findings,
    )
    result = run_watch_cycle(
        pr_number=42,
        expected_head=HEAD_A,
        expected_files=DEFAULT_FILES,
        gh_runner=runner,
        dry_run=True,
    )

    assert result.terminal == TERMINAL_MERGE_READY_CANDIDATE
    assert result.github_writes == 0
    assert result.dry_run is True
    assert result.pr_number == 42
    assert isinstance(result.to_json(), dict)


# ──────────────────────────────────────────────────────────────────────────────
# 시나리오 2 — 진짜 AutoGeminiTriage 사용 (scope_expansion finding)
# ──────────────────────────────────────────────────────────────────────────────

def test_loop_boundary_on_escalation():
    """FRESH review + scope_expansion finding(out-of-scope path) → escalated ≥1 → LOOP_BOUNDARY.
    진짜 AutoGeminiTriage 를 사용해 escalation 분류를 실증한다.
    """
    reviews = [make_gemini_review(HEAD_A)]
    # out-of-scope path → scope_expansion → escalated
    findings = [make_finding(
        rule_id="real-bug-001",
        severity="high",
        category="bug",
        path="some/other/file.py",  # expected_files 밖 경로
    )]

    runner = make_gh_runner(
        actual_head=HEAD_A,
        diff_paths=list(DEFAULT_FILES),
        ci_state="SUCCESS",
        reviews=reviews,
        findings=findings,
    )
    result = run_watch_cycle(
        pr_number=42,
        expected_head=HEAD_A,
        expected_files=DEFAULT_FILES,
        gh_runner=runner,
        dry_run=True,
    )

    assert result.terminal == TERMINAL_LOOP_BOUNDARY
    assert result.triage_summary is not None
    assert result.triage_summary["escalated_count"] >= 1
    assert result.github_writes == 0


# ──────────────────────────────────────────────────────────────────────────────
# 시나리오 3
# ──────────────────────────────────────────────────────────────────────────────

def test_hold_stale_head():
    """actual_head != expected_head → terminal HOLD_STALE_HEAD."""
    runner = make_gh_runner(actual_head=HEAD_B)  # actual != expected(HEAD_A)

    result = run_watch_cycle(
        pr_number=1,
        expected_head=HEAD_A,
        expected_files=DEFAULT_FILES,
        gh_runner=runner,
    )

    assert result.terminal == TERMINAL_HOLD_STALE_HEAD
    assert result.actual_head == HEAD_B
    assert result.github_writes == 0


def test_hold_stale_head_on_none_actual_head():
    """actual_head 가 None/빈값이면 'None' 암묵 문자열화 없이 빈 head 로 fail-closed
    → HOLD_STALE_HEAD (invalid head 가 정상 head 처럼 흐르지 않음)."""
    base = make_gh_runner()

    def runner(op: str, **params: Any) -> Any:
        return None if op == "actual_head" else base(op, **params)

    result = run_watch_cycle(
        pr_number=1,
        expected_head=HEAD_A,
        expected_files=DEFAULT_FILES,
        gh_runner=runner,
    )

    assert result.terminal == TERMINAL_HOLD_STALE_HEAD
    assert result.actual_head == ""   # 'none' 아님 — 암묵 변환 방지(fail-closed)
    assert result.github_writes == 0


# ──────────────────────────────────────────────────────────────────────────────
# 시나리오 4
# ──────────────────────────────────────────────────────────────────────────────

def test_hold_scope_unclean():
    """diff_paths 에 expected_files 밖 경로 포함 → HOLD_SCOPE_UNCLEAN."""
    diff = list(DEFAULT_FILES) + ["scripts/intruder.sh"]  # scope 외 경로
    runner = make_gh_runner(actual_head=HEAD_A, diff_paths=diff)

    result = run_watch_cycle(
        pr_number=7,
        expected_head=HEAD_A,
        expected_files=DEFAULT_FILES,
        gh_runner=runner,
    )

    assert result.terminal == TERMINAL_HOLD_SCOPE_UNCLEAN
    assert result.github_writes == 0


# ──────────────────────────────────────────────────────────────────────────────
# 시나리오 5
# ──────────────────────────────────────────────────────────────────────────────

def test_ci_failed_non_remediable():
    """CI FAILURE + remediable=False → CI_FAILED_NON_REMEDIABLE."""
    runner = make_gh_runner(
        actual_head=HEAD_A,
        diff_paths=list(DEFAULT_FILES),
        ci_state="FAILURE",
        ci_remediable=False,
    )

    result = run_watch_cycle(
        pr_number=10,
        expected_head=HEAD_A,
        expected_files=DEFAULT_FILES,
        gh_runner=runner,
    )

    assert result.terminal == TERMINAL_CI_FAILED_NON_REMEDIABLE
    assert result.ci_state == "FAILURE"
    assert result.github_writes == 0


# ──────────────────────────────────────────────────────────────────────────────
# 시나리오 6
# ──────────────────────────────────────────────────────────────────────────────

def test_allow_owner_trigger():
    """STALE review + is_owner=True/admin=True + dedupe False → ALLOW_OWNER_TRIGGER.
    decision_json 이 owner_trigger_decision.v1 schema 를 통과하는지 validate_decision 재검증.
    terminal == GEMINI_EXTERNAL_TRIGGER_REQUIRED.
    """
    # STALE: reviews commit_id != actual_head
    reviews = [make_gemini_review(HEAD_B)]  # HEAD_B != HEAD_A
    runner = make_gh_runner(
        actual_head=HEAD_A,
        diff_paths=list(DEFAULT_FILES),
        ci_state="SUCCESS",
        reviews=reviews,
    )
    owner_proof = {"is_owner": True, "admin": True}
    recorded: list[tuple] = []

    def record_trigger(pr: int, sha: str) -> None:
        recorded.append((pr, sha))

    result = run_watch_cycle(
        pr_number=20,
        expected_head=HEAD_A,
        expected_files=DEFAULT_FILES,
        gh_runner=runner,
        owner_proof=owner_proof,
        dedupe_checker=lambda pr, sha: False,
        record_trigger=record_trigger,
        task_id="task-2718",
        dry_run=True,
    )

    assert result.trigger_decision == TRIGGER_ALLOW_OWNER
    assert result.terminal == TERMINAL_GEMINI_EXTERNAL_TRIGGER_REQUIRED
    assert result.decision_json is not None
    assert result.github_writes == 0

    # decision_json 이 validate_decision 통과 검증
    validated = validate_decision(result.decision_json, current_head_actual=result.actual_head)
    assert validated is result.decision_json

    # record_trigger 호출 확인
    assert len(recorded) == 1
    assert recorded[0] == (20, HEAD_A)


# ──────────────────────────────────────────────────────────────────────────────
# 시나리오 7
# ──────────────────────────────────────────────────────────────────────────────

def test_external_trigger_required_without_owner():
    """STALE review + owner_proof=None → terminal GEMINI_EXTERNAL_TRIGGER_REQUIRED, trigger GEMINI_EXTERNAL_TRIGGER_REQUIRED."""
    reviews = [make_gemini_review(HEAD_B)]
    runner = make_gh_runner(
        actual_head=HEAD_A,
        diff_paths=list(DEFAULT_FILES),
        reviews=reviews,
    )

    result = run_watch_cycle(
        pr_number=30,
        expected_head=HEAD_A,
        expected_files=DEFAULT_FILES,
        gh_runner=runner,
        owner_proof=None,
    )

    assert result.terminal == TERMINAL_GEMINI_EXTERNAL_TRIGGER_REQUIRED
    assert result.trigger_decision == TRIGGER_EXTERNAL_REQUIRED
    assert result.decision_json is None
    assert result.github_writes == 0


def test_external_trigger_required_is_owner_false():
    """STALE + is_owner=False → GEMINI_EXTERNAL_TRIGGER_REQUIRED (owner 불충분)."""
    reviews = [make_gemini_review(HEAD_B)]
    runner = make_gh_runner(actual_head=HEAD_A, diff_paths=list(DEFAULT_FILES), reviews=reviews)

    result = run_watch_cycle(
        pr_number=31,
        expected_head=HEAD_A,
        expected_files=DEFAULT_FILES,
        gh_runner=runner,
        owner_proof={"is_owner": False, "admin": True},
    )

    assert result.terminal == TERMINAL_GEMINI_EXTERNAL_TRIGGER_REQUIRED
    assert result.trigger_decision == TRIGGER_EXTERNAL_REQUIRED


# ──────────────────────────────────────────────────────────────────────────────
# 시나리오 8
# ──────────────────────────────────────────────────────────────────────────────

def test_self_key_blocked():
    """STALE + owner_proof.self_key=True → BLOCKED_BY_CAPABILITY / SELF_GEMINI_TRIGGER_BLOCKED."""
    reviews = [make_gemini_review(HEAD_B)]
    runner = make_gh_runner(actual_head=HEAD_A, diff_paths=list(DEFAULT_FILES), reviews=reviews)

    result = run_watch_cycle(
        pr_number=40,
        expected_head=HEAD_A,
        expected_files=DEFAULT_FILES,
        gh_runner=runner,
        owner_proof={"is_owner": True, "admin": True, "self_key": True},
    )

    assert result.terminal == TERMINAL_BLOCKED_BY_CAPABILITY
    assert result.trigger_decision == TRIGGER_SELF_BLOCKED
    assert result.github_writes == 0


# ──────────────────────────────────────────────────────────────────────────────
# 시나리오 9
# ──────────────────────────────────────────────────────────────────────────────

def test_blocked_by_capability_no_gh_runner():
    """gh_runner=None → capability gate 차단 → BLOCKED_BY_CAPABILITY."""
    result = run_watch_cycle(
        pr_number=50,
        expected_head=HEAD_A,
        expected_files=DEFAULT_FILES,
        gh_runner=None,  # type: ignore[arg-type]
    )

    assert result.terminal == TERMINAL_BLOCKED_BY_CAPABILITY
    assert result.trigger_decision == TRIGGER_NONE
    assert result.github_writes == 0


# ──────────────────────────────────────────────────────────────────────────────
# 시나리오 10
# ──────────────────────────────────────────────────────────────────────────────

def test_auto_gemini_triage_called():
    """FRESH 경로에서 triage= 로 MagicMock(spec=AutoGeminiTriage) 주입 후 triage_batch 호출 검증."""
    reviews = [make_gemini_review(HEAD_A)]
    findings = [make_finding(path=DEFAULT_FILES[0])]

    runner = make_gh_runner(
        actual_head=HEAD_A,
        diff_paths=list(DEFAULT_FILES),
        ci_state="SUCCESS",
        reviews=reviews,
        findings=findings,
    )

    mock_triage = mock.MagicMock(spec=AutoGeminiTriage)
    mock_triage.triage_batch.return_value = {"applied": [], "dismissed": [], "escalated": []}

    result = run_watch_cycle(
        pr_number=60,
        expected_head=HEAD_A,
        expected_files=DEFAULT_FILES,
        gh_runner=runner,
        triage=mock_triage,
    )

    # triage_batch 가 정확히 1회 호출되었는지 검증
    mock_triage.triage_batch.assert_called_once()
    call_args = mock_triage.triage_batch.call_args
    # positional: (findings, expected_files)
    called_findings = call_args[0][0]
    called_expected = call_args[0][1]
    assert called_findings == findings
    assert list(called_expected) == DEFAULT_FILES

    # CI SUCCESS + escalated 0 → MERGE_READY_CANDIDATE
    assert result.terminal == TERMINAL_MERGE_READY_CANDIDATE
    assert result.github_writes == 0


# ──────────────────────────────────────────────────────────────────────────────
# 시나리오 11
# ──────────────────────────────────────────────────────────────────────────────

def test_owner_trigger_decision_validate_called():
    """STALE+owner 경로에서 validate_decision 이 실제 호출되어 decision_json 생성됨을 spy 로 입증."""
    reviews = [make_gemini_review(HEAD_B)]
    runner = make_gh_runner(
        actual_head=HEAD_A,
        diff_paths=list(DEFAULT_FILES),
        reviews=reviews,
    )

    with mock.patch(
        "anu_v2.ci_gemini_watcher_runner.validate_decision",
        wraps=validate_decision,
    ) as spy:
        result = run_watch_cycle(
            pr_number=70,
            expected_head=HEAD_A,
            expected_files=DEFAULT_FILES,
            gh_runner=runner,
            owner_proof={"is_owner": True, "admin": True},
            dedupe_checker=lambda pr, sha: False,
            task_id="task-2718-spy",
        )

    spy.assert_called_once()
    assert result.decision_json is not None
    assert result.trigger_decision == TRIGGER_ALLOW_OWNER


# ──────────────────────────────────────────────────────────────────────────────
# 시나리오 12
# ──────────────────────────────────────────────────────────────────────────────

def test_github_write_zero():
    """merge_ready, allow_owner, external_required 모든 경로에서 github_writes == 0."""
    # --- merge_ready 경로 ---
    r1 = run_watch_cycle(
        pr_number=80,
        expected_head=HEAD_A,
        expected_files=DEFAULT_FILES,
        gh_runner=make_gh_runner(
            actual_head=HEAD_A,
            diff_paths=list(DEFAULT_FILES),
            ci_state="SUCCESS",
            reviews=[make_gemini_review(HEAD_A)],
            findings=[make_finding(category="style")],
        ),
    )
    assert r1.github_writes == 0
    assert r1.terminal == TERMINAL_MERGE_READY_CANDIDATE

    # --- allow_owner 경로 ---
    r2 = run_watch_cycle(
        pr_number=81,
        expected_head=HEAD_A,
        expected_files=DEFAULT_FILES,
        gh_runner=make_gh_runner(
            actual_head=HEAD_A,
            diff_paths=list(DEFAULT_FILES),
            reviews=[make_gemini_review(HEAD_B)],
        ),
        owner_proof={"is_owner": True, "admin": True},
        dedupe_checker=lambda pr, sha: False,
    )
    assert r2.github_writes == 0
    assert r2.trigger_decision == TRIGGER_ALLOW_OWNER

    # --- external_required 경로 ---
    r3 = run_watch_cycle(
        pr_number=82,
        expected_head=HEAD_A,
        expected_files=DEFAULT_FILES,
        gh_runner=make_gh_runner(
            actual_head=HEAD_A,
            diff_paths=list(DEFAULT_FILES),
            reviews=[make_gemini_review(HEAD_B)],
        ),
        owner_proof=None,
    )
    assert r3.github_writes == 0
    assert r3.trigger_decision == TRIGGER_EXTERNAL_REQUIRED


# ──────────────────────────────────────────────────────────────────────────────
# 시나리오 13
# ──────────────────────────────────────────────────────────────────────────────

def test_idempotent_dedupe():
    """동일 head 2회 호출: 1회차 ALLOW_OWNER_TRIGGER → record_trigger 호출. 2회차 dedupe → OWNER_TRIGGER_DEDUPED. record_trigger 총 호출 횟수 == 1."""
    reviews = [make_gemini_review(HEAD_B)]  # STALE
    runner = make_gh_runner(
        actual_head=HEAD_A,
        diff_paths=list(DEFAULT_FILES),
        reviews=reviews,
    )
    owner_proof = {"is_owner": True, "admin": True}

    # stateful set: 실제 dedupe_checker 가 참조하는 in-memory 저장소
    triggered_set: set[tuple[int, str]] = set()

    def dedupe_checker(pr: int, sha: str) -> bool:
        return (pr, sha) in triggered_set

    recorded_calls: list[tuple] = []

    def record_trigger(pr: int, sha: str) -> None:
        triggered_set.add((pr, sha))
        recorded_calls.append((pr, sha))

    # 1회차
    r1 = run_watch_cycle(
        pr_number=90,
        expected_head=HEAD_A,
        expected_files=DEFAULT_FILES,
        gh_runner=runner,
        owner_proof=owner_proof,
        dedupe_checker=dedupe_checker,
        record_trigger=record_trigger,
        task_id="task-2718-dedupe",
    )
    assert r1.trigger_decision == TRIGGER_ALLOW_OWNER
    assert len(recorded_calls) == 1

    # 2회차 — dedupe_checker 가 이미 trigger 된 것으로 반환
    r2 = run_watch_cycle(
        pr_number=90,
        expected_head=HEAD_A,
        expected_files=DEFAULT_FILES,
        gh_runner=runner,
        owner_proof=owner_proof,
        dedupe_checker=dedupe_checker,
        record_trigger=record_trigger,
        task_id="task-2718-dedupe",
    )
    assert r2.trigger_decision == TRIGGER_OWNER_DEDUPED
    # record_trigger 는 2회차에 호출되지 않아야 함
    assert len(recorded_calls) == 1
    assert r2.github_writes == 0


# ──────────────────────────────────────────────────────────────────────────────
# 보너스: WatchResult.to_json() 직렬화 smoke
# ──────────────────────────────────────────────────────────────────────────────

def test_watch_result_to_json_serializable():
    """WatchResult.to_json() 이 dict 를 반환하고 모든 terminal 상수가 ALL_TERMINALS 에 포함됨."""
    result = run_watch_cycle(
        pr_number=99,
        expected_head=HEAD_A,
        expected_files=DEFAULT_FILES,
        gh_runner=make_gh_runner(
            actual_head=HEAD_A,
            diff_paths=list(DEFAULT_FILES),
            ci_state="SUCCESS",
            reviews=[make_gemini_review(HEAD_A)],
            findings=[],
        ),
    )
    d = result.to_json()
    assert isinstance(d, dict)
    assert d["terminal"] in ALL_TERMINALS
    assert d["github_writes"] == 0