103 lines
5.4 KiB
Python
103 lines
5.4 KiB
Python
#!/usr/bin/env python3
|
||
import importlib.util
|
||
import json
|
||
from pathlib import Path
|
||
import tempfile
|
||
import unittest
|
||
|
||
SCRIPT = Path(__file__).with_name("public_mirror_agent.py")
|
||
ROOT = SCRIPT.parents[2]
|
||
SPEC = importlib.util.spec_from_file_location("public_mirror_agent", SCRIPT)
|
||
M = importlib.util.module_from_spec(SPEC)
|
||
SPEC.loader.exec_module(M)
|
||
|
||
|
||
def contribution(**overrides):
|
||
value = {
|
||
"contribution_id":"LA-SKILL-CONTRIB-20260908-001", "module_id":"SKILL-PUBLIC-TEST-001",
|
||
"name":"事实闭环", "version":"1.0.0", "contributor_human_id":"TCS-GL-TEST∞",
|
||
"contributor_persona_id":"PER-TEST-001", "summary":"核验事实", "usage":"输入来源并运行检查",
|
||
"capabilities":["FACT_CHECK"], "provenance":{"source_kind":"PERSONA_SKILL_DERIVATION","source_id":"PRIVATE-SNAPSHOT-HASH-ONLY"},
|
||
"persona_share_decision":"SHARE", "contains_private_human_data":True, "human_data_consent":"GRANTED",
|
||
"public_payload":{"instructions":"不要读取 /Users/private/memory.json；联系 a@example.com", "input_schema":{"source":"string"}, "private_memory":"never"}
|
||
}
|
||
value.update(overrides)
|
||
return value
|
||
|
||
|
||
def review_event(item, kind, reviewer):
|
||
return {"module_id":item["module_id"], "contribution_id":item["contribution_id"], "candidate_sha256":item["candidate_sha256"], "decision":"ACCEPT", "reviewer_id":reviewer, "reviewer_kind":"TCS_MOTHER" if kind == "mother" else "GUANGHU_HUMAN_TEAM", "reviewer_authority_root":"TCS-MOTHER-LPM-0001" if kind == "mother" else "TCS-0002∞"}
|
||
|
||
|
||
class MirrorAgentTest(unittest.TestCase):
|
||
def test_architecture_mirror_allowlists_and_has_no_private_path(self):
|
||
with tempfile.TemporaryDirectory() as temp:
|
||
result = M.sync_architecture(ROOT, Path(temp))
|
||
self.assertEqual(result["outcome"], "PASS")
|
||
body = (Path(temp) / "architecture/CURRENT.json").read_text()
|
||
self.assertNotIn("/Volumes/", body)
|
||
self.assertNotIn("continuity-memory", body)
|
||
unchanged = M.sync_architecture(ROOT, Path(temp))
|
||
self.assertEqual(unchanged["state"], "MIRROR_PUBLIC_ARCHITECTURE_UNCHANGED")
|
||
|
||
def test_no_persona_decision_no_candidate(self):
|
||
with tempfile.TemporaryDirectory() as temp:
|
||
with self.assertRaisesRegex(M.MirrorError, "PERSONA_EXPLICIT"):
|
||
M.prepare_contribution(contribution(persona_share_decision="DO_NOT_SHARE"), Path(temp))
|
||
|
||
def test_private_human_data_requires_human_consent(self):
|
||
with tempfile.TemporaryDirectory() as temp:
|
||
with self.assertRaisesRegex(M.MirrorError, "HUMAN_DATA_CONSENT"):
|
||
M.prepare_contribution(contribution(human_data_consent="NOT_GRANTED"), Path(temp))
|
||
|
||
def test_sanitized_candidate_is_only_quarantined(self):
|
||
with tempfile.TemporaryDirectory() as temp:
|
||
state = Path(temp)
|
||
item = M.prepare_contribution(contribution(), state)
|
||
body = json.dumps(item, ensure_ascii=False)
|
||
self.assertNotIn("/Users/private", body)
|
||
self.assertNotIn("a@example.com", body)
|
||
self.assertFalse((state / "catalog/CURRENT.json").exists())
|
||
|
||
def test_both_independent_reviews_required_and_hash_bound(self):
|
||
with tempfile.TemporaryDirectory() as temp:
|
||
state = Path(temp)
|
||
item = M.prepare_contribution(contribution(), state)
|
||
with self.assertRaisesRegex(M.MirrorError, "MOTHER_ACCEPT"):
|
||
M.review(review_event(item, "team", "TEAM-1"), state, "team")
|
||
M.review(review_event(item, "mother", "MOTHER-1"), state, "mother")
|
||
M.review(review_event(item, "team", "TEAM-1"), state, "team")
|
||
record = M.register(item["module_id"], item["contribution_id"], state)
|
||
self.assertEqual(record["state"], "ARRIVAL_REGISTERED_LOCAL_NOT_PUBLISHED")
|
||
self.assertFalse(record["runtime_enabled"])
|
||
|
||
def test_same_reviewer_cannot_fill_both_gates(self):
|
||
with tempfile.TemporaryDirectory() as temp:
|
||
state = Path(temp)
|
||
item = M.prepare_contribution(contribution(), state)
|
||
M.review(review_event(item, "mother", "SAME"), state, "mother")
|
||
M.review(review_event(item, "team", "SAME"), state, "team")
|
||
with self.assertRaisesRegex(M.MirrorError, "INDEPENDENT_REVIEWERS"):
|
||
M.register(item["module_id"], item["contribution_id"], state)
|
||
|
||
def test_review_command_cannot_change_registered_authority_root(self):
|
||
with tempfile.TemporaryDirectory() as temp:
|
||
state = Path(temp)
|
||
item = M.prepare_contribution(contribution(), state)
|
||
event = review_event(item, "mother", "MOTHER-1")
|
||
event["reviewer_authority_root"] = "TCS-0002∞"
|
||
with self.assertRaisesRegex(M.MirrorError, "AUTHORITY_ROOT"):
|
||
M.review(event, state, "mother")
|
||
|
||
def test_hash_drift_and_replay_rejected(self):
|
||
with tempfile.TemporaryDirectory() as temp:
|
||
state = Path(temp)
|
||
item = M.prepare_contribution(contribution(), state)
|
||
with self.assertRaisesRegex(M.MirrorError, "CANDIDATE_HASH"):
|
||
M.review({**review_event(item, "mother", "M"), "candidate_sha256":"0"*64}, state, "mother")
|
||
with self.assertRaisesRegex(M.MirrorError, "REPLAY"):
|
||
M.prepare_contribution(contribution(summary="changed"), state)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|