240 lines
13 KiB
Python
240 lines
13 KiB
Python
"""Task relation graphs and atomic integration records, without Git execution."""
|
|
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, TASK_RELATION_DEFAULTS
|
|
from agent_control.server import Server
|
|
|
|
|
|
def integration(**changes):
|
|
data=dict(source_ids=['source-a','source-b'],owner='Integrator',status='waiting',
|
|
reason='Packages ready for coordinated integration',version='v1')
|
|
data.update(changes)
|
|
return data
|
|
|
|
|
|
class RelationTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp=tempfile.TemporaryDirectory()
|
|
self.path=Path(self.tmp.name)/'relations.sqlite3'
|
|
self.store=Store(self.path)
|
|
for tid in ['target','source-a','source-b','other']:
|
|
self.store.task(dict(id=tid,name=tid,status='running',current_progress='Working on '+tid))
|
|
|
|
def tearDown(self):
|
|
self.tmp.cleanup()
|
|
|
|
def task(self,tid):
|
|
return next(t for t in self.store.state()['tasks'] if t['id']==tid)
|
|
|
|
def audit_count(self):
|
|
with self.store.db() as db:
|
|
return db.execute('SELECT count(*) FROM task_integrations').fetchone()[0]
|
|
|
|
def test_defaults_legacy_and_descriptive_metadata(self):
|
|
with self.store.db() as db:
|
|
db.execute('INSERT INTO tasks(id,data) VALUES(?,?)',('legacy',json.dumps(dict(id='legacy',name='Legacy'))))
|
|
task=self.task('legacy')
|
|
for field,value in TASK_RELATION_DEFAULTS.items():
|
|
self.assertEqual(task[field],value)
|
|
task=self.store.task(dict(goal='Deliver verified result',scope='Parser only',coupling_assessment='Shared schema',
|
|
owner='Worker',parallel_scope='Read fixtures before schema is ready',
|
|
acceptance_criteria=['Pass parser checks'],
|
|
sync_points=[dict(id='schema',label='Schema ready',task_ids=['source-a'],status='waiting')]),'target')
|
|
self.assertEqual(task['goal'],'Deliver verified result')
|
|
self.assertEqual(task['sync_points'][0]['task_ids'],['source-a'])
|
|
self.assertEqual(task['integration_status'],'not_started')
|
|
|
|
def test_single_multi_dependency_and_dedup(self):
|
|
task=self.store.task(dict(depends_on=['source-b','source-a','source-a']),'target')
|
|
self.assertEqual(task['depends_on'],['source-a','source-b'])
|
|
self.store.task(dict(depends_on=['other']),'source-a')
|
|
with self.assertRaises(Conflict): self.store.task(dict(depends_on=['target']),'other')
|
|
self.assertEqual(self.task('other')['depends_on'],[])
|
|
|
|
def test_parent_hierarchy_separate_from_dependency_and_merge_graph(self):
|
|
self.store.task(dict(parent_id='target'),'source-a')
|
|
self.store.task(dict(parent_id='source-a'),'other')
|
|
with self.assertRaises(Conflict): self.store.task(dict(parent_id='other'),'target')
|
|
# Parent depends on a child, and child merges back into parent: all legitimate.
|
|
self.store.task(dict(depends_on=['source-a']),'target')
|
|
result=self.store.integrate('target',integration(source_ids=['source-a']))
|
|
self.assertEqual(result['sources'][0]['parent_id'],'target')
|
|
self.assertEqual(result['sources'][0]['merged_into_id'],'target')
|
|
self.assertEqual(result['task']['depends_on'],['source-a'])
|
|
|
|
def test_overlap_requires_parallel_scope_not_false_serial_order(self):
|
|
with self.assertRaises(Invalid):
|
|
self.store.task(dict(depends_on=['source-a'],parallel_with=['source-a']),'target')
|
|
task=self.store.task(dict(depends_on=['source-a'],parallel_with=['source-a'],
|
|
parallel_scope='Prepare fixtures now; wait for schema before integration'),'target')
|
|
self.assertEqual(task['parallel_with'],['source-a'])
|
|
with self.assertRaises(Invalid): self.store.task(dict(parallel_scope=' '),'target')
|
|
self.assertTrue(self.task('target')['parallel_scope'])
|
|
|
|
def test_multisource_integrate_preserves_states_progress_and_timestamps(self):
|
|
self.store.task(dict(status='blocked'),'source-a')
|
|
before_a=self.task('source-a')
|
|
before_b=self.task('source-b')
|
|
with patch('subprocess.run',side_effect=AssertionError('integration never executes Git')):
|
|
result=self.store.integrate('target',integration(status='integrating'))
|
|
self.assertFalse(result['duplicate'])
|
|
self.assertEqual(result['task']['merge_sources'],['source-a','source-b'])
|
|
self.assertEqual(result['task']['integration_owner'],'Integrator')
|
|
self.assertEqual(result['task']['integration_status'],'integrating')
|
|
self.assertEqual(result['task']['version'],'v1')
|
|
for before in [before_a,before_b]:
|
|
after=self.task(before['id'])
|
|
self.assertEqual(after['merged_into_id'],'target')
|
|
self.assertEqual(after['status'],before['status'])
|
|
self.assertEqual(after['current_progress'],before['current_progress'])
|
|
self.assertEqual(after['progress_updated'],before['progress_updated'])
|
|
self.assertEqual(after['eta_estimated_at'],before['eta_estimated_at'])
|
|
|
|
def test_idempotent_order_and_retry_cannot_rewind_new_status(self):
|
|
first=self.store.integrate('target',integration())
|
|
events=len(self.store.state()['events'])
|
|
second=self.store.integrate('target',integration(source_ids=['source-b','source-a','source-a']))
|
|
self.assertTrue(second['duplicate'])
|
|
self.assertEqual(first['task'],second['task'])
|
|
self.assertEqual(self.audit_count(),1)
|
|
self.assertEqual(len(self.store.state()['events']),events)
|
|
self.store.integrate('target',integration(status='verified',reason='Actual integration and acceptance completed'))
|
|
result=self.store.integrate('target',integration())
|
|
self.assertTrue(result['duplicate'])
|
|
self.assertEqual(result['task']['integration_status'],'verified')
|
|
self.assertEqual(self.audit_count(),2)
|
|
|
|
def test_source_owned_elsewhere_and_merge_cycle_reject_atomically(self):
|
|
self.store.integrate('other',integration(source_ids=['source-a']))
|
|
with self.assertRaises(Conflict): self.store.integrate('target',integration())
|
|
self.assertEqual(self.task('target')['merge_sources'],[])
|
|
self.assertIsNone(self.task('source-b')['merged_into_id'])
|
|
self.assertEqual(self.audit_count(),1)
|
|
self.store.integrate('target',integration(source_ids=['other']))
|
|
with self.assertRaises(Conflict):
|
|
self.store.integrate('source-a',integration(source_ids=['target']))
|
|
self.assertIsNone(self.task('target')['merged_into_id'])
|
|
self.assertEqual(self.audit_count(),2)
|
|
|
|
def test_plain_task_merge_edits_are_reciprocal_and_cannot_rebind(self):
|
|
self.store.task(dict(merged_into_id='target'),'source-a')
|
|
self.assertEqual(self.task('target')['merge_sources'],['source-a'])
|
|
with self.assertRaises(Conflict):
|
|
self.store.task(dict(merged_into_id='other'),'source-a')
|
|
self.store.task(dict(merge_sources=['source-a','source-b']),'target')
|
|
self.assertEqual(self.task('source-b')['merged_into_id'],'target')
|
|
self.store.task(dict(merge_sources=['source-b']),'target')
|
|
self.assertIsNone(self.task('source-a')['merged_into_id'])
|
|
self.store.task(dict(merged_into_id=None),'source-b')
|
|
self.assertEqual(self.task('target')['merge_sources'],[])
|
|
|
|
def test_schema_and_references_are_atomic(self):
|
|
updates=[dict(goal=None),dict(scope='x'*4001),dict(owner='x'*201),dict(parallel_scope='x'*2001),
|
|
dict(parent_id='missing'),dict(parent_id='target'),dict(depends_on=['target']),
|
|
dict(depends_on=['missing']),dict(parallel_with='source-a'),dict(merge_sources=['target']),
|
|
dict(acceptance_criteria=['x'*1001]),dict(acceptance_criteria=['x']*21),
|
|
dict(sync_points=[dict(id='x',label='X',task_ids=['missing'],status='waiting')]),
|
|
dict(sync_points=[dict(id='x',label='X',task_ids=[],status='done')]),
|
|
dict(sync_points=[dict(id='x',label='X',task_ids=[],status='ready',extra=True)]),
|
|
dict(sync_points=[dict(id='x',label='X',task_ids=[],status='ready')]*2),
|
|
dict(integration_status='complete'),dict(id='other')]
|
|
before=self.task('target')
|
|
for update in updates:
|
|
with self.subTest(update=update):
|
|
with self.assertRaises(Invalid): self.store.task(update,'target')
|
|
self.assertEqual(self.task('target'),before)
|
|
|
|
def test_integration_schema_unknown_and_missing_ids(self):
|
|
changes=[dict(source_ids=[]),dict(source_ids=['target']),dict(source_ids=['missing']),
|
|
dict(source_ids=['source-a']*31),dict(owner=''),dict(reason=''),
|
|
dict(reason='x'*2001),dict(status='completed'),dict(version=None),dict(shell='git merge')]
|
|
for change in changes:
|
|
with self.assertRaises(Invalid): self.store.integrate('target',integration(**change))
|
|
with self.assertRaises(Invalid): self.store.integrate('missing',integration())
|
|
self.assertEqual(self.audit_count(),0)
|
|
self.assertEqual(self.task('target')['merge_sources'],[])
|
|
|
|
def test_integration_transaction_rollback(self):
|
|
with self.store.db() as db:
|
|
db.execute("CREATE TRIGGER fail_integration BEFORE INSERT ON task_integrations BEGIN SELECT RAISE(ABORT,'synthetic'); END")
|
|
with self.assertRaises(Exception): self.store.integrate('target',integration())
|
|
self.assertEqual(self.task('target')['merge_sources'],[])
|
|
self.assertIsNone(self.task('source-a')['merged_into_id'])
|
|
self.assertIsNone(self.task('source-b')['merged_into_id'])
|
|
|
|
def test_concurrent_same_request_only_one_audit(self):
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:
|
|
results=list(pool.map(lambda _:self.store.integrate('target',integration()),range(12)))
|
|
self.assertEqual(sum(not r['duplicate'] for r in results),1)
|
|
self.assertEqual(self.audit_count(),1)
|
|
|
|
def test_source_union_and_limit(self):
|
|
self.store.integrate('target',integration(source_ids=['source-a']))
|
|
self.store.integrate('target',integration(source_ids=['source-b']))
|
|
self.assertEqual(self.task('target')['merge_sources'],['source-a','source-b'])
|
|
for i in range(29):
|
|
self.store.task(dict(id='extra-'+str(i),name='Extra'))
|
|
sources=['extra-'+str(i) for i in range(29)]
|
|
with self.assertRaises(Invalid):
|
|
self.store.integrate('target',integration(source_ids=sources))
|
|
self.assertEqual(self.task('target')['merge_sources'],['source-a','source-b'])
|
|
self.assertIsNone(self.task('extra-0')['merged_into_id'])
|
|
|
|
|
|
class RelationHTTPTests(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_endpoint_security_scope_and_success(self):
|
|
for tid in ['target','source-a','source-b']:
|
|
self.server.store.task(dict(id=tid,name=tid))
|
|
path='/api/tasks/target/integrate'
|
|
for headers,code in [({'Authorization':''},401),({'X-CSRF-Token':''},403),
|
|
({'Origin':'https://untrusted.example'},403)]:
|
|
self.assertEqual(self.request(path,'POST',integration(),headers)[0],code)
|
|
self.assertEqual(self.request(path,'POST',integration(shell='no'))[0],400)
|
|
self.assertEqual(self.request(path+'?x=1','POST',integration())[0],404)
|
|
self.assertEqual(self.request('/api/tasks/target','PATCH',dict(id='source-a'))[0],400)
|
|
code,body=self.request(path,'POST',integration())
|
|
self.assertEqual(code,200)
|
|
self.assertEqual(body['task']['merge_sources'],['source-a','source-b'])
|
|
self.assertTrue(self.request(path,'POST',integration())[1]['duplicate'])
|
|
self.assertEqual(self.request('/api/tasks/source-a/integrate','POST',
|
|
integration(source_ids=['target']))[0],409)
|