# -*- coding: utf-8 -*-
"""task-2628 §3 회귀테스트: anu_v3 dependency closure 독립 검증.

테스터: 벨레스 (dev6팀)
팀장:  페룬 (closure 분석 / 격리 패턴 검증 완료)
작성일: 2026-05-21

대상 12개 anu_v3 모듈의 dependency closure가 완전하며 (외부 의존 0),
유일한 cross-layer 의존인 dispatch.callback_owner_enforcer 도 stdlib만
import 함을 독립 검증한다. functional 테스트(§3-5 ~ §3-8)는 실제 소스를
읽어 확인한 API 시그니처로 작성한다.

§3 회귀 항목 매핑:
  test_01  §3-1  12모듈 독립 import 성공
  test_02  §3-2  conftest/WORKSPACE_ROOT 미경유 + 정적 의존성 검증
  test_03  §3-3  callback enforcement 파일의 anu_v3 의존 ⊆ TWELVE
  test_04  §3-4  missing dep → ModuleNotFoundError (false-pass 차단)
  test_05  §3-5  self_collector_guard 동작 검증
  test_06  §3-6  callback_owner_validator ANU vs self-key 구분
  test_07  §3-7  dispatch_callback_contract schema 검증
  test_08  §3-8  runtime_reconcile dry-run side-effect 0 + recovery-layer 계약
"""
from __future__ import annotations

import ast
import os
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path
from typing import Optional

import pytest

# ── 기반 상수 ─────────────────────────────────────────────────────────────────
# Gemini medium: 하드코딩 절대경로 제거. repo 컨벤션(tests/conftest.py 동형)인
# WORKSPACE_ROOT env override 사용 → live(기본 /home/jay/workspace) 동작 보존 +
# CI/worktree 이식성 확보.
SRC = os.environ.get("WORKSPACE_ROOT", "/home/jay/workspace")

# pytest conftest 경유와 무관하게 SRC 가 sys.path 에 있도록 보장
# (기존 테스트 패턴 test_self_collector_guard_runtime_2553plus49.py 동형)
# NOTE: tests/ 디렉토리도 sys.path 에 포함될 수 있어 tests/dispatch/
#       스텁이 실제 dispatch/ 를 가릴 수 있다. SRC 를 항상 [0] 에 위치시켜
#       workspace dispatch 패키지가 우선 해소되도록 한다.
if SRC in sys.path:
    sys.path.remove(SRC)
sys.path.insert(0, SRC)


def _ensure_src_dispatch() -> None:
    """functional 테스트 직전 호출: SRC 의 실제 dispatch 패키지가
    tests/dispatch 스텁에 의해 가려지지 않도록 sys.path 와 sys.modules
    를 정리한다.

    tests/ 가 sys.path 앞에 들어오면 tests/dispatch/__init__.py 가
    실제 dispatch/ 를 가려 'dispatch.callback_owner_enforcer' 를 못 찾게
    된다. SRC 를 맨 앞으로 재배치하고 캐시된 잘못된 dispatch 모듈을
    제거한다.
    """
    # sys.path 에서 SRC 를 맨 앞으로 재배치
    if SRC in sys.path:
        sys.path.remove(SRC)
    sys.path.insert(0, SRC)
    # tests/ 내 dispatch 스텁이 이미 캐시됐으면 제거 (다음 import 에서
    # SRC/dispatch 를 우선 해소)
    bad = (
        "dispatch" in sys.modules
        and sys.modules["dispatch"].__file__ != os.path.join(SRC, "dispatch", "__init__.py")
    )
    if bad:
        for k in list(sys.modules):
            if k == "dispatch" or k.startswith("dispatch."):
                del sys.modules[k]
TWELVE = [
    "active_dispatch_scanner",
    "authoritative_verdict_selector",
    "callback_4tuple_registry",
    "callback_owner_validator",
    "dispatch_callback_contract",
    "executor_callback_contract",
    "runtime_batch_state_updater",
    "runtime_next_action_resolver",
    "runtime_reconcile_checkpoint",
    "runtime_reconcile_checkpoint_recovery_layer",
    "self_collector_guard",
    "task_artifact_detector",
]
TWELVE_SET = set(TWELVE)

ANU_KEY = "c119085addb0f8b7"
SELF_KEY = "1e41a2324a3ccdd0"


# ── 격리 import 헬퍼 (페룬 검증 패턴 — 수정 없이 사용) ───────────────────────

def _build_closure_dir(tmp: str, omit=None, include_enforcer: bool = True) -> None:
    """12개 anu_v3 + (옵션)dispatch.callback_owner_enforcer 를 tmp에 격리 복사.

    dispatch/__init__.py 는 빈 stub (실제 192KB __init__ 의 무거운 의존성 회피).
    """
    omit = omit or set()
    os.makedirs(os.path.join(tmp, "anu_v3"))
    os.makedirs(os.path.join(tmp, "dispatch"))
    open(os.path.join(tmp, "anu_v3", "__init__.py"), "w").close()
    open(os.path.join(tmp, "dispatch", "__init__.py"), "w").close()
    if include_enforcer:
        shutil.copy(
            os.path.join(SRC, "dispatch", "callback_owner_enforcer.py"),
            os.path.join(tmp, "dispatch", "callback_owner_enforcer.py"),
        )
    for m in TWELVE:
        if m in omit:
            continue
        shutil.copy(
            os.path.join(SRC, "anu_v3", f"{m}.py"),
            os.path.join(tmp, "anu_v3", f"{m}.py"),
        )


def _isolated_import(tmp: str, modname: str) -> Optional[str]:
    """완전 격리 subprocess import. 성공 시 None, 실패 시 'ErrType: msg'.

    회장 지시(§2 #2): in-process sys.path 누수를 원천 차단하기 위해 별도
    파이썬 프로세스를 다음 조건으로 실행한다.
      - cwd = 빈 중립 디렉토리 → `python -c` 의 sys.path[0]("")=cwd 가
        anu_v3 를 해소하지 못하게 한다(worktree root/cwd 누수 차단).
      - PYTHONPATH = closure tmp (오직 이 경로로만 anu_v3/dispatch 해소).
        부모(pytest)의 PYTHONPATH 는 명시 override 로 차단한다.
    따라서 import 성공 여부는 'closure tmp 안에 모듈이 있는가' 에만 의존하며
    workspace/worktree/pytest rootdir/conftest 어느 경로에도 의존하지 않는다.
    conftest sys.path 주입은 자식 프로세스에 전파되지 않으므로 false-pass 불가.
    """
    neutral_cwd = tempfile.mkdtemp(prefix="iso_cwd_")
    env = dict(os.environ)
    env["PYTHONPATH"] = tmp  # 부모 PYTHONPATH override → 누수 차단, tmp 만 허용
    code = (
        "import importlib, sys\n"
        "try:\n"
        f"    importlib.import_module({modname!r})\n"
        "except Exception as e:\n"
        "    sys.stdout.write(type(e).__name__ + ': ' + str(e)[:120])\n"
        "    raise SystemExit(3)\n"
    )
    try:
        proc = subprocess.run(
            [sys.executable, "-c", code],
            cwd=neutral_cwd,
            env=env,
            capture_output=True,
            text=True,
            timeout=60,
        )
    except subprocess.TimeoutExpired:
        return "TimeoutExpired: isolated import exceeded 60s"
    finally:
        shutil.rmtree(neutral_cwd, ignore_errors=True)
    if proc.returncode == 0:
        return None
    msg = proc.stdout.strip()
    if not msg:
        err_lines = proc.stderr.strip().splitlines()
        msg = err_lines[-1] if err_lines else f"subprocess returncode={proc.returncode}"
    return msg[:120]


# ── §3-1 ──────────────────────────────────────────────────────────────────────

def test_01_anu_v3_12_modules_import_standalone():
    """§3-1: 완전 closure 디렉토리에서 12개 모듈 전부 독립 import 성공 assert.

    12개 anu_v3 모듈과 유일한 외부 의존 dispatch.callback_owner_enforcer 를
    격리 tmp에 복사 후 WORKSPACE_ROOT 를 sys.path 에서 제거한 상태로
    importlib.import_module 을 실행한다. 실패 모듈은 메시지에 포함.
    """
    tmp = tempfile.mkdtemp(prefix="t01_closure_")
    try:
        _build_closure_dir(tmp)
        failed = []
        for m in TWELVE:
            err = _isolated_import(tmp, f"anu_v3.{m}")
            if err is not None:
                failed.append(f"{m}: {err}")
        assert not failed, (
            f"격리 import 실패 모듈 {len(failed)}/12:\n" + "\n".join(failed)
        )
    finally:
        shutil.rmtree(tmp, ignore_errors=True)


# ── §3-2 ──────────────────────────────────────────────────────────────────────

def test_02_import_without_conftest_syspath_injection():
    """§3-2: (a) closure 없는 빈 tmp → ModuleNotFoundError 확인 (false-pass 차단).
             (b) 12개 소스 ast.parse → anu_v3.X import X ⊆ TWELVE 정적 검증.

    (a): WORKSPACE_ROOT/conftest 주입이 제거된 상태에서 import 가 실패해야
         한다. 이는 test_01 의 성공이 conftest 주입 덕분이 아님을 증명한다.
    (b): 12개 소스 파일에서 ImportFrom(anu_v3.*) 노드를 순회하여 참조하는
         서브모듈이 TWELVE 집합 내에만 있음을 정적으로 검증한다.
    """
    # (a) false-pass 차단: 빈 tmp 에서 import 실패 확인
    empty_tmp = tempfile.mkdtemp(prefix="t02_empty_")
    try:
        err = _isolated_import(empty_tmp, "anu_v3.runtime_reconcile_checkpoint")
        assert err is not None, (
            "빈 closure tmp 에서 ModuleNotFoundError 가 발생하지 않음 — "
            "WORKSPACE_ROOT 제거가 동작하지 않아 false-pass 위험"
        )
        assert "ModuleNotFoundError" in err, (
            f"예상 ModuleNotFoundError 대신 다른 오류: {err}"
        )
    finally:
        shutil.rmtree(empty_tmp, ignore_errors=True)

    # (b) 정적 의존성 검증: anu_v3.X → X ∈ TWELVE_SET
    extra_deps: list[str] = []
    for m in TWELVE:
        src_path = os.path.join(SRC, "anu_v3", f"{m}.py")
        tree = ast.parse(Path(src_path).read_text(encoding="utf-8"))
        for node in ast.walk(tree):
            if isinstance(node, ast.ImportFrom):
                if node.module and node.module.startswith("anu_v3."):
                    sub = node.module[len("anu_v3."):]
                    if sub not in TWELVE_SET:
                        extra_deps.append(f"{m} imports anu_v3.{sub} (outside TWELVE)")
            elif isinstance(node, ast.Import):
                for alias in node.names:
                    if alias.name.startswith("anu_v3."):
                        sub = alias.name[len("anu_v3."):]
                        if sub not in TWELVE_SET:
                            extra_deps.append(
                                f"{m} imports anu_v3.{sub} (outside TWELVE)"
                            )
    assert not extra_deps, (
        "12개 외 추가 anu_v3 의존성 발견 (closure 불완전):\n"
        + "\n".join(extra_deps)
    )


# ── §3-3 ──────────────────────────────────────────────────────────────────────

def test_03_callback_enforcement_resolvable_from_origin_main_plus_12():
    """§3-3: callback enforcement 파일 2개의 anu_v3.* import 집합 ⊆ TWELVE.

    /dispatch/cron_dispatch_guard.py 및 /dispatch/normal_fallback_callback_helper.py
    를 ast.parse 하여 anu_v3.* 의존 추출 → TWELVE 부분집합 검증.
    최소 1개 파일이 검사돼야 한다(없으면 skip).
    """
    candidates = [
        os.path.join(SRC, "dispatch", "cron_dispatch_guard.py"),
        os.path.join(SRC, "dispatch", "normal_fallback_callback_helper.py"),
    ]
    checked = 0
    extra: list[str] = []

    for fpath in candidates:
        if not os.path.isfile(fpath):
            continue
        fname = os.path.basename(fpath)
        tree = ast.parse(Path(fpath).read_text(encoding="utf-8"))
        for node in ast.walk(tree):
            if isinstance(node, ast.ImportFrom):
                if node.module and node.module.startswith("anu_v3."):
                    sub = node.module[len("anu_v3."):]
                    if sub not in TWELVE_SET:
                        extra.append(f"{fname} → anu_v3.{sub} (outside TWELVE)")
            elif isinstance(node, ast.Import):
                for alias in node.names:
                    if alias.name.startswith("anu_v3."):
                        sub = alias.name[len("anu_v3."):]
                        if sub not in TWELVE_SET:
                            extra.append(f"{fname} → anu_v3.{sub} (outside TWELVE)")
        checked += 1

    if checked == 0:
        pytest.skip("callback enforcement 파일 2개 모두 없음 — skip")

    assert not extra, (
        "origin/main + 12개로 충족 불가한 anu_v3 의존 발견:\n"
        + "\n".join(extra)
    )


# ── §3-4 ──────────────────────────────────────────────────────────────────────

def test_04_missing_anu_v3_dependency_fails():
    """§3-4: runtime_batch_state_updater 제거 시 runtime_reconcile_checkpoint
    import → ModuleNotFoundError 발생 확인 (false-pass 아님, missing dep 검출).

    내부 의존 목록: runtime_reconcile_checkpoint →
      {active_dispatch_scanner, task_artifact_detector,
       runtime_next_action_resolver, runtime_batch_state_updater}.
    하나라도 빠지면 격리 import 가 실패해야 한다.
    """
    tmp = tempfile.mkdtemp(prefix="t04_missing_dep_")
    try:
        _build_closure_dir(tmp, omit={"runtime_batch_state_updater"})
        err = _isolated_import(tmp, "anu_v3.runtime_reconcile_checkpoint")
        assert err is not None, (
            "runtime_batch_state_updater 가 없는데 import 성공 — "
            "격리 패턴이 동작하지 않아 false-pass 위험"
        )
        assert "ModuleNotFoundError" in err, (
            f"예상 ModuleNotFoundError 대신 다른 오류: {err}"
        )
    finally:
        shutil.rmtree(tmp, ignore_errors=True)


# ── §3-5 ──────────────────────────────────────────────────────────────────────

def test_05_self_collector_result_not_authoritative():
    """§3-5: self_collector_guard — executor self-session 은 FAIL·SELF_COLLECTOR_FORBIDDEN,
    독립 ANU collector session 은 PASS.

    검증 API:
      guard_self_collector_session(executor_key, collector_key, collector_role)
        → SelfCollectorGuardResult
          .verdict: PASS | FAIL
          .classification: SELF_COLLECTOR_FORBIDDEN | None
          .is_executor_self_session: bool
          .ok: bool (verdict == PASS)
    """
    _ensure_src_dispatch()
    # WORKSPACE_ROOT 에 있는 실제 모듈을 직접 import (격리 아님 — 기능 테스트)
    from anu_v3.self_collector_guard import (
        FAIL,
        PASS,
        SELF_COLLECTOR_FORBIDDEN,
        guard_self_collector_session,
    )

    # (a) executor self-session: collector_key == executor_key → FAIL
    self_res = guard_self_collector_session(
        executor_key=SELF_KEY,
        collector_key=SELF_KEY,
        collector_role="ANU",
    )
    assert self_res.verdict == FAIL, (
        f"self-session 은 FAIL 이어야 함, got {self_res.verdict}"
    )
    assert self_res.classification == SELF_COLLECTOR_FORBIDDEN, (
        f"classification={self_res.classification}"
    )
    assert self_res.is_executor_self_session is True
    assert self_res.ok is False

    # (b) 독립 ANU collector: executor=SELF_KEY, collector=ANU_KEY, role="ANU" → PASS
    anu_res = guard_self_collector_session(
        executor_key=SELF_KEY,
        collector_key=ANU_KEY,
        collector_role="ANU",
    )
    assert anu_res.verdict == PASS, (
        f"독립 ANU collector 는 PASS 이어야 함, got {anu_res.verdict}"
    )
    assert anu_res.ok is True


# ── §3-6 ──────────────────────────────────────────────────────────────────────

def test_06_callback_owner_validator_distinguishes_anu_vs_self_key():
    """§3-6: callback_owner_validator — ANU key vs executor self-key 구분,
    PASS/FAIL 판정 검증.

    검증 API:
      is_anu_key(key, DEFAULT_ANU_KEYS) → bool
      validate_callback_owner_runtime(
          task_id, executor_key, collector_key, collector_role,
          normal_collector_cron_id, fallback_callback_cron_id,
          dispatch_cron_id, entry_path, ...) → CallbackOwnerValidationResult
            .verdict: PASS | FAIL
            .registration_allowed: bool
    """
    _ensure_src_dispatch()
    from anu_v3.callback_owner_validator import (
        FAIL,
        PASS,
        validate_callback_owner_runtime,
    )
    from dispatch.callback_owner_enforcer import DEFAULT_ANU_KEYS, is_anu_key

    # is_anu_key 검증
    assert is_anu_key(ANU_KEY, DEFAULT_ANU_KEYS) is True, (
        f"ANU key {ANU_KEY!r} 는 is_anu_key=True 이어야 함"
    )
    assert is_anu_key(SELF_KEY, DEFAULT_ANU_KEYS) is False, (
        f"SELF key {SELF_KEY!r} 는 is_anu_key=False 이어야 함"
    )

    common_kwargs = dict(
        task_id="task-2628",
        executor_key=SELF_KEY,
        collector_role="ANU",
        normal_collector_cron_id="cron-normal-1",
        fallback_callback_cron_id="cron-fb-1",
        dispatch_cron_id="cron-dispatch-1",
        entry_path="cokacdir_cron_direct",
    )

    # PASS 케이스: collector_key = ANU_KEY (독립 ANU key)
    pass_res = validate_callback_owner_runtime(
        collector_key=ANU_KEY,
        **common_kwargs,
    )
    assert pass_res.verdict == PASS, (
        f"ANU collector_key 는 PASS 이어야 함, got {pass_res.verdict}\n"
        f"reasons: {pass_res.reasons}"
    )
    assert pass_res.registration_allowed is True, (
        "PASS 면 registration_allowed=True 이어야 함"
    )

    # FAIL 케이스: collector_key = SELF_KEY (executor self)
    fail_res = validate_callback_owner_runtime(
        collector_key=SELF_KEY,
        **common_kwargs,
    )
    assert fail_res.verdict == FAIL, (
        f"self-key collector 는 FAIL 이어야 함, got {fail_res.verdict}"
    )
    assert fail_res.registration_allowed is False, (
        "FAIL 이면 registration_allowed=False 이어야 함"
    )


# ── §3-7 ──────────────────────────────────────────────────────────────────────

def test_07_dispatch_callback_contract_schema_validation():
    """§3-7: dispatch_callback_contract — schema validation, key 제약, evaluate().

    검증 API:
      assert_collector_key_is_independent_anu(key) → None or raises ExecutorSelfKeyForbidden
      evaluate(observation: dict) → DispatchContractRecord or raises InvalidObservation
      run_self_check() → dict
      CONTRACT_SCHEMA, _OBSERVATION_BOOL_KEYS
    """
    _ensure_src_dispatch()
    from anu_v3.dispatch_callback_contract import (
        CONTRACT_SCHEMA,
        EXECUTOR_SELF_KEY_FORBIDDEN,
        INDEPENDENT_ANU_KEY,
        InvalidObservation,
        _OBSERVATION_BOOL_KEYS,
        assert_collector_key_is_independent_anu,
        evaluate,
        run_self_check,
        ExecutorSelfKeyForbidden,
    )

    # (1) 독립 ANU key → 예외 없음
    assert_collector_key_is_independent_anu(INDEPENDENT_ANU_KEY)  # 예외 없어야 함

    # (2) executor self-key → ExecutorSelfKeyForbidden
    with pytest.raises(ExecutorSelfKeyForbidden):
        assert_collector_key_is_independent_anu(EXECUTOR_SELF_KEY_FORBIDDEN)

    # (3) 빈 dict → InvalidObservation
    with pytest.raises(InvalidObservation):
        evaluate({})

    # (4) valid observation → DispatchContractRecord, schema 필드 일치
    valid_obs: dict = {"task_id": "task-2628"}
    for k in _OBSERVATION_BOOL_KEYS:
        valid_obs[k] = True  # bool signal 모두 채움
    record = evaluate(valid_obs)
    assert record.schema == CONTRACT_SCHEMA, (
        f"schema 필드 불일치: {record.schema!r} != {CONTRACT_SCHEMA!r}"
    )
    # DispatchContractRecord 임을 타입 이름으로 확인
    assert type(record).__name__ == "DispatchContractRecord"

    # (5) run_self_check() → dict (fixture 없어도 dict 반환)
    result = run_self_check()
    assert isinstance(result, dict), (
        f"run_self_check() 는 dict 를 반환해야 함, got {type(result).__name__}"
    )


# ── §3-8 ──────────────────────────────────────────────────────────────────────

def test_08_runtime_reconcile_dry_run_no_side_effects():
    """§3-8: runtime_reconcile_checkpoint dry-run 파일 side-effect 0 검증,
    recovery_layer 계약 검증.

    (A) RuntimeTaskObservation / classify_observation 사용 — 파일 I/O 없는 경로.
        before/after os.listdir(d) 로 side-effect 0 assert.
    (B) recovery_layer:
        checkpoint_replaces_callback_primary_path() is False
        checkpoint_discards_fallback_safety_path() is False
        recovery_layer_contract() → dict
        assert_checkpoint_is_recovery_not_primary() → [] (위반 0)
    """
    _ensure_src_dispatch()
    from anu_v3.runtime_reconcile_checkpoint import (
        RuntimeTaskObservation,
        classify_observation,
    )
    from anu_v3.runtime_reconcile_checkpoint_recovery_layer import (
        assert_checkpoint_is_recovery_not_primary,
        checkpoint_discards_fallback_safety_path,
        checkpoint_replaces_callback_primary_path,
        recovery_layer_contract,
    )

    # (A) dry-run: classify_observation 은 순수 함수 — I/O 없음
    d = tempfile.mkdtemp(prefix="t08_dryrun_")
    saved_cwd = os.getcwd()
    try:
        before = os.listdir(d)

        obs = RuntimeTaskObservation(
            task_id="task-2628-dryrun",
            dispatch_ok=True,
            result_present=True,
            done_present=True,
            normal_collector_registered=False,
            normal_collector_executed=False,
            by_design_no_normal_collector=True,
            fallback_state="NONE",
            terminal_outcome="DONE",
        )
        classification = classify_observation(obs)

        after = os.listdir(d)
        assert before == after, (
            f"classify_observation 이 파일 side-effect 를 발생시킴: "
            f"before={before}, after={after}"
        )
        assert os.getcwd() == saved_cwd, "cwd 가 변경됨"
        # 분류가 유효한 문자열임을 확인
        assert isinstance(classification, str) and classification, (
            "classify_observation 이 빈 문자열 반환"
        )
    finally:
        shutil.rmtree(d, ignore_errors=True)

    # (B) recovery-layer 계약
    assert checkpoint_replaces_callback_primary_path() is False, (
        "checkpoint_replaces_callback_primary_path() 는 항상 False 이어야 함 (§6.13)"
    )
    assert checkpoint_discards_fallback_safety_path() is False, (
        "checkpoint_discards_fallback_safety_path() 는 항상 False 이어야 함 (§6.14)"
    )

    contract = recovery_layer_contract()
    assert isinstance(contract, dict), (
        f"recovery_layer_contract() 는 dict 를 반환해야 함, got {type(contract).__name__}"
    )
    assert contract.get("replaces_primary") is False
    assert contract.get("discards_fallback") is False

    violations = assert_checkpoint_is_recovery_not_primary()
    assert violations == [], (
        f"recovery-layer 위반 발견 (§6.7/§6.13/§6.14):\n"
        + "\n".join(violations)
    )