113 lines
9.6 KiB
Python
113 lines
9.6 KiB
Python
"""Runner policy/provenance tests only: no processes, display, or input calls."""
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest.mock import patch
|
|
sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'tools'))
|
|
import headless_game as r
|
|
from test_ordinary_input_runner import records, NONCE
|
|
|
|
class ExerciseTests(unittest.TestCase):
|
|
def test_optin_preflight_before_side_effects_and_default_unchanged(self):
|
|
r.validate_input_exercise(SimpleNamespace())
|
|
for project in ('core','chat'):
|
|
args=SimpleNamespace(exercise_player_input=True,observe_player_input=True,inspect_headless_display=True,project=project,settle=50)
|
|
r.validate_input_exercise(args)
|
|
for change in ({'observe_player_input':False},{'inspect_headless_display':False},{'settle':49},{'project':r.CONTROL},{'project':'ai'},{'packed_core':Path('/x')},{'packed_addons':Path('/x')},{'packed_session':Path('/x')}):
|
|
bad=copy.copy(args)
|
|
for key,value in change.items():setattr(bad,key,value)
|
|
with self.subTest(change=change),patch.object(r,'packed_request',side_effect=AssertionError('must refuse before filesystem')):
|
|
with self.assertRaises(ValueError):r.run(bad)
|
|
|
|
def test_nested_command_keeps_original_boundary_and_proton_argv(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
root=Path(folder);(root/'tools').mkdir();profile=root/'profile';profile.mkdir()
|
|
for file in ('headless_input_proof.py','headless_display_probe.py'):(root/'tools'/file).write_text('# fake helper')
|
|
base=['gamescope','--backend','headless','--','/proton','run','/game','-profile','Z:/profile','-errOrdinaryInputObserve',NONCE]
|
|
wrapped,display=r.display_probe_command(base,root,Path('/compat'),profile,Path('/game'),NONCE,':0')
|
|
command,context=r.input_exercise_command(wrapped,root,Path('/compat'),profile,Path('/game'),NONCE,':0',display)
|
|
self.assertEqual(command[:4],base[:4]);self.assertEqual(command[-len(base[4:]):],base[4:])
|
|
child_start=command.index('--',4)+1
|
|
self.assertEqual(command[child_start:],wrapped[4:]);self.assertEqual(context['command'],wrapped[4:])
|
|
self.assertEqual(context['nonce'],NONCE)
|
|
with self.assertRaises(ValueError):r.input_exercise_command(base,root,Path('/compat'),profile,Path('/game'),NONCE,':0',display)
|
|
with self.assertRaises(ValueError):r.input_exercise_command(wrapped,root,Path('/compat'),profile,Path('/game'),NONCE,':0',None)
|
|
Path(context['report']).write_text('{}')
|
|
with self.assertRaises(ValueError):r.input_exercise_command(wrapped,root,Path('/compat'),profile,Path('/game'),NONCE,':0',display)
|
|
|
|
def test_native_movement_requires_action_and_real_motion_before_menu(self):
|
|
static=records(menus=True)
|
|
evidence=r.input_observer_evidence(static,NONCE)
|
|
self.assertTrue(evidence['sampler_complete']);self.assertFalse(evidence['movement']['action_observed']);self.assertFalse(evidence['movement']['physical_motion_observed'])
|
|
# First post-READY sample is also the native opening sample: final movement
|
|
# can precede opening during the same sample interval.
|
|
line='sample=22 real=3200 world=7200 SAMPLE origin=<1, 2, 3> forward=0 input=<0, 0, 0> velocity=<0, 0, 0>'
|
|
moving=static.replace(line,'sample=22 real=3200 world=7200 SAMPLE origin=<1.2, 2, 3> forward=1 input=<0, 0, 1> velocity=<1, 0, 0>')
|
|
good=r.input_observer_evidence(moving,NONCE)
|
|
self.assertTrue(good['sampler_complete']);self.assertTrue(good['movement']['action_observed']);self.assertTrue(good['movement']['physical_motion_observed']);self.assertTrue(good['movement']['released_input_observed'])
|
|
# Native release requires a later sample, with both controls neutral.
|
|
for held in ('forward=1 input=<0, 0, 0>', 'forward=0 input=<0, 0.2, 0>'):
|
|
after=moving.split('INVENTORY_CLOSED',1)
|
|
bad=after[0]+'INVENTORY_CLOSED'+after[1].replace('forward=0 input=<0, 0, 0>',held)
|
|
self.assertFalse(r.input_observer_evidence(bad,NONCE)['movement']['released_input_observed'])
|
|
for text,key in ((moving.replace('forward=1','forward=0'),'action_observed'),(moving.replace('input=<0, 0, 1>','input=<0, 0, 0>'),'action_observed'),(moving.replace('origin=<1.2, 2, 3>','origin=<1, 2, 3>'),'physical_motion_observed'),(moving.replace('velocity=<1, 0, 0>','velocity=<0, 0, 0>'),'physical_motion_observed')):
|
|
self.assertFalse(r.input_observer_evidence(text,NONCE)['movement'][key])
|
|
|
|
def make_proof(self,root):
|
|
session=root/'logs_session';session.mkdir()
|
|
context={'report':str(root/'report.json'),'helper_sha256':'h','ownership_helper_sha256':'d','prefix':'/compat/pfx','profile':'/profile','game_executable':'/game','command':['display-wrapper'],'nonce':NONCE}
|
|
owner={'window':10,'pid':20,'server_pid':20,'process':{'pid':20,'start':5}}
|
|
binding={'display':':99'}
|
|
(session/'display-ownership.json').write_text(json.dumps({'binding':binding,'owned_windows':[owner]}))
|
|
report={**context,'schema':1,'kind':'headless-private-keyboard-experiment','sequence_completed':True,'hard_supervision':True,'all_attempted_presses_released':True,'input_sent':True,'paired_release_errors':[], 'console':str(session/'console.log'),'binding':binding,'owner':owner,'observer':{'events':['READY','INVENTORY_OPEN','INVENTORY_CLOSED']},'events':[{'key':key,'down':down,'monotonic':1+i*.2} for i,(key,down) in enumerate([('W',True),('W',False),('Tab',True),('Tab',False),('Tab',True),('Tab',False)])]}
|
|
observation={'nonce':NONCE,'sampler_complete':True,'inventory_open_observed':True,'inventory_close_observed':True,'movement':{'action_observed':True,'physical_motion_observed':True,'released_input_observed':True}}
|
|
return session,context,report,observation
|
|
|
|
def test_live_poll_ignores_incomplete_foreign_or_success_and_latches_exact_failure(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
session,context,report,observation=self.make_proof(Path(folder));path=Path(context['report'])
|
|
self.assertFalse(r.input_exercise_poll_failure(context))
|
|
for raw in (b'',b'{',b'null',b'[]',json.dumps(report).encode()):
|
|
path.write_bytes(raw);self.assertFalse(r.input_exercise_poll_failure(context))
|
|
failed={**report,'sequence_completed':False,'error':'key release not confirmed'}
|
|
path.write_text(json.dumps({**failed,'nonce':'foreign'}));self.assertFalse(r.input_exercise_poll_failure(context))
|
|
for detail in ({'error':'native helper refused'}, {'supervision_failed':True}, {'sequence_completed':False,'whole_process_group_cleanup_verified':False}):
|
|
path.write_text(json.dumps({**report,**detail}))
|
|
self.assertTrue(r.input_exercise_poll_failure(context))
|
|
self.assertEqual(r.input_exercise_poll_result(context,None,False),'input-exercise-failed')
|
|
self.assertEqual(r.input_exercise_poll_result(context,'game-started',False),'input-exercise-failed')
|
|
self.assertEqual(r.input_exercise_poll_result(context,None,True),'game-failed')
|
|
self.assertEqual(r.input_exercise_poll_result(context,'process-exited',False),'process-exited')
|
|
self.assertIsNone(r.input_exercise_poll_result(None,None,False))
|
|
# Initial false status without the finally completion field is not terminal.
|
|
path.write_text(json.dumps({**report,'sequence_completed':False}))
|
|
self.assertFalse(r.input_exercise_poll_failure(context))
|
|
|
|
def test_exact_bytes_archived_and_verified(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
session,context,report,observation=self.make_proof(Path(folder))
|
|
raw=(json.dumps(report,indent=3)+'\n').encode();Path(context['report']).write_bytes(raw)
|
|
status,result=r.input_exercise_result('game-started',context,session,observation,{'verified':True})
|
|
self.assertEqual(status,'game-started');self.assertTrue(result['verified'],result);self.assertFalse(result['gameplay_qualified'])
|
|
self.assertEqual((session/'input-proof.json').read_bytes(),raw);self.assertEqual(result['sha256'],hashlib.sha256(raw).hexdigest())
|
|
|
|
def test_report_or_native_failure_cannot_qualify_or_replace_prior_failure(self):
|
|
mutations=[lambda p:p.update(sequence_completed=False),lambda p:p.update(hard_supervision=False),lambda p:p.update(supervision_failed=True),lambda p:p.update(all_attempted_presses_released=False),lambda p:p.update(error='failed'),lambda p:p.update(paired_release_errors=['failed']),lambda p:p.update(unconfirmed_held_key='W'),lambda p:p.update(nonce='wrong'),lambda p:p.update(helper_sha256='wrong'),lambda p:p.update(command=['wrong']),lambda p:p.update(console='/other/console.log'),lambda p:p['events'].pop(),lambda p:p['events'][0].update(down=1),lambda p:p['events'][0].update(monotonic=float('nan')),lambda p:p['events'][-1].update(monotonic=100),lambda p:p['owner'].update(pid=99),lambda p:p['observer'].update(events=['READY'])]
|
|
for change in mutations:
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
session,context,report,observation=self.make_proof(Path(folder));change(report);Path(context['report']).write_text(json.dumps(report))
|
|
status,result=r.input_exercise_result('game-started',context,session,observation,{'verified':True})
|
|
self.assertEqual(status,'input-exercise-incomplete');self.assertFalse(result['verified']);self.assertTrue((session/'input-proof.json').is_file())
|
|
for prior in ('game-failed','cleanup-failed','source-changed','timeout','display-probe-incomplete','game-started'):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
session,context,report,observation=self.make_proof(Path(folder));Path(context['report']).write_text(json.dumps(report));observation['movement']['released_input_observed']=False
|
|
status,result=r.input_exercise_result(prior,context,session,observation,{'verified':True})
|
|
self.assertEqual(status,'input-exercise-incomplete' if prior=='game-started' else prior);self.assertFalse(result['verified'])
|
|
|
|
if __name__=='__main__':unittest.main()
|