"""Actual dormant policies -> complete synthetic native log -> independent check. No parser, prior validator or policy is mocked. Native calls/input are not run. """ import copy import re from pathlib import Path import sys import unittest sys.path[:0]=[str(Path(__file__).resolve().parents[1]/'tools'),str(Path(__file__).resolve().parent)] from test_prison_action_selection_evidence import build as selected_fixture, raw as selection_raw from test_prison_held_action_proof import Scene import prison_held_action_evidence as validator def build(profile='standard'): if profile=='coarse': from unittest.mock import patch from test_prison_coarse_facing import ApproachScene,ChoiceScene,coarse,coarse_move original_run=ApproachScene.run # Fake capabilities select the explicit profile; all real policy bodies # and raw validators still run without substitution. with patch.object(ApproachScene,'run',lambda s,*a,**kw:original_run(s,*a,**dict(kw,profile='coarse'))),patch.object(ChoiceScene,'facing',coarse),patch.object(ChoiceScene,'move',coarse_move): frames,look,route,facing,selection=selected_fixture() else:frames,look,route,facing,selection=selected_fixture() s=Scene();end=selection['samples'][-1] s.now=selection['finished_monotonic'];s.next_sample=(end['real_ms']-2000)/1000+.1 s.geometry=copy.deepcopy(selection['geometry']);s.origin=list(end['origin']) s.camera=list(end['camera']);s.direction=list(end['direction']);s.actions=copy.deepcopy(selection['actions']) original_make=s.make def make(): v=original_make();v['sample']=int(round((2000+s.now*1000)/100));v['real_ms']=2000+int(round(s.now*1000));v['world_ms']=v['real_ms'] v['choices'].update(nonce='test',sample=v['sample'],real_ms=v['real_ms']) if v['phase']==1:v['choices']['selected_id']=s.actions[1]['id'] return v s.make=make;v=copy.deepcopy(end);v['turn']=[0,0,0];v['world_ms']=v['real_ms'] s.snapshot=v;s.history=[copy.deepcopy(v)];s.last_read=v['sample']-1 s.calibration.update(nonce='test',ready_real_ms=2000,ready_monotonic=0.,observer_start_real_ms=0,observer_deadline_monotonic=42.5) s.report={'action_selection_proof':selection};s.evidence={'verified':True,'last_sample':selection['final_sample'],'selected_action_id':s.actions[1]['id'],'wheel_button':5} held=s.run() for frame in s.history:frames[frame['sample']]=copy.deepcopy(frame) return frames,look,route,facing,selection,held def raw(frames): # Retain existing native-format fixture including every prerequisite. These # two fields are deliberately independently variable for timing/look tests. content=selection_raw(frames) def world(m):return m[1]+str(frames[int(m[2])].get('world_ms',frames[int(m[2])]['real_ms'])) content=re.sub(r'(ERR INPUT OBSERVE: run=test sample=(\d+) real=\d+ world=)\d+(?= SAMPLE)',world,content) def turn(m): v=frames[int(m[2])].get('turn',[0,0,0]);return m[1]+'<'+','.join(str(x) for x in v)+'>' return re.sub(r'(ERR LOOK OBSERVE: run=test sample=(\d+) real=\d+ SAMPLE[^\n]* turn=)<[^>]*>',turn,content) class HeldEvidenceTests(unittest.TestCase): @classmethod def setUpClass(cls):cls.template=build() def result(self,mutation=None,**kwargs): args=copy.deepcopy(self.template) if mutation:mutation(*args) f,*proofs=args return validator.validate_held_action_evidence(raw(f),'test',*proofs,**kwargs) def reject(self,mutation): r=self.result(mutation);self.assertFalse(r['verified'],r);self.assertTrue(r['errors']) def test_full_actual_policy_chain_and_native_phase_reset_pass(self): r=self.result();self.assertTrue(r['verified'],r) self.assertTrue(r['held_action_verified'] and r['unlock_observed'] and r['gate_motion_observed']) self.assertFalse(r['escape_verified']);self.assertFalse(r['adapter_release_confirmation_verified']) self.assertGreaterEqual(r['native_world_duration_ms'],4000) self.assertLessEqual(r['maximum_timing_allowance_seconds'],.502) def test_all_prior_proofs_revalidated_and_expected_profile_explicit(self): for which in range(1,5):self.reject(lambda *args:args[which].update(completed=False)) self.assertFalse(self.result(facing_profile='coarse')['verified']) self.assertFalse(self.result(facing_profile='unknown')['verified']) def test_coarse_full_chain_requires_expected_profile(self): f,*p=build('coarse');content=raw(f) accepted=validator.validate_held_action_evidence(content,'test',*p,facing_profile='coarse') self.assertTrue(accepted['verified'],accepted) self.assertFalse(validator.validate_held_action_evidence(content,'test',*p)['verified']) def test_one_exact_f_pair_only_and_no_uncertain_cleanup(self): mutations=[lambda h:h['events'].append(copy.deepcopy(h['events'][0])),lambda h:h['events'].pop(), lambda h:h['events'][0].update(key='W'),lambda h:h['events'][1].update(cleanup=True), lambda h:h['events'][1].update(confirmed=False),lambda h:h.update(unconfirmed_held_key='F'), lambda h:h.update(paired_release_errors=['unproved release']),lambda h:h.update(notification_errors=['supervisor lost']), lambda h:h.update(release_confirmed=False),lambda h:h.update(events=[])] for mutation in mutations:self.reject(lambda *a:mutation(a[-1])) def test_original_clock_admission_and_five_second_lease_not_renewed(self): for mutation in (lambda h:h.update(original_deadline_monotonic=50), lambda h:h.update(held_deadline_monotonic=h['press_attempted_monotonic']+5.01), lambda h:h.update(held_deadline_monotonic=h['up_monotonic']), lambda h:h.update(original_deadline_monotonic=h['down_monotonic']+5.99), lambda h:h['events'][0].update(attempted_monotonic=h['press_attempted_monotonic']-.01)): self.reject(lambda *a:mutation(a[-1])) def test_compact_references_pass_and_wrong_world_reference_refuses(self): def compact(*a): h=a[-1];h['samples']=[{k:r[k] for k in ('sample','real_ms','received_monotonic')} for r in h['samples']] self.assertTrue(self.result(compact)['verified']) self.reject(lambda *a:a[-1]['samples'][-1].update(world_ms=-1)) def test_missing_reordered_stale_and_forged_receipts_refuse(self): for mutation in (lambda h:h['samples'].pop(2),lambda h:h['samples'].reverse(), lambda h:h['samples'][-1].update(received_monotonic=float('nan')), lambda h:h['samples'][2].update(received_monotonic=-1), lambda h:h['samples'][2].update(real_ms=0), lambda h:h.update(finished_monotonic=h['samples'][-1]['received_monotonic']+.51)): self.reject(lambda *a:mutation(a[-1])) def test_native_progress_jump_and_early_transition_refuse(self): self.reject(lambda f,*a:f[a[-1]['progress_samples'][0]].update(progress=3.9)) def early(f,*a): h=a[-1];i=h['progress_samples'][2];f[i].update(phase=2,progress=0,performing=False) self.reject(early) def no_progress(f,*a): h=a[-1] for i in range(h['press_sample']+1,h['transition_sample']):f[i].update(progress=0) self.reject(no_progress) def test_reset_decrease_unbound_and_input_ack_loss_refuse(self): for key,value in [('progress',0),('progress',.001),('performing',False),('pressed',0),('progress_bound',False),('selected',False),('cached_performable',False),('context_selected',False)]: with self.subTest(key=key,value=value):self.reject(lambda f,*a:f[a[-1]['progress_samples'][5]].update({key:value})) def test_world_clock_is_independent_of_real_and_cannot_reverse_or_freeze(self): def freeze(f,*a): h=a[-1];world=f[h['press_sample']]['real_ms'] for i in range(h['press_sample']+1,h['final_sample']+1):f[i]['world_ms']=world self.reject(freeze) self.reject(lambda f,*a:f[a[-1]['progress_samples'][5]].update(world_ms=0)) def test_frame_order_allowance_uses_observed_interval_only(self): def shift(f,*a): h=a[-1] for i in range(h['press_sample']+1,h['transition_sample']): if f[i]['progress']>0:f[i]['progress']+=.025 # Actual duration resets within one frame; a small relative sampling # offset remains compatible with the documented per-frame API. self.assertTrue(self.result(shift)['verified']) def jump(f,*a):f[a[-1]['progress_samples'][3]]['progress']+=.3 self.reject(jump) def test_native_body_view_turn_menu_input_and_velocity_are_authoritative(self): for key,value in [('turn',[.02,0,0]),('input',[0,0,1]),('forward',1),('velocity',[.1,0,0]),('inventory',True),('direction',[0,0,1])]: self.reject(lambda f,*a:f[a[-1]['progress_samples'][3]].update({key:value})) self.reject(lambda f,*a:f[a[-1]['progress_samples'][3]]['origin'].__setitem__(1,.051)) def test_phase_two_moving_original_door_allowed_but_not_before_transition(self): f,*p=copy.deepcopy(self.template);h=p[-1] self.assertNotEqual(f[h['transition_sample']]['geometry']['context'],h['geometry']['context']) self.assertTrue(self.result()['verified']) self.reject(lambda f,*a:f[a[-1]['progress_samples'][3]]['geometry']['context'].__setitem__(0,200)) self.reject(lambda f,*a:f[a[-1]['released_sample']].update(phase=1)) self.reject(lambda f,*a:f[a[-1]['released_sample']].update(phase=3)) def test_no_control_or_motion_and_forged_reference_cannot_pass(self): for field in ('door_state','door_control'): def remove(f,*a): h=a[-1] for i in range(h['transition_sample'],h['final_sample']+1):f[i][field]=0 self.reject(remove) for key in ('transition_sample','control_sample','motion_sample'): self.reject(lambda *a:a[-1].update({key:a[-1][key]+1})) self.reject(lambda *a:a[-1]['progress_samples'].pop()) def test_release_needs_new_raw_neutral_sample_after_drain(self): self.reject(lambda *a:a[-1].update(released_sample=a[-1]['release_barrier_sample'])) for key,value in [('pressed',1),('progress',1),('performing',True)]: self.reject(lambda f,*a:f[a[-1]['released_sample']].update({key:value})) self.reject(lambda *a:a[-1].update(release_barrier_sample=a[-1]['transition_sample']-1)) def test_old_release_acknowledgement_cannot_credit_renewed_or_recovered_input(self): def append(f,h,pressed): i=h['final_sample'];v=copy.deepcopy(f[i]) v.update(sample=i+1,real_ms=v['real_ms']+100,world_ms=v['world_ms']+100, received_monotonic=v['received_monotonic']+.1,pressed=pressed,progress=0,performing=False) v['choices'].update(sample=i+1,real_ms=v['real_ms']);f[i+1]=v h['samples'].append(copy.deepcopy(v));h['final_sample']=i+1;h['finished_monotonic']+=.11 def old(f,*a):append(f,a[-1],1) self.reject(old) def recovered(f,*a): h=a[-1];append(f,h,1);append(f,h,0);h['released_sample']=h['final_sample'] r=self.result(recovered);self.assertFalse(r['verified'],r) self.assertIn('renewed after release',str(r['errors'])) def test_normal_release_must_follow_observed_authoritative_transition(self): def premature(*a): h=a[-1];at=h['down_monotonic']+.5 h['up_monotonic']=at;h['events'][1].update(attempted_monotonic=at-.01,monotonic=at) r=self.result(premature);self.assertFalse(r['verified'],r) self.assertIn('release preceded',str(r['errors'])) def test_native_choice_identity_permission_and_proof_binding_cannot_change(self): self.reject(lambda f,*a:f[a[-1]['progress_samples'][3]]['choices']['actions'][1].update(id=17)) self.reject(lambda f,*a:f[a[-1]['progress_samples'][3]]['choices']['actions'][1].update(permitted=False)) self.reject(lambda *a:a[-1].update(nonce='other')) self.reject(lambda *a:a[-1]['actions'][1].update(id=17)) def test_raw_intermediate_records_and_original_observer_cancellation_required(self): f,*p=copy.deepcopy(self.template);h=p[-1];content=raw(f);i=h['progress_samples'][4] for prefix in ('ERR INPUT OBSERVE:', 'ERR LOOK OBSERVE:', 'ERR GATE OBSERVE:', 'ERR GATE CHOICE:'): line=next(x for x in content.splitlines() if x.startswith(prefix) and f' sample={i} ' in x) r=validator.validate_held_action_evidence(content.replace(line+'\n',''),'test',*p) self.assertFalse(r['verified'],r) i=h['final_sample'] line=f'ERR LOOK OBSERVE: run=test sample={i} real={f[i]["real_ms"]} CANCELLED menu-opened\n' self.assertFalse(validator.validate_held_action_evidence(content+line,'test',*p)['verified']) def test_early_transition_rejected_by_world_duration_before_reference_comparison(self): def early(f,*a): h=a[-1];i=h['progress_samples'][4];f[i].update(phase=2,progress=0,performing=False) r=self.result(early);self.assertFalse(r['verified'],r) self.assertIn('before four seconds',str(r['errors'])) def test_original_duration_and_handoff_receipt_are_not_report_authority(self): self.reject(lambda f,*a:f[a[-1]['progress_samples'][3]].update(duration=1)) self.reject(lambda *a:a[-1]['samples'][0].update(received_monotonic=a[-1]['samples'][0]['received_monotonic']+.01)) def test_unknown_payload_or_success_claim_does_not_override_raw(self): self.reject(lambda *a:a[-1].update(extra='ignored input')) self.reject(lambda *a:a[-1].update(status='unlock-observed-motion-unverified')) self.reject(lambda *a:a[-1].update(completed=False)) if __name__=='__main__':unittest.main()