muse-agent-example/.claude/skills/replay-eval/scripts/test_check_snapshot.py

#!/usr/bin/env python3
"""check_snapshot.py 的无网络离线测试。"""

import pathlib
import sys
import unittest

sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent))
from check_snapshot import (  # noqa: E402
    STATUS_BLOCKED_AUTHORIZATION,
    STATUS_INVALID_ARM_DIFF,
    STATUS_READY,
    STATUS_SCHEMA_INVALID,
    STATUS_TARGET_SOURCE_FORBIDDEN,
    check_arm_manifests,
    check_authorization,
    check_candidate_output,
    check_replay,
    check_target_sources,
)


AUTH = {
    "sourceStatus": "active",
    "copyrightStatus": "licensed",
    "sourceVersion": "work-v1",
    "allowedPurpose": ["offline_evaluation"],
    "authorizationSnapshot": {
        "id": "auth-1",
        "version": "v1",
        "immutable": True,
        "sourceVersion": "work-v1",
        "sourceStatus": "active",
        "allowedPurpose": ["offline_evaluation"],
        "checkedAt": "2026-07-19T00:00:00Z",
        "revalidationAt": "2026-07-20T00:00:00Z",
    },
}


def arm(common, cards):
    return {**common, "cardInjection": cards}


class CheckSnapshotTest(unittest.TestCase):
    def test_authorization_is_fail_closed(self):
        self.assertEqual(check_authorization(None)["status"], STATUS_BLOCKED_AUTHORIZATION)
        self.assertTrue(check_authorization(AUTH)["ok"])
        self.assertEqual(check_authorization({"sourceStatus": "active"})["status"], STATUS_BLOCKED_AUTHORIZATION)
        denied = {**AUTH, "allowedPurpose": ["read"]}
        self.assertEqual(check_authorization(denied)["status"], STATUS_BLOCKED_AUTHORIZATION)
        unknown_status = {**AUTH, "sourceStatus": "temporary"}
        self.assertEqual(check_authorization(unknown_status)["status"], STATUS_BLOCKED_AUTHORIZATION)
        unlicensed = {**AUTH, "copyrightStatus": "unlicensed"}
        self.assertEqual(check_authorization(unlicensed)["status"], STATUS_BLOCKED_AUTHORIZATION)

    def test_target_source_is_blocked(self):
        allowed = check_target_sources(
            489,
            [
                {"sourceId": "chapter-488", "sourceVersion": "v1", "chapter": 488},
                {"sourceId": "outline-450-482", "sourceVersion": "v1", "from_order": 450, "to_order": 482},
            ],
        )
        self.assertEqual(allowed["status"], STATUS_READY)
        blocked = check_target_sources(489, [{"sourceId": "chapter-489", "sourceVersion": "v1", "chapter": 489}])
        self.assertEqual(blocked["status"], STATUS_TARGET_SOURCE_FORBIDDEN)

    def test_arm_common_input_must_match(self):
        common = {"snapshotVersion": "v0", "asOfChapter": 488, "l0": {"target": 489}}
        manifests = {
            "outline_only": arm(common, []),
            "outline_plus_cards": arm(common, [{"id": 1}]),
            "outline_plus_placebo_cards": arm(common, [{"id": 2}]),
        }
        self.assertTrue(check_arm_manifests(manifests)["ok"])
        extra = {**manifests, "unregistered_arm": arm(common, [])}
        self.assertEqual(check_arm_manifests(extra)["status"], STATUS_INVALID_ARM_DIFF)
        changed = dict(manifests)
        changed["outline_plus_cards"] = arm({**common, "l0": {"target": 490}}, [{"id": 1}])
        self.assertEqual(check_arm_manifests(changed)["status"], STATUS_INVALID_ARM_DIFF)

    def test_two_arm_smoke_can_be_explicit(self):
        common = {"snapshotVersion": "v0", "asOfChapter": 488}
        manifests = {"outline_only": arm(common, []), "outline_plus_cards": arm(common, [{"id": 1}])}
        self.assertTrue(check_arm_manifests(manifests, ["outline_only", "outline_plus_cards"], smoke=True)["ok"])

    def test_candidate_contract_is_structural(self):
        candidate = {
            "targetChapter": 489,
            "chapterGoal": "突破",
            "keyEvents": [],
            "entities": [],
            "foreshadowing": [],
            "stateChanges": [],
            "hook": "悬念",
            "unknowns": [],
            "assumptions": [],
        }
        self.assertTrue(check_candidate_output(candidate, 489)["ok"])
        bad = {**candidate, "正文全文": "原文"}
        self.assertEqual(check_candidate_output(bad, 489)["status"], STATUS_SCHEMA_INVALID)
        bad_type = {**candidate, "keyEvents": "事件"}
        self.assertEqual(check_candidate_output(bad_type, 489)["status"], STATUS_SCHEMA_INVALID)
        duplicate = {**candidate, "keyEvents": [{"id": "event-1"}, {"id": "event-1"}]}
        self.assertEqual(check_candidate_output(duplicate, 489)["status"], STATUS_SCHEMA_INVALID)
        missing_id = {**candidate, "keyEvents": [{"event": "没有 id"}]}
        self.assertEqual(check_candidate_output(missing_id, 489)["status"], STATUS_SCHEMA_INVALID)
        unknown_field = {**candidate, "futureSources": ["target-scaffold"]}
        self.assertEqual(check_candidate_output(unknown_field, 489)["status"], STATUS_SCHEMA_INVALID)
        future_ref = {**candidate, "sourceRefs": ["target-scaffold"]}
        self.assertEqual(
            check_candidate_output(
                future_ref,
                489,
                [{"sourceId": "target-scaffold", "chapter": 489}],
            )["status"],
            STATUS_TARGET_SOURCE_FORBIDDEN,
        )

    def test_replay_fails_closed_before_model(self):
        common = {"snapshotVersion": "v0", "asOfChapter": 488}
        manifests = {
            "outline_only": arm(common, []),
            "outline_plus_cards": arm(common, [{"id": 1}]),
            "outline_plus_placebo_cards": arm(common, [{"id": 2}]),
        }
        result = check_replay(
            authorization=AUTH,
            as_of_chapter=488,
            target_chapter=489,
            planner_sources=[{"chapter": 489}],
            arm_manifests=manifests,
        )
        self.assertFalse(result["ok"])
        self.assertEqual(result["status"], STATUS_TARGET_SOURCE_FORBIDDEN)

    def test_production_arm_semantics_and_chapter_binding(self):
        common = {"snapshotVersion": "v0", "asOfChapter": 488, "targetChapter": 489}
        manifests = {
            "outline_only": {
                **common,
                "cardInjectionCount": 0,
                "cardSourceIds": [],
                "cardStrategy": "none",
            },
            "outline_plus_cards": {
                **common,
                "cardInjectionCount": 1,
                "cardSourceIds": ["correct-1"],
                "cardStrategy": "correct",
            },
            "outline_plus_placebo_cards": {
                **common,
                "cardInjectionCount": 1,
                "cardSourceIds": ["placebo-1"],
                "cardStrategy": "placebo",
            },
        }
        self.assertTrue(
            check_replay(
                authorization=AUTH,
                as_of_chapter=488,
                target_chapter=489,
                planner_sources=[{"sourceId": "history-488", "sourceVersion": "v1", "chapter": 488}],
                arm_manifests=manifests,
            )["ok"]
        )
        wrong = {**manifests, "outline_only": {**manifests["outline_only"], "cardInjectionCount": 1}}
        self.assertEqual(
            check_replay(
                authorization=AUTH,
                as_of_chapter=488,
                target_chapter=489,
                planner_sources=[{"sourceId": "history-488", "sourceVersion": "v1", "chapter": 488}],
                arm_manifests=wrong,
            )["status"],
            STATUS_INVALID_ARM_DIFF,
        )

    def test_missing_source_chapter_is_blocked(self):
        result = check_target_sources(489, [{"sourceId": "unknown", "sourceVersion": "v1", "payload": "future"}])
        self.assertEqual(result["status"], STATUS_TARGET_SOURCE_FORBIDDEN)


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