163 lines
15 KiB
Python
163 lines
15 KiB
Python
"""Opt-in held runner wiring and independent proof boundary; no runtime or input."""
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
from types import SimpleNamespace, ModuleType
|
|
import unittest
|
|
from unittest.mock import patch
|
|
sys.path[:0]=[str(Path(__file__).resolve().parents[1]/'tools'),str(Path(__file__).resolve().parent)]
|
|
import headless_game as runner
|
|
import test_relative_look_runner as selection_fixtures
|
|
import test_headless_input_exercise_runner as exercise_fixtures
|
|
from test_ordinary_input_runner import NONCE
|
|
|
|
MODULES=('headless_input_proof.py','headless_display_probe.py','prison_approach_proof.py','ordinary_gate_evidence.py','ordinary_look_evidence.py','prison_approach_evidence.py','prison_alignment_proof.py','prison_alignment_evidence.py','ordinary_gate_choices.py','prison_action_selection_proof.py','prison_action_selection_evidence.py','prison_held_action_proof.py','prison_held_action_evidence.py')
|
|
|
|
class HeldRunnerTests(unittest.TestCase):
|
|
def options(self):
|
|
return dict(exercise_prison_held_action=True,exercise_prison_coarse_facing=True,exercise_prison_coarse_approach=True,exercise_prison_selection=True,exercise_prison_alignment=True,observe_prison_choices=True,exercise_prison_approach=True,exercise_relative_look=True,observe_prison_gate=True,exercise_player_input=True,observe_player_input=True,inspect_headless_display=True,project='chat',settle=50)
|
|
def test_all_prior_capabilities_required_before_filesystem_or_launch(self):
|
|
good=self.options();runner.validate_input_exercise(SimpleNamespace(**good));runner.validate_input_exercise(SimpleNamespace())
|
|
for change in [{key:False} for key in good if key not in ('exercise_prison_held_action','project','settle')]+[{'exercise_prison_held_action':1},{'exercise_prison_held_action':'yes'},{'settle':49},{'project':runner.CONTROL},{'packed_core':Path('/packed')},{'packed_addons':Path('/packed')},{'packed_session':Path('/packed')}]:
|
|
with self.subTest(change=change),patch.object(runner,'packed_request',side_effect=AssertionError('preflight reached filesystem')):
|
|
with self.assertRaises(ValueError):runner.run(SimpleNamespace(**(good|change)))
|
|
def test_actual_cli_parser_forwards_optin_and_preserves_default(self):
|
|
for enabled in (False,True):
|
|
argv=['headless_game.py','--compat','/private/unlaunched']
|
|
if enabled:
|
|
argv+=['--'+key.replace('_','-') for key,value in self.options().items() if value is True]
|
|
argv+=['--project','chat','--settle','50']
|
|
captured=[]
|
|
def fake_run(args):runner.validate_input_exercise(args);captured.append(args);return 17
|
|
with patch.object(sys,'argv',argv),patch.object(runner,'run',side_effect=fake_run):
|
|
self.assertEqual(runner.main(),17)
|
|
self.assertEqual(captured[0].exercise_prison_held_action,enabled)
|
|
def setup_command(self,root,held=True):
|
|
(root/'tools').mkdir();profile=root/'profile';profile.mkdir()
|
|
for name in MODULES:(root/'tools'/name).write_text('# private stub '+name)
|
|
native=['gamescope','--backend','headless','--','/proton','run','/game']
|
|
ordinary=runner.input_observer_command(native,'chat',None,NONCE,True,True,True,held_action=held)
|
|
wrapped,display=runner.display_probe_command(ordinary,root,Path('/compat'),profile,Path('/game'),NONCE,':0')
|
|
return native,ordinary,wrapped,(wrapped,root,Path('/compat'),profile,Path('/game'),NONCE,':0',display)
|
|
def test_native_and_helper_flags_exact_nonce_inside_existing_boundary(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
root=Path(folder);native,ordinary,wrapped,args=self.setup_command(root)
|
|
cmd,context=runner.input_exercise_command(*args,look=True,approach=True,alignment=True,selection=True,coarse_approach=True,coarse_facing=True,held_action=True)
|
|
self.assertEqual(ordinary[-2:],['-errOrdinaryHeldAction',NONCE]);self.assertEqual(cmd[-len(wrapped[4:]):],wrapped[4:]);self.assertEqual(context['command'],wrapped[4:])
|
|
self.assertEqual(cmd.count('--enable-private-held-action-proof'),1);self.assertTrue(context['held_action_requested'])
|
|
self.assertEqual(context['held_module_sha256'],{name:hashlib.sha256((root/'tools'/name).read_bytes()).hexdigest() for name in MODULES[-2:]})
|
|
normal=runner.input_observer_command(native,'chat',None,NONCE,True,True,True)
|
|
self.assertEqual(ordinary,normal+['-errOrdinaryHeldAction',NONCE])
|
|
for tail in ([],['-errOrdinaryHeldAction'],['-errOrdinaryHeldAction','foreign'],['-errOrdinaryHeldAction',NONCE]*2,['-errOrdinaryHeldAlias',NONCE]):
|
|
altered=wrapped[:-2]+tail;changed=(altered,*args[1:])
|
|
with self.subTest(tail=tail),self.assertRaises(ValueError):runner.input_exercise_command(*changed,look=True,approach=True,alignment=True,selection=True,coarse_approach=True,coarse_facing=True,held_action=True)
|
|
with self.assertRaises(ValueError):runner.input_exercise_command(*args,look=True,approach=True,alignment=True,selection=True,coarse_approach=True,coarse_facing=True)
|
|
with self.assertRaises(ValueError):runner.input_exercise_command(*args,look=True,approach=True,alignment=True,selection=True,coarse_approach=True,held_action=True)
|
|
def test_no_implicit_native_or_helper_activation_and_module_refusal(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
root=Path(folder);native,ordinary,wrapped,args=self.setup_command(root,False)
|
|
cmd,context=runner.input_exercise_command(*args,look=True,approach=True,alignment=True,selection=True,coarse_approach=True,coarse_facing=True)
|
|
self.assertNotIn('--enable-private-held-action-proof',cmd);self.assertNotIn('held_module_sha256',context);self.assertFalse(context.get('held_action_requested',False))
|
|
for flags in ((False,True,True),(True,False,True),(True,True,False)):
|
|
with self.assertRaises(ValueError):runner.input_observer_command(native,'chat',None,NONCE,*flags,held_action=True)
|
|
for duplicate in ('-errOrdinaryHeldAction','-errOrdinaryHeldAlias'):
|
|
with self.assertRaises(ValueError):runner.input_observer_command(native+[duplicate,NONCE],'chat',None,NONCE,True,True,True,held_action=True)
|
|
for name in MODULES[-2:]:
|
|
path=root/'tools'/name;old=path.read_bytes();path.unlink()
|
|
with self.assertRaises(ValueError):runner.held_module_hashes(root)
|
|
path.symlink_to(root/'tools'/MODULES[0])
|
|
with self.assertRaises(ValueError):runner.held_module_hashes(root)
|
|
path.unlink();path.write_bytes(old)
|
|
def fixture(self,root):
|
|
session,context,report,observation=selection_fixtures.SelectionRunner.make_selection(self,root)
|
|
context.update(held_action_requested=True,coarse_facing_requested=True,coarse_approach_requested=True,held_module_sha256=runner.held_module_hashes(runner.ROOT))
|
|
report.update({k:context[k] for k in ('held_action_requested','coarse_facing_requested','coarse_approach_requested','held_module_sha256')})
|
|
report['approach_proof']['profile']='coarse';report['facing_proof']['profile']='coarse'
|
|
held={'requested':True,'completed':True,'input_attempted':True,'status':'verified','down_monotonic':32.,'up_monotonic':36.2,'release_confirmed':True,'unconfirmed_held_key':None,'paired_release_errors':[],'notification_errors':[], 'events':[{'key':'F','down':down,'monotonic':at,'attempted_monotonic':at-.01,'confirmed':True} for down,at in ((True,32.),(False,36.2))]}
|
|
report.update(held_action_proof=held,held_release_errors=[],held_events=[{'key':'F','down':down,'monotonic':at,'sent':True} for down,at in ((True,32.),(False,36.2))])
|
|
return session,context,report,observation
|
|
def result(self,root,change=None,accepted=True,prior='game-started'):
|
|
session,context,report,observation=self.fixture(root)
|
|
if change:change(context,report,observation)
|
|
raw=(json.dumps(report,indent=2)+'\n').encode();Path(context['report']).write_bytes(raw)
|
|
approach=ModuleType('prison_approach_evidence');approach.validate_approach_evidence=lambda *a:{'verified':True}
|
|
selection=ModuleType('prison_action_selection_evidence');selection.validate_selection_evidence=lambda *a,**kw:{'verified':True}
|
|
held=ModuleType('prison_held_action_evidence');calls=[]
|
|
def validate(*args,**kwargs):calls.append((args,kwargs));return {'verified':accepted,'held_action_verified':accepted,'unlock_observed':accepted,'gate_motion_observed':accepted}
|
|
held.validate_held_action_evidence=validate
|
|
with patch.dict(sys.modules,{'prison_approach_evidence':approach,'prison_action_selection_evidence':selection,'prison_held_action_evidence':held}):
|
|
status,proof=runner.input_exercise_result(prior,context,session,observation,{'verified':True})
|
|
self.assertEqual((session/'input-proof.json').read_bytes(),raw)
|
|
return status,proof,calls,context,report
|
|
def test_raw_validator_receives_exact_archived_content_and_coarse_authority(self):
|
|
for accepted,prior in ((True,'game-started'),(False,'game-started'),(True,'game-failed'),(True,'cleanup-failed')):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
status,proof,calls,context,report=self.result(Path(folder),accepted=accepted,prior=prior)
|
|
self.assertEqual(calls,[(('native selection',NONCE,report['look_proof'],report['approach_proof'],report['facing_proof'],report['action_selection_proof'],report['held_action_proof']),{'facing_profile':'coarse'})])
|
|
self.assertEqual(proof['verified'],accepted);self.assertFalse(proof['gameplay_qualified'])
|
|
self.assertEqual(status,'input-exercise-incomplete' if not accepted and prior=='game-started' else prior)
|
|
self.assertIn('no gate passage',proof['scope'])
|
|
def test_wrong_hash_scope_or_release_and_extra_input_never_reach_raw_credit(self):
|
|
changes=[lambda c,p,o:p.update(held_action_requested=False),lambda c,p,o:p.update(held_action_requested=1),lambda c,p,o:c.update(held_action_requested=1),lambda c,p,o:p.update(held_module_sha256={}),lambda c,p,o:c.update(held_module_sha256={}),lambda c,p,o:c.update(coarse_facing_requested=False),lambda c,p,o:p.update(held_release_errors=['failed']),lambda c,p,o:p.update(unconfirmed_held_key='F'),lambda c,p,o:p.update(all_attempted_presses_released=False),lambda c,p,o:p['held_events'].append(copy.deepcopy(p['held_events'][1])),lambda c,p,o:p['held_events'][0].update(key='W'),lambda c,p,o:p['held_events'][0].update(down=1),lambda c,p,o:p['held_events'][0].update(monotonic=True),lambda c,p,o:p['held_events'][0].update(monotonic=float('nan')),lambda c,p,o:p['held_events'][0].update(sent=False),lambda c,p,o:p['held_events'][1].update(finally_=True),lambda c,p,o:p['held_action_proof'].update(release_confirmed=False),lambda c,p,o:p['held_action_proof'].update(notification_errors=['failed']),lambda c,p,o:p['held_action_proof']['events'][1].update(cleanup=True),lambda c,p,o:p['held_action_proof']['events'][1].update(monotonic=36.199),lambda c,p,o:p['held_action_proof'].update(up_monotonic=36.199),lambda c,p,o:p['held_action_proof'].update(status='failed-after-input'),lambda c,p,o:o.update(sampler_complete=False)]
|
|
for change in changes:
|
|
with self.subTest(change=change),tempfile.TemporaryDirectory() as folder:
|
|
status,proof,calls,*_=self.result(Path(folder),change)
|
|
self.assertEqual(status,'input-exercise-incomplete');self.assertFalse(proof['verified']);self.assertEqual(calls,[])
|
|
def test_late_or_reversed_pair_refuses_even_with_internally_equal_timestamps(self):
|
|
for down,up in ((32.,37.001),(32.,31.),(31.,35.)):
|
|
def change(c,p,o):
|
|
for i,(name,value) in enumerate((('down_monotonic',down),('up_monotonic',up))):
|
|
p['held_events'][i]['monotonic']=value;p['held_action_proof'][name]=value;p['held_action_proof']['events'][i]['monotonic']=value
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
status,proof,calls,*_=self.result(Path(folder),change);self.assertFalse(proof['verified']);self.assertEqual(calls,[])
|
|
def test_unrequested_F_or_held_result_refuses_and_exact_failure_poll_is_bound(self):
|
|
for payload in ({'held_events':[{'key':'F'}]},{'held_action_proof':{}},{'held_action_requested':True},{'held_release_errors':['uncertain']}):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
session,context,report,observation=exercise_fixtures.ExerciseTests.make_proof(self,Path(folder));report.update(payload);Path(context['report']).write_text(json.dumps(report))
|
|
status,proof=runner.input_exercise_result('game-started',context,session,observation,{'verified':True});self.assertFalse(proof['verified'])
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
session,context,report,observation=self.fixture(Path(folder));report.update(error='held release failed',sequence_completed=False)
|
|
Path(context['report']).write_text(json.dumps(report));self.assertTrue(runner.input_exercise_poll_failure(context));self.assertEqual(runner.input_exercise_poll_result(context,None,True),'game-failed')
|
|
for update in ({'held_action_requested':False},{'held_action_requested':1},{'held_module_sha256':{}}):
|
|
Path(context['report']).write_text(json.dumps(report|update));self.assertFalse(runner.input_exercise_poll_failure(context))
|
|
|
|
class FullRawHeldRunnerTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
from test_prison_held_action_evidence import build
|
|
cls.native_fixture=build('coarse')
|
|
def test_actual_policy_chain_through_all_raw_validators_and_runner(self):
|
|
from test_prison_held_action_evidence import raw
|
|
for fault in (None,'native_progress','native_door_motion','adapter_timestamp','expected_profile'):
|
|
with self.subTest(fault=fault),tempfile.TemporaryDirectory() as folder:
|
|
frames,look,route,facing,selected,held=copy.deepcopy(self.native_fixture)
|
|
session,context,report,observation=HeldRunnerTests.fixture(self,Path(folder))
|
|
context['nonce']='test';report['nonce']='test';observation['nonce']='test'
|
|
if fault=='native_progress':frames[held['progress_samples'][0]]['progress']=3.9
|
|
if fault=='native_door_motion':
|
|
for index in range(held['transition_sample'],held['final_sample']+1):frames[index]['door_state']=0
|
|
report.update(look_proof=look,approach_proof=route,facing_proof=facing,action_selection_proof=selected,held_action_proof=held)
|
|
report['events']=copy.deepcopy(route['events'])
|
|
report['wheel_events']=[{'button':5,'down':down,'sent':True,'monotonic':selected['wheel'][field]} for down,field in ((True,'down_monotonic'),(False,'up_monotonic'))]
|
|
report['held_events']=[{'key':'F','down':down,'sent':True,'monotonic':held[field]} for down,field in ((True,'down_monotonic'),(False,'up_monotonic'))]
|
|
if fault=='adapter_timestamp':report['held_events'][1]['monotonic']+=.001
|
|
if fault=='expected_profile':context['coarse_facing_requested']=False;report['coarse_facing_requested']=False
|
|
# Match the real wrapper's compact serialization, leaving raw native logs
|
|
# as the independent source for every state/position/progress measurement.
|
|
for record in (route,facing,selected,held):
|
|
record['samples']=[{key:x[key] for key in ('sample','real_ms','received_monotonic')} for x in record['samples']]
|
|
content=raw(frames);(session/'console.log').write_text(content)
|
|
original=(json.dumps(report,indent=1)+'\n').encode();Path(context['report']).write_bytes(original)
|
|
status,result=runner.input_exercise_result('game-started',context,session,observation,{'verified':True})
|
|
self.assertEqual((session/'input-proof.json').read_bytes(),original)
|
|
if fault is None:
|
|
self.assertEqual(status,'game-started',result);self.assertTrue(result['verified'],result)
|
|
self.assertTrue(result['held_action_proof']['held_action_verified']);self.assertFalse(result['held_action_proof']['escape_verified'])
|
|
self.assertEqual(result['held_action_proof']['transition_sample'],held['transition_sample'])
|
|
else:self.assertEqual(status,'input-exercise-incomplete',result);self.assertFalse(result['verified'],result)
|
|
|
|
if __name__=='__main__':unittest.main()
|