# -*- coding: utf-8 -*-
# pyright: reportMissingImports=false, reportFunctionMemberAccess=false
"""task-2755 P0-B scope-limited ACTIVE wiring — regression.

2 Wiring 통합 테스트(Wiring3 commit guard 는 별도 backlog 분리 — 본 파일 제외):
  Wiring1: read_provenance_path_scope_enabled(4조건 AND, default OFF) + scan_once passthrough
  Wiring2: report_path_self_ref helper(절대/상대 self-ref) + 조건5 schema relaxation(self-ref/존재파일/비어있지않은str)

절대 제약:
- 실제 ANU key literal 0건.
- pickup_once 실호출 0 — 전부 mock 주입.
- tmp_path fixture 만 사용.
"""
from __future__ import annotations

import json
import os
import sys
import time as _time
import types
from pathlib import Path

# ── sys.path 부트스트랩 (test_provenance_path_2753.py 패턴 동일) ─────────────
_ROOT = Path(__file__).resolve().parents[2]
if str(_ROOT) not in sys.path:
    sys.path.insert(0, str(_ROOT))

import importlib.util as _ilu

_real_init = _ROOT / "dispatch" / "__init__.py"
_cached = sys.modules.get("dispatch")
if _cached is None or (getattr(_cached, "__file__", "") or "") != str(_real_init):
    for _k in [k for k in list(sys.modules) if k == "dispatch" or k.startswith("dispatch.")]:
        del sys.modules[_k]
    _spec = _ilu.spec_from_file_location(
        "dispatch", _real_init, submodule_search_locations=[str(_ROOT / "dispatch")]
    )
    assert _spec is not None and _spec.loader is not None
    _pkg = _ilu.module_from_spec(_spec)
    sys.modules["dispatch"] = _pkg
    _spec.loader.exec_module(_pkg)

from dispatch import anu_pickup_driver as drv  # noqa: E402
from dispatch import anu_executor_provenance as prov  # noqa: E402

# ── 상수 ──────────────────────────────────────────────────────────────────────
TASK_ID = "task-2755"
CAPSULE_SHA = "cafebabe" * 8  # 64-hex 테스트 sha (실제 키 아님)


def _valid_result() -> dict:
    """provenance-path 통과용 valid executor_result (envelope 없음)."""
    return {
        "task_id": TASK_ID,
        "completion_signal": "EXECUTOR_RESULT_WRITTEN",
        "terminal_status": "PASS",
        "capsule_sha256": CAPSULE_SHA,
        "report_path": "report.md",
    }


def _cap_reader(*, task_id=TASK_ID, source_sha256=CAPSULE_SHA):
    """capability_reader mock."""
    return lambda tid: {"task_id": task_id, "source_sha256": source_sha256}


def _make_dirs(tmp_path: Path) -> Path:
    (tmp_path / "memory" / "events" / "p0b_inbox").mkdir(parents=True, exist_ok=True)
    (tmp_path / "memory" / "state").mkdir(parents=True, exist_ok=True)
    (tmp_path / "memory" / "capabilities").mkdir(parents=True, exist_ok=True)
    return tmp_path


def _events_dir(root: Path) -> Path:
    return root / "memory" / "events" / "p0b_inbox"


def _write_result(root: Path, name=None, payload=None) -> Path:
    if name is None:
        name = f"{TASK_ID}.result.json"
    if payload is None:
        payload = _valid_result()
    p = _events_dir(root) / name
    p.write_text(json.dumps(payload), encoding="utf-8")
    return p


def _enable_activation(root: Path) -> None:
    flag = root / "memory" / "state" / "p0b_driver_enabled"
    flag.parent.mkdir(parents=True, exist_ok=True)
    flag.write_text(drv.ACTIVATION_ENABLED + "\n", encoding="utf-8")


def _enable_provenance_flag(root: Path) -> None:
    """p0b_provenance_path_enabled flag 세팅."""
    flag = root / "memory" / "state" / "p0b_provenance_path_enabled"
    flag.parent.mkdir(parents=True, exist_ok=True)
    flag.write_text(drv.PROVENANCE_PATH_FLAG_ENABLED + "\n", encoding="utf-8")


def _enable_epoch(root: Path) -> float:
    """activation epoch 마커 쓰기. 과거 값 사용."""
    epoch_val = _time.time() - 100.0
    epoch_file = root / "memory" / "state" / "p0b_activation_epoch"
    epoch_file.parent.mkdir(parents=True, exist_ok=True)
    epoch_file.write_text(str(epoch_val) + "\n", encoding="utf-8")
    return epoch_val


def _enable_scope_id(root: Path, scope_id: str = "pilot-2755") -> None:
    """p0b_provenance_scope_id 파일 쓰기."""
    sid_file = root / "memory" / "state" / "p0b_provenance_scope_id"
    sid_file.parent.mkdir(parents=True, exist_ok=True)
    sid_file.write_text(scope_id + "\n", encoding="utf-8")


def _age(path, seconds: float = 10.0) -> None:
    past = _time.time() - seconds
    os.utime(str(path), (past, past))


_NO_SLEEP = lambda *a, **k: None  # noqa: E731


def make_collector_writer_spy():
    calls: list = []

    def _writer(collector, path):
        calls.append((collector, path))
        return path

    _writer.calls = calls  # type: ignore[attr-defined]
    return _writer


def make_pickup_mock(verdict: str = "WAKE_BUILT"):
    calls: list = []

    def _pickup(*args, **kwargs):
        calls.append((args, kwargs))
        return types.SimpleNamespace(
            verdict=verdict, ok=(verdict == "WAKE_BUILT"),
            argv=["x"], task_id=TASK_ID, reasons=[],
        )

    _pickup.calls = calls  # type: ignore[attr-defined]
    return _pickup


# ═══════════════════════════════════════════════════════════════════════════════
# Wiring1 테스트
# ═══════════════════════════════════════════════════════════════════════════════

def test_w1_scope_gate_default_off(tmp_path):
    """아무 flag 없는 빈 root → read_provenance_path_scope_enabled == False."""
    root = _make_dirs(tmp_path)
    result = drv.read_provenance_path_scope_enabled(str(root))
    assert result is False


def test_w1_scope_gate_all_on_true(tmp_path):
    """4조건 모두 충족 → True."""
    root = _make_dirs(tmp_path)
    _enable_provenance_flag(root)
    _enable_epoch(root)
    _enable_scope_id(root)
    p = _write_result(root)
    _age(p, 10.0)
    result = drv.read_provenance_path_scope_enabled(str(root))
    assert result is True


def test_w1_scope_gate_missing_flag_off(tmp_path):
    """조건1(flag) 제거 → False."""
    root = _make_dirs(tmp_path)
    # flag 미작성
    _enable_epoch(root)
    _enable_scope_id(root)
    p = _write_result(root)
    _age(p, 10.0)
    result = drv.read_provenance_path_scope_enabled(str(root))
    assert result is False


def test_w1_scope_gate_missing_epoch_off(tmp_path):
    """조건2(epoch) 제거 → False."""
    root = _make_dirs(tmp_path)
    _enable_provenance_flag(root)
    # epoch 미작성
    _enable_scope_id(root)
    p = _write_result(root)
    _age(p, 10.0)
    result = drv.read_provenance_path_scope_enabled(str(root))
    assert result is False


def test_w1_scope_gate_missing_scope_id_off(tmp_path):
    """조건3(scope_id) 제거 → False."""
    root = _make_dirs(tmp_path)
    _enable_provenance_flag(root)
    _enable_epoch(root)
    # scope_id 미작성
    p = _write_result(root)
    _age(p, 10.0)
    result = drv.read_provenance_path_scope_enabled(str(root))
    assert result is False


def test_w1_scope_gate_empty_inbox_off(tmp_path):
    """조건4(inbox 비어있음) → False."""
    root = _make_dirs(tmp_path)
    _enable_provenance_flag(root)
    _enable_epoch(root)
    _enable_scope_id(root)
    # inbox 에 파일 미작성
    result = drv.read_provenance_path_scope_enabled(str(root))
    assert result is False


def test_w1_scope_gate_injected_all_on(tmp_path):
    """reader 주입으로 4조건 모두 True → True (파일시스템 독립 단위 테스트)."""
    root = _make_dirs(tmp_path)
    p = _write_result(root)
    _age(p, 10.0)

    result = drv.read_provenance_path_scope_enabled(
        str(root),
        flag_reader=lambda: drv.PROVENANCE_PATH_FLAG_ENABLED,
        epoch_reader=lambda: str(_time.time() - 50.0),
        scope_id_reader=lambda: "pilot-injected",
        inbox_lister=lambda: [str(p)],
    )
    assert result is True


def test_w1_scan_once_forwards_provenance_flag(tmp_path):
    """scan_once에 provenance_path_enabled=True 전달 시, collector_writer_fn spy가 호출됨."""
    root = _make_dirs(tmp_path)
    _enable_activation(root)
    _enable_epoch(root)

    p = _write_result(root)
    _age(p, 10.0)

    spy_writer = make_collector_writer_spy()
    epoch_val = drv.read_activation_epoch(str(root))

    # provenance_fn spy: ok=True 반환
    prov_calls: list = []

    def _prov_fn(result, result_path, **kwargs):
        prov_calls.append((result, result_path))
        import hashlib
        sha = hashlib.sha256(Path(result_path).read_bytes()).hexdigest()
        return prov.ProvenanceResult(ok=True, source_result_sha256=sha)

    records = drv.scan_once(
        str(root),
        provenance_path_enabled=True,
        provenance_fn=_prov_fn,
        collector_writer_fn=spy_writer,
        capability_reader=_cap_reader(),
        sleep_fn=_NO_SLEEP,
        activation_epoch=epoch_val,
        legacy_cutoff=True,
        write_evidence=False,
    )
    # provenance_fn 이 호출되었는지 확인 (forward 검증)
    assert len(prov_calls) >= 1
    verdicts = [r.verdict for r in records]
    assert drv.VERDICT_PROVENANCE_ADOPTED in verdicts


def test_w1_happy_path_adopted(tmp_path):
    """scan_once(provenance_path_enabled=True) + valid dev result → PROVENANCE_ADOPTED."""
    root = _make_dirs(tmp_path)
    _enable_activation(root)
    epoch_val = _enable_epoch(root)

    p = _write_result(root)
    _age(p, 10.0)

    spy_writer = make_collector_writer_spy()

    import hashlib
    sha = hashlib.sha256(p.read_bytes()).hexdigest()

    records = drv.scan_once(
        str(root),
        provenance_path_enabled=True,
        provenance_fn=lambda res, rp, **kw: prov.ProvenanceResult(ok=True, source_result_sha256=sha),
        collector_writer_fn=spy_writer,
        capability_reader=_cap_reader(),
        sleep_fn=_NO_SLEEP,
        activation_epoch=epoch_val,
        legacy_cutoff=True,
        write_evidence=False,
    )
    adopted = [r for r in records if r.verdict == drv.VERDICT_PROVENANCE_ADOPTED]
    assert len(adopted) == 1


def test_w1_invalid_result_quarantine(tmp_path):
    """capsule_sha 불일치 invalid result → QUARANTINE (provenance FAIL)."""
    root = _make_dirs(tmp_path)
    _enable_activation(root)
    epoch_val = _enable_epoch(root)

    payload = _valid_result()
    payload["capsule_sha256"] = "WRONG_SHA" * 4
    p = _write_result(root, payload=payload)
    _age(p, 10.0)

    # provenance_fn: sha 불일치 → ok=False
    records = drv.scan_once(
        str(root),
        provenance_path_enabled=True,
        provenance_fn=lambda res, rp, **kw: prov.ProvenanceResult(ok=False, reason=prov.REASON_CAPSULE_SHA_MISMATCH),
        collector_writer_fn=make_collector_writer_spy(),
        capability_reader=_cap_reader(source_sha256="CORRECT_SHA" * 4),
        sleep_fn=_NO_SLEEP,
        activation_epoch=epoch_val,
        legacy_cutoff=True,
        write_evidence=False,
    )
    quarantined = [r for r in records if r.verdict == drv.VERDICT_QUARANTINE]
    assert len(quarantined) >= 1


# ═══════════════════════════════════════════════════════════════════════════════
# Wiring2 테스트
# ═══════════════════════════════════════════════════════════════════════════════

def test_w2_schema_complete_pass(tmp_path):
    """_valid_result() + 실제 inbox 경로 write → validate_executor_provenance ok=True."""
    root = _make_dirs(tmp_path)
    p = _write_result(root)
    pv = prov.validate_executor_provenance(
        _valid_result(), str(p), root=str(root),
        capability_reader=_cap_reader(),
    )
    assert pv.ok is True
    assert pv.reason == ""


def test_w2_missing_report_path_fail(tmp_path):
    """report_path 제거 → ok=False, reason=REASON_SCHEMA_FAIL."""
    root = _make_dirs(tmp_path)
    payload = _valid_result()
    payload.pop("report_path")
    p = _write_result(root, payload=payload)
    pv = prov.validate_executor_provenance(
        payload, str(p), root=str(root),
        capability_reader=_cap_reader(),
    )
    assert pv.ok is False
    assert pv.reason == prov.REASON_SCHEMA_FAIL


def test_w2_self_ref_pass(tmp_path):
    """report_path = result.json 자기 자신 절대경로(self-ref) → ok=True."""
    root = _make_dirs(tmp_path)
    p = _write_result(root)
    payload = _valid_result()
    payload["report_path"] = str(p)  # self-ref
    p.write_text(json.dumps(payload), encoding="utf-8")

    pv = prov.validate_executor_provenance(
        payload, str(p), root=str(root),
        capability_reader=_cap_reader(),
    )
    assert pv.ok is True
    # report_path_self_ref 함수도 True 반환해야 함
    assert prov.report_path_self_ref(str(p), str(p)) is True


def test_w2_report_path_self_ref_helper(tmp_path):
    """report_path_self_ref helper 단위 테스트."""
    root = _make_dirs(tmp_path)
    p = _write_result(root)
    path_str = str(p)
    # 동일 경로 → True
    assert prov.report_path_self_ref(path_str, path_str) is True
    # 다른 경로 → False
    assert prov.report_path_self_ref(path_str, str(root / "other.json")) is False
    # 빈 str → False
    assert prov.report_path_self_ref("", path_str) is False
    assert prov.report_path_self_ref(path_str, "") is False


def test_w2_self_ref_whitespace_strip_pass(tmp_path):
    """★ finding #2 검증: report_path/result_path에 앞뒤 공백이 있어도
    strip 후 정규화 비교하여 self-ref로 인식(공백 차이로 인한 비교 실패 방지)."""
    root = _make_dirs(tmp_path)
    p = _write_result(root)
    path_str = str(p)
    # 앞뒤 공백 패딩된 self-ref → strip 후 동일 → True
    assert prov.report_path_self_ref("  " + path_str + "  ", path_str) is True
    assert prov.report_path_self_ref(path_str, "\t" + path_str + "\n") is True
    assert prov.report_path_self_ref(" " + path_str + " ", " " + path_str + " ") is True
    # 공백만 있는 경우 → False (비어있는 것으로 취급)
    assert prov.report_path_self_ref("   ", path_str) is False


def test_w2_prompts_require_report_path(tmp_path):
    """prompts/team_prompts.py 파일 텍스트에 'report_path 필수' 포함 (validator↔task md 일치)."""
    prompts_file = _ROOT / "prompts" / "team_prompts.py"
    text = prompts_file.read_text(encoding="utf-8")
    assert "report_path 필수" in text


def test_w2_self_ref_relative_to_result_dir_pass(tmp_path):
    """★ Wiring2 HIGH fix: 상대경로 report_path(result.json 디렉토리 basename 기준)도 self-ref PASS."""
    root = _make_dirs(tmp_path)
    p = _write_result(root)
    rel = os.path.basename(str(p))  # 'task-2755.result.json'
    assert prov.report_path_self_ref(rel, str(p)) is True


def test_w2_self_ref_relative_to_root_pass(tmp_path):
    """상대경로 report_path(root 기준 repo-relative)도 root 인자 제공 시 self-ref PASS."""
    root = _make_dirs(tmp_path)
    p = _write_result(root)
    rel_to_root = os.path.relpath(str(p), str(root))  # memory/events/p0b_inbox/task-2755.result.json
    assert prov.report_path_self_ref(rel_to_root, str(p), root=str(root)) is True


def test_w2_self_ref_absolute_pass(tmp_path):
    """절대경로 self-ref → True (기존 동작 회귀 보호)."""
    root = _make_dirs(tmp_path)
    p = _write_result(root)
    assert prov.report_path_self_ref(str(p), str(p)) is True


def test_w2_self_ref_different_relative_not_selfref(tmp_path):
    """자기 자신이 아닌 다른 상대경로 report_path → self-ref 아님(False)."""
    root = _make_dirs(tmp_path)
    p = _write_result(root)
    assert prov.report_path_self_ref("other_report.md", str(p)) is False


def test_w2_self_ref_traversal_failsafe(tmp_path):
    """path traversal(root 밖) report_path → result.json 자기 자신 아니면 fail-safe(False)."""
    root = _make_dirs(tmp_path)
    p = _write_result(root)
    assert prov.report_path_self_ref("../../../../../../etc/passwd", str(p), root=str(root)) is False


def test_w2_missing_report_path_helper_false(tmp_path):
    """missing/None/빈 report_path → helper False (조건5 schema_fail 경로 유지)."""
    root = _make_dirs(tmp_path)
    p = _write_result(root)
    assert prov.report_path_self_ref(None, str(p)) is False
    assert prov.report_path_self_ref("", str(p)) is False
    assert prov.report_path_self_ref("   ", str(p)) is False


# ═══════════════════════════════════════════════════════════════════════════════
# 공통 테스트
# ═══════════════════════════════════════════════════════════════════════════════

def test_common_no_raw_key_literal():
    """guard/driver/provenance 소스 파일에 ANU key 패턴('c1190') 부재 확인.
    단, prompts/team_prompts.py 는 doctrine block 에 키가 있어 제외.
    테스트 자체에도 key literal을 직접 사용하지 않음."""
    # 감지 패턴: 실제 ANU key prefix(짧은 부분 문자열 검사로 충분).
    # "c1190"은 ANU_NORMAL_COLLECTOR_KEY 앞부분 — 소스 파일에 있으면 안 됨.
    pattern = "c1190"
    files_to_check = [
        _ROOT / "dispatch" / "anu_pickup_driver.py",
        _ROOT / "dispatch" / "anu_executor_provenance.py",
    ]
    for f in files_to_check:
        text = f.read_text(encoding="utf-8")
        # 주석/docstring 에도 패턴 없어야 함
        assert pattern not in text, f"{f.name} 에 raw key 패턴 '{pattern}' 발견"


def test_common_provenance_constants_exported():
    """Wiring1 신규 상수가 __all__ 에 포함되는지 확인."""
    assert "PROVENANCE_PATH_FLAG_REL" in drv.__all__
    assert "PROVENANCE_PATH_FLAG_ENABLED" in drv.__all__
    assert "PROVENANCE_SCOPE_ID_REL" in drv.__all__
    assert "read_provenance_path_scope_enabled" in drv.__all__


def test_common_report_path_self_ref_exported():
    """Wiring2 신규 함수가 __all__ 에 포함되는지 확인."""
    assert "report_path_self_ref" in prov.__all__


# ═══════════════════════════════════════════════════════════════════════════════
# task-2755+4 abspath 완전 정렬 — 추가 self-ref 불변 케이스
# ═══════════════════════════════════════════════════════════════════════════════

def test_w2plus4_rel_result_rel_report_same_pass(monkeypatch, tmp_path):
    """★ HIGH fix: 상대 result_path + 상대 report_path(동일 파일) → self-ref True."""
    root = _make_dirs(tmp_path)
    p = _write_result(root)
    monkeypatch.chdir(root)
    rel = os.path.relpath(str(p), str(root))  # memory/events/p0b_inbox/...result.json
    assert prov.report_path_self_ref(rel, rel) is True


def test_w2plus4_abs_result_rel_report_same_pass(monkeypatch, tmp_path):
    """절대 result_path + 상대 report_path(동일 파일, repo-relative) → True."""
    root = _make_dirs(tmp_path)
    p = _write_result(root)
    monkeypatch.chdir(root)
    rel = os.path.relpath(str(p), str(root))
    assert prov.report_path_self_ref(rel, str(p)) is True


def test_w2plus4_rel_result_abs_report_same_pass(monkeypatch, tmp_path):
    """상대 result_path + 절대 report_path(동일 파일) → True."""
    root = _make_dirs(tmp_path)
    p = _write_result(root)
    monkeypatch.chdir(root)
    rel = os.path.relpath(str(p), str(root))
    assert prov.report_path_self_ref(str(p), rel) is True


def test_w2plus4_dot_relative_report_pass(monkeypatch, tmp_path):
    """'./memory/events/...' dot-relative report_path self-ref → True."""
    root = _make_dirs(tmp_path)
    p = _write_result(root)
    monkeypatch.chdir(root)
    rel = os.path.relpath(str(p), str(root))
    assert prov.report_path_self_ref("./" + rel, rel) is True


def test_w2plus4_repo_relative_self_ref_with_root_pass(tmp_path):
    """'memory/events/p0b_inbox/...result.json' 상대 self-ref(root 인자) → True."""
    root = _make_dirs(tmp_path)
    p = _write_result(root)
    rel = os.path.relpath(str(p), str(root))
    assert prov.report_path_self_ref(rel, rel, root=str(root)) is True


def test_w2plus4_different_file_not_selfref(monkeypatch, tmp_path):
    """다른 파일 가리키는 report_path → self-ref 아님(False)."""
    root = _make_dirs(tmp_path)
    p = _write_result(root)
    monkeypatch.chdir(root)
    assert prov.report_path_self_ref("memory/events/p0b_inbox/other.result.json", str(p)) is False


def test_w2plus4_none_and_blank_schema_fail(tmp_path):
    """None/빈/공백 report_path → schema_fail(False)."""
    root = _make_dirs(tmp_path)
    p = _write_result(root)
    assert prov.report_path_self_ref(None, str(p)) is False
    assert prov.report_path_self_ref("", str(p)) is False
    assert prov.report_path_self_ref("   ", str(p)) is False


def test_w2plus4_traversal_failsafe(tmp_path):
    """path traversal(root 밖) → fail-safe(False)."""
    root = _make_dirs(tmp_path)
    p = _write_result(root)
    assert prov.report_path_self_ref("../../../../../../etc/passwd", str(p), root=str(root)) is False