# -*- coding: utf-8 -*-
"""tests/regression/test_progress_watcher_gate_2729.py — task-2729 Phase 1 회귀.

Phase 1 완료조건(회장 7) 회귀 보호:
  1. gate PASS (watcher_registered=True)
  2. fallback-only → DISPATCH_INCOMPLETE
  3. no watcher (no fallback) → DISPATCH_INCOMPLETE
  4. watcher terminal callback wired (성공 runner)
  5. WATCHER_TERMINAL_CALLBACK_NOT_WIRED (실패 runner)
  6. 6-state tracking
  7. review-settle quiet-window 골격
  8. 기존 callback/fallback behavior 무손상 smoke
  9. classify 재사용 run_once (head drift fixture)

대상 모듈(progress_watcher_gate.py, ci_watch_handoff_runner.py)은 수정하지 않고
모듈의 실제 동작에 맞춰 테스트만 작성한다. fake runner 주입 / 순수 함수 검증만 사용
(외부 네트워크 / 실제 cokacdir / gh 호출 없음).
"""
from __future__ import annotations

import importlib.util
import pathlib
import sys
import types

# ---------------------------------------------------------------------------
# worktree root sys.path 보강.
#   ★ tests/conftest.py 가 비-worktree 풀체크아웃(/home/jay/workspace)을 sys.path[0]
#     에 주입한다. 그 dispatch/utils 패키지에는 task-2729 신규 모듈
#     (dispatch.progress_watcher_gate)이 없어 ModuleNotFoundError 가 발생한다.
#     → 이 sparse worktree root 를 sys.path 최상단으로 끌어올리고, 이미 캐시된
#       stale dispatch*/utils* 모듈을 제거하여 worktree 사본이 로드되도록 강제한다.
#     (대상 모듈은 수정하지 않고 import 해석만 worktree 로 고정.)
# ---------------------------------------------------------------------------
_ROOT = pathlib.Path(__file__).resolve().parents[2]
_ROOT_STR = str(_ROOT)
# 항상 worktree root 를 최상단에 둔다(중복 제거 후 0번 삽입).
while _ROOT_STR in sys.path:
    sys.path.remove(_ROOT_STR)
sys.path.insert(0, _ROOT_STR)

# worktree 외부에서 먼저 로드된 dispatch/utils 패키지 캐시를 제거(신규 모듈 누락분 회피).
for _name in list(sys.modules):
    if _name == "dispatch" or _name.startswith("dispatch."):
        _mod = sys.modules[_name]
        _f = getattr(_mod, "__file__", None) or ""
        if not _f.startswith(_ROOT_STR):
            del sys.modules[_name]

# ---------------------------------------------------------------------------
# 대상 모듈 import
# ---------------------------------------------------------------------------
from dispatch.progress_watcher_gate import (  # noqa: E402
    DISPATCH_INCOMPLETE,
    WATCHER_TERMINAL_CALLBACK_NOT_WIRED,
    PROGRESS_WATCHER_REGISTERED_KEY,
    DISPATCH_OK,
    WATCHER_TRACKED_STATES,
    GateResult,
    evaluate_progress_watcher_gate,
    annotate_dispatch_result,
    watcher_terminal_callback_status,
    all_states_tracked,
)
from utils.pr_watcher_terminal_state_classifier import (  # noqa: E402
    PRSnapshot,
    TERMINAL_STATES,
)

# ---------------------------------------------------------------------------
# scripts/ci_watch_handoff_runner.py 는 일반 패키지가 아닐 수 있으므로
# importlib.util.spec_from_file_location 로 로드한다.
# (dataclass 평가가 sys.modules 등록에 의존하므로 exec_module 전에 등록)
# ---------------------------------------------------------------------------
_RUNNER_PATH = _ROOT / "scripts" / "ci_watch_handoff_runner.py"
_spec = importlib.util.spec_from_file_location("ci_watch_handoff_runner", _RUNNER_PATH)
assert _spec is not None and _spec.loader is not None
runner_mod = importlib.util.module_from_spec(_spec)
sys.modules["ci_watch_handoff_runner"] = runner_mod
_spec.loader.exec_module(runner_mod)


# ---------------------------------------------------------------------------
# fake runner (subprocess.run 호환 시그니처)
# register_terminal_callback 은 runner(cmd, capture_output=True, text=True,
# timeout=..., check=False) 형태로 호출하고 returncode==0 일 때 fired=True.
# ---------------------------------------------------------------------------
def _make_fake_runner(returncode: int):
    def _runner(cmd, capture_output=True, text=True, timeout=None, check=False, **kwargs):
        return types.SimpleNamespace(
            returncode=returncode,
            stdout="ok" if returncode == 0 else "",
            stderr="" if returncode == 0 else "boom",
        )

    return _runner


# ===========================================================================
# 1. gate PASS
# ===========================================================================
def test_gate_pass_watcher_registered():
    gr = evaluate_progress_watcher_gate(watcher_registered=True)
    assert isinstance(gr, GateResult)
    assert gr.incomplete is False
    assert gr.progress_watcher_registered is True
    assert gr.dispatch_status == DISPATCH_OK == "dispatched"
    assert gr.fallback_only is False
    assert gr.incomplete_reason == ""


def test_gate_pass_annotate_no_status_change():
    result = {"status": "dispatched"}
    out = annotate_dispatch_result(result, watcher_registered=True)
    # 같은 dict 반환
    assert out is result
    assert out[PROGRESS_WATCHER_REGISTERED_KEY] is True
    assert out["status"] == "dispatched"  # 무변경
    assert "progress_watcher_gate" not in out
    assert "progress_watcher_incomplete_reason" not in out


# ===========================================================================
# 2. fallback-only → DISPATCH_INCOMPLETE
# ===========================================================================
def test_fallback_only_incomplete():
    gr = evaluate_progress_watcher_gate(
        watcher_registered=False, fallback_registered=True
    )
    assert gr.incomplete is True
    assert gr.fallback_only is True
    assert gr.dispatch_status == DISPATCH_INCOMPLETE
    assert gr.incomplete_reason == "fallback_only_no_progress_watcher"
    assert gr.progress_watcher_registered is False


def test_fallback_only_annotate_inactive_keeps_status():
    result = {"status": "dispatched"}
    out = annotate_dispatch_result(
        result, watcher_registered=False, fallback_registered=True, active=False
    )
    assert out[PROGRESS_WATCHER_REGISTERED_KEY] is False
    assert out["progress_watcher_gate"] == DISPATCH_INCOMPLETE
    assert out["status"] == "dispatched"  # active=False → 무변경
    assert out["progress_watcher_incomplete_reason"] == "fallback_only_no_progress_watcher"


def test_fallback_only_annotate_active_changes_status():
    result = {"status": "dispatched"}
    out = annotate_dispatch_result(
        result, watcher_registered=False, fallback_registered=True, active=True
    )
    assert out["status"] == DISPATCH_INCOMPLETE
    assert out["progress_watcher_gate"] == DISPATCH_INCOMPLETE


# ===========================================================================
# 3. no watcher (no fallback)
# ===========================================================================
def test_no_watcher_no_fallback_incomplete():
    gr = evaluate_progress_watcher_gate(
        watcher_registered=False, fallback_registered=False
    )
    assert gr.incomplete is True
    assert gr.fallback_only is False
    assert gr.dispatch_status == DISPATCH_INCOMPLETE
    assert gr.incomplete_reason == "no_progress_watcher"
    assert gr.progress_watcher_registered is False


# ===========================================================================
# 4. watcher terminal callback wired (성공 runner)
# ===========================================================================
def test_watcher_terminal_callback_status_wired():
    assert (
        watcher_terminal_callback_status(terminal_reached=True, callback_fired=True)
        == "WATCHER_TERMINAL_CALLBACK_WIRED"
    )


def test_fire_terminal_callback_success_runner():
    state = runner_mod.WatcherState()
    assert state.normal_callback is False
    out = runner_mod.fire_terminal_callback(
        task_id="2729",
        pr_number=42,
        terminal_state=TERMINAL_STATES[0],  # MERGE_READY (valid terminal)
        reason="test_wired",
        state=state,
        runner=_make_fake_runner(0),
    )
    assert out["fired"] is True
    assert out["callback_status"] == "WATCHER_TERMINAL_CALLBACK_WIRED"
    assert out["terminal_state"] == TERMINAL_STATES[0]
    assert state.normal_callback is True


# ===========================================================================
# 5. WATCHER_TERMINAL_CALLBACK_NOT_WIRED (실패 runner)
# ===========================================================================
def test_watcher_terminal_callback_status_not_wired():
    assert (
        watcher_terminal_callback_status(terminal_reached=True, callback_fired=False)
        == WATCHER_TERMINAL_CALLBACK_NOT_WIRED
    )


def test_fire_terminal_callback_failure_runner():
    # register_terminal_callback: runner returncode != 0 → fired=False
    state = runner_mod.WatcherState()
    out = runner_mod.fire_terminal_callback(
        task_id="2729",
        pr_number=42,
        terminal_state="HOLD_FOR_CHAIR",  # valid terminal
        reason="test_not_wired",
        state=state,
        runner=_make_fake_runner(1),
    )
    assert out["fired"] is False
    assert out["callback_status"] == WATCHER_TERMINAL_CALLBACK_NOT_WIRED
    # 실패 시 state.normal_callback 은 set 되지 않음
    assert state.normal_callback is False


# ===========================================================================
# 6. 6-state tracking
# ===========================================================================
def test_six_state_tracking():
    state = runner_mod.WatcherState()
    assert state.all_tracked() is False

    runner_mod.mark_head_change(state)
    runner_mod.mark_non_force_push(state)
    runner_mod.mark_ci(state)
    runner_mod.mark_finish_done(state)
    runner_mod.mark_normal_callback(state)
    runner_mod.mark_fallback_prune(state)

    assert state.all_tracked() is True
    assert all_states_tracked(state.as_tracked_dict()) is True


def test_as_tracked_dict_keys_match_constant():
    state = runner_mod.WatcherState()
    assert tuple(state.as_tracked_dict().keys()) == WATCHER_TRACKED_STATES


# ===========================================================================
# 7. review-settle quiet-window 골격
# ===========================================================================
def test_quiet_window_not_settled():
    assert (
        runner_mod.is_quiet_window_settled(
            last_event_sec=0, now_sec=119, quiet_window_sec=120
        )
        is False
    )


def test_quiet_window_settled():
    assert (
        runner_mod.is_quiet_window_settled(
            last_event_sec=0, now_sec=120, quiet_window_sec=120
        )
        is True
    )


def test_quiet_window_default_constant():
    assert runner_mod.REVIEW_SETTLE_QUIET_WINDOW_SEC == 120


# ===========================================================================
# 8. 기존 callback/fallback behavior 무손상 smoke
# ===========================================================================
def test_normal_fallback_callback_helper_import_smoke():
    from dispatch.normal_fallback_callback_helper import (
        build_anu_owned_callback_request,
        launch_callback,
        CALLBACK_KIND_NORMAL,
        CALLBACK_KIND_FALLBACK,
    )

    assert CALLBACK_KIND_NORMAL == "normal"
    assert CALLBACK_KIND_FALLBACK == "fallback"
    assert callable(build_anu_owned_callback_request)
    assert callable(launch_callback)


def test_spawn_callback_contract_validator_import_smoke():
    from dispatch.spawn_callback_contract_validator import (
        validate_spawn_callback_contract,
        NO_OP_SPAWN_CONTRACT_FAILED,
    )

    assert NO_OP_SPAWN_CONTRACT_FAILED == "NO_OP_SPAWN_CONTRACT_FAILED"
    assert callable(validate_spawn_callback_contract)


# ===========================================================================
# 9. classify 재사용 run_once (head drift fixture)
# ===========================================================================
def test_run_once_head_drift_returns_terminal():
    # classify head-drift 조건: snap.head_ref_oid 가 truthy 이고 expected_head 와 다르면
    # HOLD_FOR_CHAIR (TERMINAL_STATES 포함) 반환.
    snap = PRSnapshot(head_ref_oid="deadbeefdrifted", merge_state_status="BLOCKED")
    state = runner_mod.WatcherState()
    terminal_state, reason = runner_mod.run_once(
        snap,
        expected_head="expectedhead000",
        elapsed_watcher_sec=10,
        state=state,
    )
    assert terminal_state in TERMINAL_STATES
    assert terminal_state == "HOLD_FOR_CHAIR"
    assert "head_drift" in reason
    assert state.polls == 1
    assert state.elapsed_sec == 10


def test_run_once_non_terminal_returns_empty():
    # head 일치 + 미수렴 (continue) → 비terminal → "" 반환.
    snap = PRSnapshot(
        head_ref_oid="samehead000",
        merge_state_status="BLOCKED",
        unresolved_thread_count=0,
    )
    state = runner_mod.WatcherState()
    terminal_state, reason = runner_mod.run_once(
        snap,
        expected_head="samehead000",
        elapsed_watcher_sec=5,
        state=state,
    )
    assert terminal_state == ""
    assert reason == "continue"


# ===========================================================================
# 10. AUTO_REPAIR round-1 방어적 가드 회귀 (HIGH-2 / HIGH-1)
# ===========================================================================
def test_annotate_none_returns_original_no_exception():
    """HIGH-2: result=None 이면 TypeError 없이 None 을 그대로 반환 (record-only 무손상)."""
    out = annotate_dispatch_result(None, watcher_registered=False)
    assert out is None


def test_annotate_non_dict_returns_original_no_exception():
    """HIGH-2: result 가 dict 가 아니면(list/str) 원본 객체를 변형 없이 그대로 반환."""
    for bad in (["not", "a", "dict"], "string-result", 42):
        out = annotate_dispatch_result(bad, watcher_registered=True, active=True)
        assert out is bad  # 동일 객체, 키 추가/변형 없음


def test_gate_call_site_exception_does_not_interrupt(monkeypatch):
    """HIGH-1: record-only 게이트 annotate 가 예외를 던져도 dispatch 호출부는 무중단,
    원래 _result 가 보존되어야 한다 (dispatch/__init__.py 호출부 try-except 계약 재현)."""
    import logging

    sentinel = {"status": "dispatched", "task_id": "task-2729"}
    _result = sentinel

    def _boom(*args, **kwargs):
        raise RuntimeError("injected gate failure")

    # dispatch/__init__.py 호출부와 동일한 방어 계약 재현
    logger = logging.getLogger("test_gate_call_site")
    try:
        annotate = _boom  # 예외 주입 (게이트 annotate 가 터지는 상황)
        _result = annotate(_result, watcher_registered=False, fallback_registered=False, active=False)
    except Exception as _pwg_err:
        logger.warning(f"[progress-watcher-gate] record-only annotate 실패 (무시): {_pwg_err}")

    # 예외가 흐름을 중단시키지 않고 원래 _result 가 그대로 보존됨
    assert _result is sentinel
    assert _result["status"] == "dispatched"


def test_dispatch_call_site_has_try_except_guard():
    """HIGH-1: dispatch/__init__.py annotate_dispatch_result 호출부 직전에 try: 가 존재."""
    import os
    base = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
    src_path = os.path.join(base, "dispatch", "__init__.py")
    s = open(src_path, encoding="utf-8").read()
    i = s.find("annotate_dispatch_result(")
    assert i != -1
    seg = s[max(0, i - 400):i]
    assert "try:" in seg