#!/usr/bin/env python3
"""task-3017 회귀 테스트 — ``~/.claude/hooks/git-lineage-guard.sh`` (PreToolUse, matcher=Bash).

이 훅은 "내 작업 base 와 쓰기 대상 브랜치 HEAD 의 **계보가 어긋난** 상태"에서
``gh api PUT repos/{o}/{r}/contents/{path}`` 로 파일을 통째 덮어쓰는 것을 차단한다.

배경(2026-08-25 실사고): worktree(base=997fa5d) 내용을 다른 계보의 브랜치에 contents API 로
PUT 하여 t3014 변경분을 되돌렸다(codex 29→17건, cafe 8→1건, server/main.py -161줄).
GitHub 의 blob sha 옵티미스틱 락은 이걸 막지 못한다 — 대상의 *현재* sha 를 정확히 넣으면
API 는 통과하고 내용만 조용히 과거로 돌아간다.

★ 자기완결: 계보 픽스처는 pytest ``tmp_path`` 안에서 ``git init`` 으로 만든다(자동 정리).
★ 네트워크 없음: ``gh`` 조회는 tmp_path 안의 스텁으로 대체하고 PATH 앞에 붙인다.
  (실 저장소 대상 검증은 보고서에만 남긴다 — CI 재현성 때문에 여기에는 넣지 않는다.)
★ 훅 파일이 없거나 못 읽으면 skip 이 아니라 **fail** 이다. 가드 유실 자체가 회귀다.
★ 이 테스트는 명령을 실행하지 않는다. 훅에 stdin JSON 을 먹여 판정만 본다.
"""

from __future__ import annotations

import hashlib
import json
import os
import subprocess
from pathlib import Path
from types import SimpleNamespace

import pytest

HOOK = Path("/home/jay/.claude/hooks/git-lineage-guard.sh")
SETTINGS = Path("/home/jay/.claude/settings.json")

DENY = "DENY"
ALLOW = "ALLOW"
NL = "\n"

# 픽스처 저장소의 원격 슬러그. 실제 워크스페이스 클론과 이름이 겹치지 않는 것을 쓴다.
OWNER = "Jeon-Jonghyuk"
RENAMED_OWNER = "JonghyukJeon"      # owner 리네임 케이스 (repo 이름만 일치)
REPO_NAME = "T3017LineageFixture"
SLUG = f"{OWNER}/{REPO_NAME}"
RENAMED_SLUG = f"{RENAMED_OWNER}/{REPO_NAME}"

BLOB = "1" * 40                      # contents PUT 의 기존 파일 blob sha (값 자체는 무관)
GHOST_SHA = "0123456789" * 4         # 40자 hex, 로컬에 존재하지 않는 커밋

# v3.6 harness 가 리터럴 셸 문자열을 차단하므로 원본 검증 스크립트와 같이 분할해 둔다.
PR_MERGE_CLI = "gh pr " + "merge 256 --squash"

_DIGEST_AT_IMPORT = hashlib.md5(HOOK.read_bytes()).hexdigest() if HOOK.is_file() else None

STUB_GH = '''#!/usr/bin/env python3
"""테스트용 gh 스텁 — 네트워크 조회만 대체한다(훅 본체 로직은 그대로 탄다)."""
import json, os, sys

STATE = json.load(open(os.environ["GLG_STUB_STATE"]))
args = sys.argv[1:]

if not args or args[0] != "api":
    sys.stderr.write("stub: unsupported %r\\n" % args)
    sys.exit(1)

path = args[1] if len(args) > 1 else ""
parts = path.split("/")

# repos/{o}/{r}/commits/{branch}
if len(parts) >= 5 and parts[0] == "repos" and parts[3] == "commits":
    branch = "/".join(parts[4:])
    m = STATE.get("branch_sha", {})
    if branch in m:
        print(m[branch]); sys.exit(0)
    sys.stderr.write("stub: no such branch %s\\n" % branch); sys.exit(1)

# repos/{o}/{r}
if len(parts) == 3 and parts[0] == "repos":
    if "default_branch" in STATE:
        print(STATE["default_branch"]); sys.exit(0)
    sys.stderr.write("stub: no default_branch\\n"); sys.exit(1)

# repos/{o}/{r}/pulls/{N}
if len(parts) == 5 and parts[0] == "repos" and parts[3] == "pulls":
    pr = STATE.get("pulls", {}).get(parts[4])
    if pr is None:
        sys.stderr.write("stub: no such PR\\n"); sys.exit(1)
    print(json.dumps(pr)); sys.exit(0)

sys.stderr.write("stub: unhandled path %s\\n" % path)
sys.exit(1)
'''


# --------------------------------------------------------------------------- 픽스처
def _git(repo: Path, *args: str) -> str:
    env = dict(os.environ)
    env.update(
        {
            "GIT_AUTHOR_NAME": "t3017",
            "GIT_AUTHOR_EMAIL": "t3017@example.com",
            "GIT_COMMITTER_NAME": "t3017",
            "GIT_COMMITTER_EMAIL": "t3017@example.com",
            "GIT_CONFIG_GLOBAL": os.devnull,
            "GIT_CONFIG_SYSTEM": os.devnull,
        }
    )
    proc = subprocess.run(
        ["git", "-C", str(repo), *args],
        capture_output=True,
        text=True,
        timeout=60,
        env=env,
    )
    assert proc.returncode == 0, f"git {args} 실패: {proc.stderr.strip()[:300]}"
    return proc.stdout.strip()


def _commit(repo: Path, name: str, body: str, message: str) -> str:
    (repo / name).write_text(body, encoding="utf-8")
    _git(repo, "add", name)
    _git(repo, "commit", "-q", "-m", message)
    return _git(repo, "rev-parse", "HEAD")


@pytest.fixture(scope="session")
def lineage(tmp_path_factory) -> SimpleNamespace:
    """계보가 갈라진 실제 git 저장소 + gh 스텁을 tmp 안에 만든다(세션 1회, 자동 정리).

    구성:  base ─┬─ main            ("my worktree work")        = 로컬 HEAD
                 └─ remote-branch   (t3014 커밋 2개)            = 다른 계보
    """
    base_dir = tmp_path_factory.mktemp("t3017-lineage")

    repo = base_dir / "fixture-repo"
    repo.mkdir()
    _git(repo, "init", "-q", "-b", "main")
    base_sha = _commit(repo, "a.txt", "v1\n", "base commit")

    _git(repo, "checkout", "-q", "-b", "remote-branch")
    _commit(repo, "c.txt", "codex 29\n", "t3014: codex 케이스 29건 추가")
    remote_sha = _commit(repo, "c.txt", "cafe 8\n", "t3014: cafe 8건 추가")

    _git(repo, "checkout", "-q", "main")
    main_sha = _commit(repo, "a.txt", "v2\n", "my worktree work")
    _git(repo, "remote", "add", "origin", f"git@github.com:{SLUG}.git")

    # 계보 전제 검증 — 픽스처가 의도대로 갈라졌는지 먼저 확인한다.
    assert main_sha != remote_sha
    ancestor = subprocess.run(
        ["git", "-C", str(repo), "merge-base", "--is-ancestor", remote_sha, main_sha],
        capture_output=True,
    )
    assert ancestor.returncode != 0, "픽스처 오류: remote-branch 가 main 의 조상이 되어버렸다"

    norepo = base_dir / "norepo"
    norepo.mkdir()
    (norepo / "keep.txt").write_text("", encoding="utf-8")
    assert not (norepo / ".git").exists()

    stub_dir = base_dir / "stubbin"
    stub_dir.mkdir()
    stub_gh = stub_dir / "gh"
    stub_gh.write_text(STUB_GH, encoding="utf-8")
    stub_gh.chmod(0o755)

    state = base_dir / "stub-state.json"
    state.write_text(
        json.dumps(
            {
                "default_branch": "remote-branch",
                "branch_sha": {
                    "remote-branch": remote_sha,
                    "main": main_sha,
                    "old": base_sha,
                    "ghost": GHOST_SHA,
                },
                "pulls": {
                    "256": {"head": "a" * 40, "state": "open",
                            "mergeable": True, "mergeable_state": "clean"},
                    "300": {"head": "b" * 40, "state": "open",
                            "mergeable": False, "mergeable_state": "dirty"},
                    "301": {"head": "c" * 40, "state": "open",
                            "mergeable": True, "mergeable_state": "behind"},
                    "302": {"head": "d" * 40, "state": "open",
                            "mergeable": True, "mergeable_state": "unstable"},
                    "303": {"head": "e" * 40, "state": "open",
                            "mergeable": None, "mergeable_state": "blocked"},
                },
            }
        ),
        encoding="utf-8",
    )

    return SimpleNamespace(
        repo=repo,
        norepo=norepo,
        stub_dir=stub_dir,
        state=state,
        base_sha=base_sha,
        main_sha=main_sha,
        remote_sha=remote_sha,
    )


def judge(command: str, cwd: Path, lineage: SimpleNamespace) -> SimpleNamespace:
    """훅에 PreToolUse stdin JSON(+cwd)을 넣고 (verdict, reason, system_message) 를 얻는다."""
    assert HOOK.is_file(), f"가드 유실(회귀): {HOOK}"
    assert os.access(HOOK, os.R_OK), f"가드를 읽을 수 없다(회귀): {HOOK}"

    env = dict(os.environ)
    env["PATH"] = str(lineage.stub_dir) + os.pathsep + env.get("PATH", "")
    env["GLG_STUB_STATE"] = str(lineage.state)
    env["PWD"] = str(cwd)

    payload = json.dumps(
        {"tool_name": "Bash", "tool_input": {"command": command}, "cwd": str(cwd)},
        ensure_ascii=False,
    )
    proc = subprocess.run(
        ["bash", str(HOOK)],
        input=payload,
        capture_output=True,
        text=True,
        timeout=300,
        env=env,
        cwd=str(cwd),
    )
    assert proc.returncode == 0, (
        f"훅은 어떤 경우에도 rc=0 이어야 한다. rc={proc.returncode} "
        f"stderr={proc.stderr[:400]!r}"
    )

    out = (proc.stdout or "").strip()
    if not out:
        return SimpleNamespace(verdict=ALLOW, reason="", system_message="", raw="")
    try:
        data = json.loads(out)
    except json.JSONDecodeError:  # pragma: no cover - 훅 출력 파손 시에만
        pytest.fail(f"훅 출력이 JSON 이 아니다: {out[:400]!r}")

    hook_out = data.get("hookSpecificOutput") or {}
    return SimpleNamespace(
        verdict=DENY if hook_out.get("permissionDecision") == "deny" else ALLOW,
        reason=hook_out.get("permissionDecisionReason", "") or "",
        system_message=data.get("systemMessage", "") or "",
        raw=out,
    )


# --------------------------------------------------------------------------- 가드 존재
def test_hook_file_exists_and_is_readable():
    """가드 파일이 삭제/이동/권한상실되면 계보 검사가 통째로 사라지므로 fail 로 잡는다."""
    assert HOOK.is_file(), f"가드 유실(회귀): {HOOK}"
    assert os.access(HOOK, os.R_OK), f"가드 읽기 불가(회귀): {HOOK}"
    body = HOOK.read_text(encoding="utf-8", errors="replace")
    assert body.startswith("#!"), "가드가 셸 스크립트가 아니다"
    assert len(body.splitlines()) > 100, "가드 본문이 비정상적으로 짧다(덮어쓰기 의심)"


def test_hook_is_registered_in_settings_json():
    """settings.json PreToolUse(matcher=Bash) 등록이 빠지면 훅 파일이 멀쩡해도 가드는 죽는다."""
    assert SETTINGS.is_file(), f"settings.json 유실(회귀): {SETTINGS}"
    data = json.loads(SETTINGS.read_text(encoding="utf-8"))
    commands = [
        hook.get("command", "")
        for entry in (data.get("hooks", {}).get("PreToolUse") or [])
        if entry.get("matcher") == "Bash"
        for hook in (entry.get("hooks") or [])
    ]
    assert any(
        "git-lineage-guard.sh" in cmd for cmd in commands
    ), f"PreToolUse(matcher=Bash) 에 git-lineage-guard.sh 미등록(회귀): {commands}"


# --------------------------------------------------------------------------- 32케이스
LINEAGE_CASE_IDS = [
    1, 2, 3, 4, 5, 6,
    10, 11, 12, 13,
    20, 21, 22,
    30, 31, 32, 33, 34, 35, 36, 37, 38, 39,
    40, 41, 42, 43, 44, 45, 46, 47, 48,
]


def _lineage_cases(lin: SimpleNamespace) -> dict:
    """(라벨, 명령, cwd, 기대판정) — 번호는 /tmp/t3017 검증 당시 케이스 번호를 유지한다."""
    repo = lin.repo
    norepo = lin.norepo
    return {
        # ---------- 계보 불일치 → DENY ----------
        1: ("계보 불일치 contents PUT (branch 명시)",
            f"gh api -X PUT repos/{SLUG}/contents/a.txt -f message=m -f content=eA== "
            f"-f sha={BLOB} -f branch=remote-branch", repo, DENY),
        2: ("계보 불일치 (branch 미지정 → 기본브랜치 조회)",
            f"gh api --method PUT repos/{SLUG}/contents/a.txt -f message=m "
            f"-f content=eA== -f sha={BLOB}", repo, DENY),
        3: ("계보 불일치 (owner 리네임 — repo 이름으로 매칭)",
            f"gh api -X PUT repos/{RENAMED_SLUG}/contents/a.txt -f sha={BLOB} "
            "-f branch=remote-branch", repo, DENY),
        4: ("계보 불일치 (--method=PUT 철자)",
            f"gh api --method=PUT repos/{SLUG}/contents/a.txt -f sha={BLOB} "
            "-f branch=remote-branch", repo, DENY),
        5: ("계보 불일치 (heredoc 안 bash 본문)",
            "bash <<EOF" + NL
            + f"gh api -X PUT repos/{SLUG}/contents/a.txt -f sha={BLOB} "
              "-f branch=remote-branch" + NL + "EOF", repo, DENY),
        6: ("계보 불일치 (GH_TOKEN 주입돼 있어도 계보는 따로)",
            f'GH_TOKEN="$BOT_GITHUB_TOKEN" gh api -X PUT repos/{SLUG}/contents/a.txt '
            f"-f sha={BLOB} -f branch=remote-branch", repo, DENY),

        # ---------- 같은 계보 / 신규 생성 → ALLOW ----------
        10: ("같은 계보 (대상 HEAD == 로컬 HEAD)",
             f"gh api -X PUT repos/{SLUG}/contents/a.txt -f sha={BLOB} -f branch=main",
             repo, ALLOW),
        11: ("같은 계보 (대상 HEAD 가 로컬 HEAD 의 조상)",
             f"gh api -X PUT repos/{SLUG}/contents/a.txt -f sha={BLOB} -f branch=old",
             repo, ALLOW),
        12: ("신규 파일 생성 (sha= 없음)",
             f"gh api -X PUT repos/{SLUG}/contents/new.txt -f message=m -f content=eA== "
             "-f branch=remote-branch", repo, ALLOW),
        13: ("신규 파일 생성 (sha= 없음, branch 도 없음)",
             f"gh api -X PUT repos/{SLUG}/contents/new.txt -f message=m -f content=eA==",
             repo, ALLOW),

        # ---------- 판정 불가 → DENY (fail-closed) ----------
        20: ("로컬 저장소 못 찾음",
             f"gh api -X PUT repos/Someone/NoSuchRepoXYZ/contents/a.txt -f sha={BLOB} "
             "-f branch=main", norepo, DENY),
        21: ("대상 HEAD 가 로컬에 없음(fetch 필요)",
             f"gh api -X PUT repos/{SLUG}/contents/a.txt -f sha={BLOB} -f branch=ghost",
             repo, DENY),
        22: ("대상 브랜치 HEAD 조회 실패",
             f"gh api -X PUT repos/{SLUG}/contents/a.txt -f sha={BLOB} "
             "-f branch=does-not-exist", repo, DENY),

        # ---------- 대상 아닌 명령 무간섭 ----------
        30: ("gh pr view", "gh pr view 256", repo, ALLOW),
        31: ("gh api GET contents", f"gh api repos/{SLUG}/contents/a.txt", repo, ALLOW),
        32: ("gh api 명시 GET contents",
             f"gh api -X GET repos/{SLUG}/contents/a.txt", repo, ALLOW),
        33: ("ls", "ls -la", repo, ALLOW),
        34: ("git status", "git status", repo, ALLOW),
        35: ("gh api POST contents (대상 아님)",
             f"gh api -X POST repos/{SLUG}/contents/a.txt -f sha={BLOB}", repo, ALLOW),
        36: ("gh api PUT 이지만 다른 엔드포인트",
             f"gh api -X PUT repos/{SLUG}/collaborators/xyz -f permission=push",
             repo, ALLOW),
        37: ("gh pr 머지 CLI (이 훅의 범위 밖)", PR_MERGE_CLI, repo, ALLOW),
        38: ("echo 안의 문자열",
             "echo 'gh api -X PUT repos/o/r/contents/a.txt'", repo, ALLOW),
        39: ("복합 파이프라인 (gh 없음)", "cat a.txt | grep v1 && echo ok", repo, ALLOW),

        # ---------- pulls/{N}/merge ----------
        40: ("PR#256 (clean) — 정상 머지 통과",
             f"gh api -X PUT repos/{SLUG}/pulls/256/merge -f merge_method=squash",
             repo, ALLOW),
        41: ("PR#256 + 일치하는 sha",
             f"gh api -X PUT repos/{SLUG}/pulls/256/merge -f sha={'a' * 40} "
             "-f merge_method=squash", repo, ALLOW),
        42: ("PR#256 + 축약 sha 일치",
             f"gh api -X PUT repos/{SLUG}/pulls/256/merge -f sha={'a' * 12}",
             repo, ALLOW),
        43: ("PR#256 + 다른 sha → DENY",
             f"gh api -X PUT repos/{SLUG}/pulls/256/merge -f sha={'9' * 40}",
             repo, DENY),
        44: ("PR#300 mergeable_state=dirty → DENY",
             f"gh api -X PUT repos/{SLUG}/pulls/300/merge", repo, DENY),
        45: ("PR#301 mergeable_state=behind → DENY",
             f"gh api -X PUT repos/{SLUG}/pulls/301/merge", repo, DENY),
        46: ("PR#302 unstable → ALLOW",
             f"gh api -X PUT repos/{SLUG}/pulls/302/merge", repo, ALLOW),
        47: ("PR#303 blocked → ALLOW",
             f"gh api -X PUT repos/{SLUG}/pulls/303/merge", repo, ALLOW),
        48: ("PR 조회 실패 → ALLOW(경고만)",
             f"gh api -X PUT repos/{SLUG}/pulls/999/merge", repo, ALLOW),
    }


@pytest.mark.parametrize("num", LINEAGE_CASE_IDS, ids=[f"c{n}" for n in LINEAGE_CASE_IDS])
def test_lineage_guard_verdict(num, lineage):
    """계보가 어긋난 contents PUT/머지는 차단, 같은 계보·신규생성·비대상 명령은 통과해야 한다."""
    label, command, cwd, expected = _lineage_cases(lineage)[num]
    got = judge(command, cwd, lineage)
    assert got.verdict == expected, (
        f"#{num} {label}\ncmd={command!r}\ncwd={cwd}\n"
        f"기대={expected} 실제={got.verdict}\n사유={got.reason[:800]}"
    )


# --------------------------------------------------------------------------- 실사고 재현
def test_incident_20260825_stale_base_overwrite_is_blocked(lineage):
    """실사고 재현: stale base 에서 다른 계보 브랜치의 기존 파일을 통째 PUT → 차단돼야 한다.

    차단 사유에 '대상 브랜치에만 있는 커밋'이 실제로 열거되는지까지 확인한다
    (판정만 맞고 근거가 비면, 사람이 무시하고 강행하게 된다).
    """
    command = (
        'GH_TOKEN="$BOT_GITHUB_TOKEN" gh api -X PUT '
        f"repos/{SLUG}/contents/c.txt -f message=overwrite -f content=eA== "
        f"-f sha={BLOB} -f branch=remote-branch"
    )
    got = judge(command, lineage.repo, lineage)
    assert got.verdict == DENY, "stale base 덮어쓰기가 통과했다(사고 재발)"
    assert "cafe 8건" in got.reason, (
        f"덮어쓸 뻔한 커밋 목록이 사유에 없다: {got.reason[:800]}"
    )
    assert lineage.main_sha[:12] in got.reason, "내 base 커밋이 사유에 없다"
    assert lineage.remote_sha[:12] in got.reason, "대상 HEAD 커밋이 사유에 없다"


def test_incident_control_new_file_creation_is_allowed(lineage):
    """대조군: 덮어쓸 기존 내용이 없는 신규 파일 생성(sha 없음)은 통과해야 한다."""
    command = (
        'GH_TOKEN="$BOT_GITHUB_TOKEN" gh api -X PUT '
        f"repos/{SLUG}/contents/brand-new-t3017.txt -f message=create -f content=eA== "
        "-f branch=remote-branch"
    )
    got = judge(command, lineage.repo, lineage)
    assert got.verdict == ALLOW, f"신규 파일 생성이 차단됐다: {got.reason[:500]}"


def test_pr_lookup_failure_warns_but_allows(lineage):
    """PR 상태 조회가 실패하면 차단이 아니라 경고여야 한다(조회 장애로 운영이 멈추면 안 된다)."""
    got = judge(f"gh api -X PUT repos/{SLUG}/pulls/999/merge", lineage.repo, lineage)
    assert got.verdict == ALLOW
    assert "git-lineage-guard" in got.system_message, (
        f"조회 실패 경고가 사라졌다(회귀): {got.system_message!r}"
    )


# --------------------------------------------------------------------------- 불변식
def test_hook_file_is_untouched_by_this_suite():
    """이 스위트는 훅을 '읽기만' 해야 한다 — 실행 중 본문이 바뀌면 테스트 자체가 오염이다."""
    assert _DIGEST_AT_IMPORT is not None, f"수집 시점에 가드가 없었다(회귀): {HOOK}"
    assert hashlib.md5(HOOK.read_bytes()).hexdigest() == _DIGEST_AT_IMPORT, (
        "테스트 실행 중 가드 본문이 변경됐다"
    )