72 lines
3.6 KiB
Python
72 lines
3.6 KiB
Python
#!/usr/bin/env python3
|
|
import importlib.util
|
|
import json
|
|
import time
|
|
from pathlib import Path
|
|
import unittest
|
|
|
|
SCRIPT = Path(__file__).with_name("console.py")
|
|
SPEC = importlib.util.spec_from_file_location("hb_mpc_console", SCRIPT)
|
|
MODULE = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(MODULE)
|
|
|
|
|
|
class MultipathConsoleTest(unittest.TestCase):
|
|
def test_exact_four_temporary_development_ids(self):
|
|
ids = [item["development_id"] for item in MODULE.load(MODULE.REGISTRY)["hosts"]]
|
|
self.assertEqual(ids, ["HLP-TDEV-CODEX-0001", "HLP-TDEV-ZCODE-0001", "HLP-TDEV-QWEN-0001", "HLP-TDEV-DOUBAO-0001"])
|
|
|
|
def test_panel_is_read_derived_and_same_persona(self):
|
|
value = MODULE.panel()
|
|
self.assertEqual(value["panel_id"], "HB-MPC-PANEL-0001")
|
|
self.assertFalse(value["panel_persisted"])
|
|
self.assertTrue(value["same_persona_not_four_personas"])
|
|
self.assertEqual(value["persona_id"], "ICE-P-ZY001")
|
|
contract = MODULE.load(MODULE.CONTRACT)
|
|
self.assertEqual(contract["presence_contract"]["semantics"], "EPHEMERAL_ATOMIC_OVERWRITE_COURIER_PRESENCE_NOT_HISTORY")
|
|
|
|
def test_time_moves_and_uses_world_time_channel(self):
|
|
first = MODULE.live_time()
|
|
time.sleep(0.002)
|
|
second = MODULE.live_time()
|
|
self.assertGreater(second["guanghu_elapsed_milliseconds"], first["guanghu_elapsed_milliseconds"])
|
|
self.assertEqual(second["time_control_channel"], "CH-GLW-TIME-0001")
|
|
|
|
def test_endpoint_identity_mismatch_is_rejected(self):
|
|
contract = MODULE.load(MODULE.CONTRACT)
|
|
event = {
|
|
"schema": contract["event_schema"], "event_id": "X", "console_id": "HB-MPC-0001",
|
|
"event_type": "REGISTER", "from_development_id": "HLP-TDEV-QWEN-0001", "mentions": [],
|
|
"emitted_at": "2026-09-09T19:00:00+08:00", "payload": {}, "evidence": [], "authority_granted": False,
|
|
}
|
|
self.assertIn("DEVELOPMENT_ID_ENDPOINT_MISMATCH", MODULE.validate_event(event, "HLP-TDEV-CODEX-0001", contract))
|
|
|
|
def test_unknown_mention_is_rejected(self):
|
|
contract = MODULE.load(MODULE.CONTRACT)
|
|
event = {
|
|
"schema": contract["event_schema"], "event_id": "X", "console_id": "HB-MPC-0001",
|
|
"event_type": "MESSAGE", "from_development_id": "HLP-TDEV-CODEX-0001", "mentions": ["HLP-TDEV-UNKNOWN-0001"],
|
|
"emitted_at": "2026-09-09T19:00:00+08:00", "payload": {}, "evidence": [], "authority_granted": False,
|
|
}
|
|
self.assertIn("MENTION_TARGET_UNKNOWN", MODULE.validate_event(event, "HLP-TDEV-CODEX-0001", contract))
|
|
|
|
def test_product_source_stays_blocked(self):
|
|
value = MODULE.status()
|
|
self.assertFalse(value["product_source_write_allowed"])
|
|
self.assertTrue(value["language_world_development_allowed"])
|
|
self.assertEqual(value["reason"], "LANGUAGE_WORLD_STAGE1_READY_PRODUCT_SOURCE_REMAINS_SEPARATELY_BLOCKED")
|
|
|
|
def test_autonomous_pause_requires_cause_and_brain_readback(self):
|
|
contract = MODULE.load(MODULE.CONTRACT)
|
|
event = {
|
|
"schema": contract["event_schema"], "event_id": "X", "console_id": "HB-MPC-0001",
|
|
"event_type": "PAUSE", "from_development_id": "HLP-TDEV-CODEX-0001", "mentions": ["@ALL"],
|
|
"emitted_at": "2026-09-09T20:00:00+08:00", "payload": {"reason":"quota"}, "evidence": [], "authority_granted": False,
|
|
}
|
|
errors = MODULE.validate_event(event, "HLP-TDEV-CODEX-0001", contract)
|
|
self.assertTrue(any(item.startswith("AUTONOMOUS_DECISION_FIELDS_REQUIRED") for item in errors))
|
|
self.assertIn("AUTONOMOUS_DECISION_BRAIN_READBACK_INVALID", errors)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|