246 lines
14 KiB
Python
246 lines
14 KiB
Python
"""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()
|