223 lines
12 KiB
Python
223 lines
12 KiB
Python
"""Atomic worker receipts and execution versus acceptance state."""
|
|
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
|
|
from agent_control.server import Server
|
|
|
|
|
|
def report(**changes):
|
|
data=dict(report_id='report-1',task_id='task',session_id='worker',stage='execution_completed',
|
|
current_progress='Implementation tested',summary='Execution finished',
|
|
evidence=['Synthetic tests passed'],version='v1')
|
|
data.update(changes)
|
|
return data
|
|
|
|
|
|
class ReportTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp=tempfile.TemporaryDirectory()
|
|
self.path=Path(self.tmp.name)/'reports.sqlite3'
|
|
self.store=Store(self.path)
|
|
for sid in ['worker','other']:
|
|
self.store.session(dict(id=sid,name=sid))
|
|
self.store.task(dict(id='task',name='Work',session_id='worker',status='running'))
|
|
self.store.task(dict(id='other-task',name='Other',session_id='other'))
|
|
|
|
def tearDown(self):
|
|
self.tmp.cleanup()
|
|
|
|
def task(self):
|
|
return next(t for t in self.store.state()['tasks'] if t['id']=='task')
|
|
|
|
def test_execution_completion_is_waiting_not_business_completion(self):
|
|
result=self.store.task_report(report())
|
|
self.assertFalse(result['duplicate'])
|
|
task=result['task']
|
|
self.assertEqual(task['execution_status'],'completed')
|
|
self.assertEqual(task['status'],'waiting')
|
|
self.assertEqual(task['phase'],'integration')
|
|
self.assertEqual(task['current_progress'],'Implementation tested')
|
|
self.assertEqual(task['report_evidence'],['Synthetic tests passed'])
|
|
self.assertEqual(task['last_report_id'],'report-1')
|
|
self.assertEqual(self.store.task_reports('task')['reports'],[result['report']])
|
|
self.assertEqual(self.store.state()['attention']['items'],[])
|
|
|
|
def test_duplicate_persistent_receipt_returns_current_task_without_reapplying(self):
|
|
first=self.store.task_report(report())
|
|
self.store.task(dict(status='completed',phase='done',current_progress='Accepted'),'task')
|
|
second=Store(self.path).task_report(report(needs_attention=False))
|
|
self.assertTrue(second['duplicate'])
|
|
self.assertEqual(second['report'],first['report'])
|
|
self.assertEqual(second['task']['status'],'completed')
|
|
self.assertEqual(second['task']['current_progress'],'Accepted')
|
|
self.assertEqual(len(self.store.task_reports('task')['reports']),1)
|
|
with self.assertRaises(Conflict): self.store.task_report(report(summary='Different'))
|
|
with self.assertRaises(Conflict): self.store.task_report(report(task_id='other-task',session_id='other'))
|
|
|
|
def test_new_late_report_does_not_roll_back_integrated_or_accepted_task(self):
|
|
self.store.task_report(report())
|
|
for status in ['integrated','verified']:
|
|
self.store.task(dict(integration_status=status,current_progress='Validated',version='accepted'),'task')
|
|
before=self.task()
|
|
result=self.store.task_report(report(report_id='late-'+status,stage='progress',
|
|
current_progress='Old work',version='old'))
|
|
self.assertEqual(result['task']['integration_status'],status)
|
|
self.assertEqual(result['task']['current_progress'],'Validated')
|
|
self.assertEqual(result['task']['execution_status'],'completed')
|
|
self.assertEqual(result['task']['version'],'accepted')
|
|
self.assertEqual(result['task']['progress_updated'],before['progress_updated'])
|
|
self.store.task(dict(status='completed',phase='done'),'task')
|
|
result=self.store.task_report(report(report_id='late-failure',stage='failed'))
|
|
self.assertEqual(result['task']['status'],'completed')
|
|
self.assertEqual(result['task']['phase'],'done')
|
|
self.assertEqual(len(self.store.task_reports('task')['reports']),4)
|
|
|
|
def test_concurrent_completion_and_late_progress_preserve_waiting_but_new_blocker_applies(self):
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:
|
|
bodies=[report()]+[report(report_id='progress-'+str(i),stage='progress',
|
|
current_progress='Older progress') for i in range(12)]
|
|
list(pool.map(self.store.task_report,bodies))
|
|
task=self.task()
|
|
self.assertEqual(task['execution_status'],'completed')
|
|
self.assertEqual(task['status'],'waiting')
|
|
self.assertEqual(task['phase'],'integration')
|
|
self.assertEqual(task['current_progress'],'Implementation tested')
|
|
progress_updated=task['progress_updated']
|
|
self.store.task_report(report(report_id='latest-arrival',stage='progress',
|
|
current_progress='Still old progress'))
|
|
task=self.task()
|
|
self.assertEqual(task['current_progress'],'Implementation tested')
|
|
self.assertEqual(task['progress_updated'],progress_updated)
|
|
self.assertEqual(len(self.store.task_reports('task')['reports']),14)
|
|
for stage in ['blocked','failed']:
|
|
task=self.store.task_report(report(report_id='new-'+stage,stage=stage,
|
|
current_progress='New issue discovered'))['task']
|
|
self.assertEqual(task['status'],stage)
|
|
self.assertEqual(task['execution_status'],stage)
|
|
self.assertEqual(task['current_progress'],'New issue discovered')
|
|
|
|
def test_blocked_failed_progress_and_server_managed_defaults(self):
|
|
self.assertEqual(self.task()['execution_status'],'pending')
|
|
self.assertIsNone(self.task()['last_report_id'])
|
|
for stage,expected in [('blocked','blocked'),('failed','failed'),('progress','running')]:
|
|
task=self.store.task_report(report(report_id=stage,stage=stage))['task']
|
|
self.assertEqual(task['execution_status'],expected)
|
|
self.assertEqual(task['status'],expected)
|
|
for field,value in [('execution_status','completed'),('last_report_id','fake'),
|
|
('last_report_at',123),('report_evidence',['fake'])]:
|
|
with self.assertRaises(Invalid): self.store.task({field:value},'task')
|
|
self.store.task(dict(id='manual',name='Manual',status='completed',phase='done'))
|
|
task=next(t for t in self.store.state()['tasks'] if t['id']=='manual')
|
|
self.assertEqual(task['execution_status'],'pending')
|
|
|
|
def test_identity_and_schema_reject_without_mutation(self):
|
|
before=self.task()
|
|
cases=[dict(session_id='other'),dict(session_id='missing'),dict(task_id='other-task'),
|
|
dict(stage='completed'),dict(current_progress=None),dict(current_progress='x'*1001),
|
|
dict(summary='x'*2001),dict(evidence=['x']*21),dict(evidence=['x'*1001]),
|
|
dict(evidence=[{}]),dict(version=None),dict(needs_attention=1),
|
|
dict(needs_attention=True,summary=''),dict(source='admin')]
|
|
for change in cases:
|
|
with self.subTest(change=change):
|
|
with self.assertRaises(Invalid): self.store.task_report(report(**change))
|
|
self.assertEqual(self.task(),before)
|
|
self.assertEqual(self.store.task_reports('task')['reports'],[])
|
|
|
|
def test_attention_only_on_explicit_true_and_same_task_item_updated(self):
|
|
first=self.store.task_report(report(needs_attention=True))
|
|
items=self.store.state()['attention']['items']
|
|
self.assertEqual(len(items),1)
|
|
aid=items[0]['id']
|
|
self.store.task_report(report(report_id='r2',stage='blocked',needs_attention=True,summary='Need review'))
|
|
items=self.store.state()['attention']['items']
|
|
self.assertEqual(len(items),1)
|
|
self.assertEqual(items[0]['id'],aid)
|
|
self.assertEqual(items[0]['reason'],'Need review')
|
|
self.store.attention(dict(id=aid,page='tasks',required=False,reason=''))
|
|
self.store.task_report(report(needs_attention=True))
|
|
self.assertEqual(self.store.state()['attention']['items'],[])
|
|
self.assertEqual(self.store.hook_notifications()['pending_count'],0)
|
|
self.assertEqual(self.store.decision_notifications()['pending_count'],0)
|
|
|
|
def test_failure_rolls_back_task_and_report_and_attention(self):
|
|
before=self.task()
|
|
with self.store.db() as db:
|
|
db.execute("CREATE TRIGGER fail_report_attention BEFORE INSERT ON attention_history BEGIN SELECT RAISE(ABORT,'synthetic'); END")
|
|
with self.assertRaises(Exception): self.store.task_report(report(needs_attention=True))
|
|
self.assertEqual(self.task(),before)
|
|
self.assertEqual(self.store.task_reports('task')['reports'],[])
|
|
self.assertEqual(self.store.state()['attention']['items'],[])
|
|
|
|
def test_concurrent_receipt_only_one_record(self):
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:
|
|
results=list(pool.map(lambda _:self.store.task_report(report()),range(12)))
|
|
self.assertEqual(sum(not r['duplicate'] for r in results),1)
|
|
self.assertEqual(len(self.store.task_reports('task')['reports']),1)
|
|
|
|
def test_recent_reports_are_bounded_and_task_scoped(self):
|
|
for i in range(53):
|
|
self.store.task_report(report(report_id='r-'+str(i),stage='progress'))
|
|
self.store.task_report(report(report_id='other',task_id='other-task',session_id='other'))
|
|
rows=self.store.task_reports('task')['reports']
|
|
self.assertEqual(len(rows),50)
|
|
self.assertEqual(rows[0]['report_id'],'r-52')
|
|
self.assertTrue(all(row['task_id']=='task' for row in rows))
|
|
self.assertEqual(len(self.store.task_reports('other-task')['reports']),1)
|
|
with self.assertRaises(Invalid): self.store.task_reports('missing')
|
|
|
|
|
|
class ReportHTTPTests(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_query_and_roundtrip(self):
|
|
self.server.store.session(dict(id='worker',name='Worker'))
|
|
self.server.store.task(dict(id='task',name='Task',session_id='worker'))
|
|
path='/api/task-reports'
|
|
for headers,code in [({'Authorization':''},401),({'X-CSRF-Token':''},403),
|
|
({'Origin':'https://untrusted.example'},403)]:
|
|
self.assertEqual(self.request(path,'POST',report(),headers)[0],code)
|
|
self.assertEqual(self.request(path+'?task_id=task',headers={'Authorization':''})[0],401)
|
|
for query in ['', '?task_id=', '?task_id=task&task_id=other', '?session_id=worker']:
|
|
self.assertEqual(self.request(path+query)[0],400)
|
|
self.assertEqual(self.request(path,'POST',report())[0],200)
|
|
self.assertTrue(self.request(path,'POST',report())[1]['duplicate'])
|
|
self.assertEqual(self.request(path,'POST',report(summary='Changed'))[0],409)
|
|
self.assertEqual(len(self.request(path+'?task_id=task')[1]['reports']),1)
|