"""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=',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()