122 lines
4.9 KiB
Python

#!/usr/bin/env python3
"""回放编排器和安全摘要的无网络测试。"""
from __future__ import annotations
import json
import os
import pathlib
import sys
import tempfile
import unittest
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent))
from run_replay import run_replay # noqa: E402
from write_report import render_report # noqa: E402
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 config():
common = {"l0": {"targetChapter": 489}, "l1": {"asOfChapter": 488}, "l2": {"mainline": "安全公共输入"}}
return {
"runId": "smoke-001",
"referenceWork": {"id": "deep-space", "version": "work-v1"},
"evaluationSetVersion": "set-v1",
"strategyVersion": "strategy-v1",
"authorization": AUTH,
"runPermissions": {"purpose": "offline_evaluation", "mode": "dry_run"},
"targetChapter": 489,
"snapshot": {
"asOfChapter": 488,
"snapshotVersion": "next_fine_outline_replay_v0",
"data": {
"milestones": [{"chapter": 488, "fact": "安全历史"}, {"chapter": 489, "fact": "未来"}],
"cards": [{"name": "已知实体", "milestones": [{"chapter": 488, "step": "历史"}]}],
},
},
"sources": [{"sourceId": "history-488", "sourceVersion": "v1", "chapter": 488}],
"commonContext": common,
"arms": {
"outline_only": {"cards": [], "cardSourceIds": [], "cardStrategy": "none"},
"outline_plus_cards": {"cards": [{"name": "正确卡"}], "cardSourceIds": ["card-correct-1"], "cardStrategy": "correct"},
"outline_plus_placebo_cards": {"cards": [{"name": "错配卡"}], "cardSourceIds": ["card-placebo-1"], "cardStrategy": "placebo"},
},
}
class ReplayRunTest(unittest.TestCase):
def test_dry_run_passes_and_writes_only_metadata(self):
with tempfile.TemporaryDirectory() as directory:
result = run_replay(config(), pathlib.Path(directory), mode="dry_run")
self.assertTrue(result["ok"])
self.assertEqual(result["status"], "ready")
manifest = json.loads((pathlib.Path(directory) / "snapshot_manifest.json").read_text())
self.assertNotIn("payload", json.dumps(manifest, ensure_ascii=False))
self.assertFalse(list(pathlib.Path(directory).glob("planner_*.raw.json")))
def test_bad_authorization_stops_before_model(self):
bad = config()
bad["authorization"] = {"sourceStatus": "active"}
with tempfile.TemporaryDirectory() as directory:
result = run_replay(bad, pathlib.Path(directory), mode="dry_run")
self.assertFalse(result["ok"])
self.assertEqual(result["status"], "blocked_authorization")
self.assertFalse((pathlib.Path(directory) / "snapshot_manifest.json").exists())
def test_report_contains_hashes_but_not_raw_fields(self):
result = run_replay(config(), pathlib.Path(tempfile.mkdtemp()), mode="dry_run")
report = render_report(result)
self.assertIn("smoke-001", report)
self.assertIn("snapshotManifestSha256", report)
self.assertNotIn('"prompt":', report)
self.assertNotIn('"payload":', report)
def test_execute_uses_external_planner_and_validates_each_candidate(self):
candidate = {
"targetChapter": 489,
"chapterGoal": "机制 smoke",
"keyEvents": [],
"entities": [],
"foreshadowing": [],
"stateChanges": [],
"hook": "下一步",
"unknowns": [],
"assumptions": [],
}
with tempfile.TemporaryDirectory() as directory:
fake = pathlib.Path(directory) / "fake-planner.py"
fake.write_text(
"#!/usr/bin/env python3\n"
"import json\n"
f"print(json.dumps({{'result': json.dumps({candidate!r}, ensure_ascii=False)}}))\n",
encoding="utf-8",
)
os.chmod(fake, 0o755)
output_dir = pathlib.Path(directory) / "run"
result = run_replay(config(), output_dir, mode="execute", planner_bin=str(fake))
self.assertTrue(result["ok"])
self.assertEqual(result["status"], "completed")
self.assertEqual(set(result["results"]), {"outline_only", "outline_plus_cards", "outline_plus_placebo_cards"})
self.assertTrue(list(output_dir.glob("candidate_*.json")))
if __name__ == "__main__":
unittest.main()