#!/usr/bin/env python3
"""task-2962: notify_claim 공용 모듈 + 성공 알림 단일 소유자 게이트 회귀 테스트.

배경: 봇 완료 성공 알림을 보내는 소비자가 done-watcher.sh / activity-watcher.py /
done-watcher.py / notify-completion.py 4개 공존 → 이중·삼중 알림 위험.
해소책:
  1. 성공 알림 단일 소유자 = done-watcher-success.service (done-watcher.py --daemon).
     done-watcher.sh / activity-watcher.py 의 성공 알림 경로는 환경변수 opt-in
     (기본 비활성)으로 차단.
  2. 전송 전 `{task_id}.done.notified` 마커를 O_EXCL 로 원자적 선점한 프로세스만 발송.
     공용 모듈: scripts/lib/notify_claim.py
     (claim_notification / release_notification / finalize_notification)

대상:
  - scripts/lib/notify_claim.py
  - scripts/activity-watcher.py 의 ACTIVITY_WATCHER_SUCCESS_NOTIFY 게이트
  - scripts/done-watcher.sh 의 DONE_WATCHER_SH_SUCCESS_NOTIFY 게이트 + stale 에스컬레이션 존치

★ 절대 규칙:
  - 실제 Telegram API 호출 금지, 네트워크 호출 금지.
  - 실제 워크스페이스 memory/events/ 절대 미접촉 — 모든 파일 조작은 pytest tmp_path 안에서만.
  - 제품 코드(scripts/*.py, scripts/*.sh)는 읽기 전용. 이 테스트 파일만 신규 생성.
"""

from __future__ import annotations

import ast
import concurrent.futures
import importlib.util
import json
import multiprocessing
import re
import subprocess
import time
from pathlib import Path

import pytest

_SCRIPTS_DIR = Path(__file__).parent.parent
_NOTIFY_CLAIM_PATH = _SCRIPTS_DIR / "lib" / "notify_claim.py"
_ACTIVITY_WATCHER_PATH = _SCRIPTS_DIR / "activity-watcher.py"
_DONE_WATCHER_SH_PATH = _SCRIPTS_DIR / "done-watcher.sh"


def _load_notify_claim():
    """notify_claim.py 를 매 테스트마다 독립된 모듈 인스턴스로 새로 로드한다.

    - importlib.util.spec_from_file_location 사용, sys.modules 에는 등록하지 않는다
      (다른 테스트 파일이 `import notify_claim` 으로 캐시해 둔 전역 상태와 공유 방지).
    """
    spec = importlib.util.spec_from_file_location(
        f"notify_claim_test_{id(object())}", _NOTIFY_CLAIM_PATH
    )
    assert spec is not None
    module = importlib.util.module_from_spec(spec)
    assert spec.loader is not None
    spec.loader.exec_module(module)
    return module


def _mp_claim_worker(marker_path_str: str, notify_claim_path_str: str, start_at: float) -> bool:
    """멀티프로세스 워커 (모듈 최상위 정의 — spawn 시작 방식으로 pickle 되려면 필수).

    각 워커는 자신의 독립 프로세스 안에서 notify_claim.py 를 파일 경로로부터
    새로 로드한다(부모 프로세스와 모듈 객체를 공유하지 않는다 — spawn 은 fork 와
    달리 인터프리터 상태를 상속하지 않으므로 이래야 진짜 프로세스 격리 조건에서
    테스트하는 것이 된다). start_at(공통 목표 시각)까지 busy-wait 한 뒤 동시에
    claim_notification() 을 호출해 커널 syscall 레벨 경쟁을 유도한다.
    """
    spec = importlib.util.spec_from_file_location(
        f"notify_claim_mp_worker_{id(object())}", notify_claim_path_str
    )
    assert spec is not None
    module = importlib.util.module_from_spec(spec)
    assert spec.loader is not None
    spec.loader.exec_module(module)

    while time.time() < start_at:
        pass

    return module.claim_notification(marker_path_str)


@pytest.fixture
def nc(tmp_path):
    """격리된 notify_claim 모듈 인스턴스. tmp_path 는 사용하지 않지만 각 테스트가
    자신만의 tmp_path 마커 파일을 만들도록 관례상 함께 요구한다."""
    return _load_notify_claim()


# ---------------------------------------------------------------------------
# 1~3: claim_notification 기본 동작
# ---------------------------------------------------------------------------


class TestClaimNotificationBasics:
    def test_first_claim_succeeds_and_creates_marker(self, nc, tmp_path):
        """1. 최초 호출 → True, 마커 파일 실제 생성됨."""
        marker = tmp_path / "task-1.done.notified"
        assert not marker.exists()

        result = nc.claim_notification(marker)

        assert result is True
        assert marker.exists()
        assert marker.is_file()

    def test_second_claim_on_same_path_fails(self, nc, tmp_path):
        """2. 같은 경로 2회차 호출 → False (선점 실패)."""
        marker = tmp_path / "task-2.done.notified"

        first = nc.claim_notification(marker)
        second = nc.claim_notification(marker)

        assert first is True
        assert second is False

    def test_missing_parent_dir_auto_created_and_claim_succeeds(self, nc, tmp_path):
        """3. 부모 디렉토리가 없어도 자동 생성되고 선점 성공."""
        marker = tmp_path / "nested" / "deeper" / "task-3.done.notified"
        assert not marker.parent.exists()

        result = nc.claim_notification(marker)

        assert result is True
        assert marker.exists()
        assert marker.parent.is_dir()


# ---------------------------------------------------------------------------
# 4~5: release_notification
# ---------------------------------------------------------------------------


class TestReleaseNotification:
    def test_release_deletes_marker_and_allows_reclaim(self, nc, tmp_path):
        """4. release_notification() → 마커 삭제 후 재선점 가능 (True)."""
        marker = tmp_path / "task-4.done.notified"
        assert nc.claim_notification(marker) is True

        nc.release_notification(marker)

        assert not marker.exists()
        assert nc.claim_notification(marker) is True

    def test_release_on_nonexistent_marker_no_exception(self, nc, tmp_path):
        """5. release_notification() 을 마커 없는 경로에 호출해도 예외 없음."""
        marker = tmp_path / "never-claimed.done.notified"
        assert not marker.exists()

        # 예외가 전파되면 이 라인에서 pytest 가 자동으로 실패시킨다.
        nc.release_notification(marker)

        assert not marker.exists()


# ---------------------------------------------------------------------------
# 6: finalize_notification
# ---------------------------------------------------------------------------


class TestFinalizeNotification:
    def test_finalize_writes_payload_json(self, nc, tmp_path):
        """6. finalize_notification(marker, payload) → 마커 내용이 payload JSON과 일치."""
        marker = tmp_path / "task-6.done.notified"
        assert nc.claim_notification(marker) is True

        payload = {"task_id": "task-6", "status": "success", "pr": 219, "sent_by": "done-watcher.py"}
        nc.finalize_notification(marker, payload)

        written = json.loads(marker.read_text(encoding="utf-8"))
        assert written == payload


# ---------------------------------------------------------------------------
# 7: fail-closed
# ---------------------------------------------------------------------------


class TestFailClosed:
    """os.open 이 FileExistsError 이외의 예외를 던지면 fail-closed(False, 예외 전파 금지)."""

    def test_permission_error_returns_false_not_raises(self, nc, tmp_path, monkeypatch):
        marker = tmp_path / "task-7a.done.notified"

        def _boom(*_args, **_kwargs):
            raise PermissionError("simulated permission denied")

        monkeypatch.setattr(nc.os, "open", _boom)

        result = nc.claim_notification(marker)

        assert result is False
        assert not marker.exists()

    def test_generic_oserror_returns_false_not_raises(self, nc, tmp_path, monkeypatch):
        marker = tmp_path / "task-7b.done.notified"

        def _boom(*_args, **_kwargs):
            raise OSError(5, "simulated I/O error")

        monkeypatch.setattr(nc.os, "open", _boom)

        result = nc.claim_notification(marker)

        assert result is False
        assert not marker.exists()


# ---------------------------------------------------------------------------
# 8: 동시성
# ---------------------------------------------------------------------------


class TestConcurrency:
    def test_exactly_one_of_twenty_threads_wins(self, nc, tmp_path):
        """8. 동일 마커에 대해 20개 스레드가 동시에 claim_notification 호출
        → 정확히 1개만 True (O_EXCL 의 OS 레벨 원자성이 GIL 유무와 무관하게 보장)."""
        marker = tmp_path / "task-8.done.notified"

        def _attempt(_i):
            return nc.claim_notification(marker)

        with concurrent.futures.ThreadPoolExecutor(max_workers=20) as pool:
            results = list(pool.map(_attempt, range(20)))

        assert len(results) == 20
        assert results.count(True) == 1
        assert results.count(False) == 19
        assert marker.exists()


# ---------------------------------------------------------------------------
# 9: activity-watcher.py 단일 소유자 게이트 (AST 구조 검증)
# ---------------------------------------------------------------------------


class TestActivityWatcherSingleOwnerGate:
    """9. activity-watcher.py: ACTIVITY_WATCHER_SUCCESS_NOTIFY 게이트가 존재하고,
    그 게이트 안쪽에서만 send_telegram_notification 이 "호출"되는지 AST 로 구조적으로 검증한다.
    (게이트 없이 무조건 전송하는 경로가 남아 있으면 FAIL)"""

    def test_gate_env_var_present_in_source(self):
        source = _ACTIVITY_WATCHER_PATH.read_text(encoding="utf-8")
        assert "ACTIVITY_WATCHER_SUCCESS_NOTIFY" in source

    def test_send_telegram_notification_calls_confined_to_gate(self):
        source = _ACTIVITY_WATCHER_PATH.read_text(encoding="utf-8")
        tree = ast.parse(source, filename=str(_ACTIVITY_WATCHER_PATH))

        # (a) send_telegram_notification 을 "호출"하는 모든 지점 수집 (def 자체는 제외)
        call_lines: list[int] = []
        for node in ast.walk(tree):
            if isinstance(node, ast.Call):
                func = node.func
                name = func.id if isinstance(func, ast.Name) else getattr(func, "attr", None)
                if name == "send_telegram_notification":
                    call_lines.append(node.lineno)

        assert call_lines, "send_telegram_notification 호출부를 하나도 찾지 못함 — 테스트 전제 붕괴"

        # (b) ACTIVITY_WATCHER_SUCCESS_NOTIFY 를 조건식에 포함하는 If 노드(들)의 라인 범위 수집
        gate_ranges: list[tuple[int, int]] = []
        for node in ast.walk(tree):
            if isinstance(node, ast.If) and "ACTIVITY_WATCHER_SUCCESS_NOTIFY" in ast.dump(node.test):
                body_start = min(n.lineno for n in node.body)
                end_lines = [
                    getattr(n, "end_lineno", None) or getattr(n, "lineno", None)
                    for n in ast.walk(node)
                ]
                body_end = max(ln for ln in end_lines if ln is not None)
                gate_ranges.append((body_start, body_end))

        assert gate_ranges, "ACTIVITY_WATCHER_SUCCESS_NOTIFY 를 조건으로 하는 If 게이트를 찾지 못함"

        for line in call_lines:
            in_gate = any(start <= line <= end for start, end in gate_ranges)
            assert in_gate, (
                f"send_telegram_notification 호출(line {line}) 이 "
                f"ACTIVITY_WATCHER_SUCCESS_NOTIFY 게이트 밖에 존재함 — 단일 소유자 원칙 위반"
            )


# ---------------------------------------------------------------------------
# 10: done-watcher.sh 단일 소유자 게이트 + 문법 검증
# ---------------------------------------------------------------------------


class TestDoneWatcherShGate:
    """10. done-watcher.sh: DONE_WATCHER_SH_SUCCESS_NOTIFY 게이트 존재 확인 +
    bash -n 문법 통과 확인."""

    def test_gate_env_var_present(self):
        source = _DONE_WATCHER_SH_PATH.read_text(encoding="utf-8")
        assert "DONE_WATCHER_SH_SUCCESS_NOTIFY" in source

    def test_bash_syntax_check_passes(self):
        result = subprocess.run(
            ["bash", "-n", str(_DONE_WATCHER_SH_PATH)],
            capture_output=True,
            text=True,
            timeout=10,
        )
        assert result.returncode == 0, (
            f"bash -n 문법 검사 실패 (returncode={result.returncode})\n"
            f"stdout={result.stdout}\nstderr={result.stderr}"
        )


# ---------------------------------------------------------------------------
# 11: 에스컬레이션 회귀 방지 — 게이트 밖에 존치되어 있는지 구조적 검증
# ---------------------------------------------------------------------------


def _find_top_level_if_span(lines: list[str], marker: str) -> tuple[int, int]:
    """marker 문자열을 포함한 라인에서 시작하는 최상위 bash if/fi 블록의
    (시작 인덱스, 종료 `fi` 인덱스) 를 0-based 로 반환한다.

    done-watcher.sh 는 내부에 `python3 -c "..."` 형태의 멀티라인 파이썬 문자열을
    임베드하는데, 그 안에 파이썬 `if`/`for` 키워드(bash 의 fi/done 짝이 없음)가
    섞여 있어 순진한 키워드 스택 카운터를 오염시킨다. 이 함수는 그 문자열이
    열리는 시점(라인이 `-c "` 로 끝남)부터 닫히는 시점(라인이 `"` 로 시작함)까지는
    깊이 계산에서 제외한다.

    또한 이 스크립트는 `*.done`, `.done.notified`, `.done.acked` 같은 파일명/확장자와
    `[done-watcher]` 로그 태그를 곳곳에서 리터럴로 사용하는데, `\bdone\b` 만으로는
    `.done`(마침표 뒤) 이나 `done-watcher`(하이픈 뒤) 처럼 실제로는 파일명/태그의
    일부인 "done" 도 진짜 bash `done` 키워드로 오탐한다. 바로 앞/뒤가 `.` 또는 `-`
    인 경우는 제외하여 이 오탐을 걸러낸다.
    """
    open_re = re.compile(r"\b(if|for|while|until)\b")
    close_re = re.compile(r"(?<![.\-])\b(fi|done)\b(?![.\-])")

    depth = 0
    start_idx: int | None = None
    in_py_string = False

    for i, raw in enumerate(lines):
        if in_py_string:
            if raw.strip().startswith('"'):
                in_py_string = False
            continue

        if start_idx is None and marker in raw:
            start_idx = i
            depth = 0

        if start_idx is not None:
            depth += len(open_re.findall(raw)) - len(close_re.findall(raw))

        if raw.rstrip().endswith('-c "'):
            in_py_string = True

        if start_idx is not None and i > start_idx and depth <= 0:
            return start_idx, i

    raise AssertionError(f"marker {marker!r} 에서 시작하는 최상위 if/fi 블록의 종료를 찾지 못함")


class TestEscalationOutsideGate:
    """11. stale .done 에스컬레이션 블록(1800초 기준, escalation_marker.py 호출) 이
    DONE_WATCHER_SH_SUCCESS_NOTIFY 게이트 **밖**에 그대로 남아 있는지 검증한다.
    단일 소유자 조치(성공 알림 opt-in화)가 실수로 에스컬레이션 경로까지
    게이트 안에 가둬버렸다면 이 테스트가 FAIL 해야 한다."""

    def test_escalation_block_present_and_outside_gate(self):
        source = _DONE_WATCHER_SH_PATH.read_text(encoding="utf-8")
        assert "escalation_marker.py" in source, "escalation_marker.py 호출이 사라짐 (에스컬레이션 경로 소실)"
        assert "1800" in source, "1800초(30분) stale 기준이 사라짐"

        lines = source.splitlines()
        gate_start, gate_end = _find_top_level_if_span(lines, "DONE_WATCHER_SH_SUCCESS_NOTIFY:-0")

        escalation_line_idx = next(
            (i for i, line in enumerate(lines) if "escalation_marker.py" in line), None
        )
        assert escalation_line_idx is not None

        assert escalation_line_idx > gate_end, (
            f"escalation_marker.py 호출부(line {escalation_line_idx + 1}) 가 "
            f"DONE_WATCHER_SH_SUCCESS_NOTIFY 게이트 블록"
            f"(line {gate_start + 1}~{gate_end + 1}) 내부에 있음 — "
            "단일 소유자 조치가 에스컬레이션 경로까지 꺼버렸을 가능성"
        )

        # 1800초 age 비교 자체도 게이트 밖(에스컬레이션 루프 안)에서 쓰이는지 함께 확인
        age_check_idx = next((i for i, line in enumerate(lines) if "1800" in line), None)
        assert age_check_idx is not None
        assert age_check_idx > gate_end, (
            f"1800초 기준 비교(line {age_check_idx + 1}) 가 게이트 블록 "
            f"(line {gate_start + 1}~{gate_end + 1}) 내부에 있음"
        )


# ---------------------------------------------------------------------------
# 12: 멀티프로세스 경쟁 (핵심 — 검증 공백 메우기)
# ---------------------------------------------------------------------------


class TestMultiProcessConcurrency:
    """12. 라이브 스모크에서 "게이트를 강제로 연 done-watcher.sh 를 데몬과 경쟁시키는"
    시나리오가 flock(/tmp/done-watcher.lock, flock -n) 에 막혀 재현되지 않았다 →
    O_EXCL 계층이 **프로세스 간** 경쟁에서도 유효한지는 아직 미실증이었다.

    TestConcurrency(위, 8번)는 스레드 기반이라 이 공백을 메우지 못한다 — 파이썬
    GIL 은 바이트코드 단위로 인터프리터 자체를 직렬화하므로, 스레드로는 여러
    os.open(O_EXCL) 호출이 커널에 "진짜 동시에" 도달하는지 구별할 수 없다(GIL이
    이미 순서를 강제하고 있어서 그 순서가 O_EXCL 덕분인지 GIL 덕분인지 이 테스트
    만으로는 알 수 없음).

    이 테스트는 20개의 **별도 OS 프로세스**(multiprocessing, spawn 시작 방식 —
    fork 와 달리 이식성이 있고 인터프리터 상태를 상속하지 않아 프로세스 격리가
    더 엄격함)가 공통 시작 시각까지 busy-wait 한 뒤 동일 마커 경로에 대해
    claim_notification() 을 동시에 호출하게 만든다. GIL 이 전혀 개입할 수 없는
    조건에서도 정확히 1개만 성공해야 커널 레벨 O_EXCL 원자성이 실증된 것이다.
    """

    def test_exactly_one_of_twenty_processes_wins(self, tmp_path):
        marker = tmp_path / "task-12.done.notified"
        assert not marker.exists()

        # 모든 워커가 이 시각까지 busy-wait 한 뒤 한꺼번에 claim 을 시도하도록
        # 공통 시작 시각을 맞춘다 (프로세스 생성 지연으로 인한 순차 도착 방지).
        start_at = time.time() + 0.5

        ctx = multiprocessing.get_context("spawn")
        with ctx.Pool(processes=20) as pool:
            results = pool.starmap(
                _mp_claim_worker,
                [(str(marker), str(_NOTIFY_CLAIM_PATH), start_at) for _ in range(20)],
            )

        assert len(results) == 20
        assert results.count(True) == 1, (
            f"True 반환 프로세스가 정확히 1개여야 하는데 {results.count(True)}개 — "
            "프로세스 간 O_EXCL 원자성 위반(이중 알림 발송 가능성)"
        )
        assert results.count(False) == 19
        assert marker.exists()
        assert marker.is_file()


# ---------------------------------------------------------------------------
# 13: done-watcher.sh flock 계약 (관측 사실의 회귀 고정)
# ---------------------------------------------------------------------------


class TestDoneWatcherShFlock:
    """13. done-watcher.sh 는 /tmp/done-watcher.lock 에 대한 flock(-n) 단일 인스턴스
    가드를 갖고 있으며, 락 획득 실패 시 즉시 종료(already running)한다. 이 flock
    가드 때문에 done-watcher.sh 는 구조적으로 동시 2인스턴스가 불가하며, 이는
    O_EXCL 위에 놓인 **추가 방어층**이다. 이 가드가 제거되면 (앞서 라이브 스모크가
    막고 있던) 실제 경쟁 시나리오가 다시 열리므로, 이 테스트는 그 전제를 회귀로
    감지하기 위해 존재한다. (테스트 12 의 멀티프로세스 O_EXCL 실증과는 별개로,
    done-watcher.sh 자체는 이 가드 때문에 애초에 경쟁 상황에 진입하지 않는다는
    사실을 고정한다.)"""

    def test_flock_single_instance_guard_present(self):
        """LOCK_FILE 변수, `flock -n` non-blocking 락 시도, 실패 시 조기 종료
        (already running 류 메시지 + exit) 분기가 모두 존재하는지 한 번에 검증한다."""
        source = _DONE_WATCHER_SH_PATH.read_text(encoding="utf-8")

        assert "LOCK_FILE=" in source, "LOCK_FILE 변수 선언이 사라짐 — 단일 인스턴스 가드 소실"
        assert "/tmp/done-watcher.lock" in source

        lines = source.splitlines()
        flock_line_idx = next((i for i, line in enumerate(lines) if re.search(r"flock\s+-n\b", line)), None)
        assert flock_line_idx is not None, (
            "`flock -n` (non-blocking 락 시도) 패턴을 찾지 못함 — "
            "동시 2인스턴스 방지 가드가 제거되었을 가능성"
        )

        flock_line = lines[flock_line_idx]
        assert "||" in flock_line, (
            f"flock 호출부(line {flock_line_idx + 1})에 `||` 실패 처리 분기가 없음: {flock_line!r}"
        )
        assert re.search(r"already running", flock_line, re.IGNORECASE), (
            f"flock 실패 시 'already running' 류 메시지가 없음: {flock_line!r}"
        )
        assert "exit" in flock_line, f"flock 실패 시 exit 로 조기 종료하지 않음: {flock_line!r}"


if __name__ == "__main__":
    pytest.main([__file__, "-v"])