# -*- coding: utf-8 -*-
"""
regression: scripts/verify_task2553_closed_accepted_2553plus59.py
            (task-2553+59 TRACK D — FINAL CLOSED_ACCEPTED marker, additive)

검증 축
-------
1. Layer A NO-CRON: AST 에 subprocess/os.system/Popen/cokacdir/cron-dispatch
   import·call·token 0.
2. read-only / additive: collect() 전후 frozen anchor 6모듈 + remaining-backlog
   3 원천 + 소비 산출물 byte-0 무변(SHA256 EQUAL).
3. write surface = §4 allowlist 한정 — 비-allowlist write 거부 · Track A/B/C
   네임스페이스 0 overlap.
4. CLOSED_ACCEPTED marker 정형 — +32~+55 scope · self-chain QUARANTINED /
   independent ANU authoritative · callback 독립 ANU key only.
5. 잔여 backlog 정직 보존(과장 0): LEGACY_PENDING_FALLBACK_FIRED_AFTER_
   CONVERGENCE(NON_BLOCKING) + DEAD_CODE_CLEANLINESS_CANDIDATE(LOW).
6. closeout 일관성 점검 실 산출물 기반 · hold_for_chair=false.
7. ★ MOCK-ONLY FAIL gate — 빈/mock root 에서 실 검증이 inconsistency 를
   *탐지* (blind PASS 아님). 실 entrypoint 강제(문서-only 금지).
8. 산출 파일 temp out-root 격리 생성 — 실 repo write 0(테스트 자체).
"""
import ast
import hashlib
import json
import os
import sys
import tempfile
import unittest

REPO = "/home/jay/workspace"
SCRIPT_REL = "scripts/verify_task2553_closed_accepted_2553plus59.py"
SCRIPT = os.path.join(REPO, SCRIPT_REL)

sys.path.insert(0, os.path.join(REPO, "scripts"))
import verify_task2553_closed_accepted_2553plus59 as V  # noqa: E402


def _sha(path):
    h = hashlib.sha256()
    with open(path, "rb") as fh:
        for ch in iter(lambda: fh.read(65536), b""):
            h.update(ch)
    return h.hexdigest()


class LayerANoCronTest(unittest.TestCase):
    def test_01_no_subprocess_cokacdir_cron_dispatch(self):
        src = open(SCRIPT, "r", encoding="utf-8").read()
        tree = ast.parse(src)
        banned_import = {"subprocess", "pty"}
        for node in ast.walk(tree):
            if isinstance(node, ast.Import):
                for n in node.names:
                    self.assertNotIn(n.name.split(".")[0], banned_import,
                                     f"banned import {n.name}")
            if isinstance(node, ast.ImportFrom):
                root = (node.module or "").split(".")[0]
                self.assertNotIn(root, banned_import,
                                 f"banned from-import {node.module}")
            if isinstance(node, ast.Call) and isinstance(node.func,
                                                         ast.Attribute):
                self.assertNotIn(node.func.attr,
                                 {"system", "Popen", "popen", "spawn",
                                  "spawnv", "fork", "exec"},
                                 f"banned call attr {node.func.attr}")
        for tok in ("subprocess.run", "os.system(", "--sendfile", "--cron",
                    "--key fedf78d1d09509f5"):
            self.assertNotIn(tok, src, f"banned token present: {tok}")
        # cokacdir 실행 토큰 0 (경로 문자열 자체 미등장)
        self.assertNotIn("/usr/local/bin/cokacdir", src)


class ReadOnlyAdditiveTest(unittest.TestCase):
    def test_02_frozen_and_sources_byte0_unchanged_after_collect(self):
        targets = list(V.FROZEN_BYTE0.keys())
        targets += [r for r in V.CONSUMED_PINNED
                    if os.path.isfile(os.path.join(REPO, r))]
        targets += [r for r in V.CONSUMED_PRESENT_ONLY
                    if os.path.isfile(os.path.join(REPO, r))]
        before = {r: _sha(os.path.join(REPO, r)) for r in targets}
        produced = V.collect(REPO)
        after = {r: _sha(os.path.join(REPO, r)) for r in targets}
        self.assertEqual(before, after,
                         "read-only/additive 위반 — 소스 산출물 변경됨")
        self.assertIsInstance(produced, dict)

    def test_03_frozen_byte0_all_equal_expected(self):
        fz = V.verify_frozen_byte0(REPO)
        self.assertEqual(len(fz), 6)
        for rel, v in fz.items():
            self.assertTrue(v["present"], f"{rel} missing")
            self.assertTrue(v["equal"], f"{rel} byte-0 drift {v}")

    def test_04_remaining_backlog_sources_pinned_equal(self):
        cb = V.verify_consumed_byte0(REPO)
        self.assertTrue(cb["pinned_all_equal"],
                        f"remaining-backlog 원천 byte-0 drift: {cb['pinned']}")
        # 3 원천(primary backlog + addendum + deadcode-low) 전수 핀 일치
        self.assertEqual(len(V.CONSUMED_PINNED), 3)
        for rel, v in cb["pinned"].items():
            self.assertTrue(v["equal"], f"{rel} drift {v}")


class AllowlistWriteSurfaceTest(unittest.TestCase):
    def test_05_reject_non_allowlisted_write(self):
        with self.assertRaises(RuntimeError):
            V._assert_allowlisted("memory/events/NOT-ALLOWED.json")
        for rel in V.ALLOWLIST:
            V._assert_allowlisted(rel)

    def test_06_write_outputs_only_allowlisted_temp_isolated(self):
        produced = V.collect(REPO)
        with tempfile.TemporaryDirectory() as td:
            written = V.write_outputs(td, produced)
            for rel in written:
                self.assertIn(rel, V.ALLOWLIST)
                self.assertTrue(os.path.isfile(os.path.join(td, rel)))
            self.assertGreaterEqual(len(written), 6)
        # 실 repo 의 +59 산출물은 본 테스트가 미접촉(temp 격리 입증)
        self.assertNotEqual(td, REPO)

    def test_07_track_abc_disjoint_no_overlap(self):
        for rel in V.ALLOWLIST:
            low = rel.lower()
            self.assertNotIn("tracka", low)
            self.assertNotIn("trackb", low)
            self.assertNotIn("trackc", low)
            self.assertIn("2553plus59", low) if rel.startswith(
                ("scripts/", "tests/")) else None


class MarkerShapeTest(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.out = V.collect(REPO)
        cls.marker = cls.out["memory/events/task-2553.closed-accepted.json"]
        cls.result = cls.out["memory/events/task-2553+59.result.json"]
        cls.decision = cls.out["memory/events/task-2553+59.decision.json"]

    def test_08_closed_accepted_scope_covers_32_to_55(self):
        refs = " ".join(s["ref"] for s in self.marker["scope_closed_accepted"])
        # 회장 §3 verbatim 라벨("+44/+46", "+37/+45") 기준
        for need in ("+32", "+37", "+44", "+46", "+47", "+48", "+49",
                     "+50~+52", "+53", "+54", "+55"):
            self.assertIn(need, refs, f"scope 누락: {need}")
        self.assertEqual(self.marker["round_disposition"], "CLOSED_ACCEPTED")
        self.assertEqual(self.marker["marker"],
                         "TASK_2553_FINAL_CLOSED_ACCEPTED_MARKER")

    def test_09_selfchain_quarantined_independent_anu_authoritative(self):
        sc = json.dumps(self.marker["self_chain_vs_independent_anu"],
                        ensure_ascii=False)
        self.assertIn("QUARANTINED", sc)
        self.assertIn("c119085addb0f8b7", sc)
        self.assertIn("fail-closed", sc)

    def test_10_callback_independent_anu_key_only(self):
        ca = self.marker["callback_a"]
        self.assertEqual(ca["collector_owner_key"], "c119085addb0f8b7")
        self.assertEqual(ca["collector_role"], "ANU")
        self.assertEqual(ca["executor_self_key_forbidden"],
                         "fedf78d1d09509f5")
        self.assertIn("c119085addb0f8b7", ca["rule"])
        self.assertIn("fedf78d1d09509f5", ca["rule"])
        # decision/result 도 동일 계약
        self.assertEqual(
            self.decision["callback_a"]["collector_owner_key"],
            "c119085addb0f8b7")

    def test_11_no_overclaim_backlog_preserved(self):
        bl = self.out[
            "memory/events/task-2553.remaining-backlog.final_260518.json"]
        cr = bl["consolidated_remaining"]
        nb_ids = {x["id"] for x in cr["non_blocking"]}
        low_ids = {x["id"] for x in cr["low"]}
        self.assertIn("LEGACY_PENDING_FALLBACK_FIRED_AFTER_CONVERGENCE",
                      nb_ids)
        self.assertIn("DEAD_CODE_CLEANLINESS_CANDIDATE", low_ids)
        # 원천 무수정 임베드 — verbatim 보존 + sha256 기록
        for k in ("primary_backlog", "legacy_pending_fallback_addendum",
                  "deadcode_cleanliness_low"):
            self.assertIn(k, bl["source_files"])
            self.assertIsNotNone(
                bl["source_files"][k]["embedded_verbatim"],
                f"{k} 원천 임베드 누락")
            self.assertIsNotNone(bl["source_files"][k]["sha256"])
        self.assertTrue(self.marker["pending_overclaim_zero"])
        self.assertTrue(bl["overclaim_zero"]["principle"])

    def test_12_closeout_consistent_and_no_hold(self):
        cons = self.result["closeout_consistency"]
        self.assertTrue(cons["all_consistent"],
                        f"closeout 불일치: {cons['checks']}")
        self.assertFalse(self.marker["hold_for_chair"])
        self.assertFalse(self.result["hold_for_chair"])
        for k, v in self.marker["hold_assessment"]["triggers"].items():
            self.assertFalse(v, f"HOLD 트리거 적중: {k}")
        inv = self.result["invariants"]
        self.assertTrue(inv["git_branch_matches_expected"])
        self.assertTrue(inv["git_head_matches_expected"])
        self.assertTrue(inv["frozen_byte0_all_equal"])
        self.assertTrue(inv["additive_only"])

    def test_13_not_documentation_only(self):
        self.assertFalse(self.result["documentation_only"])
        self.assertEqual(self.result["real_entrypoint"], SCRIPT_REL)
        self.assertIn("mock-only FAIL", self.result["regression"])
        sa = self.result["executor_self_actions"]
        self.assertEqual(sum(sa.values()), 0, "executor self-* 비-0")


class MockOnlyFailGateTest(unittest.TestCase):
    """★ 실 entrypoint 강제 — mock/빈 root 에서 blind PASS 가 아님을 입증."""

    def test_14_empty_root_detects_inconsistency_not_blind_pass(self):
        with tempfile.TemporaryDirectory() as empty_root:
            produced = V.collect(empty_root)
            marker = produced["memory/events/task-2553.closed-accepted.json"]
            res = produced["memory/events/task-2553+59.result.json"]
            # mock-only(빈 root): 소비 산출물·frozen 부재 → 일관성 FAIL +
            # HOLD 적중해야 한다(검증기가 실제로 동작한다는 증거).
            self.assertFalse(
                res["closeout_consistency"]["all_consistent"],
                "mock-only 인데 일관성 PASS — blind rubber-stamp 의심")
            self.assertTrue(
                marker["hold_for_chair"],
                "mock-only 인데 HOLD 미적중 — 실 검증 아님(문서-only)")
            self.assertFalse(marker["invariants"]["frozen_byte0_all_equal"])

    def test_15_real_root_passes_only_because_real_artifacts_exist(self):
        # 실 repo: 동일 코드 경로가 PASS — 즉 결과가 입력 의존(하드코딩 아님)
        real = V.collect(REPO)
        res = real["memory/events/task-2553+59.result.json"]
        marker = real["memory/events/task-2553.closed-accepted.json"]
        self.assertTrue(res["closeout_consistency"]["all_consistent"])
        self.assertFalse(marker["hold_for_chair"])


if __name__ == "__main__":
    unittest.main(verbosity=2)
