169 lines
13 KiB
Python
169 lines
13 KiB
Python
"""Ordinary observer runner tests: fake process boundaries; never launch a game."""
|
|
import contextlib
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest.mock import Mock, patch
|
|
sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'tools'))
|
|
import headless_game as runner
|
|
|
|
NONCE='a1'*16
|
|
|
|
def records(nonce=NONCE, menus=False):
|
|
def line(index,real,event):
|
|
return f'SCRIPT : ERR INPUT OBSERVE: run={nonce} sample={index} real={1000+real} world={5000+real} {event}\n'
|
|
result=line(0,0,'QUEUED')
|
|
for i in range(1,450):
|
|
opened=menus and 22<=i<=25
|
|
result+=line(i,i*100,f'SAMPLE origin=<1, 2, 3> forward=0 input=<0, 0, 0> velocity=<0, 0, 0> inventory={int(opened)}')
|
|
if i==21:result+=line(i,i*100,'READY')
|
|
if not menus and i==24:result+=line(i,i*100,'CAPTURE_REQUEST baseline')
|
|
if menus and i==22:result+=line(i,i*100,'INVENTORY_OPEN')
|
|
if menus and i==25:result+=line(i,i*100,'CAPTURE_REQUEST open')
|
|
if menus and i==26:result+=line(i,i*100,'INVENTORY_CLOSED')
|
|
if menus and i==29:result+=line(i,i*100,'CAPTURE_REQUEST closed')
|
|
return result+line(449,45000,'FINISHED observation-bound')
|
|
|
|
class InputRunnerTests(unittest.TestCase):
|
|
def test_default_and_explicit_command_scope(self):
|
|
args=(runner.ROOT,Path('/steam'),Path('/compat'),Path('/profile'))
|
|
for project in ('core','chat'):
|
|
original=runner.command_for(*args,project)
|
|
self.assertNotIn('-errOrdinaryInputObserve',original)
|
|
command=runner.input_observer_command(original,project,None,NONCE)
|
|
self.assertEqual(command[:-4],original)
|
|
self.assertEqual(command[-4:],['-errOrdinaryInputObserve',NONCE,'-errOrdinaryInputScreenshots','1'])
|
|
self.assertNotIn('-errOrdinaryInputObserve',original)
|
|
for project,packed,nonce in ((runner.CONTROL,None,NONCE),('core',{},NONCE),('core',None,'../bad'),('core',None,''),('core',None,'x'*33)):
|
|
with self.assertRaises(ValueError):runner.input_observer_command([],project,packed,nonce)
|
|
with self.assertRaises(ValueError):runner.input_observer_command(['-errOrdinaryInputObserve',NONCE],'core',None,NONCE)
|
|
|
|
def test_refusal_before_any_launch_or_profile_work(self):
|
|
for changes in ({'project':runner.CONTROL},{'packed_core':Path('/candidate')},{'packed_addons':Path('/candidate')},{'settle':49}):
|
|
args=SimpleNamespace(project='core',observe_player_input=True,packed_core=None,packed_addons=None,settle=50)
|
|
for key,value in changes.items():setattr(args,key,value)
|
|
with patch.object(runner,'packed_request',side_effect=AssertionError('early refusal required')):
|
|
with self.assertRaises(ValueError):runner.run(args)
|
|
runner.validate_input_observer(SimpleNamespace(observe_player_input=False))
|
|
runner.validate_input_observer(SimpleNamespace(observe_player_input=True,project='chat',settle=50))
|
|
|
|
def test_strict_complete_sampler_is_not_input_success(self):
|
|
evidence=runner.input_observer_evidence(records(),NONCE)
|
|
self.assertTrue(evidence['sampler_complete'],evidence)
|
|
self.assertEqual(evidence['sample_count'],449)
|
|
self.assertFalse(evidence['input_sent'])
|
|
self.assertFalse(evidence['inventory_open_observed'])
|
|
self.assertNotIn('movement_success',evidence)
|
|
menu=runner.input_observer_evidence(records(menus=True),NONCE)
|
|
self.assertTrue(menu['sampler_complete'],menu)
|
|
self.assertTrue(menu['inventory_open_observed'] and menu['inventory_close_observed'])
|
|
self.assertEqual(menu['capture_requests'],['open','closed'])
|
|
|
|
def test_malformed_missing_reordered_unbounded_and_foreign_records_refused(self):
|
|
good=records()
|
|
cases=['',good.replace(' QUEUED',' WRONG'),good.replace(' READY',' MISSING'),good.replace(' FINISHED observation-bound',' CANCELLED original-admission-lost'),
|
|
good.replace(NONCE,'b'*32,1),good.replace('sample=2 real=1200','sample=3 real=1200',1),good.replace('real=1200','real=1000',1),
|
|
good.replace('world=5200','world=4900',1),good.replace('origin=<1, 2, 3>','origin=<NaN, 2, 3>',1),good.replace('forward=0','forward=1e999',1),
|
|
good.replace('real=46000 world=50000','real=45999 world=50000'),good.replace('sample=449 real=46000','sample=450 real=46000'),
|
|
good.replace('FINISHED observation-bound','FINISHED sample-bound'),good+good.splitlines()[0]+'\n',good.replace('inventory=0','inventory=unknown',1),
|
|
good.replace('real=1200','real=46000',1),good.replace('real=3100 world=7100 READY','real=3000 world=7100 READY')]
|
|
for candidate in cases:
|
|
with self.subTest(candidate=candidate[-150:]):self.assertFalse(runner.input_observer_evidence(candidate,NONCE)['sampler_complete'])
|
|
self.assertFalse(runner.input_observer_evidence(good*2,NONCE)['sampler_complete'])
|
|
self.assertFalse(runner.input_observer_evidence(good,'../bad')['sampler_complete'])
|
|
|
|
def test_transition_and_capture_cannot_manufacture_inventory_observation(self):
|
|
good=records(menus=True)
|
|
for bad in (good.replace('INVENTORY_OPEN','INVENTORY_CLOSED'),good.replace('inventory=1','inventory=0'),good.replace('CAPTURE_REQUEST open','CAPTURE_REQUEST baseline')):
|
|
self.assertFalse(runner.input_observer_evidence(bad,NONCE)['sampler_complete'])
|
|
|
|
def test_delayed_capture_requires_delay_current_sample_and_transition_interval(self):
|
|
def add_after(text, anchor, suffix):
|
|
lines=text.splitlines()
|
|
index=next(i for i,line in enumerate(lines) if anchor in line)
|
|
prefix=lines[index].split(' SAMPLE ',1)[0] if ' SAMPLE ' in lines[index] else lines[index].rsplit(' ',1)[0]
|
|
lines.insert(index+1,prefix+' CAPTURE_REQUEST '+suffix)
|
|
return '\n'.join(lines)+'\n'
|
|
baseline=records()
|
|
without='\n'.join(line for line in baseline.splitlines() if 'CAPTURE_REQUEST' not in line)+'\n'
|
|
# A skipped capture is valid; a same-tick or 200ms request is not.
|
|
self.assertTrue(runner.input_observer_evidence(without,NONCE)['sampler_complete'])
|
|
for anchor in (' READY','sample=23 real=3300 world=7300 SAMPLE'):
|
|
self.assertFalse(runner.input_observer_evidence(add_after(without,anchor,'baseline'),NONCE)['sampler_complete'])
|
|
duplicate=baseline.replace('CAPTURE_REQUEST baseline','CAPTURE_REQUEST baseline\n'+next(line for line in baseline.splitlines() if 'CAPTURE_REQUEST baseline' in line))
|
|
stale=baseline.replace('sample=24 real=3400 world=7400 CAPTURE','sample=23 real=3400 world=7400 CAPTURE')
|
|
self.assertFalse(runner.input_observer_evidence(duplicate,NONCE)['sampler_complete'])
|
|
self.assertFalse(runner.input_observer_evidence(stale,NONCE)['sampler_complete'])
|
|
menus=records(menus=True)
|
|
no_caps='\n'.join(line for line in menus.splitlines() if 'CAPTURE_REQUEST' not in line)+'\n'
|
|
for anchor,suffix in (('sample=25 real=3500 world=7500 SAMPLE','baseline'),(' INVENTORY_CLOSED','open')):
|
|
self.assertFalse(runner.input_observer_evidence(add_after(no_caps,anchor,suffix),NONCE)['sampler_complete'])
|
|
wrong_state=menus.replace('sample=25 real=3500 world=7500 SAMPLE origin=<1, 2, 3> forward=0 input=<0, 0, 0> velocity=<0, 0, 0> inventory=1',
|
|
'sample=25 real=3500 world=7500 SAMPLE origin=<1, 2, 3> forward=0 input=<0, 0, 0> velocity=<0, 0, 0> inventory=0')
|
|
self.assertFalse(runner.input_observer_evidence(wrong_state,NONCE)['sampler_complete'])
|
|
|
|
def test_delayed_baseline_rejects_intervening_motion_even_if_returned(self):
|
|
good=records()
|
|
original='sample=22 real=3200 world=7200 SAMPLE origin=<1, 2, 3> forward=0 input=<0, 0, 0> velocity=<0, 0, 0>'
|
|
for changed in (original.replace('origin=<1, 2, 3>','origin=<1.2, 2, 3>'),original.replace('forward=0','forward=1'),
|
|
original.replace('input=<0, 0, 0>','input=<0, 0, 1>'),original.replace('velocity=<0, 0, 0>','velocity=<0, 0, 1>')):
|
|
with self.subTest(changed=changed):
|
|
self.assertFalse(runner.input_observer_evidence(good.replace(original,changed),NONCE)['sampler_complete'])
|
|
|
|
def test_prior_failure_results_are_never_replaced(self):
|
|
for previous in ('game-failed','cleanup-failed','source-changed','display-probe-incomplete','timeout','process-exited'):
|
|
for content in ('',records()):self.assertEqual(runner.input_observer_result(previous,content,NONCE)[0],previous)
|
|
self.assertEqual(runner.input_observer_result('game-started','',NONCE)[0],'observation-incomplete')
|
|
self.assertEqual(runner.input_observer_result('game-started',records(),NONCE)[0],'game-started')
|
|
|
|
def simulate(self,observe,content,display_ok=True,native_failed=False):
|
|
with tempfile.TemporaryDirectory(prefix='erri-input-runner-test-') as tmp:
|
|
root=Path(tmp);steam=root/'steam';compat=root/'compat';profile=root/'profile'
|
|
exe=steam/'steamapps/common/Arma Reforger/ArmaReforgerSteam.exe';proton=steam/'steamapps/common/Proton - Experimental/proton'
|
|
for path in (exe,proton):path.parent.mkdir(parents=True,exist_ok=True);path.write_bytes(b'fake never launched')
|
|
(exe.parent/'addons/data').mkdir(parents=True)
|
|
session=profile/'logs/logs_fixture';child=Mock();child.poll.return_value=None
|
|
def spawn(*args,**kwargs):session.mkdir(parents=True);(session/'console.log').write_text(content);return child
|
|
def wrap(command,*args):
|
|
if observe:self.assertEqual(command[-4:],['-errOrdinaryInputObserve',NONCE,'-errOrdinaryInputScreenshots','1'])
|
|
return command,{'fake':True}
|
|
args=SimpleNamespace(project='core',capture_prison_view=False,inspect_headless_display=True,observe_player_input=observe,timeout=180,settle=50,failure_grace=0)
|
|
monitor=SimpleNamespace(deadline=time.monotonic()+60,started=0,first_failure_at=None,observe=lambda *args:'game-started',annotate=lambda x:x)
|
|
captured=[]
|
|
def snapshot(root,command):captured.append(command.copy());return {}
|
|
with contextlib.ExitStack() as stack:
|
|
for name in ('assert_no_game','assert_prefix_idle','persist','cleanup'):stack.enter_context(patch.object(runner,name))
|
|
stack.enter_context(patch.object(runner.uuid,'uuid4',return_value=SimpleNamespace(hex=NONCE)))
|
|
stack.enter_context(patch.object(runner,'snapshot',side_effect=snapshot))
|
|
stack.enter_context(patch.object(runner,'bindings',return_value={}))
|
|
stack.enter_context(patch.object(runner,'seal_run',return_value={'source_unchanged_after_run':True}))
|
|
stack.enter_context(patch.object(runner,'StartupMonitor',return_value=monitor))
|
|
stack.enter_context(patch.object(runner,'game_evidence',return_value={'ready':not native_failed,'failed':native_failed}))
|
|
stack.enter_context(patch.object(runner,'display_probe_command',side_effect=wrap))
|
|
stack.enter_context(patch.object(runner,'display_probe_result',return_value={'verified':display_ok}))
|
|
stack.enter_context(patch.object(runner.shutil,'which',return_value='/fake'))
|
|
stack.enter_context(patch.object(runner.subprocess,'Popen',side_effect=spawn))
|
|
stack.enter_context(patch.object(runner.subprocess,'check_output',side_effect=lambda args,**kw:'[]' if '-f' in args else '11'))
|
|
stack.enter_context(patch.object(runner.subprocess,'run'))
|
|
stack.enter_context(contextlib.redirect_stdout(io.StringIO()))
|
|
status=runner.run_locked(args,steam,compat,profile)
|
|
return status,json.loads((session/'runner.json').read_text()),json.loads((session/'game-startup-report.json').read_text()),captured[0]
|
|
|
|
def test_actual_report_flow_default_success_missing_and_prior_failure(self):
|
|
status,report,final,command=self.simulate(False,'')
|
|
self.assertEqual(status,0);self.assertNotIn('input_observation',report);self.assertNotIn('-errOrdinaryInputObserve',command)
|
|
for content,display_ok,native_failed,result in ((records(),True,False,'game-started'),('',True,False,'observation-incomplete'),('',False,False,'display-probe-incomplete'),(records(),True,True,'game-failed')):
|
|
status,report,final,command=self.simulate(True,content,display_ok,native_failed)
|
|
self.assertEqual(report['result'],result)
|
|
self.assertEqual(status,0 if result=='game-started' else 1)
|
|
self.assertEqual(report['input_observation'],final['input_observation'])
|
|
self.assertEqual(report['input_observation']['nonce'],NONCE)
|
|
self.assertIn(NONCE,command)
|
|
|
|
if __name__=='__main__':unittest.main()
|