"""anu_v2.tests.test_owner_trigger_pat_phase2_2553 — Phase 2 trigger-only comment writer.

회장 §명시 Phase 2 (task-2553):
  - comment body 정확히 `/gemini review` (strict equality, 다른 body fail-fast)
  - endpoint allowlist (issue comment POST 1 endpoint 외 차단)
  - merge/approve/close/reopen/push API 호출 0 정적 차단
  - audit append-only 에 token_present + token_hash 만 (token raw 0)
  - happy path: queue-head + evidence missing + dedupe clean + token loaded → comment posted

mock 패턴은 `test_merge_queue_executor_2531.py` 와 동일 — `subprocess.CompletedProcess`
mock + audit_calls list collector.
"""

from __future__ import annotations

import json
import subprocess
import sys
from pathlib import Path
from typing import Any, Mapping, Sequence

import pytest

# workspace root → sys.path
WORKSPACE_ROOT = Path(__file__).resolve().parents[2]
if str(WORKSPACE_ROOT) not in sys.path:
    sys.path.insert(0, str(WORKSPACE_ROOT))

from anu_v2.owner_trigger_pat import (  # noqa: E402
    ACTOR_OWNER_TRIGGER_PAT,
    ALLOWED_COMMENT_BODY,
    DECISION_PASS,
    DECISION_REJECT,
    ERR_BODY_NOT_ALLOWED,
    ERR_ENDPOINT_NOT_ALLOWED,
    EVIDENCE_MISSING_FOR_CURRENT_HEAD,
    OUTCOME_FAILED,
    OUTCOME_OK,
    OUTCOME_REJECTED,
    OWNER_PAT_ENV_NAME,
    OwnerTriggerPat,
    assert_body_allowed,
    assert_endpoint_allowed,
    serialize_decision,
    write_decision_json,
)


# ─── helpers ────────────────────────────────────────────────────────────────
def _cp(returncode: int = 0, stdout: str = "", stderr: str = "") -> subprocess.CompletedProcess:
    return subprocess.CompletedProcess(args=[], returncode=returncode, stdout=stdout, stderr=stderr)


_FAKE_TOKEN = "github_pat_FAKE_OWNER_TRIGGER_TOKEN_X"
_FIXED_TS = "2026-05-11T12:00:00+00:00"


@pytest.fixture
def gh_calls() -> list[dict[str, Any]]:
    return []


@pytest.fixture
def audit_calls() -> list[dict[str, Any]]:
    return []


@pytest.fixture
def decision_calls() -> list[dict[str, Any]]:
    return []


@pytest.fixture
def trigger(
    gh_calls: list[dict[str, Any]],
    audit_calls: list[dict[str, Any]],
    decision_calls: list[dict[str, Any]],
) -> OwnerTriggerPat:
    def gh_runner(args: Sequence[str], env: Mapping[str, str]) -> subprocess.CompletedProcess:
        gh_calls.append({"args": list(args), "env": dict(env or {})})
        return _cp()

    def audit_writer(record: Mapping[str, Any]) -> None:
        audit_calls.append(dict(record))

    def decision_writer(payload: Mapping[str, Any], path: Path) -> None:
        decision_calls.append({"payload": dict(payload), "path": str(path)})
        # 실제 atomic write 도 검증 (write_decision_json 회로를 함께 검증)
        write_decision_json(payload, path)

    return OwnerTriggerPat(
        gh_runner=gh_runner,
        audit_writer=audit_writer,
        decision_writer=decision_writer,
        owner="jeon-jonghyuk",
        repo="taskctl-anu",
        clock=lambda: _FIXED_TS,
    )


# ─── A. happy path — comment posted ──────────────────────────────────────────
def test_phase2_happy_path_comment_posted(
    tmp_path: Path,
    trigger: OwnerTriggerPat,
    gh_calls: list[dict[str, Any]],
    audit_calls: list[dict[str, Any]],
    decision_calls: list[dict[str, Any]],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setenv(OWNER_PAT_ENV_NAME, _FAKE_TOKEN)
    audit_path = tmp_path / "owner_trigger_audit.jsonl"
    decision_path = tmp_path / "owner_trigger_decision.json"

    out = trigger.trigger_gemini_review(
        pr_number=81,
        head_sha="deadbeef",
        queue_position=0,
        gemini_evidence_state=EVIDENCE_MISSING_FOR_CURRENT_HEAD,
        audit_log_path=audit_path,
        decision_json_path=decision_path,
    )

    assert out.outcome == OUTCOME_OK
    assert out.decision_path == str(decision_path)
    # gh 호출 정확히 1회
    assert len(gh_calls) == 1
    call = gh_calls[0]
    # endpoint allowlist 정확히 일치
    assert call["args"] == [
        "api", "-X", "POST",
        "/repos/jeon-jonghyuk/taskctl-anu/issues/81/comments",
        "-f", f"body={ALLOWED_COMMENT_BODY}",
    ]
    # env 에 OWNER PAT 이 GH_TOKEN/GITHUB_TOKEN 으로 주입됨
    assert call["env"]["GH_TOKEN"] == _FAKE_TOKEN
    assert call["env"]["GITHUB_TOKEN"] == _FAKE_TOKEN

    # decision JSON 박제됨
    assert decision_path.exists()
    decision_data = json.loads(decision_path.read_text(encoding="utf-8"))
    assert decision_data["decision"] == DECISION_PASS
    assert decision_data["pr_number"] == 81
    assert decision_data["dedupe_key"] == "81#deadbeef"
    assert decision_data["ts"] == _FIXED_TS

    # audit record — token_present True, token_hash 박제, token raw 0
    assert len(audit_calls) == 1
    record = audit_calls[0]
    assert record["outcome"] == OUTCOME_OK
    assert record["token_present"] is True
    assert record["token_hash"]
    assert len(record["token_hash"]) == 12
    assert record["actor"] == ACTOR_OWNER_TRIGGER_PAT
    # token raw 가 어디에도 들어가지 않음
    for value in record.values():
        assert _FAKE_TOKEN not in str(value)


# ─── B. body strict equality (fail-fast) ─────────────────────────────────────
def test_phase2_assert_body_allowed_strict_equality() -> None:
    assert_body_allowed(ALLOWED_COMMENT_BODY)
    # 변형은 모두 BODY_NOT_ALLOWED
    for bad in [
        "/gemini  review",
        " /gemini review",
        "/gemini review ",
        "/Gemini review",
        "/gemini review\n",
        "/gemini review please",
        "/gemini-review",
        "",
    ]:
        with pytest.raises(RuntimeError) as exc:
            assert_body_allowed(bad)
        assert ERR_BODY_NOT_ALLOWED in str(exc.value)


def test_phase2_assert_endpoint_allowed_only_issue_comments() -> None:
    """endpoint 가 정확히 issue comments POST path 일 때만 허용."""
    assert_endpoint_allowed(
        "/repos/jeon-jonghyuk/taskctl-anu/issues/81/comments",
        "jeon-jonghyuk", "taskctl-anu", 81,
    )
    forbidden_endpoints = [
        "/repos/jeon-jonghyuk/taskctl-anu/pulls/81/merge",
        "/repos/jeon-jonghyuk/taskctl-anu/pulls/81/reviews",
        "/repos/jeon-jonghyuk/taskctl-anu/issues/82/comments",   # 다른 PR
        "/repos/other/taskctl-anu/issues/81/comments",            # 다른 owner
        "/repos/jeon-jonghyuk/other-repo/issues/81/comments",     # 다른 repo
        "/repos/jeon-jonghyuk/taskctl-anu/git/refs/heads/main",
    ]
    for ep in forbidden_endpoints:
        with pytest.raises(RuntimeError) as exc:
            assert_endpoint_allowed(ep, "jeon-jonghyuk", "taskctl-anu", 81)
        assert ERR_ENDPOINT_NOT_ALLOWED in str(exc.value)


# ─── C. queue_position != 0 → REJECT ─────────────────────────────────────────
def test_phase2_non_queue_head_rejected(
    tmp_path: Path,
    trigger: OwnerTriggerPat,
    gh_calls: list[dict[str, Any]],
    audit_calls: list[dict[str, Any]],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setenv(OWNER_PAT_ENV_NAME, _FAKE_TOKEN)
    audit_path = tmp_path / "audit.jsonl"
    decision_path = tmp_path / "decision.json"

    out = trigger.trigger_gemini_review(
        pr_number=81,
        head_sha="deadbeef",
        queue_position=2,   # not queue-head
        gemini_evidence_state=EVIDENCE_MISSING_FOR_CURRENT_HEAD,
        audit_log_path=audit_path,
        decision_json_path=decision_path,
    )
    assert out.outcome == OUTCOME_REJECTED
    assert "not_queue_head" in out.reason
    # gh 호출 0 (보안 경계)
    assert len(gh_calls) == 0
    # decision 박제 REJECT
    decision_data = json.loads(decision_path.read_text(encoding="utf-8"))
    assert decision_data["decision"] == DECISION_REJECT
    # audit token_present False (토큰 검증 단계 도달 전)
    assert audit_calls[0]["token_present"] is False


# ─── D. evidence state != missing → REJECT ───────────────────────────────────
def test_phase2_evidence_not_missing_rejected(
    tmp_path: Path,
    trigger: OwnerTriggerPat,
    gh_calls: list[dict[str, Any]],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setenv(OWNER_PAT_ENV_NAME, _FAKE_TOKEN)
    audit_path = tmp_path / "audit.jsonl"
    decision_path = tmp_path / "decision.json"

    out = trigger.trigger_gemini_review(
        pr_number=81,
        head_sha="deadbeef",
        queue_position=0,
        gemini_evidence_state="present_for_current_head",  # 이미 evidence 있음
        audit_log_path=audit_path,
        decision_json_path=decision_path,
    )
    assert out.outcome == OUTCOME_REJECTED
    assert "evidence_not_missing" in out.reason
    assert len(gh_calls) == 0


# ─── E. dedupe blocked ───────────────────────────────────────────────────────
def test_phase2_duplicate_trigger_blocked(
    tmp_path: Path,
    trigger: OwnerTriggerPat,
    gh_calls: list[dict[str, Any]],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setenv(OWNER_PAT_ENV_NAME, _FAKE_TOKEN)
    audit_path = tmp_path / "audit.jsonl"
    decision_path = tmp_path / "decision.json"

    # 1차 trigger 성공
    out1 = trigger.trigger_gemini_review(
        pr_number=81,
        head_sha="deadbeef",
        queue_position=0,
        gemini_evidence_state=EVIDENCE_MISSING_FOR_CURRENT_HEAD,
        audit_log_path=audit_path,
        decision_json_path=decision_path,
    )
    assert out1.outcome == OUTCOME_OK
    assert len(gh_calls) == 1

    # 동일 (pr, head) 2차 시도 — fail-fast. audit_log_path 가 비어있어 dedupe 가
    # 실제로 동작하려면 audit 가 jsonl 로 박제되어야 한다. 본 테스트에서는 default
    # writer 가 아니므로 수동으로 jsonl append.
    audit_path.write_text(
        json.dumps({"dedupe_key": "81#deadbeef", "outcome": OUTCOME_OK}) + "\n",
        encoding="utf-8",
    )
    out2 = trigger.trigger_gemini_review(
        pr_number=81,
        head_sha="deadbeef",
        queue_position=0,
        gemini_evidence_state=EVIDENCE_MISSING_FOR_CURRENT_HEAD,
        audit_log_path=audit_path,
        decision_json_path=decision_path,
    )
    assert out2.outcome == OUTCOME_REJECTED
    assert "duplicate_trigger" in out2.reason
    # 두 번째 trigger 는 gh 호출 X — 여전히 1
    assert len(gh_calls) == 1


def test_phase2_new_head_allows_new_trigger(
    tmp_path: Path,
    trigger: OwnerTriggerPat,
    gh_calls: list[dict[str, Any]],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """update-branch 후 새 head 면 dedupe stale — 새 trigger 허용."""
    monkeypatch.setenv(OWNER_PAT_ENV_NAME, _FAKE_TOKEN)
    audit_path = tmp_path / "audit.jsonl"
    decision_path = tmp_path / "decision.json"

    # 기존 head 박제
    audit_path.write_text(
        json.dumps({"dedupe_key": "81#OLDSHA", "outcome": OUTCOME_OK}) + "\n",
        encoding="utf-8",
    )
    # 새 head 로 trigger
    out = trigger.trigger_gemini_review(
        pr_number=81,
        head_sha="NEWSHA",
        queue_position=0,
        gemini_evidence_state=EVIDENCE_MISSING_FOR_CURRENT_HEAD,
        audit_log_path=audit_path,
        decision_json_path=decision_path,
    )
    assert out.outcome == OUTCOME_OK
    assert len(gh_calls) == 1


# ─── F. token 미설정 → REJECT (audit token_present=False) ────────────────────
def test_phase2_token_missing_rejected(
    tmp_path: Path,
    trigger: OwnerTriggerPat,
    gh_calls: list[dict[str, Any]],
    audit_calls: list[dict[str, Any]],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.delenv(OWNER_PAT_ENV_NAME, raising=False)
    # fallback 차단 검증 위해 다른 token 환경에 박아둠
    monkeypatch.setenv("GH_TOKEN", "ghp_LEAKABLE_BOT")
    monkeypatch.setenv("GITHUB_TOKEN", "ghp_LEAKABLE_BOT")

    audit_path = tmp_path / "audit.jsonl"
    decision_path = tmp_path / "decision.json"

    out = trigger.trigger_gemini_review(
        pr_number=81,
        head_sha="deadbeef",
        queue_position=0,
        gemini_evidence_state=EVIDENCE_MISSING_FOR_CURRENT_HEAD,
        audit_log_path=audit_path,
        decision_json_path=decision_path,
    )
    assert out.outcome == OUTCOME_REJECTED
    assert "token_missing" in out.reason
    # gh 호출 0 — fallback 사용 안 함
    assert len(gh_calls) == 0
    # audit 에 token_present False
    record = audit_calls[-1]
    assert record["token_present"] is False
    assert record["token_hash"] == ""
    # leak 된 GH_TOKEN 값이 audit/reason 에 들어가지 않음
    assert "ghp_LEAKABLE_BOT" not in json.dumps(record)
    assert "ghp_LEAKABLE_BOT" not in out.reason


# ─── G. gh runner 실패 시 failed outcome + token redact ──────────────────────
def test_phase2_gh_runner_failure_redacts_token(
    tmp_path: Path,
    audit_calls: list[dict[str, Any]],
    decision_calls: list[dict[str, Any]],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setenv(OWNER_PAT_ENV_NAME, _FAKE_TOKEN)

    def failing_gh(args: Sequence[str], env: Mapping[str, str]) -> subprocess.CompletedProcess:
        # stderr 에 token 값이 우연히 박힌 케이스 — redact 검증
        return _cp(
            returncode=1,
            stderr=f"401 unauthorized: token={_FAKE_TOKEN} expired",
        )

    def audit_writer(record: Mapping[str, Any]) -> None:
        audit_calls.append(dict(record))

    def decision_writer(payload: Mapping[str, Any], path: Path) -> None:
        decision_calls.append({"payload": dict(payload), "path": str(path)})
        write_decision_json(payload, path)

    t = OwnerTriggerPat(
        gh_runner=failing_gh,
        audit_writer=audit_writer,
        decision_writer=decision_writer,
        owner="jeon-jonghyuk",
        repo="taskctl-anu",
        clock=lambda: _FIXED_TS,
    )

    audit_path = tmp_path / "audit.jsonl"
    decision_path = tmp_path / "decision.json"

    out = t.trigger_gemini_review(
        pr_number=81,
        head_sha="deadbeef",
        queue_position=0,
        gemini_evidence_state=EVIDENCE_MISSING_FOR_CURRENT_HEAD,
        audit_log_path=audit_path,
        decision_json_path=decision_path,
    )
    assert out.outcome == OUTCOME_FAILED
    # outcome.reason 에는 redact 처리된 stderr 가 들어감
    assert _FAKE_TOKEN not in out.reason
    assert "***REDACTED***" in out.reason
    # audit 에도 token raw 0
    record = audit_calls[-1]
    assert _FAKE_TOKEN not in json.dumps(record)
    assert record["token_present"] is True
    assert record["token_hash"]


# ─── H. gh runner exception → failed outcome + redact ────────────────────────
def test_phase2_gh_runner_exception_redacts_token(
    tmp_path: Path,
    audit_calls: list[dict[str, Any]],
    decision_calls: list[dict[str, Any]],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setenv(OWNER_PAT_ENV_NAME, _FAKE_TOKEN)

    def raising_gh(args: Sequence[str], env: Mapping[str, str]) -> subprocess.CompletedProcess:
        # exception 메시지에 토큰이 우연히 박혀도 redact 되어야 함
        raise OSError(f"connect failed with creds {_FAKE_TOKEN}")

    def audit_writer(record: Mapping[str, Any]) -> None:
        audit_calls.append(dict(record))

    def decision_writer(payload: Mapping[str, Any], path: Path) -> None:
        decision_calls.append({"payload": dict(payload), "path": str(path)})
        write_decision_json(payload, path)

    t = OwnerTriggerPat(
        gh_runner=raising_gh,
        audit_writer=audit_writer,
        decision_writer=decision_writer,
        owner="jeon-jonghyuk",
        repo="taskctl-anu",
        clock=lambda: _FIXED_TS,
    )

    audit_path = tmp_path / "audit.jsonl"
    decision_path = tmp_path / "decision.json"

    out = t.trigger_gemini_review(
        pr_number=81,
        head_sha="deadbeef",
        queue_position=0,
        gemini_evidence_state=EVIDENCE_MISSING_FOR_CURRENT_HEAD,
        audit_log_path=audit_path,
        decision_json_path=decision_path,
    )
    assert out.outcome == OUTCOME_FAILED
    assert _FAKE_TOKEN not in out.reason
    record = audit_calls[-1]
    assert _FAKE_TOKEN not in json.dumps(record)


# ─── I. owner/repo 인자 검증 ─────────────────────────────────────────────────
def test_phase2_invalid_owner_repo_rejected() -> None:
    """owner / repo 에 path traversal / 빈 값 등 입력 시 ValueError."""
    def noop_gh(args: Sequence[str], env: Mapping[str, str]) -> subprocess.CompletedProcess:
        return _cp()

    def noop_audit(record: Mapping[str, Any]) -> None:
        return None

    def noop_decision(payload: Mapping[str, Any], path: Path) -> None:
        return None

    for owner, repo in [
        ("", "repo"),
        ("owner", ""),
        ("owner/../etc", "repo"),
        ("owner", "repo with space"),
        ("../owner", "repo"),
    ]:
        with pytest.raises(ValueError):
            OwnerTriggerPat(
                gh_runner=noop_gh,
                audit_writer=noop_audit,
                decision_writer=noop_decision,
                owner=owner,
                repo=repo,
            )


# ─── J. security boundaries — endpoint allowlist 정적 차단 ────────────────────
def test_phase2_security_only_one_endpoint_called(
    tmp_path: Path,
    trigger: OwnerTriggerPat,
    gh_calls: list[dict[str, Any]],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """모든 happy-path trigger 호출의 args 가 issue comments POST 외 다른 fragment 0."""
    monkeypatch.setenv(OWNER_PAT_ENV_NAME, _FAKE_TOKEN)
    audit_path = tmp_path / "audit.jsonl"
    decision_path = tmp_path / "decision.json"

    trigger.trigger_gemini_review(
        pr_number=81,
        head_sha="deadbeef",
        queue_position=0,
        gemini_evidence_state=EVIDENCE_MISSING_FOR_CURRENT_HEAD,
        audit_log_path=audit_path,
        decision_json_path=decision_path,
    )
    assert len(gh_calls) == 1
    args = gh_calls[0]["args"]
    # 금지 endpoint fragment 가 args 어디에도 없어야 함
    forbidden = ("/merges", "/reviews", "/pulls/", "/branches/", "/git/refs", "/merge")
    flat = " ".join(args)
    for frag in forbidden:
        assert frag not in flat
    # body 는 정확히 `/gemini review` only
    assert f"body={ALLOWED_COMMENT_BODY}" in args


def test_phase2_audit_record_token_raw_zero(
    tmp_path: Path,
    trigger: OwnerTriggerPat,
    audit_calls: list[dict[str, Any]],
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """audit record 안의 모든 값에 token raw 0 (직렬화 후 검증)."""
    monkeypatch.setenv(OWNER_PAT_ENV_NAME, _FAKE_TOKEN)
    audit_path = tmp_path / "audit.jsonl"
    decision_path = tmp_path / "decision.json"

    trigger.trigger_gemini_review(
        pr_number=81,
        head_sha="deadbeef",
        queue_position=0,
        gemini_evidence_state=EVIDENCE_MISSING_FOR_CURRENT_HEAD,
        audit_log_path=audit_path,
        decision_json_path=decision_path,
    )
    record = audit_calls[-1]
    serialized = json.dumps(record, sort_keys=True)
    assert _FAKE_TOKEN not in serialized
    # 필수 필드 박제
    assert record["token_present"] is True
    assert record["token_hash"]
    assert record["pr_number"] == 81
    assert record["head_sha"] == "deadbeef"
    assert record["dedupe_key"] == "81#deadbeef"
    assert record["outcome"] == OUTCOME_OK
    assert record["actor"] == ACTOR_OWNER_TRIGGER_PAT


def test_phase2_serialize_decision_no_token_in_output(
    tmp_path: Path,
    trigger: OwnerTriggerPat,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """decision JSON 박제 결과에 token raw 0 (PASS 케이스).

    decision 파일 내용 + serialize_decision 회로 둘 다 검증.
    """
    from anu_v2.owner_trigger_pat import OwnerTriggerDecision

    monkeypatch.setenv(OWNER_PAT_ENV_NAME, _FAKE_TOKEN)
    audit_path = tmp_path / "audit.jsonl"
    decision_path = tmp_path / "decision.json"

    trigger.trigger_gemini_review(
        pr_number=81,
        head_sha="deadbeef",
        queue_position=0,
        gemini_evidence_state=EVIDENCE_MISSING_FOR_CURRENT_HEAD,
        audit_log_path=audit_path,
        decision_json_path=decision_path,
    )
    decision_text = decision_path.read_text(encoding="utf-8")
    assert _FAKE_TOKEN not in decision_text
    # PASS 필드 확인
    payload = json.loads(decision_text)
    assert payload["decision"] == DECISION_PASS
    # serialize_decision 회로 — frozen instance reconstruct 후 재직렬화
    reconstructed = OwnerTriggerDecision(
        pr_number=payload["pr_number"],
        head_sha=payload["head_sha"],
        decision=payload["decision"],
        reason=payload["reason"],
        gemini_evidence_state=payload["gemini_evidence_state"],
        queue_position=payload["queue_position"],
        dedupe_key=payload["dedupe_key"],
        ts=payload["ts"],
    )
    serialized_again = json.dumps(serialize_decision(reconstructed))
    assert _FAKE_TOKEN not in serialized_again