import importlib.util import multiprocessing import os import sys import tempfile import unittest import uuid from pathlib import Path spec = importlib.util.spec_from_file_location('task_execution_journal', Path(__file__).resolve().parents[1] / 'scripts/task_execution_journal.py') module = importlib.util.module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) Binding, Journal, Refused = module.Binding, module.Journal, module.Refused def reserve_worker(directory, binding, gate, results): journal = Journal(Path(directory)) gate.wait() try: with journal.locked(): journal.reserve(binding) results.put('admitted') except Refused: results.put('refused') finally: journal.close() class Tests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.path = Path(self.tmp.name) self.binding = Binding('approval-1', 'sha256:'+'a'*64, str(uuid.uuid4()), 'b'*64) self.journal = Journal(self.path) self.addCleanup(self.journal.close) def test_actual_multi_process_race_admits_once(self): context = multiprocessing.get_context('fork') gate, results = context.Event(), context.Queue() workers = [context.Process(target=reserve_worker,args=(str(self.path),self.binding,gate,results)) for _ in range(4)] for w in workers: w.start() gate.set() for w in workers: w.join(10) if w.is_alive(): w.terminate(); w.join(); self.fail('worker exceeded bound') self.assertEqual(w.exitcode, 0) self.assertEqual(sorted(results.get(timeout=2) for _ in workers), ['admitted','refused','refused','refused']) def test_reservation_persists_after_process_abandonment(self): with self.journal.locked(): self.journal.reserve(self.binding) reopened = Journal(self.path) try: self.assertEqual(reopened.observe(self.binding)['state'], 'reserved') self.assertFalse(reopened.observe(self.binding)['dispatch_allowed']) with reopened.locked(): with self.assertRaises(Refused): reopened.reserve(self.binding) with self.assertRaises(Refused): reopened.transition(self.binding,'reserved','consume-confirmed') finally: reopened.close() def test_same_approval_different_digest_refused(self): with self.journal.locked(): self.journal.reserve(self.binding) changed = Binding('approval-1','sha256:'+'c'*64,str(uuid.uuid4()),'b'*64) with self.journal.locked(): with self.assertRaises(Refused): self.journal.reserve(changed) with self.assertRaises(Refused): self.journal.observe(changed) def test_same_operation_different_approval_refused(self): with self.journal.locked(): self.journal.reserve(self.binding) with self.journal.locked(): with self.assertRaises(Refused): self.journal.reserve(Binding('approval-2','sha256:'+'a'*64,self.binding.operation_id,'b'*64)) def test_full_path_cleanup_and_terminal_replay(self): states=['reserved','consume-confirmed','authenticating','grant-issuing','grant-issued','applying','readback','verified','cleanup-confirmed'] with self.journal.locked(): self.journal.reserve(self.binding) for before,after in zip(states,states[1:]): self.journal.transition(self.binding,before,after,'c'*64) with self.assertRaises(Refused): self.journal.transition(self.binding,'cleanup-confirmed','applying') self.assertEqual(self.journal.observe(self.binding),{'state':'cleanup-confirmed','evidence_digest':'c'*64,'dispatch_allowed':False,'external_outcome_unknown':False}) with self.journal.locked(): with self.assertRaises(Refused): self.journal.reserve(self.binding) def test_unknown_is_terminal(self): with self.journal.locked(): self.journal.reserve(self.binding) self.journal.transition(self.binding,'reserved','unknown') with self.assertRaises(Refused): self.journal.transition(self.binding,'unknown','reserved') def test_early_dispatch_wrong_state_and_cleanup_without_evidence_refused(self): with self.journal.locked(): self.journal.reserve(self.binding) with self.assertRaises(Refused): self.journal.transition(self.binding,'reserved','applying') with self.assertRaises(Refused): self.journal.transition(self.binding,'consume-confirmed','authenticating') self.assertEqual(self.journal.observe(self.binding)['state'],'reserved') def test_other_live_invocation_cannot_obtain_lock(self): second=Journal(self.path) try: with self.journal.locked(): with self.assertRaises(Refused): with second.locked(): pass finally: second.close() def test_symlink_and_hardlink_files_refused(self): for name,link in [('journal.sqlite','hard'),('execution.lock','symbolic')]: with tempfile.TemporaryDirectory() as d: p=Path(d); target=p/'other'; target.touch(mode=0o600) if link=='hard': os.link(target,p/name) else: (p/name).symlink_to(target) with self.assertRaises((Refused,OSError)): Journal(p) def test_nonprivate_directory_refused(self): with tempfile.TemporaryDirectory() as d: os.chmod(d,0o755) with self.assertRaises(Refused): Journal(Path(d)) def test_binding_rejects_unbounded_or_malformed_metadata(self): for value in ['token'+ 'x'*200, '../approval', 'approval\n1']: with self.assertRaises(Refused): Binding(value,'sha256:'+'a'*64,self.binding.operation_id,'b'*64).validate() for value in ['a'*64,'sha256:'+'a'*63,'sha256:'+'A'*64,'not-a-digest']: with self.assertRaises(Refused): Binding('a',value,self.binding.operation_id,'b'*64).validate() def test_pending_external_call_blocks_next_and_survives_reopen(self): with self.journal.locked(): self.journal.reserve(self.binding) self.journal.before_call(self.binding,0,'d'*64) with self.assertRaises(Refused): self.journal.before_call(self.binding,1,'e'*64) with self.assertRaises(Refused): self.journal.transition(self.binding,'reserved','consume-confirmed') reopened=Journal(self.path) try: self.assertTrue(reopened.observe(self.binding)['external_outcome_unknown']) self.assertFalse(reopened.observe(self.binding)['dispatch_allowed']) finally: reopened.close() def test_call_order_and_exact_confirmation(self): with self.journal.locked(): self.journal.reserve(self.binding) with self.assertRaises(Refused): self.journal.before_call(self.binding,1,'d'*64) self.journal.before_call(self.binding,0,'d'*64) with self.assertRaises(Refused): self.journal.after_call(self.binding,0,'e'*64,'f'*64) self.journal.after_call(self.binding,0,'d'*64,'f'*64) with self.assertRaises(Refused): self.journal.after_call(self.binding,0,'d'*64,'f'*64) with self.assertRaises(Refused): self.journal.before_call(self.binding,0,'d'*64) self.journal.before_call(self.binding,1,'e'*64) def test_nonstring_binding_and_boolean_ordinal_refused(self): with self.assertRaises(Refused): Binding(True,'sha256:'+'a'*64,self.binding.operation_id,'b'*64).validate() with self.journal.locked(): self.journal.reserve(self.binding) with self.assertRaises(Refused): self.journal.before_call(self.binding,False,'d'*64) def test_confirmation_rejects_sqlite_ordinal_coercion(self): with self.journal.locked(): self.journal.reserve(self.binding) self.journal.before_call(self.binding,0,'d'*64) for ordinal in (False,0.0,'0',None,-1,64): with self.subTest(ordinal=ordinal): with self.assertRaises(Refused): self.journal.after_call(self.binding,ordinal,'d'*64,'f'*64) self.assertTrue(self.journal.observe(self.binding)['external_outcome_unknown']) for digest in (None,False,'D'*64,'d'*63): with self.assertRaises(Refused): self.journal.after_call(self.binding,0,digest,'f'*64) self.assertTrue(self.journal.observe(self.binding)['external_outcome_unknown']) self.journal.after_call(self.binding,0,'d'*64,'f'*64) self.assertFalse(self.journal.observe(self.binding)['external_outcome_unknown']) if __name__=='__main__': unittest.main()