#!/usr/bin/env python3 """Runner integration policy tests: fake native processes and no display connections.""" import contextlib import copy import hashlib import io import json import os 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 import headless_display_probe as helper DEFAULT_REPORT = object() class DisplayRunnerTests(unittest.TestCase): def setUp(self): self.tmp=tempfile.TemporaryDirectory(prefix='erri-display-runner-test-') self.base=Path(self.tmp.name);self.root=self.base/'root';(self.root/'tools').mkdir(parents=True) (self.root/'tools/headless_display_probe.py').write_text('synthetic helper bytes for provenance test') self.compat=self.base/'compat';self.profile=self.base/'profile';self.profile.mkdir() self.steam=self.base/'Steam';self.exe=self.steam/'steamapps/common/Arma Reforger/ArmaReforgerSteam.exe' self.proton=self.steam/'steamapps/common/Proton - Experimental/proton' for path in (self.exe,self.proton):path.parent.mkdir(parents=True,exist_ok=True);path.write_bytes(b'synthetic file, never launched') self.command=['gamescope','--backend','headless','-W','1280','--',str(self.proton),'run',str(self.exe),'-noBackend'] self.wrapped,self.context=runner.display_probe_command(self.command,self.root,self.compat,self.profile,self.exe,'fixed-run-id',':0') self.session=self.base/'session';self.session.mkdir() def ident(pid):return {'pid':pid,'start':100,'uid':os.getuid()} self.report={**{k:self.context[k] for k in ('helper_sha256','prefix','profile','game_executable','report','command')}, 'schema':1,'kind':'headless-display-read-only','owned_game_window_verified':True, 'read_only':True,'input_sent':False,'observation_finished':True,'wrapper':ident(10), 'binding':{'display':':2','gamescope':ident(20),'xwayland':ident(30), 'sockets':[{'path':'@/tmp/.X11-unix/X2','inode':1000}], 'relationship':{'mode':'descendant','descendant':True}}, 'owned_windows':[{'window':12345,'pid':40,'server_pid':40,'process':ident(40),'title':'fixture game'}]} def tearDown(self):self.tmp.cleanup() def save(self,value=DEFAULT_REPORT): raw=(json.dumps(self.report if value is DEFAULT_REPORT else value,indent=3)+'\n').encode() Path(self.context['report']).write_bytes(raw);return raw def test_exact_gamescope_boundary_and_unmodified_proton_child(self): separator=self.command.index('--') self.assertEqual(self.wrapped[:separator+1],self.command[:separator+1]) self.assertEqual(self.wrapped[-len(self.context['command']):],self.command[separator+1:]) self.assertEqual(self.wrapped[separator+1],sys.executable) self.assertEqual(self.wrapped[separator+2],str(self.root/'tools/headless_display_probe.py')) for flag,value in (('--report',self.context['report']),('--host-display',':0'),('--prefix',str(self.compat/'pfx'))): self.assertEqual(self.wrapped[self.wrapped.index(flag)+1],value) self.assertEqual(self.command[separator+1],str(self.proton)) # caller array untouched def test_default_command_has_no_wrapper_and_remains_ordinary(self): command=runner.command_for(runner.ROOT,self.steam,self.compat,self.profile,'core') separator=command.index('--') self.assertEqual(command[separator+1:separator+4],[str(self.proton),'run',str(self.exe)]) self.assertNotIn('--host-display',command);self.assertNotIn('--query-worker',command) self.assertIn('-noSound',command);self.assertIn('-noBackend',command) def test_bad_boundary_missing_helper_and_existing_report_refused(self): wrong=self.command.copy();wrong[2]='wayland' with self.assertRaises(ValueError):runner.display_probe_command(wrong,self.root,self.compat,self.profile,self.exe,'fixed-run-id',':0') self.save() with self.assertRaises(ValueError):runner.display_probe_command(self.command,self.root,self.compat,self.profile,self.exe,'fixed-run-id',':0') Path(self.context['report']).unlink() helper=self.root/'tools/headless_display_probe.py';helper.unlink();helper.symlink_to(self.base/'absent') with self.assertRaises(ValueError):runner.display_probe_command(self.command,self.root,self.compat,self.profile,self.exe,'fixed-run-id',':0') def test_packed_and_vanilla_reject_before_resolution_or_launch(self): for project,packed in [('core',{'session':self.base/'packed'}),(runner.CONTROL,None)]: args=SimpleNamespace(project=project,inspect_headless_display=True) with patch.object(runner,'packed_request',return_value=packed),patch.object(runner.subprocess,'Popen') as spawn,patch.object(runner,'assert_no_game') as process_check: with self.assertRaisesRegex(ValueError,'loose Core or Chat'):runner.run(args) spawn.assert_not_called();process_check.assert_not_called() def test_exact_bytes_archived_with_hash_without_promoting_gameplay(self): raw=self.save();result=runner.display_probe_result(self.context,self.session) self.assertTrue(result['verified']);self.assertTrue(result['read_only']);self.assertFalse(result['input_sent']) self.assertEqual((self.session/'display-ownership.json').read_bytes(),raw) self.assertEqual(result['sha256'],hashlib.sha256(raw).hexdigest()) self.assertEqual(result['archive_file'],'display-ownership.json') self.assertFalse(runner.display_probe_result(self.context,self.session)['verified']) # exclusive archive copy self.assertEqual((self.session/'display-ownership.json').read_bytes(),raw) def test_each_launch_binding_mismatch_fails(self): for key in ('helper_sha256','prefix','profile','game_executable','report','command'): altered=copy.deepcopy(self.report);altered[key]=['foreign'] if key=='command' else 'foreign' self.save(altered) with self.subTest(key=key):self.assertFalse(runner.display_probe_result(self.context,None)['verified']) def test_absent_malformed_unverified_and_input_claims_fail(self): self.assertFalse(runner.display_probe_result(self.context,None)['verified']) for value in ([],None,{}, {**self.report,'owned_game_window_verified':False}, {**self.report,'read_only':False}, {**self.report,'input_sent':True}, {**self.report,'observation_finished':False}, {**self.report,'schema':2}): self.save(value);self.assertFalse(runner.display_probe_result(self.context,None)['verified']) Path(self.context['report']).write_text('{truncated') self.assertFalse(runner.display_probe_result(self.context,None)['verified']) Path(self.context['report']).unlink();Path(self.context['report']).symlink_to(self.base/'absent') self.assertFalse(runner.display_probe_result(self.context,None)['verified']) def test_host_display_uid_pid_start_or_window_disagreement_fails(self): faults=[] for target in ('wrapper','gamescope','xwayland','window'): for key,value in [('uid',os.getuid()+1),('pid',0),('start',0),('pid',True)]: bad=copy.deepcopy(self.report) item=(bad['wrapper'] if target=='wrapper' else bad['owned_windows'][0]['process'] if target=='window' else bad['binding'][target]) item[key]=value;faults.append(bad) for display in (':0',':0.1','unix:0','remote:2',''): bad=copy.deepcopy(self.report);bad['binding']['display']=display;faults.append(bad) for key,value in [('pid',41),('server_pid',41),('window',0),('window',None),('window',True)]: bad=copy.deepcopy(self.report);bad['owned_windows'][0][key]=value;faults.append(bad) bad=copy.deepcopy(self.report);bad['owned_windows']=[];faults.append(bad) for bad in faults: self.save(bad) with self.subTest(report=bad):self.assertFalse(runner.display_probe_result(self.context,None)['verified']) def test_missing_foreign_malformed_duplicate_socket_evidence_fails(self): for sockets in (None,[],[{'path':'@/tmp/.X11-unix/X0','inode':1000}], [{'path':'@/tmp/.X11-unix/X2','inode':0}], [{'path':'@/tmp/.X11-unix/X2','inode':True}], [{'path':'@/tmp/.X11-unix/X2','inode':1000}]*2): bad=copy.deepcopy(self.report);bad['binding']['sockets']=sockets;self.save(bad) with self.subTest(sockets=sockets):self.assertFalse(runner.display_probe_result(self.context,None)['verified']) def test_missing_unknown_or_unproven_relationship_refused(self): for relationship in (None,{}, {'mode':'unknown'}, {'mode':'descendant'}, {'mode':'descendant','descendant':False},{'mode':'descendant','descendant':1}, {'mode':'shared-listeners'}, {'mode':'shared-listeners','checks':{}},'descendant'): bad=copy.deepcopy(self.report);bad['binding']['relationship']=relationship;self.save(bad) with self.subTest(relationship=relationship):self.assertFalse(runner.display_probe_result(self.context,None)['verified']) bad=copy.deepcopy(self.report);del bad['binding']['relationship'];self.save(bad) self.assertFalse(runner.display_probe_result(self.context,None)['verified']) def test_shared_relationship_requires_every_check_and_exact_socket_inodes(self): checks={key:True for key in helper.SHARED_CHECKS};checks['listener_inodes']=[1000] good=copy.deepcopy(self.report);good['binding']['relationship']={'mode':'shared-listeners','checks':checks} self.save(good);self.assertTrue(runner.display_probe_result(self.context,None)['verified']) for key in helper.SHARED_CHECKS: for value in (False,None,1): bad=copy.deepcopy(good);bad['binding']['relationship']['checks'][key]=value;self.save(bad) with self.subTest(key=key,value=value):self.assertFalse(runner.display_probe_result(self.context,None)['verified']) bad=copy.deepcopy(good);del bad['binding']['relationship']['checks'][key];self.save(bad) self.assertFalse(runner.display_probe_result(self.context,None)['verified']) for inodes in (None,[],[1001],[True],[1000,1000],[1000,1001]): bad=copy.deepcopy(good);bad['binding']['relationship']['checks']['listener_inodes']=inodes;self.save(bad) with self.subTest(inodes=inodes):self.assertFalse(runner.display_probe_result(self.context,None)['verified']) bad=copy.deepcopy(good);bad['binding']['xwayland']['start']=99;self.save(bad) self.assertFalse(runner.display_probe_result(self.context,None)['verified']) good['binding']['sockets'].append({'path':'/tmp/.X11-unix/X2','inode':1001}) good['binding']['relationship']['checks']['listener_inodes']=[1000,1001] self.save(good);self.assertTrue(runner.display_probe_result(self.context,None)['verified']) def test_boolean_server_pid_is_not_an_integer_process_identity(self): bad=copy.deepcopy(self.report);window=bad['owned_windows'][0] window['pid']=1;window['process']['pid']=1;window['server_pid']=True self.save(bad);self.assertFalse(runner.display_probe_result(self.context,None)['verified']) def simulate_run_locked(self,display_ok,ordinary_failed=False): (self.exe.parent/'addons/data').mkdir(parents=True,exist_ok=True) args=SimpleNamespace(project='core',capture_prison_view=False,inspect_headless_display=True,timeout=120,settle=15,failure_grace=0) session=self.profile/'logs/logs_fixture' child=Mock();child.poll.return_value=None def spawn(*args,**kwargs): session.mkdir(parents=True);(session/'console.log').write_text('synthetic ordinary startup fixture') return child def output(args,**kwargs):return '[]' if '-f' in args else '11' monitor=SimpleNamespace(deadline=time.monotonic()+60,started=0,first_failure_at=None, observe=lambda *args:'game-started',annotate=lambda value:value) evidence={'ready':not ordinary_failed,'failed':ordinary_failed,'scope':'synthetic test only'} 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,'snapshot',return_value={})) 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=evidence)) stack.enter_context(patch.object(runner,'display_probe_result',return_value={'verified':display_ok})) stack.enter_context(patch.object(runner.shutil,'which',return_value='/usr/bin/synthetic')) stack.enter_context(patch.object(runner.subprocess,'Popen',side_effect=spawn)) stack.enter_context(patch.object(runner.subprocess,'check_output',side_effect=output)) stack.enter_context(patch.object(runner.subprocess,'run')) stack.enter_context(contextlib.redirect_stdout(io.StringIO())) status=runner.run_locked(args,self.steam,self.compat,self.profile) return status,json.loads((session/'runner.json').read_text()) def test_missing_display_proof_turns_actual_runner_path_into_failure(self): status,report=self.simulate_run_locked(False) self.assertEqual(status,1);self.assertEqual(report['result'],'display-probe-incomplete') self.assertEqual(report['startup_result_before_display_check'],'game-started') def test_valid_display_does_not_clear_native_failure(self): status,report=self.simulate_run_locked(True,ordinary_failed=True) self.assertEqual(status,1);self.assertEqual(report['result'],'game-failed') def test_valid_ownership_preserves_clean_startup_only(self): status,report=self.simulate_run_locked(True) self.assertEqual(status,0);self.assertEqual(report['result'],'game-started') self.assertTrue(report['display_ownership']['verified']) self.assertNotIn('multiplayer_verified',report) if __name__=='__main__':unittest.main()