63 lines
3.5 KiB
Python
63 lines
3.5 KiB
Python
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 bridge_service
|
|
import worker
|
|
|
|
class Process:
|
|
def __init__(self,code):self.code=code
|
|
def poll(self):return self.code
|
|
|
|
class RecoveryTests(unittest.TestCase):
|
|
def fixture(self,root):
|
|
path=Path(root)/'worker.json'
|
|
cfg={'control':{},'session_id':'session','thread_id':'thread','state_file':str(path)}
|
|
proof={'input_sha256':'hash','thread_id':'thread','turn_id':'turn'}
|
|
state={'inflight':{'message_id':'original','submission':proof},'needs_reconciliation':True}
|
|
worker.atomic_json(path,state)
|
|
rows=[{'id':'original','session_id':'session','status':'completed','response_id':'reply','submission':proof},
|
|
{'id':'reply','source':'agent','reply_to':'original','text':'answer'}]
|
|
return cfg,path,rows
|
|
def test_stopped_worker_terminal_proof_converges_without_submission(self):
|
|
with tempfile.TemporaryDirectory() as root:
|
|
cfg,path,rows=self.fixture(root)
|
|
with patch.object(bridge_service.Client,'call',return_value={'messages':rows}) as call:
|
|
self.assertTrue(bridge_service.recover_worker(cfg,Process(1)))
|
|
self.assertEqual(call.call_count,1);self.assertTrue(call.call_args.args[0].startswith('/api/messages?'))
|
|
state=json.loads(path.read_text());self.assertNotIn('inflight',state)
|
|
self.assertIn('original',state['delivery_receipts'])
|
|
def test_running_worker_and_locked_orphan_are_never_touched(self):
|
|
with tempfile.TemporaryDirectory() as root:
|
|
cfg,path,rows=self.fixture(root);before=path.read_bytes()
|
|
with patch.object(bridge_service.Client,'call') as call:
|
|
self.assertFalse(bridge_service.recover_worker(cfg,Process(None)));call.assert_not_called()
|
|
identity=json.dumps({'control':cfg['control'],'session_id':cfg['session_id']},sort_keys=True)
|
|
with worker.session_lock(identity):
|
|
self.assertFalse(bridge_service.recover_worker(cfg,Process(1)))
|
|
call.assert_not_called()
|
|
self.assertEqual(path.read_bytes(),before)
|
|
|
|
def test_unrelated_reply_cannot_release_the_queue(self):
|
|
with tempfile.TemporaryDirectory() as root:
|
|
cfg,path,rows=self.fixture(root);before=path.read_bytes()
|
|
rows[1]['reply_to']='other-input'
|
|
with patch.object(bridge_service.Client,'call',return_value={'messages':rows}):
|
|
self.assertFalse(bridge_service.recover_worker(cfg,Process(1)))
|
|
self.assertEqual(path.read_bytes(),before)
|
|
def test_offline_or_unmatched_proof_preserves_original(self):
|
|
with tempfile.TemporaryDirectory() as root:
|
|
cfg,path,rows=self.fixture(root);before=path.read_bytes()
|
|
with patch.object(bridge_service.Client,'call',side_effect=OSError('offline')):
|
|
self.assertFalse(bridge_service.recover_worker(cfg,Process(1)))
|
|
for field in ('input_sha256','thread_id','turn_id'):
|
|
changed=json.loads(json.dumps(rows));changed[0]['submission'][field]='other'
|
|
with patch.object(bridge_service.Client,'call',return_value={'messages':changed}):
|
|
self.assertFalse(bridge_service.recover_worker(cfg,Process(1)))
|
|
self.assertEqual(path.read_bytes(),before)
|
|
|
|
if __name__=='__main__':unittest.main()
|