101 lines
5.3 KiB
Python
101 lines
5.3 KiB
Python
"""Decision bridge boundaries without spawning a model or modifying host sessions."""
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'scripts'))
|
|
import decision
|
|
import decision_fleet
|
|
from worker import atomic_json
|
|
|
|
|
|
class Process:
|
|
pid=42
|
|
code=None
|
|
def poll(self):return self.code
|
|
def terminate(self):self.code=0
|
|
def wait(self,timeout=None):return self.code
|
|
def kill(self):self.code=-1
|
|
|
|
|
|
class DecisionBridgeTests(unittest.TestCase):
|
|
def fixture(self,root):
|
|
cfg={'control':{},'codex':'codex','state_file':str(Path(root)/'worker.json'),
|
|
'decision_fleet':{'runtime_root':str(Path(root)/'decisions'),'enabled':True,'max_active':1}}
|
|
binding={'decision_id':'d','session_id':'s','thread_id':'11111111-1111-4111-8111-111111111111',
|
|
'worktree':root,'status':'open','archive_status':'none'}
|
|
return cfg,binding
|
|
|
|
def test_only_bound_threads_and_bounded_startup_are_bridged(self):
|
|
with tempfile.TemporaryDirectory() as root:
|
|
cfg,binding=self.fixture(root);fleet=decision_fleet.Fleet(cfg)
|
|
second={**binding,'decision_id':'other','session_id':'other','thread_id':'22222222-2222-4222-8222-222222222222'}
|
|
with patch.object(decision_fleet.subprocess,'Popen',side_effect=lambda *a,**k:Process()) as spawn:
|
|
status=fleet.tick([binding,second]);fleet.tick([binding,second])
|
|
self.assertEqual(spawn.call_count,2)
|
|
self.assertEqual(status['waiting'],1)
|
|
self.assertEqual(status['active'],1)
|
|
self.assertTrue(all('app-server' not in call.args[0] for call in spawn.call_args_list))
|
|
self.assertEqual(fleet.children['d']['cfg']['thread_id'],binding['thread_id'])
|
|
self.assertNotIn('notification_targets',fleet.children['d']['cfg'])
|
|
fleet.close()
|
|
|
|
def test_binding_change_and_duplicate_target_never_redirect_running_work(self):
|
|
with tempfile.TemporaryDirectory() as root:
|
|
cfg,b=self.fixture(root);fleet=decision_fleet.Fleet(cfg)
|
|
with patch.object(decision_fleet.subprocess,'Popen',side_effect=lambda *a,**k:Process()) as spawn:
|
|
fleet.tick([b]);old=fleet.children['d']['cfg']['thread_id']
|
|
result=fleet.tick([{**b,'thread_id':'22222222-2222-4222-8222-222222222222'}])
|
|
self.assertEqual(result['status'],'failed')
|
|
self.assertEqual(spawn.call_count,2)
|
|
self.assertEqual(fleet.children['d']['cfg']['thread_id'],old)
|
|
with self.assertRaises(ValueError):fleet.tick([b,{**b,'decision_id':'other'}])
|
|
fleet.close()
|
|
|
|
def test_archive_waits_for_pending_completion_and_does_not_kill_model(self):
|
|
with tempfile.TemporaryDirectory() as root:
|
|
cfg,b=self.fixture(root);fleet=decision_fleet.Fleet(cfg)
|
|
with patch.object(decision_fleet.subprocess,'Popen',side_effect=lambda *a,**k:Process()):
|
|
fleet.tick([b]);child=fleet.children['d'];state=Path(child['cfg']['state_file'])
|
|
atomic_json(state,{'inflight':{'message_id':'message'}})
|
|
result=fleet.tick([{**b,'status':'completed','archive_status':'archived'}])
|
|
self.assertEqual(result['status'],'failed')
|
|
self.assertIn('d',fleet.children)
|
|
self.assertTrue(all(p.poll() is None for p in child['processes'].values()))
|
|
atomic_json(state,{})
|
|
fleet.tick([{**b,'status':'completed','archive_status':'archived'}])
|
|
self.assertEqual(fleet.children,{})
|
|
self.assertTrue(state.exists())
|
|
|
|
def test_unknown_inflight_is_never_restarted(self):
|
|
with tempfile.TemporaryDirectory() as root:
|
|
cfg,b=self.fixture(root);fleet=decision_fleet.Fleet(cfg)
|
|
with patch.object(decision_fleet.subprocess,'Popen',side_effect=lambda *a,**k:Process()) as spawn:
|
|
fleet.tick([b]);child=fleet.children['d']
|
|
atomic_json(Path(child['cfg']['state_file']),{'needs_reconciliation':True})
|
|
child['processes']['worker'].code=1
|
|
self.assertEqual(fleet.tick([b])['status'],'failed')
|
|
self.assertEqual(spawn.call_count,2)
|
|
fleet.close()
|
|
|
|
def test_consultation_helper_cannot_choose_other_decision_or_review(self):
|
|
calls=[]
|
|
class Client:
|
|
def __init__(self,*args):pass
|
|
def call(self,*args):calls.append(args);return {}
|
|
cfg={'control':{},'decision_id':'decision-one'}
|
|
with patch.object(decision,'Client',Client):
|
|
decision.execute(cfg,'show')
|
|
decision.execute(cfg,'ask',{'question':'What is the dependency?'})
|
|
decision.execute(cfg,'ask',{'question':'What is the dependency?'})
|
|
self.assertEqual(calls[1][1]['idempotency_key'],calls[2][1]['idempotency_key'])
|
|
self.assertEqual(calls[0][0],'/api/decisions/decision-one')
|
|
with self.assertRaises(ValueError):decision.execute(cfg,'review',{})
|
|
with self.assertRaises(ValueError):decision.execute({'control':{},'decision_id':'../other'},'show')
|
|
with self.assertRaises(ValueError):decision.execute(cfg,'edit',{'title':'No revision'})
|
|
|
|
|
|
if __name__=='__main__':unittest.main()
|