"""test_owner_trigger_2553_plus1_high_fix.py — task-2553+1 F1-solo regression.

F1 = `/gemini review` comment trigger 의 endpoint/args allowlist 강화. token transport 무관 (task-2553+1 §2). clean replacement PR 의 6-file effective diff #2 (회장 §5 single authority).
"""

from __future__ import annotations

import importlib.util
import sys
from pathlib import Path

import pytest

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

_OTP = _WS / 'anu_v2/owner_trigger_pat.py'


def _load_otp():
    spec = importlib.util.spec_from_file_location(
        "anu_v2_owner_trigger_pat_2553p1_highfix", _OTP
    )
    mod = importlib.util.module_from_spec(spec)
    sys.modules[spec.name] = mod  # task-2553+14: py3.12 @dataclass(dataclasses.py:749) sys.modules.get(cls.__module__) None 가드
    spec.loader.exec_module(mod)
    return mod


def test_allowed_comment_body_is_exactly_gemini_review():
    """F1 invariant: 허용 comment body 는 정확히 `/gemini review`."""
    otp = _load_otp()
    assert otp.ALLOWED_COMMENT_BODY == "/gemini review"


def test_args_allowlist_rejects_foreign_endpoint():
    """F1 invariant: 다른 endpoint → ENDPOINT_NOT_ALLOWED 정적 차단."""
    otp = _load_otp()
    with pytest.raises(Exception) as ei:
        otp._assert_args_allowlist(
            ["api", "-X", "POST", "/repos/o/r/issues/1/labels",
             "-f", "body=/gemini review"],
            "o", "r", 1,
        )
    assert otp.ERR_ENDPOINT_NOT_ALLOWED in str(ei.value)


def test_args_allowlist_rejects_foreign_body():
    """F1 invariant: 다른 body → BODY_NOT_ALLOWED 정적 차단."""
    otp = _load_otp()
    with pytest.raises(Exception) as ei:
        otp._assert_args_allowlist(
            ["api", "-X", "POST", "/repos/o/r/issues/1/comments",
             "-f", "body=please merge"],
            "o", "r", 1,
        )
    assert otp.ERR_BODY_NOT_ALLOWED in str(ei.value)


# ─── task-2553+10: is_duplicate_trigger 스트리밍 리팩터 동치 회귀 ──────────────
# Gemini medium thread(owner_trigger_pat.py:263) 해소 — read_text()+splitlines()
# → with path.open() as f: for line in f 스트리밍 전환. behavior-preserving 증명.
# 기존 assertion 수정 0 (신규 helper/케이스만 추가).

import inspect
import json


def _load_otp_streaming():
    """sys.modules 등록형 로더 — Python 3.12 frozen dataclass 문자열 어노 해소.

    기존 `_load_otp` 는 sys.modules 미등록이라 3.12 에서 @dataclass(line 77)
    문자열 어노 해소가 깨진다(본 task 무관·사전존재). 본 helper 는 동일 모듈을
    sys.modules 선등록 후 로드하여 is_duplicate_trigger 동치 검증을 가능케 한다.
    """
    spec = importlib.util.spec_from_file_location(
        "anu_v2_owner_trigger_pat_2553p10_streaming", _OTP
    )
    assert spec is not None and spec.loader is not None
    mod = importlib.util.module_from_spec(spec)
    sys.modules[spec.name] = mod
    spec.loader.exec_module(mod)
    return mod


def _write(tmp_path, content_bytes: bytes):
    p = tmp_path / "audit.jsonl"
    p.write_bytes(content_bytes)
    return p


def test_is_duplicate_streaming_equivalence_core(tmp_path):
    """매칭/미매칭/깨진라인-skip/dedupe 마커우선 — 스트리밍 전후 동일 결과."""
    otp = _load_otp_streaming()
    key = "128#6c443d87"
    ok = json.dumps({"dedupe_key": key, "outcome": otp.OUTCOME_OK})
    pend = json.dumps({"dedupe_key": key, "outcome": otp.OUTCOME_PENDING})
    # 매칭(ok) → True
    assert otp.is_duplicate_trigger(_write(tmp_path, (ok + "\n").encode()), key) is True
    # 매칭(pending) → True
    assert otp.is_duplicate_trigger(_write(tmp_path, (pend + "\n").encode()), key) is True
    # 미매칭(다른 key) → False
    other = json.dumps({"dedupe_key": "999#x", "outcome": otp.OUTCOME_OK})
    assert otp.is_duplicate_trigger(_write(tmp_path, (other + "\n").encode()), key) is False
    # 깨진 라인 skip 후 정상 라인 매칭 → True
    assert otp.is_duplicate_trigger(_write(tmp_path, (b"{not json\n" + (ok + "\n").encode())), key) is True
    # 파일 미존재 → False
    assert otp.is_duplicate_trigger(tmp_path / "absent.jsonl", key) is False
    # 디렉토리(OSError) → False
    assert otp.is_duplicate_trigger(tmp_path, key) is False


def test_is_duplicate_streaming_adversarial_parity(tmp_path):
    """9-R.5 적대적 동치: trailing-newline無 / BOM첫라인 / 다수+빈줄혼재 / 빈파일."""
    otp = _load_otp_streaming()
    key = "128#6c443d87"
    ok = json.dumps({"dedupe_key": key, "outcome": otp.OUTCOME_OK})
    pend = json.dumps({"dedupe_key": key, "outcome": otp.OUTCOME_PENDING})
    # (1) 최종 라인 trailing newline 無 → 매칭 True
    assert otp.is_duplicate_trigger(_write(tmp_path, ok.encode()), key) is True
    # (2) BOM-prefixed 첫 라인(utf-8 BOM 미strip → json 실패 → skip) → False
    assert otp.is_duplicate_trigger(_write(tmp_path, b"\xef\xbb\xbf" + (ok + "\n").encode()), key) is False
    # (3) 다수 라인 + 빈 줄 혼재 → 후행 매칭 True
    blob = (b"\n\n" + (json.dumps({"dedupe_key": "a", "outcome": "ok"}) + "\n").encode()
            + b"   \n" + (pend + "\n").encode() + b"\n")
    assert otp.is_duplicate_trigger(_write(tmp_path, blob), key) is True
    # (4) 빈 파일 → False
    assert otp.is_duplicate_trigger(_write(tmp_path, b""), key) is False
    # CRLF / CR-only 종결도 매칭 True (universal newline parity)
    assert otp.is_duplicate_trigger(_write(tmp_path, (ok + "\r\n").encode()), key) is True
    assert otp.is_duplicate_trigger(_write(tmp_path, (ok + "\r").encode()), key) is True


def test_is_duplicate_decode_error_propagates_not_false(tmp_path):
    """9-R.3 decode-error parity: invalid utf-8 → UnicodeDecodeError 전파(False 변환 0)."""
    otp = _load_otp_streaming()
    p = _write(tmp_path, b"\xff\xfe invalid utf8 \xff")
    with pytest.raises(UnicodeDecodeError):
        otp.is_duplicate_trigger(p, "128#6c443d87")


def test_is_duplicate_no_read_text_streaming_static():
    """§12.5 정적 증명(AST): is_duplicate_trigger 가 read_text/splitlines 호출 부재 +

    파일객체 스트리밍(open + iter) 사용 + except 핸들러는 OSError 만(9-R.3 미확대).
    AST 기반 — 설명 주석/docstring 의 'read_text' 문자열 언급에 오탐되지 않는다.
    """
    import ast
    import textwrap

    otp = _load_otp_streaming()
    src = textwrap.dedent(inspect.getsource(otp.is_duplicate_trigger))
    tree = ast.parse(src)
    fn = tree.body[0]
    assert isinstance(fn, ast.FunctionDef)

    called_attrs = {
        n.func.attr
        for n in ast.walk(fn)
        if isinstance(n, ast.Call) and isinstance(n.func, ast.Attribute)
    }
    # 전량 적재 호출 부재 (실제 call 만 — 주석 언급 무관).
    assert "read_text" not in called_attrs, "is_duplicate_trigger 가 여전히 read_text 호출"
    assert "splitlines" not in called_attrs, "splitlines() 전량 분할 호출 잔존"
    # 스트리밍: path.open(...) 호출 존재.
    assert "open" in called_attrs, "스트리밍 path.open() 호출 부재"
    # 파일객체 line-by-line 순회: `for line in f` (Name 'f' iter) 존재.
    has_file_iter = any(
        isinstance(n, ast.For)
        and isinstance(n.iter, ast.Name)
        and isinstance(n.target, ast.Name)
        for n in ast.walk(fn)
    )
    assert has_file_iter, "파일객체 for-line 스트리밍 순회 부재"
    # 9-R.3: except 핸들러 정확히 OSError 만 (Exception/ValueError/UnicodeDecodeError 확대 금지).
    handler_types = set()
    for n in ast.walk(fn):
        if isinstance(n, ast.ExceptHandler) and isinstance(n.type, ast.Name):
            handler_types.add(n.type.id)
    assert "OSError" in handler_types
    assert handler_types.isdisjoint(
        {"Exception", "ValueError", "UnicodeDecodeError", "BaseException"}
    ), f"except 핸들러 부당 확대: {handler_types}"