152 lines
9.6 KiB
Python
152 lines
9.6 KiB
Python
"""Actual policies -> complete synthetic native RAW -> independent validators.
|
|
|
|
No raw parser, prerequisite validator, display, or native API is mocked.
|
|
"""
|
|
import copy
|
|
import math
|
|
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_alignment_evidence import build as alignment_fixture, raw as gate_raw
|
|
from test_prison_action_selection_proof import ChoiceScene
|
|
import prison_alignment_evidence as facing_validator
|
|
import prison_action_selection_evidence as validator
|
|
|
|
|
|
def build():
|
|
frames,look,route,unused=alignment_fixture()
|
|
frames={i:s for i,s in frames.items() if i<=route['final_sample']}
|
|
s=ChoiceScene();end=route['samples'][-1]
|
|
s.now=route['finished_monotonic'];s.next_sample=(end['real_ms']-2000)/1000+.1
|
|
s.geometry=copy.deepcopy(route['geometry']);s.origin=list(end['origin'])
|
|
s.camera=[s.origin[0],s.origin[1]+1.7,s.origin[2]-.1]
|
|
s.angle=math.degrees(math.atan2(end['direction'][0],end['direction'][2]));s.guard_cost=.18
|
|
make=s.make
|
|
def snapshot():
|
|
v=make();v['sample']=int(round((2000+s.now*1000)/100));v['real_ms']=2000+int(round(s.now*1000))
|
|
v['choices'].update(nonce='test',sample=v['sample'],real_ms=v['real_ms'])
|
|
return v
|
|
s.make=snapshot;v=snapshot();v.update(sample=end['sample'],real_ms=end['real_ms'],received_monotonic=end['received_monotonic'])
|
|
v['choices'].update(sample=v['sample'],real_ms=v['real_ms'])
|
|
s.snapshot=v;s.history=[copy.deepcopy(v)];s.last_read=v['sample']-1;s.pending=[]
|
|
s.calibration.update(nonce='test',ready_real_ms=2000,ready_monotonic=0.,observer_start_real_ms=0,
|
|
observer_deadline_monotonic=42.5,alignment_deadline_monotonic=s.now+8)
|
|
s.report={'approach_proof':route};s.accepted={'verified':True,'last_sample':route['final_sample']}
|
|
facing=s.facing();proof=s.select_action()
|
|
for frame in s.history:frames[frame['sample']]=copy.deepcopy(frame)
|
|
return frames,look,route,facing,proof
|
|
|
|
|
|
def raw(frames):
|
|
content=gate_raw(frames)
|
|
lines=[]
|
|
for i,s in sorted(frames.items()):
|
|
c=s.get('choices',{'selected_id':-1,'selected_class':'none','actions':[]})
|
|
a=c['actions'];entries=','.join(f'{n}={e["class"]}:{e["id"]}:{int(e["permitted"])}' for n,e in enumerate(a))
|
|
lines.append(f'ERR GATE CHOICE: run=test sample={i} real={s["real_ms"]} selected={c["selected_class"]}:{c["selected_id"]} context={int(s["context_selected"])} actions={len(a)} permissions={len(a)} shown={len(a)} truncated=0 ordered=[{entries}]')
|
|
return content+'\n'.join(lines)+'\n'
|
|
|
|
|
|
class SelectionEvidenceTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):cls.template=build()
|
|
def result(self,mutation=None):
|
|
frames,look,route,facing,proof=copy.deepcopy(self.template)
|
|
if mutation:mutation(frames,look,route,facing,proof)
|
|
return validator.validate_selection_evidence(raw(frames),'test',look,route,facing,proof)
|
|
def reject(self,mutation):
|
|
r=self.result(mutation);self.assertFalse(r['verified'],r);self.assertTrue(r['errors'])
|
|
def test_actual_policy_chain_passes_complete_unmocked_native_parsers(self):
|
|
f,l,a,p,s=copy.deepcopy(self.template)
|
|
r=facing_validator.validate_facing_evidence(raw(f),'test',l,a,p)
|
|
self.assertTrue(r['verified'],r);self.assertFalse(r['lock_selection_verified'])
|
|
r=self.result();self.assertTrue(r['verified'],r)
|
|
self.assertEqual(r['selected_action_id'],0);self.assertFalse(r['held_action_verified']);self.assertFalse(r['escape_verified'])
|
|
def test_prior_look_approach_or_facing_cannot_be_asserted(self):
|
|
for which in (1,2,3):
|
|
self.reject(lambda *args:args[which].update(completed=False))
|
|
def test_extra_or_wrong_wheel_and_unconfirmed_release_refuse(self):
|
|
self.reject(lambda f,l,a,p,s:s['wheel'].update(button=4))
|
|
self.reject(lambda f,l,a,p,s:s['wheel'].update(released=False))
|
|
self.reject(lambda f,l,a,p,s:s['wheel'].update(confirmed=False))
|
|
self.reject(lambda f,l,a,p,s:s.update(events=[{'key':'F','down':True}]))
|
|
self.reject(lambda f,l,a,p,s:s.update(motions=[{'dx':1,'dy':0}]))
|
|
self.reject(lambda f,l,a,p,s:s['wheel'].update(events=[4,5]))
|
|
self.reject(lambda f,l,a,p,s:s.update(wheel_events=[{'button':4}]))
|
|
def test_full_3d_body_drift_and_capture_context_refuse(self):
|
|
def drift(f,l,a,p,s):
|
|
i=s['initial_sample']+1;f[i]['origin'][0]+=.04;f[i]['origin'][1]+=.04
|
|
self.reject(drift)
|
|
self.reject(lambda f,l,a,p,s:f[s['initial_sample']+1].update(context_selected=False))
|
|
self.reject(lambda f,l,a,p,s:f[s['initial_sample']+1].update(phase=2))
|
|
def test_original_choice_ids_order_and_permission_cannot_change(self):
|
|
self.reject(lambda f,l,a,p,s:f[s['final_sample']]['choices']['actions'][1].update(id=8))
|
|
self.reject(lambda f,l,a,p,s:f[s['final_sample']]['choices']['actions'].reverse())
|
|
self.reject(lambda f,l,a,p,s:f[s['final_sample']]['choices']['actions'][0].update(permitted=True))
|
|
self.reject(lambda f,l,a,p,s:f[s['final_sample']]['choices'].update(selected_id=1))
|
|
self.reject(lambda f,l,a,p,s:f[s['final_sample']]['choices'].update(selected_class='ERR_PrisonDoorAction'))
|
|
def test_facing_pin_cannot_change_before_final_pair(self):
|
|
def change(f,l,a,p,s):
|
|
for i in range(p['initial_sample'],p['final_sample']):
|
|
if f[i]['context_selected']:
|
|
f[i]['choices']['actions'][1]['id']=8;return
|
|
raise AssertionError('simulation must expose early native context')
|
|
self.reject(change)
|
|
def test_facing_is_not_lock_selection_and_requires_context(self):
|
|
self.reject(lambda f,l,a,p,s:p.update(selection_samples=p['facing_samples']))
|
|
self.reject(lambda f,l,a,p,s:f[p['final_sample']].update(context_selected=False))
|
|
self.reject(lambda f,l,a,p,s:f[p['final_sample']].update(progress_bound=True))
|
|
def test_deadlines_hold_cap_and_ack_bound_are_not_renewable(self):
|
|
self.reject(lambda f,l,a,p,s:s.update(deadline_monotonic=s['deadline_monotonic']+.1))
|
|
self.reject(lambda f,l,a,p,s:s.update(finished_monotonic=s['deadline_monotonic']+.1))
|
|
self.reject(lambda f,l,a,p,s:s['wheel'].update(up_monotonic=s['wheel']['attempted_monotonic']+1.6))
|
|
self.reject(lambda f,l,a,p,s:s['wheel'].update(down_monotonic=s['started_monotonic']-.1))
|
|
self.reject(lambda f,l,a,p,s:s.update(finished_monotonic=s['wheel']['up_monotonic']+1.6))
|
|
def test_pre_dispatch_insufficient_remaining_time_refuses(self):
|
|
def exhausted(f,l,a,p,s):
|
|
# Jointly shortening a valid upstream lease must not bypass the
|
|
# selection policy's >=1.5s admission requirement.
|
|
d=s['wheel']['attempted_monotonic']+1.49
|
|
p['deadline_monotonic']=d;s['deadline_monotonic']=d
|
|
self.reject(exhausted)
|
|
def test_sample_receipts_stale_missing_or_nonfinite_refuse(self):
|
|
self.reject(lambda f,l,a,p,s:s['samples'][-1].update(received_monotonic=float('nan')))
|
|
self.reject(lambda f,l,a,p,s:s['samples'].pop(1))
|
|
self.reject(lambda f,l,a,p,s:s.update(finished_monotonic=s['samples'][-1]['received_monotonic']+.6))
|
|
self.reject(lambda f,l,a,p,s:s['wheel'].update(attempted_monotonic=s['started_monotonic']-.2))
|
|
def test_selected_lock_needs_consecutive_bound_neutral_samples(self):
|
|
self.reject(lambda f,l,a,p,s:f[s['selection_samples'][0]].update(progress_bound=False))
|
|
self.reject(lambda f,l,a,p,s:f[s['selection_samples'][0]].update(pressed=1))
|
|
self.reject(lambda f,l,a,p,s:s.update(selection_samples=[s['final_sample'],s['final_sample']]))
|
|
self.reject(lambda f,l,a,p,s:f[s['selection_samples'][0]].update(input=[0,0,1]))
|
|
def test_native_lock_selected_before_wheel_is_not_acknowledgement(self):
|
|
def early(f,l,a,p,s):
|
|
i=s['wheel']['after_sample'];f[i].update(selected=True,progress_bound=True)
|
|
f[i]['choices'].update(selected_id=0,selected_class='ERR_UnlockPrisonAction')
|
|
self.reject(early)
|
|
def test_selected_then_dropped_then_reacquired_cannot_hide_in_batch(self):
|
|
def dropped(f,l,a,p,s):
|
|
first=s['wheel']['after_sample']+1
|
|
f[first].update(selected=True,progress_bound=True);f[first]['choices'].update(selected_id=0,selected_class='ERR_UnlockPrisonAction')
|
|
f[first+1].update(selected=False,progress_bound=False);f[first+1]['choices'].update(selected_id=1,selected_class='ERR_PrisonDoorAction')
|
|
self.reject(dropped)
|
|
def test_final_bearing_and_stability_are_native(self):
|
|
def angle(f,l,a,p,s):
|
|
i=s['selection_samples'][0];v=f[i]['direction'];r=math.atan2(v[0],v[2])+math.radians(.3)
|
|
f[i]['direction']=[math.sin(r),0,math.cos(r)]
|
|
self.reject(angle)
|
|
self.reject(lambda f,l,a,p,s:f[s['final_sample']].update(direction=[0,0,1]))
|
|
def test_report_geometry_cannot_replace_native_measurement(self):
|
|
def forged(f,l,a,p,s):
|
|
p['geometry']['context'][0]+=.2;s['geometry']['context'][0]+=.2
|
|
self.reject(forged)
|
|
def test_missing_choice_record_or_reversed_choice_clock_refuses(self):
|
|
f,l,a,p,s=copy.deepcopy(self.template);content=raw(f)
|
|
line=next(x for x in content.splitlines() if x.startswith(f'ERR GATE CHOICE: run=test sample={s["final_sample"]} '))
|
|
self.assertFalse(validator.validate_selection_evidence(content.replace(line+'\n',''),'test',l,a,p,s)['verified'])
|
|
self.assertFalse(validator.validate_selection_evidence(content.replace(line,line.replace('run=test','run=other')),'test',l,a,p,s)['verified'])
|
|
|
|
|
|
if __name__=='__main__':unittest.main()
|