69 lines
3.2 KiB
Python
69 lines
3.2 KiB
Python
"""Transport evidence and demand survive storage reopen without resending."""
|
|
import json
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
from agent_control.store import Store
|
|
from agent_control.conversation import project
|
|
from test_decision_workflow import bind
|
|
from test_decisions import definition
|
|
|
|
|
|
class DemandTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp=tempfile.TemporaryDirectory()
|
|
self.path=Path(self.tmp.name)/'state.sqlite3'
|
|
self.store=Store(self.path)
|
|
self.store.decision(definition())
|
|
bind(self.store)
|
|
|
|
def tearDown(self):
|
|
self.tmp.cleanup()
|
|
|
|
def demand(self):
|
|
return Store(self.path).decision_bindings()['bindings'][0]['demand']
|
|
|
|
def test_persisted_queue_and_expired_claims_have_distinct_demand(self):
|
|
m=self.store.send(dict(session_id='consult',text='Question',idempotency_key='one'))
|
|
self.assertEqual(m['delivery_stage'],'queued')
|
|
self.assertEqual(self.demand()['queued'],1)
|
|
self.assertEqual(self.demand()['human_queued'],1)
|
|
self.assertIsNotNone(self.demand()['oldest_queued_at'])
|
|
claim=self.store.claim(dict(session_id='consult',worker_id='worker',lease_seconds=10))['message']
|
|
self.assertEqual(self.demand()['inflight'],1)
|
|
with self.store.db() as db:
|
|
db.execute('UPDATE messages SET lease_until=? WHERE id=?',(time.time()-1,m['id']))
|
|
self.assertEqual(self.demand()['queued'],1)
|
|
with self.store.db() as db:
|
|
db.execute('UPDATE messages SET started=1 WHERE id=?',(m['id'],))
|
|
self.assertEqual(self.demand()['uncertain'],1)
|
|
self.assertEqual(self.demand()['queued'],0)
|
|
self.assertEqual(Store(self.path).send(dict(session_id='consult',text='Question',idempotency_key='one'))['id'],m['id'])
|
|
self.assertEqual(len(self.store.messages('consult')),1)
|
|
|
|
def test_safe_projection_uses_host_evidence_not_local_start(self):
|
|
m=self.store.send(dict(session_id='consult',text='Question',idempotency_key='one'))
|
|
stages=[('claimed',None,None,'queued'),
|
|
('claimed',dict(state='submitting'),None,'queued'),
|
|
('claimed',dict(state='accepted',receipt_id='private'),None,'delivered'),
|
|
('claimed',dict(state='observed',turn_id='private'),None,'processing'),
|
|
('uncertain',dict(state='accepted'),None,'blocked'),
|
|
('completed',dict(state='observed'),'reply','replied')]
|
|
for status,submission,response,expected in stages:
|
|
with self.store.db() as db:
|
|
db.execute('UPDATE messages SET status=?,started=1,submission=?,response_id=? WHERE id=?',
|
|
(status,json.dumps(submission) if submission else None,response,m['id']))
|
|
view=Store(self.path).conversation('consult')[0]
|
|
self.assertEqual(view['delivery_stage'],expected)
|
|
self.assertNotIn('submission',view)
|
|
self.assertNotIn('private',json.dumps(view))
|
|
|
|
def test_mirrored_history_does_not_fabricate_delivery_proof(self):
|
|
row=dict(id='history',text='Question',source='human',origin='session_history',status='delivered',created=1)
|
|
self.assertNotIn('delivery_stage',project([row],[])[0])
|
|
|
|
|
|
if __name__=='__main__':
|
|
unittest.main()
|