28 lines
2.1 KiB
Python
28 lines
2.1 KiB
Python
#!/usr/bin/env python3
|
|
import importlib.util,json,tempfile,unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
S=Path(__file__).with_name('command_router.py'); spec=importlib.util.spec_from_file_location('router',S); M=importlib.util.module_from_spec(spec); spec.loader.exec_module(M)
|
|
|
|
def event(persona='ICE-P-ZY001', session='s1'):
|
|
return {"parent_persona_id":persona,"session_id":session,"channel_id":"ICE-CH-HB001","current_intent_sha256":"a"*64,"model_runtime":"codex"}
|
|
|
|
class Tests(unittest.TestCase):
|
|
def test_every_slot_is_model_backed_and_owned_by_parent(self):
|
|
with tempfile.TemporaryDirectory() as t, patch.object(M,'runtime_available',return_value=True):
|
|
v=M.activate(event(),Path(t)); self.assertEqual(len(v['team']),4)
|
|
self.assertTrue(all(x['model_cognition_required'] and x['controller']=='ICE-P-ZY001' for x in v['team']))
|
|
def test_persona_switch_rebinds_all_slots_and_invalidates_generation(self):
|
|
personas=iter(sorted(M.registered_personas()))
|
|
first=next(personas); second=next(x for x in personas if x!=first)
|
|
with tempfile.TemporaryDirectory() as t, patch.object(M,'runtime_available',return_value=True):
|
|
state=Path(t); a=M.activate(event(first,'s1'),state); b=M.activate(event(second,'s2'),state)
|
|
self.assertEqual(b['generation'],a['generation']+1); self.assertTrue(b['prior_generation_invalidated'])
|
|
self.assertTrue(all(x['parent_persona_id']==second and second in x['child_agent_id'] for x in b['team']))
|
|
with self.assertRaisesRegex(M.CommandError,'STALE_TEAM_GENERATION'): M.dispatch({"generation":a['generation'],"child_agent_id":a['team'][0]['child_agent_id'],"task":"x"},state)
|
|
def test_unknown_persona_and_missing_provider_fail_closed(self):
|
|
with tempfile.TemporaryDirectory() as t:
|
|
with self.assertRaisesRegex(M.CommandError,'NOT_REGISTERED'): M.activate(event('PER-NOT-REAL'),Path(t))
|
|
with patch.object(M,'runtime_available',return_value=False), self.assertRaisesRegex(M.CommandError,'PROVIDER_UNAVAILABLE'): M.activate(event(),Path(t))
|
|
|
|
if __name__=='__main__': unittest.main()
|