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