"""
scripts/update-system-spec.py 봉인 테스트 (task-3056).

봉인 대상 3가지:
  1) 유령 변경 제거 — new_state 에 없는 섹션은 변경 판정에서 제외
  2) cron 섹션 철회 + 부재 선언 (마커는 생존, 함수는 생존)
  3) 5개 렌더 섹션의 수집시각 KST 헤더 (★ 렌더 출력에만. 수집 결과에는 금지)

★ 모든 파일 쓰기는 pytest tmp_path 안에서만 수행한다.
  실제 워크스페이스 파일은 읽기 전용으로만 접근한다.
"""

from __future__ import annotations

import copy
import hashlib
import importlib.util
import json
import os
import re
import shutil
import subprocess
import sys
from pathlib import Path

import pytest

WORKSPACE = Path("/home/jay/workspace")
SCRIPT_PATH = WORKSPACE / "scripts" / "update-system-spec.py"
REAL_SPEC = WORKSPACE / "memory" / "specs" / "anu-system-spec.md"

KST_HEADER_RE = re.compile(r"^_수집: \d{4}-\d{2}-\d{2} \d{2}:\d{2} KST_$")


def _load_module():
    """하이픈이 들어간 파일명이라 importlib 로 직접 로드한다."""
    spec = importlib.util.spec_from_file_location("update_system_spec_under_test", SCRIPT_PATH)
    assert spec is not None and spec.loader is not None
    mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(mod)
    return mod


@pytest.fixture(scope="module")
def mod():
    return _load_module()


def _sha256(p: Path) -> str:
    return hashlib.sha256(p.read_bytes()).hexdigest()


# ══════════════════════════════════════════════════════════════════════════════
# 1. 유령 변경 회귀 — new_state 에 없는 섹션은 항목을 만들지 않는다
# ══════════════════════════════════════════════════════════════════════════════


def test_missing_section_produces_no_ghost_change(mod):
    """수집 실패로 new_state 에 없는 섹션(cron)이 old_state 에만 있어도 변경 0건."""
    old_state = {
        "cron": {"cron_jobs": {"kind": "cron_list", "schedules": [{"id": "EE825F5E"}]}},
        "skills": {"skills": ["a", "b"], "count": 2},
    }
    new_state = {"skills": {"skills": ["a", "b"], "count": 2}}

    changes = mod.detect_changes(old_state, new_state)

    assert "cron" not in changes, f"유령 변경 발생: {changes}"
    assert changes == {}, f"변경 없어야 하는데 감지됨: {changes}"


def test_missing_arbitrary_section_produces_no_ghost_change(mod):
    """SECTIONS 에 있으나 이번 회차 수집이 실패한 임의 섹션도 동일하게 제외된다."""
    old_state = {
        "services": {"running_services": ["x.service"], "count": 1},
        "skills": {"skills": ["a"], "count": 1},
    }
    new_state = {"skills": {"skills": ["a"], "count": 1}}  # services 수집 실패 가정

    changes = mod.detect_changes(old_state, new_state)

    assert "services" not in changes, f"유령 변경 발생: {changes}"


# ══════════════════════════════════════════════════════════════════════════════
# 2. 조기종료 실증 — 2회차는 spec/changelog 를 재작성하지 않는다 (sha256 + mtime)
# ══════════════════════════════════════════════════════════════════════════════


def _build_sandbox(tmp_path: Path) -> Path:
    root = tmp_path / "ws"
    (root / "memory" / "specs").mkdir(parents=True)
    (root / "scripts").mkdir(parents=True)

    for name in ("anu-system-spec.md", ".spec-state-cache.json", "anu-system-spec-changelog.md"):
        src = WORKSPACE / "memory" / "specs" / name
        if src.exists():
            shutil.copy2(src, root / "memory" / "specs" / name)
    shutil.copy2(WORKSPACE / "memory" / "task-timers.json", root / "memory" / "task-timers.json")

    # collect_scripts() 가 스캔할 대상. 두 회차 사이에 변하지 않아야 한다.
    for name in ("alpha.py", "beta.sh"):
        (root / "scripts" / name).write_text("# fixture\n", encoding="utf-8")
    shutil.copy2(SCRIPT_PATH, root / "scripts" / SCRIPT_PATH.name)
    return root


def _run(root: Path) -> subprocess.CompletedProcess:
    env = dict(os.environ)
    env["WORKSPACE_ROOT"] = str(root)
    env.pop("ANU_KEY", None)
    env.pop("ANU_CHAT", None)
    return subprocess.run(
        [sys.executable, str(root / "scripts" / SCRIPT_PATH.name)],
        capture_output=True,
        text=True,
        env=env,
        timeout=180,
    )


def test_second_run_short_circuits_without_rewriting_files(tmp_path):
    root = _build_sandbox(tmp_path)
    spec = root / "memory" / "specs" / "anu-system-spec.md"
    changelog = root / "memory" / "specs" / "anu-system-spec-changelog.md"
    cache = root / "memory" / "specs" / ".spec-state-cache.json"

    r1 = _run(root)
    assert r1.returncode == 0, f"1회차 exit {r1.returncode}\nSTDOUT:{r1.stdout}\nSTDERR:{r1.stderr}"

    before = {
        "spec_sha": _sha256(spec),
        "spec_mtime": spec.stat().st_mtime_ns,
        "changelog_sha": _sha256(changelog),
        "changelog_mtime": changelog.stat().st_mtime_ns,
        "cache_sha": _sha256(cache),
        "cache_mtime": cache.stat().st_mtime_ns,
    }

    r2 = _run(root)
    assert r2.returncode == 0, f"2회차 exit {r2.returncode}\nSTDOUT:{r2.stdout}\nSTDERR:{r2.stderr}"
    assert "변경 없음 - 파일 업데이트 생략" in r2.stdout, f"조기종료 경로 미진입:\n{r2.stdout}"

    after = {
        "spec_sha": _sha256(spec),
        "spec_mtime": spec.stat().st_mtime_ns,
        "changelog_sha": _sha256(changelog),
        "changelog_mtime": changelog.stat().st_mtime_ns,
        "cache_sha": _sha256(cache),
        "cache_mtime": cache.stat().st_mtime_ns,
    }

    assert after["spec_sha"] == before["spec_sha"], "spec 파일이 재작성됨 (sha256 변화)"
    assert after["spec_mtime"] == before["spec_mtime"], "spec 파일이 재작성됨 (mtime 변화)"
    assert after["changelog_sha"] == before["changelog_sha"], "changelog 가 재작성됨 (sha256 변화)"
    assert after["changelog_mtime"] == before["changelog_mtime"], "changelog 가 재작성됨 (mtime 변화)"
    assert after["cache_sha"] == before["cache_sha"], "상태 캐시가 재작성됨 (sha256 변화)"
    assert after["cache_mtime"] == before["cache_mtime"], "상태 캐시가 재작성됨 (mtime 변화)"


def test_second_run_changelog_has_no_cron_entry(tmp_path):
    """조기종료 이전에도, 1회차 changelog 최신 항목에 [cron] 이 없어야 한다."""
    root = _build_sandbox(tmp_path)
    changelog = root / "memory" / "specs" / "anu-system-spec-changelog.md"

    r1 = _run(root)
    assert r1.returncode == 0, r1.stderr

    head = changelog.read_text(encoding="utf-8").split("\n\n", 1)[0]
    assert "[cron]" not in head, f"유령 [cron] 항목 재발:\n{head}"


# ══════════════════════════════════════════════════════════════════════════════
# 3. 정상 섹션 무손상 (양방향)
# ══════════════════════════════════════════════════════════════════════════════


def _skills(items):
    return {"skills": {"skills": list(items), "count": len(items)}}


def test_skills_addition_detected(mod):
    changes = mod.detect_changes(_skills(["a", "b"]), _skills(["a", "b", "c"]))
    assert "skills" in changes
    assert any("추가" in m and "`c`" in m for m in changes["skills"]), changes


def test_skills_removal_detected(mod):
    changes = mod.detect_changes(_skills(["a", "b", "c"]), _skills(["a", "b"]))
    assert "skills" in changes
    assert any("삭제" in m and "`c`" in m for m in changes["skills"]), changes


def test_no_change_returns_empty_dict(mod):
    assert mod.detect_changes(_skills(["a", "b"]), _skills(["a", "b"])) == {}


def test_new_section_creation_still_detected(mod):
    """old 에 없고 new 에 있는 경우 '신규 생성' 경로가 살아있어야 한다."""
    changes = mod.detect_changes({}, _skills(["a"]))
    assert "skills" in changes
    assert any("신규 생성" in m for m in changes["skills"]), changes


def test_scalar_dict_change_detected(mod):
    """task_stats 처럼 리스트가 아닌 dict 값 변경 → '데이터 변경' 경로 생존 확인."""
    old = {"task_stats": {"total_tasks": 10, "completed_tasks": 5}}
    new = {"task_stats": {"total_tasks": 11, "completed_tasks": 5}}
    changes = mod.detect_changes(old, new)
    assert "task_stats" in changes
    assert any("데이터 변경" in m for m in changes["task_stats"]), changes


# ══════════════════════════════════════════════════════════════════════════════
# 4. 부재 선언 + 마커 생존 (실제 문서, 읽기 전용)
# ══════════════════════════════════════════════════════════════════════════════


def _cron_block() -> str:
    text = REAL_SPEC.read_text(encoding="utf-8")
    start, end = "<!-- AUTO:cron:START -->", "<!-- AUTO:cron:END -->"
    i, j = text.find(start), text.find(end)
    assert i != -1, "AUTO:cron:START 마커 소실"
    assert j != -1, "AUTO:cron:END 마커 소실"
    assert i < j, "AUTO:cron 마커 순서 이상"
    return text[i + len(start) : j]


def test_cron_markers_survive():
    _cron_block()  # 마커 3조건은 위 assert 로 검증됨


def test_cron_block_has_no_april_json_residue():
    block = _cron_block()
    for token in ("EE825F5E", "8C2B2814", "8F9C55BC", "050ECF5D", "0BCB9FBD", "1063C2BC"):
        assert token not in block, f"4월 JSON 잔재 발견: {token}"
    assert "cron_list" not in block, "4월 JSON(cron_list) 잔재 발견"
    assert "schedule_type" not in block, "4월 JSON(schedule_type) 잔재 발견"


def test_cron_block_has_absence_declaration():
    block = _cron_block()
    assert "자동수집하지 않습니다" in block, block
    assert "부분집합" in block, block
    assert "schedule_history" in block, block


def test_cron_block_leaks_no_credential():
    """자격증명 값이 문서에 새지 않아야 한다."""
    block = _cron_block()
    assert "ANU_KEY" not in block
    assert not re.search(r"\b[0-9a-f]{16}\b", block), "16자리 hex 키 형태 문자열 노출"


# ══════════════════════════════════════════════════════════════════════════════
# 5. 수집시각 KST 헤더 — 5개 렌더 함수 출력 첫 줄
# ══════════════════════════════════════════════════════════════════════════════


RENDER_CASES = {
    "skills": {"skills": ["alpha"], "count": 1},
    "services": {"running_services": ["a.service"], "count": 1},
    "projects": {"projects": ["p1"], "count": 1},
    "scripts": {"scripts": ["s1.py"], "count": 1},
    "task_stats": {
        "total_tasks": 3,
        "completed_tasks": 2,
        "avg_duration_sec": 10.0,
        "max_duration_sec": 20.0,
        "min_duration_sec": 5.0,
    },
}

EMPTY_CASES = {
    "services": {"running_services": [], "count": 0},
    "projects": {"projects": [], "count": 0},
    "scripts": {"scripts": [], "count": 0},
}


@pytest.mark.parametrize("section", sorted(RENDER_CASES))
def test_render_first_line_is_kst_timestamp(mod, section):
    out = mod.RENDERERS[section](RENDER_CASES[section])
    first = out.splitlines()[0]
    assert "KST" in first, f"{section}: 첫 줄에 KST 없음 → {first!r}"
    assert KST_HEADER_RE.match(first), f"{section}: 수집시각 형식 불일치 → {first!r}"


@pytest.mark.parametrize("section", sorted(EMPTY_CASES))
def test_render_empty_branch_also_has_kst_header(mod, section):
    """빈 목록 조기 반환 경로에서도 수집시각이 빠지면 안 된다."""
    out = mod.RENDERERS[section](EMPTY_CASES[section])
    first = out.splitlines()[0]
    assert KST_HEADER_RE.match(first), f"{section}(빈 목록): {first!r}"


def test_kst_header_is_utc_plus_9_not_server_tz(mod):
    """서버 TZ 에 의존하지 않고 명시적으로 UTC+9 로 계산하는지."""
    import datetime as _dt

    expected = _dt.datetime.now(_dt.timezone.utc) + _dt.timedelta(hours=9)
    got = mod._collected_at_kst()
    assert got.endswith(" KST")
    got_dt = _dt.datetime.strptime(got[: -len(" KST")], "%Y-%m-%d %H:%M")
    assert abs((got_dt - expected.replace(tzinfo=None)).total_seconds()) < 120, (got, expected)


def test_timestamp_never_enters_collected_state(tmp_path):
    """★ 수집시각은 렌더 출력 전용. collect_* 반환값에 들어가면 조기종료가 다시 죽는다."""
    root = _build_sandbox(tmp_path)
    _run(root)  # 캐시 생성
    cache = json.loads((root / "memory" / "specs" / ".spec-state-cache.json").read_text(encoding="utf-8"))
    blob = json.dumps(cache, ensure_ascii=False)
    assert "KST" not in blob, "수집시각이 상태 캐시(new_state)에 유입됨"
    assert "수집:" not in blob, "수집시각이 상태 캐시(new_state)에 유입됨"


def test_renderer_does_not_mutate_input(mod):
    """렌더가 입력 dict 를 건드리면 new_state 가 오염된다."""
    for section, data in RENDER_CASES.items():
        payload = copy.deepcopy(data)
        mod.RENDERERS[section](payload)
        assert payload == data, f"{section}: 렌더가 입력을 변형함"


# ══════════════════════════════════════════════════════════════════════════════
# 6. SECTIONS 철회 + 함수 생존
# ══════════════════════════════════════════════════════════════════════════════


def test_cron_removed_from_sections(mod):
    assert "cron" not in mod.SECTIONS, mod.SECTIONS
    assert mod.SECTIONS == ["skills", "services", "projects", "scripts", "task_stats"]


def test_cron_functions_still_exist(mod):
    assert callable(getattr(mod, "collect_cron", None)), "collect_cron 삭제됨"
    assert callable(getattr(mod, "render_cron", None)), "render_cron 삭제됨"


def test_cron_collector_renderer_entries_preserved(mod):
    """되살릴 근거 보존: dict 엔트리는 남기되 SECTIONS 를 통해 호출되지 않는다."""
    assert "cron" in mod.COLLECTORS
    assert "cron" in mod.RENDERERS
    assert "cron" not in mod.SECTIONS


def test_anu_key_not_defined_to_revive_collection():
    """ANU_KEY/ANU_CHAT 를 스크립트가 스스로 채워 넣지 않아야 한다."""
    src = SCRIPT_PATH.read_text(encoding="utf-8")
    assert 'os.environ.get("ANU_KEY", "")' in src
    assert 'os.environ.get("ANU_CHAT", "")' in src


# ══════════════════════════════════════════════════════════════════════════════
# 7. 실제 실행 후 문서 파싱 — 5개 섹션 **전부** 블록 첫 줄에 KST 수집시각
#    (main() 이 changes 가 아닌 SECTIONS 를 순회해야만 통과한다)
# ══════════════════════════════════════════════════════════════════════════════

RENDERED_SECTIONS = ("skills", "services", "projects", "scripts", "task_stats")


def _block_body(text: str, section: str) -> str:
    start, end = f"<!-- AUTO:{section}:START -->", f"<!-- AUTO:{section}:END -->"
    i, j = text.find(start), text.find(end)
    assert i != -1, f"AUTO:{section}:START 마커 소실"
    assert j != -1, f"AUTO:{section}:END 마커 소실"
    assert i < j, f"AUTO:{section} 마커 순서 이상"
    return text[i + len(start) : j].strip("\n")


def test_all_five_sections_get_kst_header_after_real_run(tmp_path):
    """실행 후 문서에서 직접 파싱: 5개 섹션 전부 블록 첫 줄이 KST 수집시각이어야 한다."""
    root = _build_sandbox(tmp_path)
    spec = root / "memory" / "specs" / "anu-system-spec.md"

    r1 = _run(root)
    assert r1.returncode == 0, f"exit {r1.returncode}\nSTDOUT:{r1.stdout}\nSTDERR:{r1.stderr}"
    assert "업데이트 완료" in r1.stdout, f"갱신 경로 미진입:\n{r1.stdout}"

    text = spec.read_text(encoding="utf-8")
    missing = []
    for section in RENDERED_SECTIONS:
        body = _block_body(text, section)
        first = body.splitlines()[0] if body else "(empty)"
        if not KST_HEADER_RE.match(first):
            missing.append((section, first))
    assert not missing, f"KST 수집시각 없는 섹션: {missing}"


def test_cron_block_untouched_by_full_section_sweep(tmp_path):
    """SECTIONS 전체 순회로 바뀌어도 cron 블록(부재 선언)은 건드리지 않는다."""
    root = _build_sandbox(tmp_path)
    spec = root / "memory" / "specs" / "anu-system-spec.md"

    before = _block_body(spec.read_text(encoding="utf-8"), "cron")
    r1 = _run(root)
    assert r1.returncode == 0, r1.stderr
    after = _block_body(spec.read_text(encoding="utf-8"), "cron")

    assert after == before, "cron 블록이 변경됨"
    assert "자동수집하지 않습니다" in after, after


def test_failed_section_does_not_raise_keyerror_in_sweep(tmp_path):
    """수집 실패 섹션이 있어도 렌더 순회가 KeyError 로 죽지 않는다 (가드 생존)."""
    root = _build_sandbox(tmp_path)
    # projects 수집을 실패시킨다 → new_state 에 'projects' 키가 없다
    (root / "scripts" / SCRIPT_PATH.name).write_text(
        SCRIPT_PATH.read_text(encoding="utf-8").replace(
            'PROJECTS_DIR = Path("/home/jay/projects/")',
            'PROJECTS_DIR = Path("/nonexistent/task-3056/projects/")',
            1,
        ),
        encoding="utf-8",
    )
    r = _run(root)

    assert "KeyError" not in r.stderr, r.stderr
    assert "[FAIL] projects" in r.stderr, r.stderr
    assert r.returncode == 1, f"수집 실패 회차는 exit 1 이어야 한다: {r.returncode}"

    text = (root / "memory" / "specs" / "anu-system-spec.md").read_text(encoding="utf-8")
    for section in ("skills", "services", "scripts", "task_stats"):
        first = _block_body(text, section).splitlines()[0]
        assert KST_HEADER_RE.match(first), f"{section}: {first!r}"