237 lines
12 KiB
Python
237 lines
12 KiB
Python
"""Persistent explicit attention; opening pages never dismisses it."""
|
|
import concurrent.futures
|
|
import http.client
|
|
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, ATTENTION_PAGES
|
|
from agent_control.server import Server
|
|
|
|
|
|
def item(**changes):
|
|
data = dict(id='item-a',page='tasks',required=True,reason='Needs a human check')
|
|
data.update(changes)
|
|
return data
|
|
|
|
|
|
def hook(**changes):
|
|
data = dict(source='synthetic',event_id='incident',severity='warning',status='firing',
|
|
summary='Connection unavailable')
|
|
data.update(changes)
|
|
return data
|
|
|
|
|
|
class AttentionTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.path = Path(self.tmp.name)/'attention.sqlite3'
|
|
self.store = Store(self.path)
|
|
|
|
def tearDown(self):
|
|
self.tmp.cleanup()
|
|
|
|
def history_count(self):
|
|
with self.store.db() as db:
|
|
return db.execute('SELECT count(*) FROM attention_history').fetchone()[0]
|
|
|
|
def test_empty_explicit_update_revoke_and_persist(self):
|
|
self.assertEqual(self.store.state()['attention'],dict(items=[],pages={p:False for p in ATTENTION_PAGES}))
|
|
with patch('agent_control.store.time.time',return_value=100):
|
|
created = self.store.attention(item())
|
|
self.assertFalse(created['duplicate'])
|
|
self.assertEqual(created['href'],'/tasks')
|
|
self.assertEqual(created['updated'],100)
|
|
self.assertTrue(Store(self.path).state()['attention']['pages']['tasks'])
|
|
for _ in range(3):
|
|
self.store.state()
|
|
self.assertEqual(self.history_count(),1)
|
|
with patch('agent_control.store.time.time',return_value=200):
|
|
again = self.store.attention(item())
|
|
self.assertTrue(again['duplicate'])
|
|
self.assertEqual(again['updated'],100)
|
|
self.assertEqual(self.history_count(),1)
|
|
self.store.attention(item(reason='Evidence changed',href='/logs?incident=5'))
|
|
self.assertEqual(self.history_count(),2)
|
|
self.store.attention(item(required=False,reason=''))
|
|
self.assertFalse(self.store.state()['attention']['pages']['tasks'])
|
|
self.assertEqual(self.history_count(),3)
|
|
self.assertTrue(self.store.attention(item(required=False,reason=''))['duplicate'])
|
|
self.assertEqual(self.history_count(),3)
|
|
|
|
def test_many_independent_items_any_aggregate(self):
|
|
self.store.attention(item())
|
|
self.store.attention(item(id='item-b'))
|
|
self.store.attention(item(id='item-c',page='cluster'))
|
|
self.store.attention(item(required=False,reason='Done'))
|
|
state = self.store.state()['attention']
|
|
self.assertEqual(len(state['items']),2)
|
|
self.assertTrue(state['pages']['tasks'])
|
|
self.assertTrue(state['pages']['cluster'])
|
|
self.store.attention(item(id='item-b',required=False,reason=''))
|
|
self.assertFalse(self.store.state()['attention']['pages']['tasks'])
|
|
self.assertTrue(self.store.state()['attention']['pages']['cluster'])
|
|
self.assertEqual(set(state['items'][0]),{'id','page','required','reason','href','updated'})
|
|
|
|
def test_subject_cannot_change_and_associations_can_be_omitted_on_withdrawal(self):
|
|
self.store.task(dict(id='a',name='A'))
|
|
self.store.task(dict(id='b',name='B'))
|
|
self.store.attention(item(task_id='a'))
|
|
for data in [item(task_id='b'),item(page='cluster'),item(task_id='b',required=False,reason='')]:
|
|
with self.assertRaises(Conflict): self.store.attention(data)
|
|
self.assertEqual(self.history_count(),1)
|
|
self.store.attention(item(required=False,reason=''))
|
|
with self.store.db() as db:
|
|
saved=json.loads(db.execute("SELECT data FROM attention WHERE id='item-a'").fetchone()[0])
|
|
self.assertEqual(saved['task_id'],'a')
|
|
self.assertFalse(saved['required'])
|
|
|
|
def test_decision_automatic_attention_response_retires_every_associated_item(self):
|
|
self.store.decision(dict(id='d',title='Choose a plan',question='Which plan?'))
|
|
state = self.store.state()['attention']
|
|
self.assertTrue(state['pages']['decisions'])
|
|
auto=state['items'][0]
|
|
self.assertEqual(auto['href'],'/decisions?decision=d')
|
|
self.store.attention(item(decision_id='d'))
|
|
self.store.decision_respond('d',dict(option_id=None,comment='Chosen',idempotency_key='reply'))
|
|
state=self.store.state()['attention']
|
|
self.assertEqual(state['items'],[])
|
|
self.assertFalse(any(state['pages'].values()))
|
|
with self.store.db() as db:
|
|
self.assertTrue(all(not json.loads(r['data'])['required'] for r in db.execute('SELECT data FROM attention')))
|
|
self.store.attention(item(decision_id='d',reason='Attempt to revive already answered'))
|
|
self.assertFalse(self.store.state()['attention']['pages']['tasks'])
|
|
|
|
def test_manual_early_decision_withdrawal_never_recreated_by_retry_read_or_restart(self):
|
|
definition=dict(id='d',title='Choose',question='What next?')
|
|
self.store.decision(definition)
|
|
auto=self.store.state()['attention']['items'][0]
|
|
self.store.attention(dict(id=auto['id'],page='decisions',required=False,reason=''))
|
|
count=self.history_count()
|
|
for _ in range(3):
|
|
self.store.decision(definition)
|
|
self.assertFalse(Store(self.path).state()['attention']['pages']['decisions'])
|
|
self.assertEqual(self.history_count(),count)
|
|
|
|
def test_hook_only_upgrade_marks_attention_and_resolution_clears_related(self):
|
|
self.store.hook(hook())
|
|
self.assertFalse(self.store.state()['attention']['pages']['logs'])
|
|
escalation=dict(source='synthetic',event_id='incident',reason='Persistent outage')
|
|
result=self.store.hook_escalate(escalation)
|
|
state=self.store.state()['attention']
|
|
self.assertTrue(state['pages']['logs'])
|
|
auto=state['items'][0]
|
|
self.assertEqual(auto['href'],'/logs?incident='+result['notifications'][0]['hook']['log_id'])
|
|
self.store.attention(item(hook_source='synthetic',hook_event_id='incident'))
|
|
self.store.hook(hook(status='resolved'))
|
|
self.assertEqual(self.store.state()['attention']['items'],[])
|
|
self.store.hook(hook(event_id='ordinary'))
|
|
self.store.hook(hook(event_id='ordinary',status='resolved'))
|
|
self.assertEqual(self.store.state()['attention']['items'],[])
|
|
|
|
def test_early_upgrade_withdrawal_is_not_recreated(self):
|
|
self.store.hook(hook())
|
|
escalation=dict(source='synthetic',event_id='incident',reason='Persistent outage')
|
|
self.store.hook_escalate(escalation)
|
|
auto=self.store.state()['attention']['items'][0]
|
|
self.store.attention(dict(id=auto['id'],page='logs',required=False,reason='Acknowledged separately'))
|
|
self.store.hook_escalate(escalation)
|
|
self.assertFalse(Store(self.path).state()['attention']['pages']['logs'])
|
|
|
|
def test_task_completion_keeps_attention_cancellation_retires(self):
|
|
self.store.task(dict(id='task',name='Task'))
|
|
self.store.attention(item(task_id='task'))
|
|
self.store.task(dict(status='completed',phase='done'),'task')
|
|
self.assertTrue(self.store.state()['attention']['pages']['tasks'])
|
|
self.store.task(dict(status='cancelled'),'task')
|
|
self.assertFalse(self.store.state()['attention']['pages']['tasks'])
|
|
self.store.attention(item(task_id='task',reason='Try to revive cancelled task'))
|
|
self.assertFalse(self.store.state()['attention']['pages']['tasks'])
|
|
|
|
def test_schema_paths_and_unknown_targets(self):
|
|
for data in [item(required=1),item(reason=''),item(reason=None),item(reason='x'*241),
|
|
item(page='other'),item(thread_id='arbitrary'),item(html='<b>hi</b>'),
|
|
item(task_id='missing'),item(decision_id='missing'),
|
|
item(hook_source='synthetic'),item(hook_event_id='incident'),
|
|
item(hook_source='missing',hook_event_id='incident'),
|
|
item(id='attention-decision-forged')]:
|
|
with self.subTest(data=data):
|
|
with self.assertRaises(Invalid): self.store.attention(data)
|
|
for href in ['https://example.invalid','//example.invalid/logs',r'/logs\evil','/other',
|
|
'/logs/../tasks','/logs#x','/logs?next=https://example.invalid',
|
|
'/logs?x=%3Cscript%3E','/%6cogs','/logs?x=%255Cevil','/logs\n','javascript:alert(1)']:
|
|
with self.subTest(href=href):
|
|
with self.assertRaises(Invalid): self.store.attention(item(href=href))
|
|
self.assertEqual(self.history_count(),0)
|
|
self.store.attention(item(href='/logs?incident=123&status=active'))
|
|
self.assertEqual(self.store.state()['attention']['items'][0]['href'],'/logs?incident=123&status=active')
|
|
|
|
def test_concurrent_duplicate_requests_one_history_record(self):
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:
|
|
results=list(pool.map(lambda _:self.store.attention(item()),range(16)))
|
|
self.assertEqual(sum(not r['duplicate'] for r in results),1)
|
|
self.assertEqual(self.history_count(),1)
|
|
self.assertEqual(len(self.store.state()['attention']['items']),1)
|
|
|
|
def test_auto_attention_failure_rolls_back_decision_and_upgrade(self):
|
|
with self.store.db() as db:
|
|
db.execute("CREATE TRIGGER fail_attention BEFORE INSERT ON attention_history BEGIN SELECT RAISE(ABORT,'synthetic'); END")
|
|
with self.assertRaises(Exception):
|
|
self.store.decision(dict(id='d',title='Choose',question='What next?'))
|
|
self.assertEqual(self.store.state()['decisions'],[])
|
|
self.store.hook(hook())
|
|
with self.assertRaises(Exception):
|
|
self.store.hook_escalate(dict(source='synthetic',event_id='incident',reason='Persistent'))
|
|
self.assertEqual(self.store.logs()['logs'][0]['level'],'warning')
|
|
self.assertEqual(self.store.hook_notifications()['pending_count'],1)
|
|
|
|
|
|
class AttentionHTTPTests(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_auth_csrf_origin_and_reads_never_clear(self):
|
|
for headers,status in [({'Authorization':''},401),({'X-CSRF-Token':''},403),
|
|
({'Origin':'https://untrusted.example'},403)]:
|
|
self.assertEqual(self.request('/api/attention','POST',item(),headers)[0],status)
|
|
self.assertEqual(self.request('/api/attention','POST',item())[0],200)
|
|
self.assertEqual(self.request('/api/attention','POST',item(thread_id='no'))[0],400)
|
|
for _ in range(3):
|
|
self.assertTrue(self.request('/api/state')[1]['attention']['pages']['tasks'])
|
|
self.assertTrue(self.request('/api/attention','POST',item())[1]['duplicate'])
|
|
self.assertEqual(self.request('/api/attention','POST',item(page='logs'))[0],409)
|
|
self.assertEqual(self.request('/api/attention','POST',item(required=False,reason=''))[0],200)
|
|
self.assertFalse(self.request('/api/state')[1]['attention']['pages']['tasks'])
|