# -*- coding: utf-8 -*-
"""task-2779_a — Pre-push Terminal Artifact Guard 회귀 테스트.

pre_push_terminal_guard.run_guard 및 main 진입점의 불변식을 격리 검증:
  - artifact 존재 시 push_attempt=False, surfacing 정상 동작.
  - artifact 부재 시 push_attempt=True, PUSH_ATTEMPT_WITHOUT_TERMINAL_MARKER 기록.
  - warn mode: 항상 비차단(blocked=False), exit_code=0.
  - 내부 오류 시 fail-open: GUARD_INTERNAL_ERROR_FAIL_OPEN, 크래시 없음.
  - traversal/절대경로 입력은 거부 처리됨(부재 취급).
  - 금지 리터럴(raw key, subprocess, git push 실행 패턴) 소스에 없음.
  - 기존 dispatch.terminal_artifact_index 동작 무회귀 스모크.

모든 테스트는 tmp_path만 사용. 실제 /home/jay/workspace canonical 절대 미접촉.
"""
from __future__ import annotations

import importlib.util as _ilu
import json
import re
import sys
from pathlib import Path

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

# 실제 dispatch 패키지를 파일 위치로 직접 로드해 sys.modules 에 고정.
_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)

# dispatch 패키지가 sys.modules 에 고정된 후 guard 모듈 로드 (동일 인스턴스 공유).
_guard_path = _ROOT / "scripts" / "pre_push_terminal_guard.py"
_gspec = _ilu.spec_from_file_location("pre_push_terminal_guard", _guard_path)
assert _gspec is not None and _gspec.loader is not None
guard = _ilu.module_from_spec(_gspec)
_gspec.loader.exec_module(guard)

# dispatch 심볼 임포트 (부트스트랩 후 안전)
from dispatch.terminal_artifact_index import (  # noqa: E402
    build_manifest,
    surface_manifest,
    STATUS_WORKTREE_LOCAL_PRESENT,
    STATUS_SURFACED_TO_CANONICAL,
    STATUS_TERMINAL_CALLBACK_MISSING,
    STATUS_PUSH_ATTEMPT_WITHOUT_TERMINAL,
    CANONICAL_INDEX_DIR_REL,
)


# ─────────────────────────────────────────────────────────────────────────────
# 테스트 1: worktree에 report 존재 → push_attempt=False, surfacing 정상
# ─────────────────────────────────────────────────────────────────────────────
def test_terminal_artifact_present_surfaces_manifest(tmp_path):
    """worktree report 존재 시 push_attempt=False, STATUS_WORKTREE_LOCAL_PRESENT in statuses, canonical index 파일 생성 확인."""
    task_id = "task-2779-test-present"
    canonical = tmp_path / "canonical"
    worktree = tmp_path / "worktree"

    # worktree에 report 파일 생성
    report_dir = worktree / "memory" / "reports"
    report_dir.mkdir(parents=True)
    (report_dir / f"{task_id}.md").write_text("# terminal report\n", encoding="utf-8")

    result = guard.run_guard(
        task_id,
        canonical_root=str(canonical),
        worktree_root=str(worktree),
    )

    assert result["push_attempt"] is False, "artifact 존재 시 push_attempt는 False여야 한다"
    assert STATUS_WORKTREE_LOCAL_PRESENT in result["statuses"], \
        "worktree에 report 있을 때 STATUS_WORKTREE_LOCAL_PRESENT가 statuses에 있어야 한다"
    assert result["surfaced_index_path"] != "", "surfaced_index_path는 비어 있으면 안 된다"

    # canonical index 파일 실제 생성 확인
    expected_index = canonical / CANONICAL_INDEX_DIR_REL / f"{task_id}.json"
    assert expected_index.exists(), f"canonical index 파일이 생성되어야 한다: {expected_index}"

    # JSON 내용 검증
    data = json.loads(expected_index.read_text(encoding="utf-8"))
    assert data["task_id"] == task_id
    assert STATUS_WORKTREE_LOCAL_PRESENT in data["statuses"]


# ─────────────────────────────────────────────────────────────────────────────
# 테스트 2: artifact 전무 → push_attempt=True, PUSH_ATTEMPT_WITHOUT_TERMINAL_MARKER 기록
# ─────────────────────────────────────────────────────────────────────────────
def test_terminal_artifact_absent_push_attempt_marker(tmp_path):
    """artifact 부재 시 push_attempt=True, status/statuses에 마커 설정, canonical index에도 기록됨."""
    task_id = "task-2779-test-absent"
    canonical = tmp_path / "canonical"
    worktree = tmp_path / "worktree"
    worktree.mkdir(parents=True)

    result = guard.run_guard(
        task_id,
        canonical_root=str(canonical),
        worktree_root=str(worktree),
    )

    assert result["push_attempt"] is True, "artifact 부재 시 push_attempt는 True여야 한다"
    assert result["status"] == "PUSH_ATTEMPT_WITHOUT_TERMINAL_MARKER", \
        "status가 PUSH_ATTEMPT_WITHOUT_TERMINAL_MARKER여야 한다"
    assert "PUSH_ATTEMPT_WITHOUT_TERMINAL_MARKER" in result["statuses"], \
        "statuses에 PUSH_ATTEMPT_WITHOUT_TERMINAL_MARKER가 있어야 한다"

    # canonical index 파일에 마커 기록 확인
    expected_index = canonical / CANONICAL_INDEX_DIR_REL / f"{task_id}.json"
    assert expected_index.exists(), "canonical index 파일이 생성되어야 한다"
    data = json.loads(expected_index.read_text(encoding="utf-8"))
    assert "PUSH_ATTEMPT_WITHOUT_TERMINAL_MARKER" in data["statuses"], \
        "canonical index에도 마커가 기록되어야 한다"


# ─────────────────────────────────────────────────────────────────────────────
# 테스트 3: warn mode — 항상 비차단, exit_code 0
# ─────────────────────────────────────────────────────────────────────────────
def test_warn_mode_non_blocking_exit_zero(tmp_path):
    """artifact 부재 시에도 warn mode에서 blocked=False, exit_code=0 (비차단) 보장."""
    task_id = "task-2779-test-warn"
    canonical = tmp_path / "canonical"
    worktree = tmp_path / "worktree"
    worktree.mkdir(parents=True)

    result = guard.run_guard(
        task_id,
        canonical_root=str(canonical),
        worktree_root=str(worktree),
        mode="warn",
    )

    assert result["blocked"] is False, "warn mode에서 blocked는 항상 False여야 한다"
    assert result["exit_code"] == 0, "warn mode에서 exit_code는 항상 0이어야 한다"


# ─────────────────────────────────────────────────────────────────────────────
# 테스트 4: 내부 오류 주입 → fail-open, GUARD_INTERNAL_ERROR_FAIL_OPEN
# ─────────────────────────────────────────────────────────────────────────────
def test_internal_error_fail_open(tmp_path, monkeypatch):
    """build_manifest에 예외 주입 시 fail-open: GUARD_INTERNAL_ERROR_FAIL_OPEN, 크래시 없음, exit_code=0."""
    task_id = "task-2779-test-error"
    canonical = tmp_path / "canonical"
    worktree = tmp_path / "worktree"
    worktree.mkdir(parents=True)

    def _raise(*args, **kwargs):
        raise RuntimeError("강제 내부 오류 주입")

    monkeypatch.setattr(guard, "build_manifest", _raise)

    result = guard.run_guard(
        task_id,
        canonical_root=str(canonical),
        worktree_root=str(worktree),
    )

    assert result["failed_open"] is True, "내부 오류 시 failed_open=True여야 한다"
    assert result["status"] == "GUARD_INTERNAL_ERROR_FAIL_OPEN", \
        "내부 오류 시 status가 GUARD_INTERNAL_ERROR_FAIL_OPEN이어야 한다"
    assert result["blocked"] is False, "내부 오류 시에도 blocked는 False여야 한다"
    assert result["exit_code"] == 0, "내부 오류 시에도 exit_code는 0이어야 한다"


# ─────────────────────────────────────────────────────────────────────────────
# 테스트 5: traversal/절대경로 report_rel → 거부(부재 취급), crash 없음
# ─────────────────────────────────────────────────────────────────────────────
def test_malicious_rel_path_rejected_by_containment(tmp_path):
    """traversal 경로(../../../../etc/passwd) 및 절대경로는 containment helper가 거부, push_attempt=True (부재 취급), crash 없음."""
    task_id = "task-2779-test-traversal"
    canonical = tmp_path / "canonical"
    worktree = tmp_path / "worktree"
    worktree.mkdir(parents=True)

    # traversal 경로로 시도
    result_traversal = guard.run_guard(
        task_id,
        canonical_root=str(canonical),
        worktree_root=str(worktree),
        report_rel="../../../../etc/passwd",
    )

    assert result_traversal["push_attempt"] is True, \
        "traversal 경로는 거부되어 부재 취급(push_attempt=True)이어야 한다"
    assert result_traversal["status"] == "PUSH_ATTEMPT_WITHOUT_TERMINAL_MARKER", \
        "traversal 경로 거부 시 status가 PUSH_ATTEMPT_WITHOUT_TERMINAL_MARKER여야 한다"

    # 절대경로로 시도
    result_abs = guard.run_guard(
        task_id + "-abs",
        canonical_root=str(canonical),
        worktree_root=str(worktree),
        report_rel="/etc/passwd",
    )

    assert result_abs["push_attempt"] is True, \
        "절대경로는 거부되어 부재 취급(push_attempt=True)이어야 한다"
    assert result_abs["status"] == "PUSH_ATTEMPT_WITHOUT_TERMINAL_MARKER", \
        "절대경로 거부 시 status가 PUSH_ATTEMPT_WITHOUT_TERMINAL_MARKER여야 한다"


# ─────────────────────────────────────────────────────────────────────────────
# 테스트 6: 소스 파일 정적 검증 — 금지 raw key 리터럴 부재
# ─────────────────────────────────────────────────────────────────────────────
def _get_docstring_nodes(tree) -> set:
    """AST tree에서 docstring에 해당하는 Constant 노드를 수집하여 id set으로 반환.

    모듈/함수/클래스 정의 body의 첫 번째 Expr>Constant(str)는 docstring으로 취급.
    """
    import ast as _ast
    docstring_ids: set[int] = set()

    def _collect(body):
        if not body:
            return
        first = body[0]
        if isinstance(first, _ast.Expr) and isinstance(first.value, _ast.Constant) \
                and isinstance(first.value.value, str):
            docstring_ids.add(id(first.value))

    # 모듈 body
    if isinstance(tree, _ast.Module):
        _collect(tree.body)
    # 함수/클래스 body
    for node in _ast.walk(tree):
        if isinstance(node, (_ast.FunctionDef, _ast.AsyncFunctionDef, _ast.ClassDef)):
            _collect(node.body)
    return docstring_ids


def test_no_self_key_no_raw_key_static():
    """scripts/pre_push_terminal_guard.py 소스에 ANU key, self/chat key, raw key 의심 리터럴이 없음 (docstring 언급 허용, 실제 코드 상수만 검증)."""
    import ast as _ast

    src = _guard_path.read_text(encoding="utf-8")

    # 알려진 두 키 리터럴은 전체 소스에서도 절대 허용 안 됨 (docstring에도 없어야 함)
    assert "c119085addb0f8b7" not in src, "ANU key 리터럴이 소스에 있으면 안 된다"
    assert "fedf78d1d09509f5" not in src, "self/chat key 리터럴이 소스에 있으면 안 된다"

    # AST로 실제 코드 노드의 문자열 상수만 추출하여 검증 (docstring 제외)
    tree = _ast.parse(src)
    docstring_ids = _get_docstring_nodes(tree)
    code_string_literals: list[str] = []
    for node in _ast.walk(tree):
        if isinstance(node, _ast.Constant) and isinstance(node.value, str):
            if id(node) not in docstring_ids:
                code_string_literals.append(node.value)

    # 코드 상수에 cokacdir, --key 패턴이 없어야 함
    for lit in code_string_literals:
        assert "cokacdir" not in lit, \
            f"코드 문자열 상수에 cokacdir가 있으면 안 된다: {lit!r}"
        assert lit not in ("--key",), \
            f"코드 문자열 상수에 --key 직접 값이 있으면 안 된다: {lit!r}"

    # 코드 상수에 raw key 의심 패턴(16자리 이상 hex) 없음
    raw_key_pattern = re.compile(r'^[0-9a-fA-F]{16,}$')
    for lit in code_string_literals:
        assert not raw_key_pattern.match(lit), \
            f"코드 문자열 상수에 raw key 의심 16진수 리터럴이 있으면 안 된다: {lit!r}"


# ─────────────────────────────────────────────────────────────────────────────
# 테스트 7: 소스 파일 정적 검증 — subprocess/git push/hooks 실행 패턴 부재
# ─────────────────────────────────────────────────────────────────────────────
def test_no_real_hook_install_no_push_static():
    """scripts/pre_push_terminal_guard.py 소스에 subprocess import, os.system, subprocess. 호출이 없음 (docstring 금지어 언급은 허용, 실제 import/호출 AST 노드만 검증)."""
    import ast as _ast

    src = _guard_path.read_text(encoding="utf-8")

    # subprocess import 여부: AST import 노드로만 판정
    tree = _ast.parse(src)
    imported_names: list[str] = []
    for node in _ast.walk(tree):
        if isinstance(node, _ast.Import):
            for alias in node.names:
                imported_names.append(alias.name)
        elif isinstance(node, _ast.ImportFrom):
            if node.module:
                imported_names.append(node.module)

    assert "subprocess" not in imported_names, \
        "subprocess가 import되어 있으면 안 된다"

    # 실제 호출 여부: AST의 Attribute 노드에서 subprocess 호출 검사
    for node in _ast.walk(tree):
        if isinstance(node, _ast.Attribute):
            if isinstance(node.value, _ast.Name) and node.value.id == "subprocess":
                raise AssertionError(
                    f"subprocess.{node.attr} 호출이 코드에 있으면 안 된다"
                )

    # os.system 호출 여부: AST에서 직접 검사
    for node in _ast.walk(tree):
        if isinstance(node, _ast.Attribute):
            if (isinstance(node.value, _ast.Name)
                    and node.value.id == "os"
                    and node.attr == "system"):
                raise AssertionError("os.system 호출이 코드에 있으면 안 된다")

    # .git/hooks/pre-push 설치 패턴: 코드 문자열 상수에서만 검사 (docstring 제외)
    docstring_ids = _get_docstring_nodes(tree)
    for node in _ast.walk(tree):
        if isinstance(node, _ast.Constant) and isinstance(node.value, str):
            if id(node) not in docstring_ids:
                assert ".git/hooks/pre-push" not in node.value, \
                    f"코드 상수에 .git/hooks/pre-push가 있으면 안 된다: {node.value!r}"


# ─────────────────────────────────────────────────────────────────────────────
# 테스트 8: 기존 2778 dispatch.terminal_artifact_index 무회귀 스모크
# ─────────────────────────────────────────────────────────────────────────────
def test_existing_2778_regression_intact(tmp_path):
    """dispatch.terminal_artifact_index build_manifest/surface_manifest가 여전히 정상 동작하는지 스모크 검증 (2778 무회귀)."""
    task_id = "task-2778-smoke"
    canonical = tmp_path / "canonical"
    worktree = tmp_path / "worktree"

    # worktree에 report 파일 생성
    report_dir = worktree / "memory" / "reports"
    report_dir.mkdir(parents=True)
    (report_dir / f"{task_id}.md").write_text("# smoke test report\n", encoding="utf-8")

    # build_manifest 동작 확인
    manifest = build_manifest(
        task_id,
        canonical_root=str(canonical),
        worktree_root=str(worktree),
    )
    assert manifest.status == STATUS_WORKTREE_LOCAL_PRESENT, \
        "worktree에 report 있을 때 STATUS_WORKTREE_LOCAL_PRESENT여야 한다"

    # surface_manifest 동작 확인
    index_path = surface_manifest(manifest, canonical_root=str(canonical))
    assert index_path != "", "surface_manifest는 비어 있지 않은 경로를 반환해야 한다"

    expected = canonical / CANONICAL_INDEX_DIR_REL / f"{task_id}.json"
    assert expected.exists(), "surface 후 canonical index 파일이 생성되어야 한다"

    data = json.loads(expected.read_text(encoding="utf-8"))
    assert data["task_id"] == task_id
    assert STATUS_WORKTREE_LOCAL_PRESENT in data["statuses"]
    assert STATUS_SURFACED_TO_CANONICAL in manifest.statuses, \
        "surface_manifest 후 manifest.statuses에 STATUS_SURFACED_TO_CANONICAL이 추가되어야 한다"