319 lines
15 KiB
Python
319 lines
15 KiB
Python
"""Meaningful workflow/schema tests on temporary projects, never on shipped state."""
|
|
import copy
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
SPEC = importlib.util.spec_from_file_location('luma_tasks', ROOT / 'scripts/tasks.py')
|
|
MODULE = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(MODULE)
|
|
Tracker = MODULE.Tracker
|
|
|
|
|
|
class WorkflowTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.root = Path(self.temp.name)
|
|
for folder in ['planning/tasks', 'state/evidence', 'state/handoffs', 'roles', 'specs']:
|
|
(self.root / folder).mkdir(parents=True, exist_ok=True)
|
|
source = '# Spec\n\n## 1. Core\n\n**REQ-001.** One real behavior.\n'
|
|
(self.root / 'specs/tz_clear.md').write_text(source)
|
|
self.write('specs/manifest.json', {'sha256': {'tz_clear.md': hashlib.sha256(source.encode()).hexdigest()}})
|
|
self.write('roles/index.json', {'roles': {'front': {}, 'qa': {}}})
|
|
for role in ['front', 'qa']:
|
|
(self.root / f'roles/{role}.md').write_text('# Role\n')
|
|
self.write('planning/requirements.json', [{'id': 'REQ-001', 'source': 'specs/tz_clear.md', 'text': 'One real behavior.', 'applicability': 'web', 'reason': 'Applies'}])
|
|
tasks = []
|
|
for i in range(1, 4):
|
|
task_id = f'S00-{i:03}'
|
|
tasks.append({'id': task_id, 'stage': 'S00', 'role': 'front', 'reviewer_role': 'qa', 'title': task_id, 'outcome': 'One bounded result', 'depends_on': ['S00-001'] if i == 2 else [], 'requirements': ['REQ-001'], 'scope': [f'src/{i}.*'], 'inputs': ['roles/front.md'], 'steps': ['Implement accepted result'], 'acceptance': ['Behavior checked'], 'checks': ['Positive and negative scenario'], 'spec_sections': [1], 'ui_sections': []})
|
|
(self.root / f'planning/tasks/{task_id}.md').write_text('# Placeholder\n')
|
|
self.write('planning/tasks.json', {'stages': [{'id': 'S00', 'title': 'Stage', 'goal': 'Goal', 'entry_gate': None}], 'tasks': tasks})
|
|
self.write('state/tasks.json', {'revision': 0, 'tasks': {t['id']: {'status': 'todo', 'owner': None} for t in tasks}, 'events': []})
|
|
self.tracker = Tracker(self.root)
|
|
|
|
def tearDown(self):
|
|
self.temp.cleanup()
|
|
|
|
def write(self, path, data):
|
|
(self.root / path).write_text(json.dumps(data, ensure_ascii=False, indent=2) + '\n')
|
|
|
|
def evidence(self, name='owner.md', folder='state/evidence'):
|
|
path = f'{folder}/{name}'
|
|
(self.root / path).write_text('# Real evidence\n\nCommand: test scenario; actual exit code 0; checked fixture and negative case.\n')
|
|
return path
|
|
|
|
def claim(self, task='S00-001', agent='owner'):
|
|
return self.tracker.mutate('claim', task, agent, role='front')
|
|
|
|
def submit(self):
|
|
return self.tracker.mutate('submit', 'S00-001', 'owner', evidence=self.evidence(), handoff=self.evidence('S00-001.md', 'state/handoffs'))
|
|
|
|
def complete(self):
|
|
self.claim()
|
|
self.submit()
|
|
return self.tracker.mutate('approve', 'S00-001', 'reviewer', role='qa', evidence=self.evidence('review.md'))
|
|
|
|
def test_fixture_validates(self):
|
|
self.assertEqual(self.tracker.validate()['tasks'], 3)
|
|
|
|
def test_next_filters_dependencies_and_role(self):
|
|
self.assertEqual([x['id'] for x in self.tracker.ready('front')], ['S00-001', 'S00-003'])
|
|
self.assertEqual(self.tracker.ready('qa'), [])
|
|
|
|
def test_claim_dependency_denied_without_state_mutation(self):
|
|
before = self.tracker.state()
|
|
with self.assertRaises(ValueError):
|
|
self.claim('S00-002')
|
|
self.assertEqual(self.tracker.state(), before)
|
|
|
|
def test_claim_wrong_role_denied(self):
|
|
with self.assertRaises(ValueError):
|
|
self.tracker.mutate('claim', 'S00-001', 'owner', role='qa')
|
|
|
|
def test_double_claim_and_multiple_owned_tasks_denied(self):
|
|
self.claim()
|
|
with self.assertRaises(ValueError):
|
|
self.claim(agent='other')
|
|
with self.assertRaises(ValueError):
|
|
self.claim('S00-003')
|
|
|
|
def test_submit_requires_owner_and_files(self):
|
|
self.claim()
|
|
with self.assertRaises(ValueError):
|
|
self.tracker.mutate('submit', 'S00-001', 'other')
|
|
with self.assertRaises(ValueError):
|
|
self.tracker.mutate('submit', 'S00-001', 'owner')
|
|
self.assertEqual(self.tracker.state()['tasks']['S00-001']['status'], 'in_progress')
|
|
|
|
def test_submit_and_review_queue(self):
|
|
self.claim()
|
|
self.submit()
|
|
self.assertEqual(self.tracker.ready('qa')[0]['action'], 'review')
|
|
self.assertEqual(self.tracker.ready('front'), [{'id': 'S00-003', 'action': 'claim', 'role': 'front', 'title': 'S00-003'}])
|
|
|
|
def test_self_review_and_wrong_reviewer_role_denied(self):
|
|
self.claim()
|
|
self.submit()
|
|
path = self.evidence('review.md')
|
|
with self.assertRaises(ValueError):
|
|
self.tracker.mutate('approve', 'S00-001', 'owner', role='qa', evidence=path)
|
|
with self.assertRaises(ValueError):
|
|
self.tracker.mutate('approve', 'S00-001', 'reviewer', role='front', evidence=path)
|
|
|
|
def test_approval_unlocks_dependency_and_keeps_history(self):
|
|
self.complete()
|
|
self.assertIn('S00-002', [x['id'] for x in self.tracker.ready('front')])
|
|
self.assertEqual([e['action'] for e in self.tracker.state()['events']], ['claim', 'submit', 'approve'])
|
|
self.assertEqual(self.tracker.state()['revision'], 3)
|
|
self.tracker.validate()
|
|
|
|
def test_reject_requeues_for_new_owner(self):
|
|
self.claim()
|
|
self.submit()
|
|
self.tracker.mutate('reject', 'S00-001', 'reviewer', role='qa', reason='Fix observed case')
|
|
row = self.tracker.state()['tasks']['S00-001']
|
|
self.assertEqual((row['status'], row['owner']), ('todo', None))
|
|
|
|
def test_bare_block_unblock_denied(self):
|
|
self.claim()
|
|
for action in ['block', 'unblock']:
|
|
with self.assertRaisesRegex(ValueError, 'Use escalate'):
|
|
self.tracker.mutate(action, 'S00-001', 'owner', reason='No remediation')
|
|
self.assertEqual(self.tracker.state()['tasks']['S00-001']['status'], 'in_progress')
|
|
|
|
def test_reassign_explicit_and_one_owner(self):
|
|
self.claim()
|
|
self.claim('S00-003', 'busy')
|
|
with self.assertRaises(ValueError):
|
|
self.tracker.mutate('reassign', 'S00-001', 'owner', target='busy', reason='Authorized transfer')
|
|
self.tracker.mutate('reassign', 'S00-001', 'owner', target='new-owner', reason='Authorized transfer')
|
|
self.assertEqual(self.tracker.state()['tasks']['S00-001']['owner'], 'new-owner')
|
|
|
|
def test_path_traversal_and_external_evidence_denied(self):
|
|
path = self.root / 'outside.md'
|
|
path.write_text('x' * 80)
|
|
with self.assertRaises(ValueError):
|
|
self.tracker.safe_file('state/evidence/../../outside.md', 'state/evidence')
|
|
with self.assertRaises(ValueError):
|
|
self.tracker.safe_file(str(path), 'state/evidence')
|
|
|
|
def test_symlink_evidence_denied(self):
|
|
path = self.root / 'outside.md'
|
|
path.write_text('x' * 80)
|
|
link = self.root / 'state/evidence/link.md'
|
|
try:
|
|
link.symlink_to(path)
|
|
except OSError:
|
|
self.skipTest('Symlink unavailable on this platform')
|
|
with self.assertRaises(ValueError):
|
|
self.tracker.safe_file('state/evidence/link.md', 'state/evidence')
|
|
|
|
def test_lock_excludes_concurrent_mutation(self):
|
|
with self.tracker.lock():
|
|
with self.assertRaises(ValueError):
|
|
self.claim()
|
|
self.assertFalse(self.tracker.lock_path.exists())
|
|
self.claim()
|
|
|
|
def test_two_processes_cannot_claim_same_task(self):
|
|
code = "import importlib.util,sys; s=importlib.util.spec_from_file_location('t',sys.argv[1]);m=importlib.util.module_from_spec(s);s.loader.exec_module(m);m.Tracker(sys.argv[2]).mutate('claim','S00-001',sys.argv[3],role='front')"
|
|
processes = [subprocess.Popen([sys.executable, '-c', code, str(ROOT / 'scripts/tasks.py'), str(self.root), a], stdout=subprocess.PIPE, stderr=subprocess.PIPE) for a in ['race-1', 'race-2']]
|
|
for process in processes:
|
|
process.communicate()
|
|
exits = [p.returncode for p in processes]
|
|
self.assertEqual(exits.count(0), 1)
|
|
self.assertEqual(len(self.tracker.state()['events']), 1)
|
|
|
|
def test_dependency_cycle_detected(self):
|
|
defs = self.tracker.definitions()
|
|
defs['tasks'][0]['depends_on'] = ['S00-002']
|
|
self.write('planning/tasks.json', defs)
|
|
with self.assertRaisesRegex(ValueError, 'cycle'):
|
|
self.tracker.validate()
|
|
|
|
def test_unknown_dependency_detected(self):
|
|
defs = self.tracker.definitions()
|
|
defs['tasks'][0]['depends_on'] = ['S99-999']
|
|
self.write('planning/tasks.json', defs)
|
|
with self.assertRaisesRegex(ValueError, 'Missing dependency'):
|
|
self.tracker.validate()
|
|
|
|
def test_duplicate_task_and_missing_coverage_detected(self):
|
|
defs = self.tracker.definitions()
|
|
defs['tasks'].append(copy.deepcopy(defs['tasks'][0]))
|
|
self.write('planning/tasks.json', defs)
|
|
with self.assertRaisesRegex(ValueError, 'Duplicate'):
|
|
self.tracker.validate()
|
|
|
|
def test_missing_requirement_allocation_detected(self):
|
|
defs = self.tracker.definitions()
|
|
for task in defs['tasks']:
|
|
task['requirements'] = []
|
|
self.write('planning/tasks.json', defs)
|
|
with self.assertRaisesRegex(ValueError, 'Unallocated'):
|
|
self.tracker.validate()
|
|
|
|
def test_original_spec_edit_detected(self):
|
|
with (self.root / 'specs/tz_clear.md').open('a') as f:
|
|
f.write('Changed\n')
|
|
with self.assertRaisesRegex(ValueError, 'snapshot changed'):
|
|
self.tracker.validate()
|
|
|
|
def test_sync_adds_without_erasing_status(self):
|
|
self.claim()
|
|
defs = self.tracker.definitions()
|
|
new = copy.deepcopy(defs['tasks'][2])
|
|
new['id'] = 'S00-004'
|
|
defs['tasks'].append(new)
|
|
self.write('planning/tasks.json', defs)
|
|
self.assertEqual(self.tracker.sync(), ['S00-004'])
|
|
self.assertEqual(self.tracker.state()['tasks']['S00-001']['owner'], 'owner')
|
|
|
|
def test_sync_refuses_history_removal(self):
|
|
defs = self.tracker.definitions()
|
|
defs['tasks'].pop()
|
|
self.write('planning/tasks.json', defs)
|
|
with self.assertRaisesRegex(ValueError, 'remove task history'):
|
|
self.tracker.sync()
|
|
|
|
def test_packet_contains_requirements_and_dependency_state(self):
|
|
packet = self.tracker.packet('S00-002', 'owner')
|
|
self.assertIn('One real behavior.', packet)
|
|
self.assertIn('S00-001', packet)
|
|
self.assertIn('status=todo', packet)
|
|
self.assertLess(len(packet), 48000)
|
|
|
|
def test_render_deterministic_matrix_csv_and_cards(self):
|
|
self.tracker.render()
|
|
first = (self.root / 'planning/TASK_MATRIX.md').read_bytes()
|
|
self.tracker.render()
|
|
self.assertEqual(first, (self.root / 'planning/TASK_MATRIX.md').read_bytes())
|
|
self.assertIn('depends_on', (self.root / 'planning/task_matrix.csv').read_text())
|
|
self.assertIn('Один результат', (self.root / 'planning/tasks/S00-001.md').read_text())
|
|
|
|
def test_invalid_done_detected(self):
|
|
state = self.tracker.state()
|
|
state['tasks']['S00-001'] = {'status': 'done', 'owner': 'owner'}
|
|
self.write('state/tasks.json', state)
|
|
with self.assertRaises(ValueError):
|
|
self.tracker.validate()
|
|
|
|
|
|
def test_unsafe_scope_detected(self):
|
|
defs = self.tracker.definitions()
|
|
defs['tasks'][0]['scope'] = ['../other-repo/**']
|
|
self.write('planning/tasks.json', defs)
|
|
with self.assertRaisesRegex(ValueError, 'Unsafe task scope'):
|
|
self.tracker.validate()
|
|
|
|
def test_todo_cannot_be_approved(self):
|
|
with self.assertRaises(ValueError):
|
|
self.tracker.mutate('approve', 'S00-001', 'reviewer', role='qa', evidence=self.evidence())
|
|
|
|
def test_unknown_operator_role_rejected(self):
|
|
with self.assertRaisesRegex(ValueError, 'Unknown operator'):
|
|
self.tracker.ready('frnot')
|
|
|
|
def test_duplicate_stage_rejected(self):
|
|
defs = self.tracker.definitions()
|
|
defs['stages'].append(copy.deepcopy(defs['stages'][0]))
|
|
self.write('planning/tasks.json', defs)
|
|
with self.assertRaisesRegex(ValueError, 'Duplicate stage'):
|
|
self.tracker.validate()
|
|
|
|
def test_packet_refuses_oversized_context(self):
|
|
defs = self.tracker.definitions()
|
|
defs['tasks'][0]['outcome'] = 'x' * 49000
|
|
self.write('planning/tasks.json', defs)
|
|
with self.assertRaisesRegex(ValueError, '>48k'):
|
|
self.tracker.packet('S00-001', 'owner')
|
|
|
|
def test_failed_submit_keeps_revision_and_removes_short_lock(self):
|
|
self.claim()
|
|
before = self.tracker.state()
|
|
with self.assertRaises(ValueError):
|
|
self.tracker.mutate('submit', 'S00-001', 'owner', evidence='missing.md', handoff='missing.md')
|
|
self.assertEqual(before, self.tracker.state())
|
|
self.assertFalse(self.tracker.lock_path.exists())
|
|
|
|
|
|
class FoundationTests(unittest.TestCase):
|
|
def test_real_foundation_validates(self):
|
|
result = Tracker(ROOT).validate()
|
|
self.assertGreaterEqual(result['tasks'], 200)
|
|
self.assertGreaterEqual(result['stages'], 20)
|
|
|
|
def test_every_packet_fits_budget(self):
|
|
tracker = Tracker(ROOT)
|
|
lengths = [len(tracker.packet(t['id'], 'context-test')) for t in tracker.definitions()['tasks']]
|
|
self.assertLessEqual(max(lengths), 48000)
|
|
|
|
def test_cli_list_filters_future_tasks_without_claiming(self):
|
|
result = subprocess.run([sys.executable, str(ROOT / 'scripts/tasks.py'), 'list', '--role', 'crypto', '--stage', 'S12', '--status', 'todo', '--query', 'OMEMO', '--limit', '2'], capture_output=True, text=True)
|
|
self.assertEqual(result.returncode, 0)
|
|
rows = json.loads(result.stdout)
|
|
self.assertGreater(len(rows), 0)
|
|
self.assertLessEqual(len(rows), 2)
|
|
self.assertTrue(all(x['stage'] == 'S12' and x['role'] == 'crypto' for x in rows))
|
|
self.assertTrue(all(x['unmet_dependencies'] for x in rows))
|
|
|
|
def test_cli_next_and_invalid_command_exit(self):
|
|
good = subprocess.run([sys.executable, str(ROOT / 'scripts/tasks.py'), 'next', '--role', 'orchestrator', '--json'], capture_output=True, text=True)
|
|
self.assertEqual(good.returncode, 0)
|
|
self.assertIsInstance(json.loads(good.stdout), list)
|
|
bad = subprocess.run([sys.executable, str(ROOT / 'scripts/tasks.py'), 'show', 'MISSING'], capture_output=True, text=True)
|
|
self.assertEqual(bad.returncode, 2)
|
|
self.assertIn('Unknown task', bad.stderr)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|