115 lines
5.8 KiB
Python
115 lines
5.8 KiB
Python
"""Explicit delivery ownership without inferred merge or historical timestamps."""
|
|
import json
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
from agent_control.store import Store, Invalid
|
|
|
|
|
|
class DeliveryGraphTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.path = Path(self.tmp.name)/'delivery.sqlite3'
|
|
self.store = Store(self.path)
|
|
self.store.session(dict(id='worker',name='Worker'))
|
|
|
|
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 create(self, tid, **fields):
|
|
return self.store.task(dict(id=tid,name=tid,session_id='worker',**fields))
|
|
|
|
def report(self, **fields):
|
|
return self.store.task_report(dict(report_id='r',task_id='w',session_id='worker',
|
|
stage='execution_completed',current_progress='Checks passed',summary='Execution finished',evidence=[],**fields))
|
|
|
|
def test_create_update_clear_and_persistence(self):
|
|
self.create('d',kind='delivery')
|
|
work=self.create('w',delivery_id='d')
|
|
self.assertEqual(work['kind'],'work')
|
|
self.assertEqual(work['delivery_id'],'d')
|
|
self.assertEqual(Store(self.path).state()['tasks'],self.store.state()['tasks'])
|
|
self.store.task(dict(delivery_id=None),'w')
|
|
self.assertIsNone(self.task('w')['delivery_id'])
|
|
self.store.task(dict(kind='delivery'),'w')
|
|
self.assertEqual(self.task('w')['kind'],'delivery')
|
|
|
|
def test_invalid_references_and_kind_are_atomic(self):
|
|
self.create('d',kind='delivery')
|
|
self.create('w')
|
|
before=self.task('w')
|
|
for fields in [dict(kind='release'),dict(kind=None),dict(kind=[]),dict(delivery_id='missing'),
|
|
dict(delivery_id='w'),dict(delivery_id=True),dict(delivery_id=''),
|
|
dict(kind='delivery',delivery_id='d')]:
|
|
with self.subTest(fields=fields):
|
|
with self.assertRaises(Invalid): self.store.task(fields,'w')
|
|
self.create('other')
|
|
with self.assertRaises(Invalid): self.store.task(dict(delivery_id='other'),'w')
|
|
self.assertEqual(self.task('w'),before)
|
|
with self.assertRaises(Invalid): self.create('self',delivery_id='self')
|
|
self.assertFalse(any(t['id']=='self' for t in self.store.state()['tasks']))
|
|
|
|
def test_referenced_delivery_cannot_downgrade(self):
|
|
self.create('d',kind='delivery')
|
|
self.create('w',delivery_id='d')
|
|
with self.assertRaises(Invalid): self.store.task(dict(kind='work'),'d')
|
|
self.assertEqual(self.task('d')['kind'],'delivery')
|
|
self.store.task(dict(delivery_id=None),'w')
|
|
self.store.task(dict(kind='work'),'d')
|
|
self.assertEqual(self.task('d')['kind'],'work')
|
|
|
|
def test_old_data_default_does_not_guess_from_version_or_merge(self):
|
|
with self.store.db() as db:
|
|
db.execute('INSERT INTO tasks(id,data) VALUES(?,?)',('old',json.dumps(dict(id='old',name='Old',version='abc123'))))
|
|
old=self.task('old')
|
|
self.assertEqual(old['kind'],'work')
|
|
self.assertIsNone(old['delivery_id'])
|
|
self.assertIsNone(old['created_at'])
|
|
self.store.task(dict(current_progress='Reviewed existing work'),'old')
|
|
self.assertIsNone(self.task('old')['created_at'])
|
|
self.assertEqual(self.task('old')['kind'],'work')
|
|
|
|
def test_created_at_is_server_owned_stable_across_update_and_report(self):
|
|
with patch('agent_control.store.time.time',return_value=100):
|
|
self.create('d',kind='delivery')
|
|
self.create('w',delivery_id='d')
|
|
with patch('agent_control.store.time.time',return_value=200):
|
|
self.store.task(dict(name='Renamed'),'w')
|
|
self.report(version='candidate')
|
|
self.assertEqual(self.task('w')['created_at'],100)
|
|
self.assertEqual(self.task('d')['created_at'],100)
|
|
for fields in [dict(created_at=1),dict(created_at=None),dict(created=1)]:
|
|
with self.assertRaises(Invalid): self.store.task(fields,'w')
|
|
with self.assertRaises(Invalid): self.create('forged',created_at=0)
|
|
for fields in [dict(kind='delivery'),dict(delivery_id=None),dict(created_at=1)]:
|
|
with self.assertRaises(Invalid): self.report(**fields)
|
|
self.assertEqual(self.task('w')['kind'],'work')
|
|
self.assertEqual(self.task('w')['delivery_id'],'d')
|
|
|
|
def test_assignment_and_completion_never_invent_merge_or_verification(self):
|
|
self.create('d',kind='delivery')
|
|
self.create('w',delivery_id='d',status='completed',phase='done')
|
|
self.assertIsNone(self.task('w')['merged_into_id'])
|
|
self.assertEqual(self.task('d')['merge_sources'],[])
|
|
self.assertEqual(self.task('w')['integration_status'],'not_started')
|
|
self.store.integrate('d',dict(source_ids=['w'],owner='Integrator',status='verified',reason='Actual target acceptance'))
|
|
self.assertEqual(self.task('w')['merged_into_id'],'d')
|
|
self.assertEqual(self.task('d')['merge_sources'],['w'])
|
|
self.assertEqual(self.task('w')['integration_status'],'not_started')
|
|
self.assertEqual(self.task('d')['integration_status'],'verified')
|
|
self.store.task(dict(integration_status='verified'),'w')
|
|
self.assertEqual(self.task('w')['integration_status'],'verified')
|
|
|
|
def test_existing_work_to_work_merge_remains_compatible(self):
|
|
self.create('d',kind='delivery')
|
|
self.create('w',delivery_id='d')
|
|
self.create('target')
|
|
self.store.integrate('target',dict(source_ids=['w'],owner='Integrator',status='waiting',reason='Existing relationship'))
|
|
self.assertEqual(self.task('w')['merged_into_id'],'target')
|
|
self.assertEqual(self.task('w')['delivery_id'],'d')
|
|
self.assertEqual(self.task('target')['kind'],'work')
|