"""Synthetic joined native snapshots; no display, engine, or input backend.""" import copy import math from pathlib import Path import sys import unittest sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'tools')) from test_prison_alignment_proof import Scene import prison_alignment_proof as alignment import prison_action_selection_proof as selection class ChoiceScene(Scene): def __init__(self): self.selected_at = None; self.choices_fault = None; self.wheels = [] self.wheel_delay = .15; self.wheel_ignored = False super().__init__() def make(self): s = super().make() target = math.degrees(math.atan2(self.geometry['context'][0]-self.camera[0],self.geometry['context'][2]-self.camera[2])) visible = abs(alignment.delta(target,self.angle)) <= 3 chosen = self.selected_at is not None and self.now >= self.selected_at s.update(selected=chosen,context_selected=visible,cached_performable=visible,progress_bound=chosen) s['choices']={'nonce':'N','sample':s['sample'],'real_ms':s['real_ms'], 'selected_id':0 if chosen else 1,'selected_class':'ERR_UnlockPrisonAction' if chosen else 'ERR_PrisonDoorAction', 'actions':[{'class':'ERR_PrisonDoorAction','id':1,'permitted':False}, {'class':'ERR_UnlockPrisonAction','id':0,'permitted':True}]} if self.choices_fault: self.choices_fault(s) return s def facing(self): return alignment.perform_facing(self,self.idle,self,self.observe,self.geometry,self.calibration,self.accepted,self.report, clock=self.clock,sleep=self.sleep,notify=self.notifications.append) def pulse(self,button,guard,*,deadline): if button != 5: raise AssertionError('Only one explicit logical5 capability') guard.check();down=self.clock() self.sleep(.01);guard.check();up=self.clock() if up >= deadline: raise selection.Refused('owned adapter deadline') self.wheels.append((button,down,up,deadline)) if not self.wheel_ignored:self.selected_at=down+self.wheel_delay return {'button':button,'down_monotonic':down,'up_monotonic':up,'released':True} def select_action(self): facing=self.report['facing_proof'] self.last_read=facing['final_sample']-1 return selection.perform_selection(self,self.idle,self,self.observe,self.geometry,self.calibration, {'verified':True,'last_sample':facing['final_sample']},self.report, clock=self.clock,sleep=self.sleep,notify=self.notifications.append) class SelectionTests(unittest.TestCase): def aligned_pair(self, cls=ChoiceScene): scene=cls();scene.guard_cost=0 scene.angle=math.degrees(math.atan2(scene.geometry['context'][0]-scene.camera[0],scene.geometry['context'][2]-scene.camera[2])) scene.snapshot=scene.make();scene.history=[copy.deepcopy(scene.snapshot)] scene.sleep(.1) return scene def test_facing_commits_existing_fresh_pair_without_second_drain(self): scene=self.aligned_pair();reads=[];original=scene.observe def observe():reads.append(scene.clock());return original() scene.observe=observe result=scene.facing() self.assertTrue(result['completed']);self.assertEqual(len(reads),1) self.assertEqual(result['facing_samples'],[310,311]) self.assertEqual([x['sample'] for x in result['samples']],[310,311]) self.assertEqual(result['final_sample'],311) self.assertEqual(result['deadline_monotonic'],39) self.assertEqual(scene.motions,[]);self.assertEqual(scene.wheels,[]) # The existing strict alignment path still performs its original reread. strict=self.aligned_pair(Scene);strict_reads=[];original=strict.observe def strict_observe():strict_reads.append(strict.clock());return original() strict.observe=strict_observe self.assertTrue(strict.run()['completed']);self.assertEqual(len(strict_reads),2) def test_final_facing_idle_freshness_clock_and_deadline_rechecked(self): for fault in ('not-idle','stale','clock-reversed','deadline'): scene=self.aligned_pair();calls=[] if fault=='deadline':scene.calibration['alignment_deadline_monotonic']=31.15 def idle(): calls.append(scene.now) if len(calls)==2: if fault=='not-idle':raise alignment.Refused('native keys are not idle') if fault=='stale':scene.sleep(.51) if fault=='clock-reversed':scene.now-=.2 if fault=='deadline':scene.sleep(.06) scene.idle=idle with self.subTest(fault=fault),self.assertRaises(alignment.Refused):scene.facing() self.assertEqual(len(calls),2);self.assertFalse(scene.report['facing_proof']['completed']) self.assertEqual(scene.motions,[]);self.assertEqual(scene.wheels,[]) def test_facing_complete_pair_does_not_hide_intermediate_native_reversal(self): scene=self.aligned_pair();scene.history[1]['real_ms']=scene.history[0]['real_ms']-1 scene.history[1]['choices']['real_ms']=scene.history[1]['real_ms'] with self.assertRaisesRegex(ValueError,'reversed'):scene.facing() self.assertEqual(scene.motions,[]);self.assertEqual(scene.wheels,[]) self.assertFalse(scene.report['facing_proof']['completed']) def test_selection_consumes_every_later_record_after_early_facing_commit(self): scene=self.aligned_pair();facing=scene.facing();self.assertEqual(facing['final_sample'],311) scene.sleep(.3) bad=next(x for x in scene.history if x['sample']==312) bad['pressed']=1 # Later records are clean: latest-only selection would incorrectly pass. self.assertEqual(scene.history[-1]['pressed'],0) with self.assertRaisesRegex(ValueError,'Action began'):scene.select_action() self.assertEqual(scene.wheels,[]) self.assertEqual(scene.report['action_selection_proof']['initial_sample'],311) def test_actual_pose_facing_then_one_wheel_selects_with_same_lease(self): s=ChoiceScene();f=s.facing();r=s.select_action() self.assertTrue(f['completed']);self.assertNotIn('alignment_proof',s.report) self.assertNotIn('selection_samples',f);self.assertEqual(len(f['facing_samples']),2) self.assertFalse(f['samples'][-1]['selected']);self.assertFalse(f['samples'][-1]['progress_bound']) self.assertTrue(r['completed']);self.assertEqual(len(s.wheels),1) self.assertEqual(r['deadline_monotonic'],f['deadline_monotonic']) self.assertEqual([x['class'] for x in r['actions']],['ERR_PrisonDoorAction','ERR_UnlockPrisonAction']) a,b=r['selection_samples'];raw={x['sample']:x for x in r['samples']} self.assertEqual(b,a+1);self.assertGreaterEqual(raw[b]['real_ms']-raw[a]['real_ms'],100) self.assertTrue(raw[a]['progress_bound'] and raw[b]['progress_bound']) self.assertLess(s.now,39);self.assertEqual(s.origin,[6092.42,211.297,7155.81]) def test_facing_stops_without_waiting_for_lock_selection(self): s=ChoiceScene();f=s.facing() self.assertTrue(f['completed']);self.assertLess(s.now,38) self.assertEqual(s.wheels,[]);self.assertFalse(f['samples'][-1]['selected']) def test_wrong_direction_or_ignored_wheel_never_retries(self): s=ChoiceScene();s.facing();s.wheel_ignored=True with self.assertRaisesRegex(selection.Refused,'Single wheel|deadline|acknowledgement'):s.select_action() self.assertEqual(len(s.wheels),1) self.assertFalse(s.report['action_selection_proof']['completed']) def test_actual_delayed_selection_waits_for_two_fresh_bound_samples(self): s=ChoiceScene();s.facing();s.wheel_delay=.65 r=s.select_action();self.assertTrue(r['completed']) raw={x['sample']:x for x in r['samples']} self.assertGreaterEqual(raw[r['selection_samples'][0]]['received_monotonic'],s.wheels[0][1]+.65) def test_insufficient_shared_budget_refuses_before_wheel(self): s=ChoiceScene();s.facing();s.sleep(39-s.now-1.4) with self.assertRaisesRegex(selection.Refused,'Insufficient'):s.select_action() self.assertEqual(s.wheels,[]) def test_explicit_handoff_cannot_renew_facing_or_original_deadline(self): s=ChoiceScene();s.facing();s.report['facing_proof']['deadline_monotonic']=50 with self.assertRaisesRegex(selection.Refused,'renewed'):s.select_action() self.assertEqual(s.wheels,[]) def test_filtered_list_and_join_faults_refuse_facing_without_wheel(self): def third(x):x['choices']['actions'].append({'class':'Other','id':12,'permitted':True}) def order(x):x['choices']['actions'].reverse() def identity(x):x['choices']['actions'][1]['id']=1 def nonce(x):x['choices']['nonce']='other' def timestamp(x):x['choices']['real_ms']-=1 def permission(x):x['choices']['actions'][0]['permitted']=True def selected_class(x):x['choices']['selected_class']='Other' def missing(x):x.pop('choices') for fault in (third,order,identity,nonce,timestamp,permission,selected_class,missing): s=ChoiceScene();s.choices_fault=fault with self.subTest(fault=fault.__name__),self.assertRaises(ValueError):s.facing() self.assertEqual(s.wheels,[]) def test_ids_cannot_be_readopted_between_facing_and_selection(self): s=ChoiceScene();s.facing() def replaced(x):x['choices']['actions'][1]['id']=15 s.choices_fault=replaced;s.sleep(.2) with self.assertRaisesRegex(selection.Refused,'identities'):s.select_action() self.assertEqual(s.wheels,[]) def test_transient_intermediate_movement_or_action_cannot_be_hidden(self): for field,value in [('origin',[6092.46,211.337,7155.81]),('pressed',1.),('performing',True),('input',[0,0,1]),('phase',2)]: s=ChoiceScene();s.facing();s.sleep(.3) i=next(i for i,x in enumerate(s.history) if x['sample']>s.report['facing_proof']['final_sample']) s.history[i][field]=value with self.subTest(field=field),self.assertRaises(ValueError):s.select_action() self.assertEqual(s.wheels,[]) def test_focus_loss_or_unconfirmed_pair_is_not_success(self): s=ChoiceScene();s.facing();s.focus=False with self.assertRaisesRegex(ValueError,'focus'):s.select_action() self.assertEqual(s.wheels,[]) s=ChoiceScene();s.facing();original=s.pulse def bad(*a,**kw):r=original(*a,**kw);r['released']=False;return r s.pulse=bad with self.assertRaisesRegex(selection.Refused,'release'):s.select_action() self.assertEqual(len(s.wheels),1);self.assertFalse(s.report['action_selection_proof']['wheel']['confirmed']) def test_already_selected_is_not_wheel_proof(self): s=ChoiceScene();s.facing();s.selected_at=s.now;s.sleep(.2) with self.assertRaisesRegex(selection.Refused,'before wheel'):s.select_action() self.assertEqual(s.wheels,[]) def test_one_selected_sample_followed_by_drop_cannot_qualify(self): s=ChoiceScene();s.facing();original=s.pulse def pulse(*a,**kw): r=original(*a,**kw) def drop(x): if x['received_monotonic'] > s.selected_at+.11: x.update(selected=False,progress_bound=False) x['choices'].update(selected_id=1,selected_class='ERR_PrisonDoorAction') s.choices_fault=drop return r s.pulse=pulse with self.assertRaises(selection.Refused):s.select_action() self.assertEqual(len(s.wheels),1) def test_unbound_selected_lock_refuses(self): s=ChoiceScene();s.facing();original=s.pulse def pulse(*a,**kw): r=original(*a,**kw) s.choices_fault=lambda x:x.update(progress_bound=False) return r s.pulse=pulse with self.assertRaisesRegex(selection.Refused,'binding'):s.select_action() def test_guarded_wheel_cannot_extend_its_uncertain_button_lease(self): s=ChoiceScene();s.facing() def slow(button,guard,*,deadline): down=s.now;s.sleep(1.6) return {'button':5,'down_monotonic':down,'up_monotonic':s.now,'released':True} s.pulse=slow with self.assertRaisesRegex(selection.Refused,'lease'):s.select_action() self.assertFalse(s.report['action_selection_proof']['completed']) def test_no_extra_mouse_requests_after_facing_handoff(self): s=ChoiceScene();s.facing();count=len(s.motions) s.select_action() self.assertEqual(len(s.motions),count) self.assertTrue(all(event['kind'].startswith(('facing','action-selection')) for event in s.notifications)) def test_facing_ids_pin_on_first_visible_context_not_just_final_pair(self): s=ChoiceScene() first=[None] def swap(x): if x['context_selected']: if first[0] is None:first[0]=x['sample'] elif x['sample'] > first[0]:x['choices']['actions'][1]['id']=7 s.choices_fault=swap with self.assertRaisesRegex(ValueError,'identities'):s.facing() self.assertEqual(s.wheels,[]) def test_selected_but_oscillating_view_is_not_stable_pair(self): s=ChoiceScene();s.facing();base=s.angle;original=s.pulse def pulse(*a,**kw): r=original(*a,**kw) def oscillate(x): radians=math.radians(base+(.18 if x['sample']%2 else -.18)) x['direction']=[math.sin(radians),0.,math.cos(radians)] s.choices_fault=oscillate return r s.pulse=pulse with self.assertRaises(selection.Refused):s.select_action() self.assertFalse(s.report['action_selection_proof']['completed']) def test_attempt_report_is_single_use(self): s=ChoiceScene();s.facing();s.select_action() with self.assertRaisesRegex(selection.Refused,'already exists'):s.select_action() self.assertEqual(len(s.wheels),1) if __name__=='__main__':unittest.main()