#!/usr/bin/env python3
"""test_post186_callback_finalize_dogfood_2729p11.py

PR#186 audit 하니스 회귀 테스트.
테스터: 하누만 (개발4팀)
task: 2729+11

대상 하니스:
    scripts/harness/v36/post186_callback_finalize_dogfood_audit.py

규칙:
    - 읽기 전용 + 부작용 0 (g4-fix-loop-count 쓰기 금지, .done 생성 금지,
      canonical write 금지)
    - raw ANU key 출력 금지 (분할 상수로만 참조)
"""
from __future__ import annotations

import json
import subprocess
import sys
from pathlib import Path

import pytest

# ---------------------------------------------------------------------------
# 경로 상수
# ---------------------------------------------------------------------------

WORKTREE = Path("/home/jay/workspace/.worktrees/task-2729+11-dev4")
HARNESS = WORKTREE / "scripts/harness/v36/post186_callback_finalize_dogfood_audit.py"
GUARD_186 = WORKTREE / "scripts/start_task_guard.py"
WORKSPACE = Path("/home/jay/workspace")

# raw ANU key 는 테스트 코드에 하드코딩 금지 — 분할 상수로 패턴만 보유
_ANU_KEY_PART_A = "c1190"
_ANU_KEY_PART_B = "85addb0f8b7"
_ANU_KEY_PATTERN = _ANU_KEY_PART_A + _ANU_KEY_PART_B  # 매칭 전용, 출력 금지


# ---------------------------------------------------------------------------
# 헬퍼
# ---------------------------------------------------------------------------

def _run_harness_json(timeout: int = 120) -> dict:
    """하니스를 --json 모드로 실행하고 파싱된 dict 반환."""
    proc = subprocess.run(
        [sys.executable, str(HARNESS), "--json"],
        cwd=str(WORKTREE),
        capture_output=True,
        text=True,
        timeout=timeout,
    )
    return {
        "returncode": proc.returncode,
        "stdout": proc.stdout,
        "stderr": proc.stderr,
        "data": json.loads(proc.stdout) if proc.returncode == 0 and proc.stdout.strip() else {},
    }


# ---------------------------------------------------------------------------
# 픽스처: 하니스 결과를 세션 단위로 1회만 실행 (속도 최적화)
# ---------------------------------------------------------------------------

@pytest.fixture(scope="session")
def harness_result() -> dict:
    """하니스 --json 실행 결과 (세션 전체 공유)."""
    return _run_harness_json()


# ---------------------------------------------------------------------------
# TEST 1: 하니스 실행 + 최종 verdict
# ---------------------------------------------------------------------------

def test_harness_runs_and_verdict_pass(harness_result: dict) -> None:
    """하니스를 subprocess --json 실행 → exit 0 + final_verdict 확인."""
    assert harness_result["returncode"] == 0, (
        f"하니스 exit code != 0 (code={harness_result['returncode']})\n"
        f"stderr: {harness_result['stderr'][:500]}"
    )
    data = harness_result["data"]
    assert data, "--json stdout이 비어 있거나 파싱 불가"
    assert data.get("final_verdict") == "CALLBACK_FINALIZE_DOGFOOD_PASS_ACTIVE_FALSE", (
        f"final_verdict 불일치: {data.get('final_verdict')}"
    )


# ---------------------------------------------------------------------------
# TEST 2: check_a1 worktree-root PASS
# ---------------------------------------------------------------------------

def test_a1_worktree_root_pass(harness_result: dict) -> None:
    """check_a1 결과가 PASS 또는 STATIC_VERIFIED 여야 함.

    canonical parked 하에서 worktree-root #7이 정상 통과함을 확인.
    동적 PASS 또는 static_ok=True 중 하나 이상 충족해야 한다.
    """
    data = harness_result["data"]
    a1 = data.get("check_a1", {})

    verdict = a1.get("verdict")
    static_ok = a1.get("static", {}).get("static_ok", False)

    assert verdict in ("PASS", "STATIC_VERIFIED"), (
        f"A1 verdict 불일치: {verdict}"
    )
    # 동적 PASS 또는 정적 검증 OK 중 하나는 반드시 참
    dynamic_pass = a1.get("dynamic", {}).get("passed_check7", False)
    assert dynamic_pass or static_ok, (
        f"A1: 동적 PASS ({dynamic_pass}) 및 static_ok ({static_ok}) 모두 False"
    )


# ---------------------------------------------------------------------------
# TEST 3: #186 #7 섹션에 canonical branch 가정 없음을 파일 읽어 직접 검증
# ---------------------------------------------------------------------------

def test_a1_no_canonical_branch_assumption() -> None:
    """GUARD_186 소스에 #7 섹션이 cwd=cwd 기반 origin/main·HEAD 비교를 쓰고
    'main_current_branch != "main"' 패턴이 없음을 직접 검증.

    이것이 PR#186의 핵심 하드닝: canonical branch 상태와 무관하게
    worktree cwd 기준으로만 #7을 판단한다.
    """
    assert GUARD_186.exists(), f"GUARD_186 미존재: {GUARD_186}"
    source = GUARD_186.read_text(encoding="utf-8")

    # (a) #7 섹션에 cwd=cwd 기반 origin/main 비교 존재
    assert '["rev-parse", "origin/main"]' in source, (
        "#7 섹션에 origin/main rev-parse 없음"
    )
    assert "cwd=cwd" in source, (
        "cwd=cwd 패턴 없음 — worktree cwd 기준 실행이 아님"
    )
    assert "검증 #7" in source, (
        "검증 #7 섹션 주석 없음"
    )

    # (b) 구버전 canonical branch 가정 패턴이 없어야 함 (PR#186 핵심)
    assert 'main_current_branch != "main"' not in source, (
        "구버전 canonical branch 체크 패턴('main_current_branch != \"main\"')이 "
        "GUARD_186 소스에 잔존 — PR#186 하드닝 미적용"
    )

    # (c) HEAD vs origin/main SHA 비교 또는 merge-base 사용 확인
    has_sha_compare = (
        "head_sha == origin_main_sha" in source
        or "merge-base" in source
    )
    assert has_sha_compare, (
        "#7 섹션에 HEAD SHA vs origin/main 비교 또는 merge-base 없음"
    )

    # (d) WORKSPACE_ROOT 환경변수 기반 오버라이드 가능 확인
    assert 'os.environ.get("WORKSPACE_ROOT"' in source, (
        "WORKSPACE_ROOT env 기반 오버라이드 없음"
    )


# ---------------------------------------------------------------------------
# TEST 4: check_a2 nullglob-safe 검증
# ---------------------------------------------------------------------------

def test_a2_nullglob_safe(harness_result: dict) -> None:
    """check_a2 nullglob_safe 결과 PASS assert.

    - worktree 디렉토리 부재 시 _WT_ISOLATED=0(fail-closed)
    - 존재 시 _WT_ISOLATED=1
    """
    data = harness_result["data"]
    a2 = data.get("check_a2", {})
    assert a2.get("check_a2_nullglob_safe") == "PASS", (
        f"A2 nullglob_safe != PASS: {a2.get('check_a2_nullglob_safe')}\n"
        f"nullglob_results: {a2.get('nullglob_results')}"
    )
    # nullglob_results 세부 검증
    nr = a2.get("nullglob_results", {})
    assert nr.get("wt_exists_isolated_is_1") is True, (
        "worktree 존재 시 _WT_ISOLATED=1 아님"
    )
    assert nr.get("wt_absent_isolated_is_0_default") is True, (
        "worktree 부재 시 _WT_ISOLATED=0(fail-closed) 아님"
    )
    assert nr.get("wt_absent_nullglob_on_isolated_is_0") is True, (
        "nullglob ON에서 worktree 부재 시 _WT_ISOLATED=0 아님"
    )


# ---------------------------------------------------------------------------
# TEST 5: check_a2 EXTERNAL_DIRTY → EXEMPT
# ---------------------------------------------------------------------------

def test_a2_external_dirty_exempt(harness_result: dict) -> None:
    """EXTERNAL_DIRTY + worktree 격리(WT_ISOLATED=1) → EXEMPT 확인."""
    data = harness_result["data"]
    a2 = data.get("check_a2", {})
    assert a2.get("check_a2_external_exempt") == "PASS", (
        f"A2 external_exempt != PASS: {a2.get('check_a2_external_exempt')}"
    )
    # 매트릭스에서 EXEMPT 케이스 직접 확인
    matrix = a2.get("matrix", [])
    exempt_cases = [
        m for m in matrix
        if m.get("dirty_class") == "EXTERNAL_DIRTY_BLOCKER" and m.get("wt_isolated") == 1
    ]
    assert exempt_cases, "EXTERNAL_DIRTY + WT_ISOLATED=1 매트릭스 케이스 없음"
    for case in exempt_cases:
        assert case.get("actual") == "EXEMPT", (
            f"EXTERNAL_DIRTY + WT_ISOLATED=1 → 기대 EXEMPT, 실제: {case.get('actual')}"
        )


# ---------------------------------------------------------------------------
# TEST 6: check_a2 OWN_DIRTY → BLOCKED (면제 불가)
# ---------------------------------------------------------------------------

def test_a2_own_dirty_failclosed(harness_result: dict) -> None:
    """OWN_DIRTY(expected_files dirty) → BLOCKED assert (면제 불가)."""
    data = harness_result["data"]
    a2 = data.get("check_a2", {})
    assert a2.get("check_a2_own_failclosed") == "PASS", (
        f"A2 own_failclosed != PASS: {a2.get('check_a2_own_failclosed')}"
    )
    # 매트릭스에서 OWN_DIRTY BLOCKED 케이스 직접 확인
    matrix = a2.get("matrix", [])
    own_cases = [
        m for m in matrix
        if m.get("dirty_class") == "OWN_DIRTY_FAIL"
    ]
    assert own_cases, "OWN_DIRTY_FAIL 매트릭스 케이스 없음"
    for case in own_cases:
        assert case.get("actual") == "BLOCKED", (
            f"OWN_DIRTY → 기대 BLOCKED (면제 불가), 실제: {case.get('actual')}"
        )


# ---------------------------------------------------------------------------
# TEST 7: check_b G4 stale-vs-live 판정
# ---------------------------------------------------------------------------

def test_b_g4_stale_residue(harness_result: dict) -> None:
    """check_b verdict == STALE_RESIDUE + 증거 확인.

    - cap_before_pr186_merge == True
    - precedent에 'task-2729+1' 포함
    - counter reset 0 확인 (테스트가 g4-fix-loop-count 파일 절대 쓰지 않음)
    """
    data = harness_result["data"]
    b = data.get("check_b", {})

    assert b.get("check_b_g4_verdict") == "STALE_RESIDUE", (
        f"B verdict 불일치: {b.get('check_b_g4_verdict')}\n"
        f"reason: {b.get('verdict_reason', '')[:300]}"
    )

    evidence = b.get("evidence", {})

    # cap이 PR#186 머지 이전 발생했음을 확인
    assert evidence.get("cap_before_pr186_merge") is True, (
        f"cap_before_pr186_merge != True: {evidence.get('cap_before_pr186_merge')}"
    )

    # 선례에 task-2729+1 포함 확인
    precedent = evidence.get("precedent", "")
    assert "task-2729+1" in precedent, (
        f"precedent에 'task-2729+1' 없음: {precedent}"
    )

    # 하니스 자체가 counter를 reset하지 않는다는 증거: counter_reset == False
    assert b.get("counter_reset") is False, (
        "check_b.counter_reset이 False가 아님 — 하니스가 카운터를 초기화함!"
    )

    # 실제 g4-fix-loop-count 파일이 있다면 읽기만 했는지 확인
    # (이 테스트 자체도 해당 파일을 절대 수정하지 않는다)
    count_file = WORKSPACE / "memory" / "events" / "task-2729+10.g4-fix-loop-count"
    if count_file.exists():
        count_val = count_file.read_text(encoding="utf-8").strip()
        # 값이 비어있지 않아야 하며, 우리가 쓰지 않았으므로 원래 값 유지
        assert count_val != "", "g4-fix-loop-count 파일이 비어있음 (예상 외)"


# ---------------------------------------------------------------------------
# TEST 8: check_c seed vs robust 분리
# ---------------------------------------------------------------------------

def test_c_seed_vs_robust_separated(harness_result: dict) -> None:
    """check_c separation_verdict=="SEPARATED" + 분류 검증."""
    data = harness_result["data"]
    c = data.get("check_c", {})

    assert c.get("separation_verdict") == "SEPARATED", (
        f"C separation_verdict 불일치: {c.get('separation_verdict')}"
    )

    seed = c.get("check_c_seed", {})
    robust = c.get("check_c_robust", {})

    assert seed.get("classification") == "SEED_PLACEHOLDER", (
        f"seed classification 불일치: {seed.get('classification')}"
    )
    assert robust.get("classification") == "ROBUST_FRAMEWORK_REGISTRATION_REQUEST", (
        f"robust classification 불일치: {robust.get('classification')}"
    )

    # 세부 seed 조건 확인
    assert seed.get("is_seed_placeholder") is True, "seed.is_seed_placeholder != True"
    assert seed.get("registration_result_status") == "REGISTERED"
    assert seed.get("callback_delivery_status") == "PENDING"
    assert seed.get("collector_receipt_status") == "UNCONFIRMED"

    # 세부 robust 조건 확인
    assert robust.get("is_robust_framework") is True, "robust.is_robust_framework != True"
    assert robust.get("verdict") == "PASS"
    assert robust.get("enforcement_verdict") == "PASS"
    assert robust.get("owner_is_anu_key") is True


# ---------------------------------------------------------------------------
# TEST 9: raw ANU key 노출 없음 (stdout 전체 검사)
# ---------------------------------------------------------------------------

def test_no_anu_key_raw_leak() -> None:
    """하니스 --json stdout 전체에 raw ANU key가 나타나지 않음을 확인.

    REDACTED/마스킹만 허용. _ANU_KEY_PATTERN 분할 상수로 매칭.
    """
    proc = subprocess.run(
        [sys.executable, str(HARNESS), "--json"],
        cwd=str(WORKTREE),
        capture_output=True,
        text=True,
        timeout=120,
    )
    stdout = proc.stdout

    # raw key 패턴이 stdout에 나타나면 FAIL
    # (_ANU_KEY_PATTERN은 매칭용으로만 사용, 출력하지 않음)
    assert _ANU_KEY_PATTERN not in stdout, (
        "하니스 --json stdout에 raw ANU key 패턴이 노출됨 — 마스킹 누락"
    )

    # 추가: stderr에도 노출 없어야 함
    stderr = proc.stderr
    assert _ANU_KEY_PATTERN not in stderr, (
        "하니스 --json stderr에 raw ANU key 패턴이 노출됨"
    )


# ---------------------------------------------------------------------------
# TEST 10: canonical/done 부작용 0 확인
# ---------------------------------------------------------------------------

def test_no_canonical_or_done_side_effects(harness_result: dict) -> None:
    """하니스 footer에 부작용 0 assert.

    - manual_done_created == 0
    - canonical_write == 0
    - production_activation == 0
    - real_ANU_spawn == 0
    - ACTIVE == "false"
    """
    data = harness_result["data"]
    footer = data.get("footer", {})

    assert footer.get("manual_done_created") == 0, (
        f"footer.manual_done_created != 0: {footer.get('manual_done_created')}"
    )
    assert footer.get("canonical_write") == 0, (
        f"footer.canonical_write != 0: {footer.get('canonical_write')}"
    )
    assert footer.get("production_activation") == 0, (
        f"footer.production_activation != 0: {footer.get('production_activation')}"
    )
    assert footer.get("real_ANU_spawn") == 0, (
        f"footer.real_ANU_spawn != 0: {footer.get('real_ANU_spawn')}"
    )
    assert footer.get("ACTIVE") == "false", (
        f"footer.ACTIVE != 'false': {footer.get('ACTIVE')}"
    )

    # 최상위 active 필드도 False여야 함
    assert data.get("active") is False, (
        f"data.active != False: {data.get('active')}"
    )

    # 최상위 부작용 플래그들도 확인
    assert data.get("canonical_write") is False, (
        f"data.canonical_write != False: {data.get('canonical_write')}"
    )
    assert data.get("done_created") is False, (
        f"data.done_created != False: {data.get('done_created')}"
    )
    assert data.get("anu_spawned") is False, (
        f"data.anu_spawned != False: {data.get('anu_spawned')}"
    )
    assert data.get("production_activation") is False, (
        f"data.production_activation != False: {data.get('production_activation')}"
    )