railiance-platform/tests/test_task_execution_cleanup.py

"""Disposable local SQLite/owner-adapter fixtures. No OpenBao or credential access."""
import multiprocessing
import os
from pathlib import Path
import sys
import tempfile
import unittest
import uuid

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'scripts'))
from task_execution_journal import Binding, Cleanup, Journal, Refused
from task_execution_cleanup import CleanupResult, run_cleanup

OBLIGATIONS = (Cleanup('task-token-revoke', '1'*64),
               Cleanup('identity-token-revoke', '2'*64),
               Cleanup('grant-policy-delete', '3'*64))


class OwnerFixture:
    def __init__(self, fail=None): self.calls = []; self.fail = fail
    def call(self, kind, digest):
        self.calls.append((kind, digest))
        if kind == self.fail: raise TimeoutError('SYNTHETIC-PRIVATE-BACKEND-OUTPUT')
        return CleanupResult(True, 'e'*64)
    def revoke_task_token(self, digest): return self.call('task-token-revoke', digest)
    def revoke_identity_token(self, digest): return self.call('identity-token-revoke', digest)
    def delete_grant_policy(self, digest): return self.call('grant-policy-delete', digest)


def crash_worker(directory, binding, after_effect):
    journal = Journal(Path(directory))
    with journal.locked():
        journal.reserve(binding, OBLIGATIONS)
        journal.transition(binding, 'reserved', 'consume-confirmed')
        journal.before_cleanup(binding, 0, OBLIGATIONS[0])
        if after_effect:
            # A real durable synthetic effect, separate from the admission DB.
            fd = os.open(Path(directory)/'synthetic-effect', os.O_CREAT | os.O_WRONLY | os.O_APPEND, 0o600)
            os.write(fd, b'one fixture revocation\n'); os.fsync(fd); os.close(fd)
        os._exit(23)


def cleanup_racer(directory, binding, gate, results):
    journal = Journal(Path(directory))
    gate.wait()
    try:
        with journal.locked():
            journal.reserve(binding, OBLIGATIONS)
            journal.transition(binding, 'reserved', 'consume-confirmed')
            run_cleanup(journal, binding, OwnerFixture())
        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-cleanup', 'sha256:'+'a'*64, str(uuid.uuid4()), 'b'*64)
        self.journal = Journal(self.path); self.addCleanup(self.journal.close)

    def consumed(self):
        self.journal.reserve(self.binding, OBLIGATIONS)
        self.journal.transition(self.binding, 'reserved', 'consume-confirmed')

    def test_timeout_business_call_keeps_unknown_while_all_cleanup_runs(self):
        with self.journal.locked():
            self.consumed()
            self.journal.before_call(self.binding, 0, 'd'*64)
            self.journal.transition(self.binding, 'consume-confirmed', 'unknown')
            owner = OwnerFixture()
            status = run_cleanup(self.journal, self.binding, owner)
            self.assertTrue(status['confirmed'])
            self.assertEqual(owner.calls, [(c.kind,c.call_digest) for c in OBLIGATIONS])
            self.assertEqual(self.journal.observe(self.binding)['state'], 'unknown')
            self.assertTrue(self.journal.observe(self.binding)['external_outcome_unknown'])
            with self.assertRaises(Refused): self.journal.before_call(self.binding, 1, 'f'*64)

    def test_failed_business_outcome_preserved(self):
        with self.journal.locked():
            self.consumed(); self.journal.transition(self.binding,'consume-confirmed','failed')
            self.assertTrue(run_cleanup(self.journal,self.binding,OwnerFixture())['confirmed'])
            self.assertEqual(self.journal.observe(self.binding)['state'], 'failed')

    def test_one_revoke_failure_never_passes_but_independent_cleanup_continues(self):
        with self.journal.locked():
            self.consumed(); owner = OwnerFixture('task-token-revoke')
            result = run_cleanup(self.journal,self.binding,owner)
            self.assertFalse(result['confirmed']); self.assertEqual(len(owner.calls),3)
            self.assertEqual([r['outcome'] for r in result['obligations']], ['unconfirmed','confirmed','confirmed'])
            self.assertNotIn('SYNTHETIC-PRIVATE', str(result))
            with self.assertRaises(Refused): run_cleanup(self.journal,self.binding,owner)
            self.assertEqual(len(owner.calls),3)
        self.assertNotIn(b'SYNTHETIC-PRIVATE', (self.path/'journal.sqlite').read_bytes())

    def test_cleanup_closes_business_before_native_attempt(self):
        with self.journal.locked():
            self.consumed(); self.journal.before_cleanup(self.binding,0,OBLIGATIONS[0])
            with self.assertRaises(Refused): self.journal.before_call(self.binding,0,'d'*64)
            with self.assertRaises(Refused): self.journal.transition(self.binding,'consume-confirmed','authenticating')

    def test_no_preconsume_cleanup_even_when_unknown(self):
        with self.journal.locked():
            self.journal.reserve(self.binding,OBLIGATIONS)
            self.journal.transition(self.binding,'reserved','unknown')
            owner = OwnerFixture()
            with self.assertRaises(Refused): run_cleanup(self.journal,self.binding,owner)
            self.assertEqual(owner.calls,[])

    def test_changed_kind_digest_ordinal_or_binding_never_dispatches(self):
        with self.journal.locked():
            self.consumed()
            for ordinal,item in [(False,OBLIGATIONS[0]),(3,OBLIGATIONS[0]),(0,Cleanup('task-token-revoke','9'*64)),(0,Cleanup('kv-delete','1'*64)),(0,{'kind':'task-token-revoke'})]:
                with self.subTest(ordinal=ordinal,item=item):
                    with self.assertRaises(Refused): self.journal.before_cleanup(self.binding,ordinal,item)
            changed = Binding('another',self.binding.request_digest,self.binding.operation_id,self.binding.plan_digest)
            with self.assertRaises(Refused): self.journal.before_cleanup(changed,0,OBLIGATIONS[0])
            self.assertTrue(all(x['outcome']=='owed' for x in self.journal.cleanup_status(self.binding)['obligations']))

    def test_cleanup_plan_must_be_closed_before_reservation(self):
        for plan in [list(OBLIGATIONS), (OBLIGATIONS[0],OBLIGATIONS[0]), (Cleanup('kv-delete','1'*64),), (Cleanup('task-token-revoke',False),)]:
            with self.journal.locked():
                with self.assertRaises(Refused): self.journal.reserve(self.binding,plan)
                self.assertIsNone(self.journal.observe(self.binding))

    def test_crash_before_and_after_real_synthetic_effect_does_not_resume(self):
        context = multiprocessing.get_context('fork')
        for after_effect in (False, True):
            with tempfile.TemporaryDirectory() as directory:
                worker = context.Process(target=crash_worker,args=(directory,self.binding,after_effect))
                worker.start(); worker.join(10)
                if worker.is_alive(): worker.terminate(); worker.join(); self.fail('crash fixture exceeded bound')
                self.assertEqual(worker.exitcode,23)
                reopened = Journal(Path(directory))
                try:
                    self.assertEqual(reopened.cleanup_status(self.binding)['obligations'][0]['outcome'],'pending')
                    self.assertTrue(reopened.observe(self.binding)['external_outcome_unknown'])
                    owner = OwnerFixture()
                    with reopened.locked():
                        with self.assertRaises(Refused): reopened.reserve(self.binding,OBLIGATIONS)
                        with self.assertRaises(Refused): run_cleanup(reopened,self.binding,owner)
                        with self.assertRaises(Refused): reopened.before_cleanup(self.binding,1,OBLIGATIONS[1])
                    self.assertEqual(owner.calls,[])
                    effect = Path(directory)/'synthetic-effect'
                    self.assertEqual(effect.read_bytes() if effect.exists() else b'',b'one fixture revocation\n' if after_effect else b'')
                    self.assertFalse(reopened.cleanup_status(self.binding)['confirmed'])
                finally: reopened.close()

    def test_actual_process_race_admits_one_cleanup_driver(self):
        context = multiprocessing.get_context('fork'); gate = context.Event(); results = context.Queue()
        workers = [context.Process(target=cleanup_racer,args=(str(self.path),self.binding,gate,results)) for _ in range(4)]
        for worker in workers: worker.start()
        gate.set()
        for worker in workers:
            worker.join(10)
            if worker.is_alive(): worker.terminate(); worker.join(); self.fail('race exceeded bound')
            self.assertEqual(worker.exitcode,0)
        self.assertEqual(sorted(results.get(timeout=2) for _ in workers),['admitted','refused','refused','refused'])
        self.assertTrue(self.journal.cleanup_status(self.binding)['confirmed'])

    def test_empty_or_malformed_native_result_never_proves_cleanup(self):
        with self.journal.locked():
            self.journal.reserve(self.binding)
            self.journal.transition(self.binding,'reserved','consume-confirmed')
            with self.assertRaises(Refused): run_cleanup(self.journal,self.binding,OwnerFixture())
            self.assertFalse(self.journal.cleanup_status(self.binding)['confirmed'])

    def test_bad_native_confirmation_still_attempts_remaining_cleanup(self):
        class BadOwner(OwnerFixture):
            def revoke_task_token(self, digest):
                self.calls.append(('task-token-revoke',digest))
                return CleanupResult('true','e'*64)
        with self.journal.locked():
            self.consumed(); owner=BadOwner()
            self.assertFalse(run_cleanup(self.journal,self.binding,owner)['confirmed'])
            self.assertEqual(len(owner.calls),3)

    def test_confirmation_exact_types_and_no_duplicate_receipts(self):
        with self.journal.locked():
            self.consumed(); self.journal.before_cleanup(self.binding,0,OBLIGATIONS[0])
            for confirmed,digest in [('true','e'*64),(1,'e'*64),(True,None),(True,'E'*64)]:
                with self.assertRaises(Refused): self.journal.after_cleanup(self.binding,0,OBLIGATIONS[0],confirmed=confirmed,evidence_digest=digest)
            self.journal.after_cleanup(self.binding,0,OBLIGATIONS[0],confirmed=True,evidence_digest='e'*64)
            with self.assertRaises(Refused): self.journal.after_cleanup(self.binding,0,OBLIGATIONS[0],confirmed=True,evidence_digest='e'*64)
            with self.assertRaises(Refused): self.journal.before_cleanup(self.binding,0,OBLIGATIONS[0])

    def test_cleanup_state_is_independent_of_legacy_business_transition(self):
        with self.journal.locked():
            self.consumed()
            states=['consume-confirmed','authenticating','grant-issuing','grant-issued','applying','readback','verified']
            for before,after in zip(states,states[1:]): self.journal.transition(self.binding,before,after)
            with self.assertRaises(Refused): self.journal.transition(self.binding,'verified','cleanup-confirmed','e'*64)
            run_cleanup(self.journal,self.binding,OwnerFixture())
            self.journal.transition(self.binding,'verified','cleanup-confirmed','e'*64)
            self.assertEqual(self.journal.observe(self.binding)['state'],'cleanup-confirmed')


if __name__ == '__main__': unittest.main()