# -*- coding: utf-8 -*-
# pyright: reportMissingImports=false, reportFunctionMemberAccess=false
#   (runtime importlib shim 으로 dispatch 패키지를 로드하고, mock 함수에 .calls 속성을
#    동적 부착하는 테스트 패턴 — pyright 정적 분석은 따라가지 못하므로 파일 레벨 억제.
#    tests/regression/test_anu_pickup_driver_2721.py 와 동일한 동적 mock 계약.)
"""task-2753 Option B provenance-path — regression (16).

대상 모듈:
- dispatch.anu_executor_provenance (validate_executor_provenance + ProvenanceResult + REASON_*)
- dispatch.anu_pickup_driver (process_one provenance-path 분기 + provenance_path_enabled flag)
- dispatch.anu_collector_result (CollectorResult.adopted_via)

design packet §5 test plan 매핑(번호는 packet §5 기준):
  (A) 순수함수 단위테스트 — validate_executor_provenance 직접 호출 (조건 1,3~13,15)
  (B) process_one 통합테스트 — driver 분기 (조건 2,14,16)

절대 제약:
- 실제 ANU 키 literal('c119085...') 0건. raw key 테스트는 가짜 키 sentinel 사용.
- pickup_once 실호출 0 — 전부 mock 주입(pickup_fn/verify_fn/provenance_fn/collector_writer_fn).
  실제 wake/cron/callback fire 0.
- tmp_path fixture 만 사용. 실제 memory 디렉토리에 쓰지 않는다.

기존 패턴(test_anu_pickup_driver_2721.py)의 sys.path/dispatch import 보강 헤더,
_make_dirs/_events_dir/_write_result/_enable_activation/_age/_NO_SLEEP/make_pickup_mock 를
그대로 재사용/모방한다.
"""
from __future__ import annotations

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

# regression/conftest.py 가 worktree root 를 sys.path 에 보장하지만,
# 단독 실행/순서 변동 대비로 한 번 더 보강한다.
_ROOT = Path(__file__).resolve().parents[2]
if str(_ROOT) not in sys.path:
    sys.path.insert(0, str(_ROOT))

# pytest prepend 모드에서 tests/dispatch(테스트용 빈 패키지)가 실제 dispatch 패키지를
# 가리는 것을 방지: 실제 dispatch 패키지를 파일 위치로 직접 로드해 sys.modules 에 고정.
import importlib.util as _ilu  # noqa: E402

_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
from dispatch import anu_collector_result as cr  # noqa: E402


# ─────────────────────────────────────────────────────────────────────────────
# 상수
# ─────────────────────────────────────────────────────────────────────────────
TASK_ID = "task-2753"
# capsule sha(테스트 상수). valid result 의 capsule_sha256 == capability snapshot.source_sha256.
CAPSULE_SHA = "deadbeef" * 8  # 64-hex 형태의 임의 테스트 sha (실제 키 아님)
# ★ raw key 테스트용 가짜 sentinel — 실제 ANU/dev key literal 이 아니다.
FAKE_RAW_KEY = "FAKE_RAW_KEY_SENTINEL_DO_NOT_USE"


def _valid_result() -> dict:
    """provenance-path 통과용 valid executor_result (envelope 없음 = provenance 진입 조건)."""
    return {
        "task_id": TASK_ID,
        "completion_signal": "EXECUTOR_RESULT_WRITTEN",
        "terminal_status": "PASS",
        "capsule_sha256": CAPSULE_SHA,
        "report_path": "report.md",
        # collector_envelope 없음 → provenance-path 진입.
        # claim 필드들은 전부 부재(=falsy) → 조건11/12/13 통과.
    }


def _cap_reader(*, task_id=TASK_ID, source_sha256=CAPSULE_SHA):
    """capability_reader mock: capability_reader(task_id)->dict|None.
    snapshot dict = {"task_id", "source_sha256"} 최소 필드."""
    return lambda tid: {"task_id": task_id, "source_sha256": source_sha256}


# ─────────────────────────────────────────────────────────────────────────────
# 헬퍼 (test_anu_pickup_driver_2721.py 패턴 재사용)
# ─────────────────────────────────────────────────────────────────────────────
def _make_dirs(tmp_path: Path) -> Path:
    """tmp_path 안에 inbox/state/capabilities 디렉토리를 만들고 root 반환."""
    (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:
    """inbox(memory/events/p0b_inbox) 아래에 <task_id>.result.json 을 쓰고 경로 반환.
    조건4/조건8 의 source-path/sha 계산이 실제 파일을 요구하므로 정규 경로에 쓴다."""
    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 _age(path, seconds: float = 10.0) -> None:
    """result 파일 mtime/atime 을 과거로 설정 — readiness 'aged' 게이트 통과."""
    past = _time.time() - seconds
    os.utime(str(path), (past, past))


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


def make_pickup_mock(verdict: str = "WAKE_BUILT"):
    """pickup_fn mock — provenance-path 는 pickup_fn 을 호출하지 않아야 하므로,
    이 mock 의 calls 가 비어있는지 검증하는 데 사용한다."""
    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


def make_verify_mock(verdict: str = "AUTHORITATIVE"):
    """verify_fn mock (기존 envelope 경로 진입 검증용)."""
    calls: list = []

    def _verify(*args, **kwargs):
        calls.append((args, kwargs))
        return types.SimpleNamespace(
            verdict=verdict, ok=(verdict == "AUTHORITATIVE"),
            owner_resolution={}, classification="", reasons=[],
        )

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


def make_launcher_mock():
    """launcher_fn/callback fire mock — provenance-path 에서 fire 0(미호출)을 검증."""
    calls: list = []

    def _launch(*args, **kwargs):
        calls.append((args, kwargs))
        return types.SimpleNamespace(decision="LAUNCHED")

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


def make_provenance_spy(pass_result=True):
    """provenance_fn spy — 호출 인자 기록 + ProvenanceResult 반환.
    integration 테스트(#14/#16)에서 '미호출' 검증, #2 에서 pass 주입에 사용."""
    calls: list = []

    def _fn(result, result_path, **kwargs):
        calls.append((result, result_path, kwargs))
        if pass_result:
            return prov.ProvenanceResult(ok=True, source_result_sha256="spysha123")
        return prov.ProvenanceResult(ok=False, reason="spy_fail")

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


def make_collector_writer_spy():
    """collector_writer_fn spy — (collector, path) 인자 기록."""
    calls: list = []

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

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


# ═════════════════════════════════════════════════════════════════════════════
# (A) 순수함수 단위테스트 — validate_executor_provenance
# ═════════════════════════════════════════════════════════════════════════════

# §5-1: valid result → ok=True, source_result_sha256 비어있지 않음(파일 sha 일치)
def test_unit_valid_result_ok(tmp_path):
    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 == ""
    assert pv.skip is False
    assert pv.duplicate is False
    # 조건8: source_result_sha256 = 실제 파일 sha256 와 일치
    assert pv.source_result_sha256
    import hashlib
    expected = hashlib.sha256(p.read_bytes()).hexdigest()
    assert pv.source_result_sha256 == expected


# §5-3: missing dispatch record(capability_reader→None) → reason=no_dispatch_record
def test_unit_missing_dispatch_record(tmp_path):
    root = _make_dirs(tmp_path)
    p = _write_result(root)
    pv = prov.validate_executor_provenance(
        _valid_result(), str(p), root=str(root),
        capability_reader=lambda tid: None,
    )
    assert pv.ok is False
    assert pv.reason == prov.REASON_NO_DISPATCH_RECORD


# §5-4: task_id mismatch(snapshot.task_id 다름) → reason=task_id_mismatch
def test_unit_task_id_mismatch(tmp_path):
    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(task_id="task-OTHER"),
    )
    assert pv.ok is False
    assert pv.reason == prov.REASON_TASK_ID_MISMATCH


# §5-5: capsule_sha mismatch(capsule_sha256 ≠ snapshot.source_sha256) → capsule_sha_mismatch
def test_unit_capsule_sha_mismatch(tmp_path):
    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(source_sha256="MISMATCHED_SHA"),
    )
    assert pv.ok is False
    assert pv.reason == prov.REASON_CAPSULE_SHA_MISMATCH


# §5-6: source path outside(result_path 가 inbox 밖) → reason=source_path_invalid
def test_unit_source_path_invalid(tmp_path):
    root = _make_dirs(tmp_path)
    # inbox 밖(root 직속)에 파일을 쓴다 → 정규 inbox 경로와 불일치.
    outside = root / f"{TASK_ID}.result.json"
    outside.write_text(json.dumps(_valid_result()), encoding="utf-8")
    pv = prov.validate_executor_provenance(
        _valid_result(), str(outside), root=str(root),
        capability_reader=_cap_reader(),
    )
    assert pv.ok is False
    assert pv.reason == prov.REASON_SOURCE_PATH_INVALID


# §5-7: invalid schema(report_path 없음) → reason=schema_fail
def test_unit_schema_fail_missing_report_path(tmp_path):
    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


# §5-8: invalid terminal_status("FAIL") → reason=terminal_invalid
def test_unit_terminal_invalid(tmp_path):
    root = _make_dirs(tmp_path)
    payload = _valid_result()
    payload["terminal_status"] = "FAIL"
    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_TERMINAL_INVALID


# §5-9: stale epoch(파일 mtime < activation_epoch) → skip=True, reason=pre_activation_epoch
def test_unit_stale_epoch_skip(tmp_path):
    root = _make_dirs(tmp_path)
    p = _write_result(root)
    _age(p, seconds=10)  # 파일 mtime 을 과거로
    future_epoch = _time.time() + 10_000  # activation_epoch 을 미래로 → stale 유발
    pv = prov.validate_executor_provenance(
        _valid_result(), str(p), root=str(root),
        capability_reader=_cap_reader(),
        activation_epoch=future_epoch,
    )
    assert pv.ok is False
    assert pv.skip is True
    assert pv.reason == prov.REASON_PRE_ACTIVATION_EPOCH


# §5-10: callback schedule claim(schedule_created_by_executor=True) → reason=callback_claim
def test_unit_callback_schedule_claim(tmp_path):
    root = _make_dirs(tmp_path)
    payload = _valid_result()
    payload["schedule_created_by_executor"] = True
    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_CALLBACK_CLAIM


# §5-11: self-collector claim(self_collector=True) → reason=self_collector_claim
def test_unit_self_collector_claim(tmp_path):
    root = _make_dirs(tmp_path)
    payload = _valid_result()
    payload["self_collector"] = True
    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_SELF_COLLECTOR_CLAIM


# §5-12: dev self-key claim(owner_key_proof_present=True) → reason=self_key_claim
def test_unit_self_key_claim(tmp_path):
    root = _make_dirs(tmp_path)
    payload = _valid_result()
    payload["owner_key_proof_present"] = True
    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_SELF_KEY_CLAIM


# §5-13: duplicate(ledger_reader→True) → duplicate=True
def test_unit_duplicate(tmp_path):
    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(),
        ledger_reader=lambda tid, sha: True,  # 중복
    )
    assert pv.ok is False
    assert pv.duplicate is True
    assert pv.reason == prov.REASON_DUPLICATE


# §5-15: raw key(forbidden_key_literals 가 result 어딘가에 포함) → reason=raw_key,
#         + valid result 의 출력 ProvenanceResult 에 raw key literal 0 확인.
def test_unit_raw_key_quarantine_and_zero_leak(tmp_path):
    root = _make_dirs(tmp_path)
    # (a) result 에 가짜 raw key sentinel 을 심는다 → raw_key quarantine.
    payload = _valid_result()
    payload["leaked_field"] = FAKE_RAW_KEY
    p = _write_result(root, payload=payload)
    pv = prov.validate_executor_provenance(
        payload, str(p), root=str(root),
        capability_reader=_cap_reader(),
        forbidden_key_literals=(FAKE_RAW_KEY,),
    )
    assert pv.ok is False
    assert pv.reason == prov.REASON_RAW_KEY

    # (b) valid result(센티넬 없음) 의 출력 ProvenanceResult 에 raw key literal 0.
    p2 = _write_result(root)
    pv_ok = prov.validate_executor_provenance(
        _valid_result(), str(p2), root=str(root),
        capability_reader=_cap_reader(),
        forbidden_key_literals=(FAKE_RAW_KEY,),
    )
    assert pv_ok.ok is True
    dumped = json.dumps({
        "ok": pv_ok.ok, "reason": pv_ok.reason,
        "source_result_sha256": pv_ok.source_result_sha256,
        "skip": pv_ok.skip, "duplicate": pv_ok.duplicate,
    })
    assert FAKE_RAW_KEY not in dumped
    # 실제 ANU key literal 도 출력에 없어야 함(분할 문자열로 literal 직접 기재 회피).
    forbidden_real = "c119085" + "addb0f8b7"
    assert forbidden_real not in dumped


# ═════════════════════════════════════════════════════════════════════════════
# (B) process_one 통합테스트 — driver 분기
# ═════════════════════════════════════════════════════════════════════════════

# §5-2: valid real dev result(envelope 없음) + provenance_path_enabled=True
#        + capability_reader/collector_writer_fn 주입
#        → verdict=PROVENANCE_ADOPTED,
#          collector_writer_fn 이 adopted_via="provenance" + source_result_sha256 일치하는
#          CollectorResult 로 호출됨.
#          + callback fire(launcher_fn/callback_launch_fn) 미호출(fire 0) — gate 우회/약화 없음.
def test_integration_provenance_adopted(tmp_path):
    root = _make_dirs(tmp_path)
    _enable_activation(root)
    pickup = make_pickup_mock()
    launcher = make_launcher_mock()
    callback_launch = make_launcher_mock()
    writer = make_collector_writer_spy()

    p = _write_result(root)  # envelope 없는 valid real dev result
    _age(p)

    rec = drv.process_one(
        str(p), root=str(root),
        pickup_fn=pickup,
        launcher_fn=launcher,
        callback_launch_fn=callback_launch,
        sleep_fn=_NO_SLEEP,
        provenance_path_enabled=True,
        capability_reader=_cap_reader(),
        collector_writer_fn=writer,
    )

    # verdict = PROVENANCE_ADOPTED, owner_key_class = PROVENANCE_ADOPTED 라벨.
    assert rec.verdict == drv.VERDICT_PROVENANCE_ADOPTED
    assert rec.owner_key_class == drv.OKC_PROVENANCE
    assert rec.quarantined is False

    # collector_writer_fn 정확히 1회 호출 + adopted_via="provenance" + sha 일치.
    assert len(writer.calls) == 1
    collector, _cr_path = writer.calls[0]
    assert collector.adopted_via == cr.ADOPTED_VIA_PROVENANCE
    # ★ 원본 파일은 processed 로 이동됐을 수 있으므로 이동 전 내용 sha 를 재계산하지 않고,
    #   provenance 가 계산한 source_result_sha256 가 collector 에 그대로 실렸는지 본다.
    assert collector.source_result_sha256
    assert collector.task_id == TASK_ID

    # ★ callback fire 0 — owner_proof_pickup_gate 우회/약화 없음(test plan #2).
    assert len(launcher.calls) == 0
    assert len(callback_launch.calls) == 0
    # provenance-path 는 pickup_once 도 호출하지 않는다(collector_result 생성만).
    assert len(pickup.calls) == 0


# §5-14: fixture 기존 경로 회귀 — envelope 있는 result + provenance_path_enabled=True
#         → 기존 경로(verify_fn 호출)로 진입, provenance_fn 미호출.
def test_integration_envelope_path_regression(tmp_path):
    root = _make_dirs(tmp_path)
    _enable_activation(root)
    pickup = make_pickup_mock(verdict="WAKE_BUILT")
    verify = make_verify_mock(verdict="AUTHORITATIVE")
    prov_spy = make_provenance_spy(pass_result=True)

    # envelope 있는 result → provenance 분기 진입 안 함.
    payload = _valid_result()
    payload["collector_envelope"] = {"task_id": TASK_ID, "schedule_id": "sch-1"}
    p = _write_result(root, payload=payload)
    _age(p)

    rec = drv.process_one(
        str(p), root=str(root),
        pickup_fn=pickup, verify_fn=verify,
        sleep_fn=_NO_SLEEP,
        provenance_path_enabled=True,      # flag ON 이어도 envelope 있으면 기존 경로.
        provenance_fn=prov_spy,
        capability_reader=_cap_reader(),
    )

    # 기존 envelope 경로 = verify_fn 호출 + WAKE_BUILT.
    assert rec.verdict == drv.VERDICT_WAKE_BUILT
    assert rec.owner_key_class == drv.OKC_ANU
    assert len(verify.calls) == 1
    # ★ provenance_fn 미호출 (envelope 있으면 provenance 분기 안 탐).
    assert len(prov_spy.calls) == 0


# §5-16(+1): flag OFF 회귀 — envelope 없는 result + provenance_path_enabled=False(기본)
#             → verdict=QUARANTINE, quarantine_reason="owner_unprovable",
#               owner_key_class=OKC_FOREIGN. provenance_fn 미호출.
def test_integration_flag_off_owner_unprovable_regression(tmp_path):
    root = _make_dirs(tmp_path)
    _enable_activation(root)
    pickup = make_pickup_mock()
    prov_spy = make_provenance_spy(pass_result=True)

    p = _write_result(root)  # envelope 없는 result
    _age(p)

    # provenance_path_enabled 미지정 → 기본 OFF.
    rec = drv.process_one(
        str(p), root=str(root),
        pickup_fn=pickup,
        sleep_fn=_NO_SLEEP,
        provenance_fn=prov_spy,
        capability_reader=_cap_reader(),
    )

    assert rec.verdict == drv.VERDICT_QUARANTINE
    assert rec.quarantine_reason == "owner_unprovable"
    assert rec.owner_key_class == drv.OKC_FOREIGN
    # ★ provenance_fn 미호출 (flag OFF → 기존 동작 100% 보존).
    assert len(prov_spy.calls) == 0
    assert len(pickup.calls) == 0


# ═════════════════════════════════════════════════════════════════════════════
# task-2753+1 Finding ① ② ③ — Gemini hardening 회귀 (14)
# ═════════════════════════════════════════════════════════════════════════════

# ── Finding ① — task_id path traversal (validate_executor_provenance 단위) ──

def test_unit_task_id_traversal_dotdot_slash(tmp_path):
    """task_id='../evil' → ok=False, reason=REASON_TASK_ID_PATH_TRAVERSAL."""
    root = _make_dirs(tmp_path)
    payload = _valid_result()
    payload["task_id"] = "../evil"
    pv = prov.validate_executor_provenance(
        payload, str(tmp_path / "any.json"), root=str(root),
        capability_reader=_cap_reader(),
    )
    assert pv.ok is False
    assert pv.reason == prov.REASON_TASK_ID_PATH_TRAVERSAL


def test_unit_task_id_traversal_absolute(tmp_path):
    """task_id='/tmp/evil' → ok=False, reason=REASON_TASK_ID_PATH_TRAVERSAL."""
    root = _make_dirs(tmp_path)
    payload = _valid_result()
    payload["task_id"] = "/tmp/evil"
    pv = prov.validate_executor_provenance(
        payload, str(tmp_path / "any.json"), root=str(root),
        capability_reader=_cap_reader(),
    )
    assert pv.ok is False
    assert pv.reason == prov.REASON_TASK_ID_PATH_TRAVERSAL


def test_unit_task_id_traversal_backslash(tmp_path):
    r"""task_id='a\b' (백슬래시) → ok=False, reason=REASON_TASK_ID_PATH_TRAVERSAL."""
    root = _make_dirs(tmp_path)
    payload = _valid_result()
    payload["task_id"] = "a\\b"
    pv = prov.validate_executor_provenance(
        payload, str(tmp_path / "any.json"), root=str(root),
        capability_reader=_cap_reader(),
    )
    assert pv.ok is False
    assert pv.reason == prov.REASON_TASK_ID_PATH_TRAVERSAL


def test_unit_task_id_traversal_forward_slash(tmp_path):
    """task_id='a/b' → ok=False, reason=REASON_TASK_ID_PATH_TRAVERSAL."""
    root = _make_dirs(tmp_path)
    payload = _valid_result()
    payload["task_id"] = "a/b"
    pv = prov.validate_executor_provenance(
        payload, str(tmp_path / "any.json"), root=str(root),
        capability_reader=_cap_reader(),
    )
    assert pv.ok is False
    assert pv.reason == prov.REASON_TASK_ID_PATH_TRAVERSAL


def test_unit_task_id_traversal_url_encoded(tmp_path):
    """task_id='..%2fevil' → ok=False, reason=REASON_TASK_ID_PATH_TRAVERSAL.
    '%' 가 whitelist([A-Za-z0-9._+-]) 밖이므로 거부됨."""
    root = _make_dirs(tmp_path)
    payload = _valid_result()
    payload["task_id"] = "..%2fevil"
    pv = prov.validate_executor_provenance(
        payload, str(tmp_path / "any.json"), root=str(root),
        capability_reader=_cap_reader(),
    )
    assert pv.ok is False
    assert pv.reason == prov.REASON_TASK_ID_PATH_TRAVERSAL


def test_unit_task_id_traversal_dotdot_bare(tmp_path):
    """task_id='..' → ok=False, reason=REASON_TASK_ID_PATH_TRAVERSAL."""
    root = _make_dirs(tmp_path)
    payload = _valid_result()
    payload["task_id"] = ".."
    pv = prov.validate_executor_provenance(
        payload, str(tmp_path / "any.json"), root=str(root),
        capability_reader=_cap_reader(),
    )
    assert pv.ok is False
    assert pv.reason == prov.REASON_TASK_ID_PATH_TRAVERSAL


# ── Finding ① — _safe_capability_path / _validate_task_id_safe 직접 단위 ──

def test_unit_safe_capability_path_escape_returns_none(tmp_path):
    """caps_root 밖으로 escape 하는 task_id 는 None 반환.

    _safe_capability_path 의 책임은 realpath+commonpath 에 의한 caps_root 밖 escape 방어.
    separator 포함('a/b') 은 caps_root 내부 경로로 resolve 되므로 None 이 아니며,
    separator 거부는 _validate_task_id_safe(whitelist 계층)가 담당한다.
    """
    root = _make_dirs(tmp_path)
    # '../../etc/passwd' 는 realpath 후 caps_root 밖으로 escape → None
    assert prov._safe_capability_path("../../etc/passwd", str(root)) is None


def test_unit_safe_capability_path_safe_resolves_inside(tmp_path):
    """safe task_id 는 caps_root 안으로 resolve 됨."""
    root = _make_dirs(tmp_path)
    p = prov._safe_capability_path("task-2753", str(root))
    assert p is not None
    caps_root = os.path.realpath(os.path.join(str(root), "memory", "capabilities"))
    assert os.path.commonpath([caps_root, p]) == caps_root


def test_unit_validate_task_id_safe_whitelist(tmp_path):
    """_validate_task_id_safe whitelist/거부 패턴 총망라."""
    assert prov._validate_task_id_safe("task-2753") is True
    assert prov._validate_task_id_safe("task-2753+1") is True
    assert prov._validate_task_id_safe("../evil") is False
    assert prov._validate_task_id_safe("a/b") is False
    assert prov._validate_task_id_safe("a\\b") is False
    assert prov._validate_task_id_safe("") is False
    assert prov._validate_task_id_safe("..") is False


# ── Finding ① — safe task_id 정상 resolve (default reader 통합) ──

def test_unit_safe_task_id_default_reader_resolves(tmp_path):
    """safe task_id + capability JSON 실재 → capability_reader 미주입(default) → ok=True."""
    root = _make_dirs(tmp_path)
    cap_json = root / "memory" / "capabilities" / f"{TASK_ID}.json"
    cap_json.write_text(
        json.dumps({"task_id": TASK_ID, "source_sha256": CAPSULE_SHA}),
        encoding="utf-8",
    )
    p = _write_result(root)
    # capability_reader 미주입 — default reader 사용
    pv = prov.validate_executor_provenance(_valid_result(), str(p), root=str(root))
    assert pv.ok is True


# ── Finding ② — chunked hash 동일성 + 대용량 ──

def test_unit_chunked_hash_matches_full_read(tmp_path):
    """_default_hash_fn 이 작은 파일과 >1 MiB 파일 모두 full-read sha256 과 일치."""
    import hashlib
    root = _make_dirs(tmp_path)

    # 작은 파일
    small = root / "small.bin"
    small.write_bytes(b"hello-provenance-2753")
    assert prov._default_hash_fn(str(small)) == hashlib.sha256(small.read_bytes()).hexdigest()

    # 대용량(>1 MiB, 여러 chunk): 약 2 MiB + 123 bytes
    big = root / "big.bin"
    data = b"A" * (1024 * 1024 * 2 + 123)
    big.write_bytes(data)
    assert prov._default_hash_fn(str(big)) == hashlib.sha256(data).hexdigest()


def test_unit_chunked_hash_large_stream_no_full_buffer(tmp_path, monkeypatch):
    """chunk 크기를 7 byte 로 monkeypatch 하여도 hash 결과가 동일 = streaming 정확성 증명."""
    import hashlib
    root = _make_dirs(tmp_path)
    monkeypatch.setattr(prov, "_HASH_CHUNK_SIZE", 7)
    f = root / "stream.bin"
    data = b"X" * 100
    f.write_bytes(data)
    assert prov._default_hash_fn(str(f)) == hashlib.sha256(data).hexdigest()


# ── Finding ③ — collector_result write 실패 시 source 이동 안 됨 (process_one 통합) ──

def test_integration_move_blocked_when_collector_write_fails(tmp_path):
    """collector_writer_fn 이 OSError 를 raise 하면 source 파일은 inbox 원위치에 보존된다."""
    root = _make_dirs(tmp_path)
    _enable_activation(root)
    p = _write_result(root)
    _age(p)

    def _failing_writer(collector, path):
        raise OSError("disk full simulated")

    rec = drv.process_one(
        str(p), root=str(root),
        pickup_fn=make_pickup_mock(),
        launcher_fn=make_launcher_mock(),
        callback_launch_fn=make_launcher_mock(),
        sleep_fn=_NO_SLEEP,
        provenance_path_enabled=True,
        capability_reader=_cap_reader(),
        collector_writer_fn=_failing_writer,
    )

    # source 가 inbox 원위치에 그대로 존재해야 함
    assert p.exists()
    # error 에 collector_result write 실패 메시지 포함
    assert rec.error and "collector_result write 실패" in rec.error


# ── Finding ③ — collector_result write 성공 후에만 source 이동 (process_one 통합) ──

def test_integration_move_after_collector_write_success(tmp_path):
    """collector_writer_fn 성공 시 source 가 inbox 에서 이동되고 verdict=PROVENANCE_ADOPTED."""
    root = _make_dirs(tmp_path)
    _enable_activation(root)
    p = _write_result(root)
    _age(p)

    writer = make_collector_writer_spy()

    rec = drv.process_one(
        str(p), root=str(root),
        pickup_fn=make_pickup_mock(),
        launcher_fn=make_launcher_mock(),
        callback_launch_fn=make_launcher_mock(),
        sleep_fn=_NO_SLEEP,
        provenance_path_enabled=True,
        capability_reader=_cap_reader(),
        collector_writer_fn=writer,
    )

    # writer 정확히 1회 호출
    assert len(writer.calls) == 1
    # source 가 inbox 에서 이동됨(원위치 부재)
    assert not p.exists()
    # verdict = PROVENANCE_ADOPTED
    assert rec.verdict == drv.VERDICT_PROVENANCE_ADOPTED