219 lines
12 KiB
Python
219 lines
12 KiB
Python
"""Service-owned queue diagnostics; no claims about a host private queue."""
|
|
import http.client
|
|
import concurrent.futures
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
from unittest.mock import patch
|
|
from agent_control.store import Store, Invalid, Conflict
|
|
from agent_control.server import Server
|
|
|
|
|
|
def firing(**changes):
|
|
data=dict(source='synthetic',event_id='event',severity='warning',status='firing',summary='Synthetic failure')
|
|
data.update(changes)
|
|
return data
|
|
|
|
|
|
class QueueTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp=tempfile.TemporaryDirectory()
|
|
self.store=Store(Path(self.tmp.name)/'queue.sqlite3')
|
|
|
|
def tearDown(self):
|
|
self.tmp.cleanup()
|
|
|
|
def test_empty_real_counts_and_explicit_scope(self):
|
|
health=self.store.queue_health(100)
|
|
self.assertEqual(health['scope'],'service_owned_outboxes')
|
|
self.assertEqual(health['delivery_semantics'],'host_accepted_not_model_completed')
|
|
for queue in ['hooks','eta','decisions']:
|
|
self.assertEqual(health[queue]['pending_count'],0)
|
|
self.assertIsNone(health[queue]['oldest_pending_at'])
|
|
self.assertEqual(health[queue]['created_last_60s'],0)
|
|
self.assertEqual(health[queue]['delivered_last_60s'],0)
|
|
|
|
def test_sql_counts_timestamps_and_receipts_are_real(self):
|
|
with patch('agent_control.store.time.time',return_value=10):
|
|
first=self.store.hook(firing())
|
|
self.store.session(dict(id='worker',name='Worker'))
|
|
self.store.task(dict(id='task',name='Task',session_id='worker',
|
|
eta_min_minutes=1,eta_max_minutes=2,eta_basis='Measured'))
|
|
self.store.decision(dict(id='decision',title='Choose',question='Which approach?'))
|
|
for notice in self.store.decision_notifications()['notifications']:
|
|
self.store.decision_ack(dict(notification_id=notice['notification_id']))
|
|
self.store.eta_tick(70)
|
|
with patch('agent_control.store.time.time',return_value=80):
|
|
self.store.hook(firing(event_id='recent',severity='critical'))
|
|
self.store.hook_ack(dict(notification_id=first['notifications'][0]['notification_id']))
|
|
self.store.decision_respond('decision',dict(option_id=None,comment='Selected',idempotency_key='choice',expected_revision=1))
|
|
self.store.task_report(dict(report_id='receipt',task_id='task',session_id='worker',
|
|
stage='progress',current_progress='Verified',summary='Progress',evidence=[]))
|
|
health=self.store.queue_health(100)
|
|
self.assertEqual(health['hooks']['pending_count'],2)
|
|
self.assertEqual(health['hooks']['oldest_pending_at'],80)
|
|
self.assertEqual(health['hooks']['created_last_60s'],2)
|
|
self.assertEqual(health['hooks']['delivered_last_60s'],1)
|
|
self.assertEqual(health['eta']['pending_count'],1)
|
|
self.assertEqual(health['eta']['oldest_pending_at'],70)
|
|
self.assertEqual(health['decisions']['pending_count'],1)
|
|
self.assertEqual(health['decisions']['oldest_pending_at'],80)
|
|
self.assertEqual(health['task_reports'],dict(received_total=1,received_last_60s=1))
|
|
with patch('agent_control.store.time.time',return_value=101):
|
|
self.store.task(dict(status='cancelled'),'task')
|
|
self.assertEqual(self.store.queue_health(102)['eta']['cancelled_count'],1)
|
|
self.assertEqual(self.store.queue_health(102)['eta']['pending_count'],0)
|
|
|
|
def test_coalesce_resolved_backlog_idempotent_without_delivery_claim(self):
|
|
with patch('agent_control.store.time.time',return_value=10):
|
|
first=self.store.hook(firing())
|
|
for _ in range(10):
|
|
self.store.hook(firing())
|
|
with patch('agent_control.store.time.time',return_value=20):
|
|
self.store.hook(firing(status='resolved'))
|
|
before=self.store.state()['hooks']['recent']
|
|
with patch('agent_control.store.time.time',return_value=30):
|
|
result=self.store.hook_coalesce({})
|
|
self.assertEqual(result['superseded_count'],2)
|
|
self.assertEqual(self.store.hook_notifications()['pending_count'],0)
|
|
self.assertEqual(self.store.hook_coalesce({})['superseded_count'],0)
|
|
self.assertEqual(self.store.state()['hooks']['recent'],before)
|
|
self.assertEqual(self.store.queue_health(40)['hooks']['delivered_last_60s'],0)
|
|
self.assertEqual(self.store.queue_health(40)['hooks']['superseded_count'],2)
|
|
repeated=self.store.hook(firing())
|
|
self.assertEqual(repeated['notifications'][0]['status'],'superseded')
|
|
self.assertEqual(len(self.store.state()['events']),2)
|
|
|
|
def test_coalesce_keeps_active_major_and_recovery_for_delivered_target(self):
|
|
warning=self.store.hook(firing())
|
|
self.store.hook_ack(dict(notification_id=warning['notifications'][0]['notification_id']))
|
|
self.store.hook_escalate(dict(source='synthetic',event_id='event',reason='Persistent'))
|
|
self.store.hook(firing(status='resolved'))
|
|
active=self.store.hook(firing(event_id='active-major',severity='critical'))
|
|
self.store.hook_coalesce({})
|
|
notices=self.store.hook_notifications()['notifications']
|
|
recoveries=[n for n in notices if n['hook']['event_id']=='event']
|
|
self.assertEqual(len(recoveries),1)
|
|
self.assertEqual(recoveries[0]['target'],'control')
|
|
self.assertEqual(recoveries[0]['hook']['status'],'resolved')
|
|
self.assertEqual(sum(n['hook']['event_id']=='active-major' for n in notices),2)
|
|
|
|
def test_coalesce_does_not_touch_user_decision_or_message(self):
|
|
self.store.session(dict(id='s',name='Synthetic'))
|
|
self.store.decision(dict(id='d',title='Choose',question='Which approach?'))
|
|
self.store.decision_respond('d',dict(option_id=None,comment='User choice',idempotency_key='choice',expected_revision=1))
|
|
before=self.store.decision_notifications()
|
|
self.store.send(dict(session_id='s',text='Real user request',idempotency_key='user-message'))
|
|
messages=self.store.messages('s')
|
|
self.store.hook(firing())
|
|
self.store.hook(firing(status='resolved'))
|
|
self.store.hook_coalesce({})
|
|
self.assertEqual(self.store.decision_notifications(),before)
|
|
self.assertEqual(self.store.messages('s'),messages)
|
|
with self.assertRaises(Invalid): self.store.hook_coalesce({'clear_all':True})
|
|
|
|
def test_unknown_submission_marker_protects_firing_and_its_recovery(self):
|
|
first=self.store.hook(firing())
|
|
nid=first['notifications'][0]['notification_id']
|
|
with patch('agent_control.store.time.time',return_value=10):
|
|
marker=self.store.hook_submitting(dict(notification_id=nid))
|
|
with patch('agent_control.store.time.time',return_value=20):
|
|
again=self.store.hook_submitting(dict(notification_id=nid))
|
|
self.assertFalse(marker['duplicate'])
|
|
self.assertTrue(again['duplicate'])
|
|
self.assertEqual(again['submission_started_at'],10)
|
|
self.assertEqual(again['status'],'submitting')
|
|
self.store.hook(firing(status='resolved'))
|
|
self.assertEqual(self.store.hook_coalesce({})['superseded_count'],0)
|
|
self.assertEqual(self.store.hook_notifications()['pending_count'],2)
|
|
health=self.store.queue_health(30)['hooks']
|
|
self.assertEqual(health['submission_started_unconfirmed_count'],1)
|
|
self.assertEqual(health['delivered_last_60s'],0)
|
|
self.assertEqual(self.store.hook_notifications()['notifications'][0]['submission_started_at'],10)
|
|
|
|
def test_superseded_cannot_begin_submission_but_real_late_ack_is_recorded(self):
|
|
first=self.store.hook(firing())
|
|
nid=first['notifications'][0]['notification_id']
|
|
self.store.hook(firing(status='resolved'))
|
|
self.store.hook_coalesce({})
|
|
with self.assertRaises(Conflict): self.store.hook_submitting(dict(notification_id=nid))
|
|
# An actual accepted receipt can arrive late; never erase that fact.
|
|
self.store.hook_ack(dict(notification_id=nid))
|
|
self.assertEqual(self.store.hook(firing())['notifications'][0]['status'],'delivered')
|
|
with self.assertRaises(Invalid): self.store.hook_submitting(dict(notification_id='missing'))
|
|
with self.assertRaises(Invalid): self.store.hook_submitting(dict(notification_id=nid,accepted=True))
|
|
|
|
def test_submission_and_coalesce_are_serialized(self):
|
|
notice=self.store.hook(firing())['notifications'][0]
|
|
self.store.hook(firing(status='resolved'))
|
|
def submit():
|
|
try:
|
|
return self.store.hook_submitting(dict(notification_id=notice['notification_id']))
|
|
except Conflict:
|
|
return None
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
|
|
attempt=pool.submit(submit)
|
|
cleanup=pool.submit(self.store.hook_coalesce,{})
|
|
result=attempt.result()
|
|
cleanup.result()
|
|
with self.store.db() as db:
|
|
row=db.execute('SELECT * FROM hook_notifications WHERE id=?',(notice['notification_id'],)).fetchone()
|
|
if result is not None:
|
|
self.assertIsNotNone(row['submission_started_at'])
|
|
self.assertIsNone(row['superseded_at'])
|
|
else:
|
|
self.assertIsNone(row['submission_started_at'])
|
|
self.assertIsNotNone(row['superseded_at'])
|
|
|
|
|
|
class QueueHTTPTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
root = Path(self.tmp.name)
|
|
root.chmod(0o700)
|
|
token = root / 'token'
|
|
self.secret = 'h'*48
|
|
token.write_text(self.secret)
|
|
token.chmod(0o600)
|
|
self.metrics = patch('agent_control.server.MetricsCache')
|
|
mock = self.metrics.start()
|
|
mock.return_value.snapshot.return_value = {'status':'synthetic'}
|
|
self.server = Server(dict(bind='127.0.0.1',port=0,runtime_root=str(root),token_file=str(token)))
|
|
self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
|
|
self.thread.start()
|
|
|
|
def tearDown(self):
|
|
self.server.shutdown()
|
|
self.server.server_close()
|
|
self.thread.join()
|
|
self.metrics.stop()
|
|
self.tmp.cleanup()
|
|
|
|
def request(self,path,method='GET',payload=None,headers=None):
|
|
conn = http.client.HTTPConnection(*self.server.server_address,timeout=3)
|
|
merged = {'Authorization':'Bearer '+self.secret,'X-CSRF-Token':'1','Content-Type':'application/json'}
|
|
merged.update(headers or {})
|
|
conn.request(method,path,json.dumps(payload) if payload is not None else None,merged)
|
|
response = conn.getresponse()
|
|
status, body = response.status,json.loads(response.read())
|
|
conn.close()
|
|
return status, body
|
|
|
|
|
|
def test_security_and_coalesce_routes(self):
|
|
first=self.server.store.hook(firing())
|
|
self.server.store.hook(firing(status='resolved'))
|
|
self.assertEqual(self.request('/api/queue-health',headers={'Authorization':''})[0],401)
|
|
self.assertEqual(self.request('/api/queue-health?x=1')[0],404)
|
|
for path,body in [('/api/hook-notifications/coalesce',{}),
|
|
('/api/hook-notifications/submitting',dict(notification_id=first['notifications'][0]['notification_id']))]:
|
|
for headers,status in [({'Authorization':''},401),({'X-CSRF-Token':''},403),
|
|
({'Origin':'https://untrusted.example'},403)]:
|
|
self.assertEqual(self.request(path,'POST',body,headers)[0],status)
|
|
self.assertEqual(self.request('/api/hook-notifications/coalesce','POST',{'clear':True})[0],400)
|
|
self.assertEqual(self.request('/api/hook-notifications/coalesce','POST',{})[1]['superseded_count'],2)
|
|
self.assertEqual(self.request('/api/queue-health')[1]['hooks']['pending_count'],0)
|
|
self.assertEqual(self.request('/api/queue-health')[1]['hooks']['superseded_count'],2)
|