"""Synthetic custody/transport tests. No native data, Docker or cluster action."""
import copy
import datetime
import io
import json
import os
from pathlib import Path
import subprocess
import sys
import tempfile
import unittest
from unittest.mock import patch
from contextlib import redirect_stdout

import actuator
from authority_journal import Journal, authority, fresh_scan
from native_program import PROGRAM
from receipt_safety import counts

CASE = Path(__file__).resolve().parent.parent
NOW = datetime.datetime(2026, 10, 11, 2, 0, tzinfo=datetime.timezone.utc)
REVISION = 'e' * 40
COUNT = {'schema_version': 5, 'integrity': 'ok', 'read_only_pre_open': True, 'store_constructed': False,
         'lifecycle': {'records': 3, 'events': 8, 'pending': 1},
         'outbox': {'pending': 2, 'delivered': 10}, 'submissions': {'confirmed': 1, 'unresolved': 1}}


def accepted():
    approval = '11111111-1111-4111-8111-111111111111'
    review = 'review-22222222-2222-4222-8222-222222222222'
    disposition = 'disp-33333333-3333-4333-8333-333333333333'
    receipt = {'proposal_id': 'INFD-WP-0009-T05/zlib85091-r1',
               'memo_id': 'infd-20261011-zlib-remediation-r1', 'memo_version': 1,
               'memo_sha256': 'a' * 64, 'source_revision': REVISION, 'activation': 'APPROVED',
               'authority_mode': 'ADMINISTER', 'target': 'realm:kubernetes/railiance01/informed-decision',
               'verification_method': 'root-fixed-origin-native-readback', 'runtime_authenticated_grant': False, 'outcome_query': '/lifecycle/memos/infd-20261011-zlib-remediation-r1/outcomes?version=1',
               'verified_at': '2026-10-11T02:00:00Z', 'accepted_at': '2026-10-11T01:59:00Z',
               'review_memo_id': review, 'approval_id': approval, 'disposition_id': disposition}
    outcome = {'memo_id': receipt['memo_id'], 'review': {'memo_id': review, 'stale': False, 'intake_state': 'ready'},
               'dispositions': [{'approval_id': None, 'submission_state': 'confirmed',
                                 'disposition': {'verb': 'accept', 'memo_version': 1, 'memo_id': review,
                                                 'approval_id': approval, 'id': disposition,
                                                 'actor': {'kind': 'person', 'sub': 'uid=platform-root,ou=people,dc=netkingdom,dc=local'},
                                                 'at': receipt['accepted_at']}}]}
    return receipt, outcome


class AuthorityAndCustody(unittest.TestCase):
    def test_exact_nested_founder_receipt_not_top_level_approval(self):
        receipt, outcome = accepted()
        result = authority(receipt, outcome, 'a' * 64, REVISION, 'drill-copy', NOW)
        self.assertEqual(result['approval_id'], receipt['approval_id'])

    def test_pending_stale_wrong_actor_version_and_binding_refuse(self):
        mutations = [lambda r, o: r.update(activation='PENDING'),
                     lambda r, o: r.update(memo_sha256='changed'),
                     lambda r, o: r.update(source_revision='changed'),
                     lambda r, o: r.update(target='other'),
                     lambda r, o: r.update(verified_at='2026-10-11T01:00:00Z'),
                     lambda r, o: r.update(verification_method='caller-asserted'),
                     lambda r, o: r.update(runtime_authenticated_grant=True),
                     lambda r, o: o['review'].update(stale=True),
                     lambda r, o: o['review'].update(intake_state='pending'),
                     lambda r, o: o['dispositions'][0].update(submission_state='unresolved'),
                     lambda r, o: o['dispositions'][0]['disposition'].update(memo_version=3),
                     lambda r, o: o['dispositions'][0]['disposition'].update(actor={'kind': 'agent', 'sub': 'platform-root'}),
                     lambda r, o: o['dispositions'].append(copy.deepcopy(o['dispositions'][0]))]
        for mutate in mutations:
            with self.subTest(mutate=mutate):
                r, o = accepted()
                mutate(r, o)
                with self.assertRaises(ValueError):
                    authority(r, o, 'a' * 64, REVISION, 'release', NOW)

    def test_expired_window_and_unknown_phase_refuse(self):
        r, o = accepted()
        with self.assertRaises(ValueError):
            authority(r, o, 'a' * 64, REVISION, 'arbitrary', NOW)
        late = datetime.datetime(2026, 10, 13, tzinfo=datetime.timezone.utc)
        r['verified_at'] = late.isoformat()
        with self.assertRaises(ValueError):
            authority(r, o, 'a' * 64, REVISION, 'rollback', late)

    def test_durable_intent_precedes_dispatch_and_never_retries(self):
        with tempfile.TemporaryDirectory() as directory:
            parent = Path(directory)
            os.chmod(parent, 0o700)
            journal = Journal(parent, {'case': 'fixture'})
            journal.start('drill-backup', {'approval_id': 'fixture'})
            self.assertEqual(json.loads(journal.path.read_text())['phases']['drill-backup']['state'], 'RUNNING')
            journal.close()  # simulated crash before result
            recovered = Journal(parent, {'case': 'fixture'})
            with self.assertRaises(ValueError):
                recovered.start('drill-backup', {})
            with self.assertRaises(ValueError):
                recovered.start('drill-backup', {})
            recovered.close()

    def test_phase_order_and_uncertain_release_allows_only_distinct_recovery(self):
        with tempfile.TemporaryDirectory() as directory:
            os.chmod(directory, 0o700)
            j = Journal(directory, {'case': 'fixture'})
            with self.assertRaises(ValueError):
                j.start('release', {})
            for phase in actuator.PHASES[:6]:
                j.start(phase, {})
                j.finish(phase, {'all_checks_pass': True})
            j.start('release', {})
            j.finish('release', {'all_checks_pass': False})
            with self.assertRaises(ValueError):
                j.start('release', {})
            with self.assertRaises(ValueError):
                j.start('verify', {})
            j.start('rollback', {})
            j.finish('rollback', {'all_checks_pass': True})
            j.close()

    def test_secret_or_unknown_child_output_is_withheld(self):
        for body in ('not-json', '{"all_checks_pass":true,"token":"private-value"}',
                     '{"all_checks_pass":true,"snapshot_readonly":{"schema_version":5}}'):
            with self.assertRaises(ValueError):
                actuator.json_result(subprocess.CompletedProcess([], 0, body, ''))
        wrong = copy.deepcopy(COUNT)
        wrong['outbox'] = {'private-row-value': 1}
        with self.assertRaises(ValueError):
            counts(wrong)
        counts(COUNT)  # blocked/unresolved counts are not required to be zero.

    def test_binary_copy_is_bounded_private_and_never_echoed(self):
        blob = b'synthetic private fixture bytes'
        digest = __import__('hashlib').sha256(blob).hexdigest()
        metadata = {'all_checks_pass': True, 'snapshot_file': 'drill.sqlite', 'snapshot_bytes': len(blob),
                    'snapshot_sha256': digest, 'snapshot_readonly': COUNT,
                    'pod_name': 'fixture-pod', 'pod_uid': 'fixture-uid'}
        journal = type('FixtureJournal', (), {'data': {'phases': {'drill-backup': {'result': metadata}}}})()
        real_popen = subprocess.Popen
        for extra in (b'', b'oversized'):
            with self.subTest(extra=extra), tempfile.TemporaryDirectory() as folder:
                os.chmod(folder, 0o700)
                output = io.StringIO()

                def child(*args, **kwargs):
                    return real_popen([sys.executable, '-c', 'import sys;sys.stdout.buffer.write(' + repr(blob + extra) + ')'],
                                      stdout=subprocess.PIPE, stderr=subprocess.DEVNULL)

                with patch.object(actuator, 'CUSTODY', Path(folder)), patch.object(actuator, 'remote', return_value=metadata), \
                        patch('subprocess.Popen', side_effect=child), redirect_stdout(output):
                    if extra:
                        with self.assertRaises(ValueError):
                            actuator.copy_snapshot('drill-copy', REVISION, {}, journal)
                    else:
                        result = actuator.copy_snapshot('drill-copy', REVISION, {}, journal)
                        self.assertTrue(result['off_cluster_private_copy_verified'])
                        self.assertEqual((Path(folder) / 'drill.sqlite').read_bytes(), blob)
                self.assertEqual(output.getvalue(), '')
                self.assertEqual((Path(folder) / 'drill.sqlite').stat().st_mode & 0o777, 0o600)

    def test_recreated_offline_volume_cannot_be_deleted(self):
        identity = {'Name': actuator.VOLUME, 'Labels': {'nonce': 'old'}, 'CreatedAt': 'old'}
        journal = type('FixtureJournal', (), {'data': {'phases': {
            'offline-verify': {'result': {'created_volume_identity': identity}}}}})()
        inspected = [{**identity, 'CreatedAt': 'recreated'}]
        with patch.object(actuator, 'docker', return_value=subprocess.CompletedProcess([], 0, json.dumps(inspected), '')) as mocked:
            with self.assertRaises(ValueError):
                actuator.cleanup_offline(journal)
            self.assertEqual(mocked.call_count, 1)

    def test_offline_post_admin_integrity_and_original_hash_are_rechecked(self):
        for defect in ('none', 'post-integrity', 'original-hash'):
            with self.subTest(defect=defect), tempfile.TemporaryDirectory() as directory:
                folder=Path(directory);os.chmod(folder,0o700)
                snapshot=folder/'drill.sqlite';snapshot.write_bytes(b'private synthetic snapshot');os.chmod(snapshot,0o600)
                digest=actuator.sha(snapshot)
                prior={'snapshot_sha256':digest,'snapshot_readonly':COUNT}
                journal=type('FixtureJournal',(),{'data':{'phases':{'drill-copy':{'result':prior},'offline-verify':{'attempt_id':'nonce'}}},'note':lambda *args:None})()
                reads=[]
                volume={'Name':actuator.VOLUME,'Labels':{'informed-decision.case':'infd-zlib-20261011-r1','informed-decision.use':'offline-restore-only','informed-decision.attempt':'nonce'},'Driver':'local','Scope':'local','CreatedAt':'fixture','Options':None}
                def docker(argv, body=None, **kwargs):
                    if argv[0]=='info':body=json.dumps('bnt-lap001')
                    elif argv[:2]==['volume','ls']:body=''
                    elif argv[:2]==['volume','create']:body=actuator.VOLUME
                    elif argv[:2]==['volume','inspect']:body=json.dumps([volume])
                    elif 'informed-decision-admin' in argv:
                        body=json.dumps({key:COUNT[key] for key in ('schema_version','lifecycle','outbox','submissions')})
                        if defect=='original-hash':snapshot.write_bytes(b'changed synthetic original')
                    elif '--entrypoint' in argv and 'python' in argv:
                        reads.append(argv);proof=copy.deepcopy(COUNT)
                        if len(reads)==2 and defect=='post-integrity':proof['integrity']='corrupt'
                        body=json.dumps(proof)
                    else:raise AssertionError('unexpected offline command')
                    return subprocess.CompletedProcess(argv,0,body,'')
                with patch.object(actuator,'CUSTODY',folder),patch.object(actuator,'CASE',CASE),patch.object(actuator,'docker',side_effect=docker),patch('subprocess.run',return_value=subprocess.CompletedProcess([],0,'{"all_checks_pass":true,"private_copy_verified":true}','')):
                    if defect=='none':
                        result=actuator.offline_verify(journal)
                        self.assertEqual(result['offline_post_admin_readonly'],COUNT)
                        self.assertTrue(result['original_snapshot_hash_unchanged'])
                    else:
                        with self.assertRaises(ValueError):actuator.offline_verify(journal)
                self.assertEqual(len(reads),2)

    def test_fresh_full_scan_db_expiry_and_new_findings_refuse(self):
        receipt={'image':actuator.IMAGE,'artifact_source_revision':'9b4bb25d325698ae543e7e8a8a3c3592db1f58fc','scanner':'aquasec/trivy@sha256:62b1e65e8869bc4b4c6aa4fa2b21595256c7c2f6018a9d9ad61caf87187c1969','no_suppression':True,'full_allseverity_scan':True,'exit_code':0,'completed_at':'2026-10-11T01:59:00Z','db_metadata':{'UpdatedAt':'2026-10-10T19:00:00Z','NextUpdate':'2026-10-11T19:00:00Z','Version':2},'db_sha256':'a'*64,'full_report_sha256':'b'*64}
        contract=json.loads((CASE/'fresh-scan-contract.json').read_text())
        report=json.loads((CASE/'fresh-scan-report-fixture.json').read_text())
        self.assertEqual(fresh_scan(receipt,report,contract,NOW)['allseverity_findings'],0)
        mutations=[lambda r,p:r.update(no_suppression=False),lambda r,p:r.update(image='foreign'),lambda r,p:r.update(scanner='other'),lambda r,p:r.update(completed_at='2026-10-10T21:00:00Z'),lambda r,p:r['db_metadata'].update(NextUpdate='2026-10-11T01:59:00Z'),lambda r,p:p['Results'][0].update(Vulnerabilities=[{'Severity':'LOW'}]),lambda r,p:p.update(Results=[]),lambda r,p:p.update(Results=[{}]),lambda r,p:p['Results'].pop(),lambda r,p:p['Results'][0]['Packages'].pop(),lambda r,p:p['Results'][0]['Packages'][0].update(Version='old'),lambda r,p:p['Metadata'].update(ImageID='other'),lambda r,p:p['Metadata']['DiffIDs'].pop(),lambda r,p:p['Metadata']['ImageConfig']['config']['Labels'].update(**{'org.opencontainers.image.revision':'other'}),lambda r,p:p.update(SchemaVersion=1),lambda r,p:p.update(ArtifactType='filesystem')]
        for mutate in mutations:
            r,p=copy.deepcopy(receipt),copy.deepcopy(report);mutate(r,p)
            with self.assertRaises(ValueError):fresh_scan(r,p,contract,NOW)


class FixedNativeProgram(unittest.TestCase):
    def fixture(self):
        sealed = json.loads((CASE / 'sealed-live-deployment.json').read_text())
        live = copy.deepcopy(sealed)
        live['metadata'].update(generation=42)
        live['status'] = {'readyReplicas': 1, 'updatedReplicas': 1, 'observedGeneration': 42}
        dependencies = json.loads((CASE / 'sealed-native-dependencies.json').read_text())
        cm = copy.deepcopy(dependencies['configmap'])
        cm.pop('canonical_data_sha256')
        service = json.loads((CASE / 'candidate-deployment.json').read_text())
        source_service = json.loads((Path('/home/worsch/informed-decision/deploy/lifecycle/service.candidate.json')).read_text())
        cm['data'] = next(x for x in source_service['items'] if x['kind'] == 'ConfigMap')['data']
        pvc = copy.deepcopy(dependencies['pvc'])
        pvc['status'] = {'phase': 'Bound'}
        app = {'metadata': {'uid': '499868f2-eabc-4df4-ae31-b89ce58d750d', 'resourceVersion': '17'},
               'spec': {'source': {'repoURL': 'https://forgejo.coulomb.social/coulomb/informed-decision.git',
                                   'path': 'deploy/lifecycle', 'targetRevision': 'c96b0c6d881e855d1d158003c6de90ef390262e2'},
                        'destination': {'namespace': 'informed-decision', 'server': 'https://kubernetes.default.svc'}}}
        objects = {'deployment': live, 'configmap': cm, 'pvc': pvc, 'application': app}
        payload = {'phase': 'release', 'source_revision': REVISION, 'approval': {}, 'sealed': sealed,
                   'dependencies': dependencies, 'candidate': service,
                   'rollback': json.loads((CASE / 'rollback-deployment.json').read_text()),
                   'readonly': (CASE / 'sqlite_readonly.py').read_text(),'purpose_checks':(CASE/'check_installed.py').read_text(),
                   'installed_zlib': json.loads((CASE / 'installed-zlib-proof.json').read_text()), 'receipt_safety': (Path(__file__).parent / 'receipt_safety.py').read_text()}
        return objects, payload

    def execute_fake(self, objects, payload, ambiguous=False, schema_refused=False):
        writes = []
        calls = []

        def fake(argv, input=None, **kwargs):
            calls.append(argv)
            if argv[:2] == ['kubectl', 'get']:
                kind = argv[2]
                if kind in objects:
                    body = objects[kind]
                elif kind == 'pods':
                    body = {'items': [{'metadata': {'name': 'fixture-pod', 'uid': 'c4cb1550-feda-4097-9de4-d891a657b662',
                                                   'ownerReferences': [{'kind': 'ReplicaSet', 'name': 'fixture-rs', 'uid': 'rs-uid'}]},
                                      'spec': objects['deployment']['spec']['template']['spec'],
                                      'status': {'phase': 'Running', 'containerStatuses': [{'name':'informed-decision','ready': True,'imageID':objects['deployment']['spec']['template']['spec']['containers'][0]['image']}] }}]}
                elif kind == 'replicaset':
                    body = {'metadata': {'uid': 'rs-uid', 'ownerReferences': [
                        {'kind': 'Deployment', 'uid': payload['sealed']['metadata']['uid']}]}}
                elif kind == 'nodes':
                    body = {'items': [{'metadata': {'name': 'fixture-node'}, 'status': {
                        'addresses': [{'address': '92.205.62.239'}], 'allocatable': {'cpu': '4'},
                        'conditions': [{'type': key, 'status': 'True' if key == 'Ready' else 'False'}
                                       for key in ('Ready', 'MemoryPressure', 'DiskPressure', 'PIDPressure')]}}]}
                else:
                    raise AssertionError('unexpected fixed metadata route')
                return subprocess.CompletedProcess(argv, 0, json.dumps(body), '')
            if argv[:2] == ['kubectl', 'exec']:
                if schema_refused:
                    return subprocess.CompletedProcess(argv, 1, '', 'withheld')
                return subprocess.CompletedProcess(argv, 0, json.dumps({
                    'all_checks_pass': True, 'live_readonly_schema_integrity_counts': COUNT}), '')
            if argv[:2] == ['kubectl', 'apply']:
                admitted = copy.deepcopy(payload['sealed'])
                admitted['spec']['template']['spec']['containers'][0]['image'] = (actuator.IMAGE if payload['phase']=='release' else payload['sealed']['spec']['template']['spec']['containers'][0]['image'])
                return subprocess.CompletedProcess(argv, 0, json.dumps(admitted), '')
            if argv[:2] == ['kubectl', 'describe']:
                return subprocess.CompletedProcess(argv, 0, 'Allocated resources:\n cpu 3460m (86%)\n', '')
            if argv[:2] == ['kubectl', 'diff']:
                return subprocess.CompletedProcess(argv, 1, '-image: base\n+image: candidate\n', '')
            if argv[:2] == ['kubectl', 'patch']:
                if '--dry-run=server' not in argv:
                    writes.append(json.loads(input))
                    if ambiguous:
                        return subprocess.CompletedProcess(argv, 1, '', 'withheld')
                return subprocess.CompletedProcess(argv, 0, 'application.argoproj.io/informed-decision\n', '')
            raise AssertionError('unexpected command')

        source = PROGRAM.replace('__PAYLOAD__', repr(json.dumps(payload))).replace(
            '__GUARD__', repr((CASE / 'change_guard.py').read_text()))
        output = io.StringIO()
        with patch('subprocess.run', side_effect=fake), redirect_stdout(output):
            try:
                exec(compile(source, 'fixed-native-fixture', 'exec'), {'__name__': '__fixture__'})
            except (ValueError, SystemExit):
                pass
        receipt = json.loads(output.getvalue()) if output.getvalue() else {'all_checks_pass': False}
        return receipt, writes, calls

    def test_only_existing_application_operation_and_deployment_are_requested(self):
        objects, payload = self.fixture()
        receipt, writes, calls = self.execute_fake(objects, payload)
        self.assertTrue(receipt['all_checks_pass'])
        self.assertFalse(receipt['rollout_verified'])
        self.assertEqual(len(writes), 1)
        operation = writes[0][-1]['value']
        self.assertFalse(operation['sync']['prune'])
        self.assertEqual(operation['sync']['resources'], [
            {'group': 'apps', 'kind': 'Deployment', 'name': 'informed-decision', 'namespace': 'informed-decision'}])
        self.assertTrue(any('--dry-run=server' in argv and argv[:2] == ['kubectl', 'patch'] for argv in calls))
        self.assertEqual(writes[0][0]['path'], '/metadata/uid')

    def test_source_overrides_automation_and_custody_drift_stop_before_write(self):
        mutations = [lambda o: o['application']['metadata'].update(uid='recreated'),
                     lambda o: o['application']['spec'].update(syncPolicy={'automated': {}}),
                     lambda o: o['application']['spec']['source'].update(kustomize={'images': ['other']}),
                     lambda o: o['application']['spec'].update(sources=[{}]),
                     lambda o: o['configmap']['metadata'].update(uid='recreated'),
                     lambda o: o['pvc']['spec'].update(volumeName='replaced'),
                     lambda o: o['deployment']['status'].update(readyReplicas=0)]
        for mutate in mutations:
            with self.subTest(mutate=mutate):
                objects, payload = self.fixture()
                mutate(objects)
                receipt, writes, calls = self.execute_fake(objects, payload)
                self.assertFalse(receipt['all_checks_pass'])
                self.assertEqual(writes, [])

    def test_preopen_schema_refusal_prevents_image_operation(self):
        objects, payload = self.fixture()
        receipt, writes, calls = self.execute_fake(objects, payload, schema_refused=True)
        self.assertFalse(receipt['all_checks_pass'])
        self.assertEqual(writes, [])


    def test_rollback_unready_candidate_never_reads_pod_or_database(self):
        objects, payload = self.fixture()
        payload['phase'] = 'rollback'
        objects['deployment']['status'].update(readyReplicas=0, updatedReplicas=0)
        objects['deployment']['spec']['template']['spec']['containers'][0]['image'] = actuator.IMAGE
        objects['application']['spec']['source']['targetRevision'] = REVISION
        receipt, writes, calls = self.execute_fake(objects, payload, schema_refused=True)
        self.assertTrue(receipt['all_checks_pass'])
        self.assertFalse(receipt['rollout_verified'])
        self.assertEqual(len(writes), 1)
        self.assertFalse(any(argv[:2] == ['kubectl', 'exec'] or argv[:3] == ['kubectl', 'get', 'pods'] for argv in calls))
        self.assertEqual(writes[0][-1]['value']['sync']['revision'], 'c96b0c6d881e855d1d158003c6de90ef390262e2')

    def test_rollback_still_refuses_custody_or_nonpair_image_drift(self):
        for defect in ('cm', 'pvc', 'image'):
            with self.subTest(defect=defect):
                objects, payload = self.fixture();payload['phase'] = 'rollback'
                objects['deployment']['status']['readyReplicas'] = 0
                if defect == 'cm':objects['configmap']['metadata']['uid'] = 'recreated'
                elif defect == 'pvc':objects['pvc']['spec']['volumeName'] = 'replaced'
                else:objects['deployment']['spec']['template']['spec']['containers'][0]['image'] = 'foreign-image'
                receipt, writes, calls = self.execute_fake(objects, payload)
                self.assertFalse(receipt['all_checks_pass']);self.assertEqual(writes, [])

    def test_recovery_verification_is_separate_ready_old_image_schema_proof(self):
        objects, payload = self.fixture();payload['phase'] = 'rollback-verify'
        resource={'group':'apps','kind':'Deployment','name':'informed-decision','namespace':'informed-decision'}
        target='c96b0c6d881e855d1d158003c6de90ef390262e2'
        payload['operation_intent']={'state':'COMPLETE','started_at':'2026-10-11T01:00:00Z','result':{'operation_requested':True,'source_revision':target,'prune':False,'resources':[resource]}}
        objects['application']['status']={'operationState':{'phase':'Succeeded','startedAt':'2026-10-11T01:00:01Z','finishedAt':'2026-10-11T01:01:00Z','operation':{'initiatedBy':{'username':'custodian-codex'},'sync':{'revision':target,'resources':[resource]}},'syncResult':{'revision':target,'resources':[{**resource,'status':'Synced'}]}}}
        receipt, writes, calls = self.execute_fake(objects, payload)
        self.assertTrue(receipt['all_checks_pass']);self.assertTrue(receipt['recovery_verification'])
        self.assertEqual(writes, [])
        self.assertTrue(any(argv[:2]==['kubectl','exec'] for argv in calls))
        objects['deployment']['status']['readyReplicas'] = 0
        receipt, writes, calls = self.execute_fake(objects,payload)
        self.assertFalse(receipt['all_checks_pass'])

    def test_ambiguous_patch_is_never_retried(self):
        objects, payload = self.fixture()
        receipt, writes, calls = self.execute_fake(objects, payload, ambiguous=True)
        self.assertFalse(receipt['all_checks_pass'])
        self.assertEqual(len(writes), 1)

    def test_false_omitempty_needs_bound_current_request_and_exact_completion(self):
        objects,payload=self.fixture();payload['phase']='rollback-verify'
        resource={'group':'apps','kind':'Deployment','name':'informed-decision','namespace':'informed-decision'}
        target='c96b0c6d881e855d1d158003c6de90ef390262e2'
        payload['operation_intent']={'state':'COMPLETE','started_at':'2026-10-11T01:00:00Z','result':{'operation_requested':True,'source_revision':target,'prune':False,'resources':[resource]}}
        objects['application']['status']={'operationState':{'phase':'Succeeded','startedAt':'2026-10-11T01:00:01Z','finishedAt':'2026-10-11T01:01:00Z','operation':{'initiatedBy':{'username':'custodian-codex'},'sync':{'revision':target,'resources':[resource]}},'syncResult':{'revision':target,'resources':[{**resource,'status':'Synced'}]}}}
        self.assertTrue(self.execute_fake(objects,payload)[0]['all_checks_pass'])
        mutations=[lambda o,p:o['application']['status']['operationState']['operation']['sync'].update(prune=True),
                   lambda o,p:o['application']['status']['operationState']['operation']['sync'].update(prune=None),
                   lambda o,p:o['application']['status']['operationState'].update(startedAt='2026-10-11T00:59:00Z'),
                   lambda o,p:o['application']['status']['operationState']['operation'].update(initiatedBy={'username':'other'}),
                   lambda o,p:o['application']['status']['operationState']['syncResult'].update(revision='foreign'),
                   lambda o,p:o['application']['status']['operationState']['syncResult']['resources'][0].update(status='Pruned'),
                   lambda o,p:p['operation_intent']['result'].update(prune=True),
                   lambda o,p:p.update(operation_intent=None)]
        for mutate in mutations:
            o,p=copy.deepcopy(objects),copy.deepcopy(payload);mutate(o,p)
            receipt,writes,calls=self.execute_fake(o,p)
            self.assertFalse(receipt['all_checks_pass']);self.assertEqual(writes,[])

    def test_optimized_executor_refuses_before_any_input(self):
        result=subprocess.run([sys.executable,'-O',str(Path(__file__).parent/'actuator.py'),'--phase','release','--root-authority-receipt','/missing','--root-native-outcome','/missing','--source-revision',REVISION],capture_output=True,text=True)
        self.assertNotEqual(result.returncode,0)
        self.assertIn('optimized interpreter refused',result.stderr)


class FullCliDispatch(unittest.TestCase):
    def invoke(self,phase,corrupt_source=False):
        import hashlib,shutil
        import authority_journal
        realpath=Path;called=[];output=io.StringIO()
        with tempfile.TemporaryDirectory() as folder:
            base=Path(folder);custody=base/'private';custody.mkdir(mode=0o700)
            empty=custody/'empty-docker-config';empty.mkdir(mode=0o700)
            owner=base/'owner';owner.mkdir();case=owner/'case';case.mkdir();execution=case/'execution';execution.mkdir()
            for name in ('actuator.py','authority_journal.py','native_program.py','receipt_safety.py'):
                shutil.copyfile(Path(__file__).parent/name,execution/name)
            (case/'memo.json').write_text('{}')
            for name in ('sealed-live-deployment.json','sealed-native-dependencies.json','candidate-deployment.json','rollback-deployment.json','check_installed.py','sqlite_readonly.py','change_guard.py','installed-zlib-proof.json','fresh-scan-contract.json'):
                shutil.copyfile(CASE/name,case/name)
            names=('deploy/lifecycle/admission.candidate.json','deploy/lifecycle/service.candidate.json','deploy/lifecycle/kustomization.yaml','deploy/lifecycle/registries.candidate.json')
            for name in names:
                target=owner/name;target.parent.mkdir(parents=True,exist_ok=True);target.write_text('sourcefixture')
            plan={'state':'prepared-unapproved','published_artifact_verified':True,'proposal_id':'INFD-WP-0009-T05/zlib85091-r1','executor_files':{n:actuator.sha(execution/n) for n in ('actuator.py','authority_journal.py','native_program.py','receipt_safety.py')},'case_files':{p.name:actuator.sha(p) for p in case.iterdir() if p.is_file() and p.name!='memo.json'},'owner_source_files':{n:actuator.sha(owner/n) for n in names}}
            (execution/'execution-plan.json').write_text(json.dumps(plan))
            receipt,outcome=accepted();receipt.update(native_outcome_sha256=hashlib.sha256(json.dumps(outcome).encode()).hexdigest(),executor_plan_sha256=actuator.sha(execution/'execution-plan.json'))
            mapping={}
            for fixed,data in [('/tmp/security-infd-zlib-root-authority.json',receipt),('/tmp/security-infd-zlib-native-outcome-latest.json',outcome),('/tmp/security-infd-zlib-fresh-scan-receipt.json',{'fixture':True}),('/tmp/security-infd-zlib-fresh-scan-full.json',{'fixture':True})]:
                target=base/Path(fixed).name;target.write_text(json.dumps(data));target.chmod(0o600);mapping[fixed]=target
            scanreceipt=mapping['/tmp/security-infd-zlib-fresh-scan-receipt.json'];scanreceipt.write_text(json.dumps({'full_report_sha256':actuator.sha(mapping['/tmp/security-infd-zlib-fresh-scan-full.json'])}))
            identity={'proposal':plan['proposal_id'],'source_revision':REVISION,'memo_sha256':actuator.sha(case/'memo.json'),'plan_sha256':actuator.sha(execution/'execution-plan.json')}
            blob=b'private synthetic full-cli snapshot';metadata={'all_checks_pass':True,'snapshot_file':'drill.sqlite','snapshot_bytes':len(blob),'snapshot_sha256':hashlib.sha256(blob).hexdigest(),'snapshot_readonly':COUNT,'pod_name':'fixture-pod','pod_uid':'fixture-poduid'}
            j=Journal(custody,identity)
            predecessors=list(actuator.PHASES[:actuator.PHASES.index(phase)])
            for predecessor in predecessors:
                j.start(predecessor,{})
                result=metadata if predecessor=='drill-backup' else {'all_checks_pass':True}
                if predecessor=='release':result={'all_checks_pass':True,'operation_requested':True,'prune':False,'source_revision':REVISION,'resources':[{'group':'apps','kind':'Deployment','name':'informed-decision','namespace':'informed-decision'}]}
                j.finish(predecessor,result)
            j.close()
            if corrupt_source:(owner/names[0]).write_text('changed')
            def mapped(value):return mapping.get(str(value),realpath(value))
            def run(argv,body=None,**kwargs):
                if argv[:4]==['git','-C',str(owner),'rev-parse']:return subprocess.CompletedProcess(argv,0,REVISION+'\n','')
                if argv[:4]==['git','-C',str(base),'check-ignore']:return subprocess.CompletedProcess(argv,0,'ignored','')
                if argv[:2]==['ssh','railiance01']:
                    import ast
                    parsed=ast.parse(body);node=next(x for x in parsed.body if isinstance(x,ast.Assign) and any(isinstance(t,ast.Name) and t.id=='payload' for t in x.targets));payload=json.loads(ast.literal_eval(node.value.args[0]));called.append(payload)
                    self.assertEqual(json.loads(j.path.read_text())['phases'][phase]['state'],'RUNNING')
                    if phase=='drill-copy':self.assertEqual(payload['snapshot_identity'],metadata)
                    if phase=='verify':self.assertEqual(payload['operation_intent']['state'],'COMPLETE');self.assertFalse(payload['operation_intent']['result']['prune'])
                    result=metadata if phase=='drill-copy' else {'all_checks_pass':True,'phase':phase}
                    return subprocess.CompletedProcess(argv,0,json.dumps(result),'')
                raise AssertionError('unexpected CLI transport')
            def git_show(argv,**kwargs):
                self.assertEqual(argv[:4],['git','-C',str(owner),'show']);data=(owner/argv[4].split(':',1)[1]).read_bytes();return subprocess.CompletedProcess(argv,0,data,b'')
            real_popen=subprocess.Popen
            def binary(*args,**kwargs):return real_popen([sys.executable,'-c','import sys;sys.stdout.buffer.write('+repr(blob)+')'],stdout=subprocess.PIPE,stderr=subprocess.DEVNULL)
            argv=['actuator.py','--phase',phase,'--root-authority-receipt','/tmp/security-infd-zlib-root-authority.json','--root-native-outcome','/tmp/security-infd-zlib-native-outcome-latest.json','--root-fresh-scan-receipt','/tmp/security-infd-zlib-fresh-scan-receipt.json','--source-revision',REVISION]
            with patch.object(actuator,'ROOT',execution),patch.object(actuator,'CASE',case),patch.object(actuator,'OWNER',owner),patch.object(actuator,'CUSTODIAN',base),patch.object(actuator,'CUSTODY',custody),patch.object(actuator,'Path',side_effect=mapped),patch.object(actuator,'run',side_effect=run),patch('subprocess.run',side_effect=git_show),patch('subprocess.Popen',side_effect=binary),patch.object(authority_journal,'authority',return_value={'fixture':True}),patch.object(authority_journal,'fresh_scan',return_value={'fixture':True}),patch('sys.argv',argv),redirect_stdout(output):
                if corrupt_source:
                    with self.assertRaises(ValueError):actuator.main()
                else:actuator.main()
            state=json.loads(j.path.read_text())['phases'].get(phase)
            return called,state,output.getvalue()

    def test_main_cli_copy_binds_snapshot_and_retains_private_bytes(self):
        calls,state,output=self.invoke('drill-copy');self.assertEqual(len(calls),2);self.assertEqual(state['state'],'COMPLETE');self.assertNotIn('private synthetic',output)

    def test_main_cli_verify_binds_completed_release_intent(self):
        calls,state,output=self.invoke('verify');self.assertEqual(len(calls),1);self.assertEqual(state['state'],'COMPLETE')

    def test_main_cli_release_dispatches_only_after_durable_phase(self):
        calls,state,output=self.invoke('release');self.assertEqual(len(calls),1);self.assertEqual(state['state'],'COMPLETE')

    def test_main_cli_source_hash_refuses_before_journal_or_transport(self):
        calls,state,output=self.invoke('release',corrupt_source=True);self.assertEqual(calls,[]);self.assertIsNone(state);self.assertEqual(output,'')


class PacketBindingAndBodyLimit(unittest.TestCase):
    def test_lossless_gzip_bundle_retains_exact_source_hashes(self):
        import gzip,base64,hashlib
        bundle=json.loads((CASE/'execution-bundle.json').read_text())
        for group in ('files','case_evidence'):
            for relative,item in bundle[group].items():
                data=gzip.decompress(base64.b64decode(item['content_base64'],validate=True))
                self.assertEqual(len(data),item['uncompressed_bytes']);self.assertEqual(hashlib.sha256(data).hexdigest(),item['sha256'])
                self.assertEqual(data,(CASE/relative).read_bytes())
        item=next(iter(bundle['files'].values()));changed=bytearray(base64.b64decode(item['content_base64']));changed[-1]^=1
        with self.assertRaises((OSError,EOFError)):gzip.decompress(changed)

    def test_actual_source_http_256kib_limit_before_register_side_effect(self):
        import os
        script="""import io,json,sys
sys.dont_write_bytecode=True
sys.path.insert(0,'/home/worsch/informed-decision')
from informed_decision.lifecycle_http import LifecycleAPI,MAX_BODY
from informed_decision.lifecycle_service import Caller
from informed_decision.lifecycle_service import LifecycleError
from informed_decision.intake import validate_submission,SubmitterGrant
from pathlib import Path
root=Path(sys.argv[1]);commands=json.loads((root/'lifecycle-commands.json').read_text());envelope=json.loads((root/'intake.json').read_text())
validate_submission(envelope,submitter_id='state-hub',grant=SubmitterGrant('state-hub',frozenset({'decision-record'}),frozenset({'net-kingdom-admins'})))
class Service:
 calls=0
 def _caller(self,caller):pass
 def register(self,*args,**kwargs):self.calls+=1;return 'fixture-event'
service=Service();api=LifecycleAPI(service,lambda token:object.__new__(Caller))
body=json.dumps(commands[0]['register']).encode();assert len(body)<240*1024
base={'REQUEST_METHOD':'POST','PATH_INFO':'/v1/decisions','QUERY_STRING':'','CONTENT_TYPE':'application/json','HTTP_AUTHORIZATION':'Bearer synthetic-fixture','HTTP_IDEMPOTENCY_KEY':'fixture'}
status,result=api.dispatch({**base,'CONTENT_LENGTH':str(len(body)),'wsgi.input':io.BytesIO(body)});assert status==200 and service.calls==1
oversize=b' '+body+b' '*(MAX_BODY+1-len(body)-1)
try:api.dispatch({**base,'CONTENT_LENGTH':str(len(oversize)),'wsgi.input':io.BytesIO(oversize)})
except LifecycleError as e:assert e.status==413
else:raise AssertionError('oversize accepted')
assert service.calls==1
print(json.dumps({'synthetic_body_limit_verified':True,'intake_validated':True,'register_bytes':len(body)}))
"""
        result=subprocess.run([sys.executable,'-c',script,str(CASE)],capture_output=True,text=True,env={**os.environ,'PYTHONDONTWRITEBYTECODE':'1'})
        self.assertEqual(result.returncode,0,result.stderr);self.assertTrue(json.loads(result.stdout)['synthetic_body_limit_verified'])


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