178 lines
12 KiB
Python
178 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""Runner cleanup integration with fake processes; no native launch or real signals."""
|
|
import ast
|
|
import contextlib
|
|
import io
|
|
import json
|
|
from pathlib import Path
|
|
import runpy
|
|
import sys
|
|
import tempfile
|
|
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 game
|
|
import native_provenance
|
|
import private_prefix_cleanup as private
|
|
import test_headless_display_runner as display_tests
|
|
|
|
WB=Path(__file__).resolve().parents[1]/'tools/headless_workbench.py'
|
|
|
|
|
|
class CleanupIntegrationTests(unittest.TestCase):
|
|
def test_canonical_cleanup_order_then_idle_verified(self):
|
|
order=[];proof={'cleared':True,'term_sent':[12],'kill_sent':[]}
|
|
with patch.object(game,'cleanup',side_effect=lambda *a:order.append('normal')),\
|
|
patch.object(private,'cleanup_private_prefix',side_effect=lambda *a:(order.append('private') or proof)),\
|
|
patch.object(game,'assert_prefix_idle',side_effect=lambda *a:order.append('idle')):
|
|
report=game.cleanup_owned_run(Mock(),private.INSTALLED_PROTON,{},private.COMPAT)
|
|
self.assertEqual(order,['normal','private','idle']);self.assertTrue(report['cleanup_confirmed'])
|
|
self.assertEqual(report['orphan_cleanup']['report'],proof);self.assertTrue(report['idle_verified'])
|
|
def test_noncanonical_private_prefix_never_gets_orphan_signalling(self):
|
|
with patch.object(game,'cleanup'),patch.object(game,'assert_prefix_idle') as idle,\
|
|
patch.object(private,'cleanup_private_prefix') as narrow:
|
|
result=game.cleanup_owned_run(Mock(),Path('/other/proton'),{},Path('/tmp/other-private-compat'))
|
|
narrow.assert_not_called();idle.assert_called_once_with(Path('/tmp/other-private-compat'))
|
|
self.assertTrue(result['cleanup_confirmed']);self.assertFalse(result['orphan_cleanup']['attempted'])
|
|
def test_errors_preserve_partial_helper_report_and_still_run_idle_check(self):
|
|
partial={'cleared':False,'term_sent':[12],'kill_sent':[]};order=[]
|
|
def narrow(*args):order.append('private');raise private.CleanupRefused('remaining identity unverified',partial)
|
|
with patch.object(game,'cleanup',side_effect=OSError('normal shutdown failed')),\
|
|
patch.object(private,'cleanup_private_prefix',side_effect=narrow),\
|
|
patch.object(game,'assert_prefix_idle',side_effect=lambda *a:(order.append('idle') or None)):
|
|
result=game.cleanup_owned_run(Mock(),private.INSTALLED_PROTON,{},private.COMPAT)
|
|
self.assertEqual(order,['private','idle']);self.assertFalse(result['cleanup_confirmed'])
|
|
self.assertTrue(result['idle_verified']);self.assertEqual(result['orphan_cleanup']['report'],partial)
|
|
self.assertEqual(len(result['errors']),2)
|
|
def test_false_helper_completion_or_busy_final_prefix_never_qualifies(self):
|
|
for details in ({'cleared':False},{'cleared':1},None):
|
|
with patch.object(game,'cleanup'),patch.object(private,'cleanup_private_prefix',return_value=details),patch.object(game,'assert_prefix_idle'):
|
|
self.assertFalse(game.cleanup_owned_run(Mock(),private.INSTALLED_PROTON,{},private.COMPAT)['cleanup_confirmed'])
|
|
with patch.object(game,'cleanup'),patch.object(private,'cleanup_private_prefix',return_value={'cleared':True}),\
|
|
patch.object(game,'assert_prefix_idle',side_effect=ValueError('private prefix still busy')):
|
|
result=game.cleanup_owned_run(Mock(),private.INSTALLED_PROTON,{},private.COMPAT)
|
|
self.assertFalse(result['cleanup_confirmed']);self.assertFalse(result['idle_verified'])
|
|
|
|
def test_game_runner_cleanup_failure_is_recorded_and_nonzero(self):
|
|
fixture=display_tests.DisplayRunnerTests();fixture.setUp()
|
|
proof={'cleanup_confirmed':False,'idle_verified':False,'errors':['idle verification: still busy'],
|
|
'orphan_cleanup':{'attempted':True,'report':{'cleared':False}}}
|
|
try:
|
|
with patch.object(game,'cleanup_owned_run',return_value=proof):
|
|
status,summary=fixture.simulate_run_locked(True)
|
|
self.assertEqual(status,1);self.assertEqual(summary['result'],'cleanup-failed')
|
|
self.assertFalse(summary['cleanup_confirmed']);self.assertEqual(summary['cleanup_report'],proof)
|
|
self.assertIn('idle verification',summary['runner_error'])
|
|
self.assertNotIn('source_unchanged_after_run',summary)
|
|
finally:fixture.tearDown()
|
|
|
|
|
|
class WorkbenchIntegrationTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp=tempfile.TemporaryDirectory(prefix='erri-cleanup-runner-test-');self.base=Path(self.tmp.name)
|
|
self.steam=self.base/'Steam';self.compat=self.base/'compat';self.profile=self.base/'profile'
|
|
(self.compat/'pfx').mkdir(parents=True);(self.compat/'pfx/system.reg').write_text('synthetic fixture')
|
|
self.proton=self.steam/'steamapps/common/Proton - Experimental/proton'
|
|
self.exe=self.steam/'steamapps/common/Arma Reforger Tools/Workbench/ArmaReforgerWorkbenchSteamDiag.exe'
|
|
for file in (self.proton,self.exe):file.parent.mkdir(parents=True,exist_ok=True);file.write_text('not executable')
|
|
(self.steam/'steamapps/common/Arma Reforger/addons/data').mkdir(parents=True)
|
|
self.open_locks=[]
|
|
original_open=Path.open
|
|
def tracked_open(path,*args,**kwargs):
|
|
stream=original_open(path,*args,**kwargs)
|
|
if path.name=='.erri-workbench.lock':self.open_locks.append(stream)
|
|
return stream
|
|
self.tracked_open=tracked_open
|
|
self.argv=[str(WB),'--steam',str(self.steam),'--compat',str(self.compat),'--profile',str(self.profile),'--project','core','--rounds','1','--timeout','60']
|
|
def tearDown(self):
|
|
for stream in self.open_locks:stream.close()
|
|
self.tmp.cleanup()
|
|
def test_busy_prefix_refused_under_lock_before_any_launch_or_audio(self):
|
|
events=[]
|
|
def busy(*a):events.append('idle');raise ValueError('existing private Workbench')
|
|
with patch.object(Path,'open',self.tracked_open),patch.object(sys,'argv',self.argv),patch('fcntl.flock',side_effect=lambda *a:events.append('lock')),\
|
|
patch.object(game,'assert_prefix_idle',side_effect=busy),patch.object(game,'cleanup_owned_run') as cleanup,\
|
|
patch.object(game.subprocess,'Popen') as launch,patch.object(game.subprocess,'check_output') as audio,\
|
|
contextlib.redirect_stderr(io.StringIO()):
|
|
with self.assertRaises(SystemExit) as ended:runpy.run_path(str(WB),run_name='__main__')
|
|
self.assertEqual(ended.exception.code,2);self.assertEqual(events,['lock','idle'])
|
|
launch.assert_not_called();audio.assert_not_called();cleanup.assert_not_called();self.assertFalse(self.profile.exists())
|
|
def execute(self,cleanup_report,native_failure=False,launch_error=False):
|
|
events=[];session=self.profile/'logs/logs_fixture';child=Mock();child.pid=12345;child.poll.return_value=None
|
|
def spawn(*args,**kwargs):
|
|
events.append('launch')
|
|
if launch_error:raise OSError('synthetic launch failure')
|
|
session.mkdir(parents=True);(session/'console.log').write_text('ERR SMOKE FAIL: fixture failure\n' if native_failure else 'ERR SMOKE COMPLETE:\n')
|
|
return child
|
|
def audio(args,**kwargs):
|
|
if 'sink-inputs' in args:return '[]'
|
|
return json.dumps([{'name':'erri_tests_silent','index':7}])
|
|
def persist(path,snapshot):events.append('persist');(path/'launch-source.json').write_text('retained launch fixture')
|
|
old_iterdir=Path.iterdir
|
|
def iterdir(path):return iter(()) if path==Path('/proc') else old_iterdir(path)
|
|
clock=[0]
|
|
def now():clock[0]+=3;return clock[0]
|
|
with contextlib.ExitStack() as stack:
|
|
stack.enter_context(patch.object(Path,'open',self.tracked_open))
|
|
stack.enter_context(patch.object(sys,'argv',self.argv))
|
|
stack.enter_context(patch('fcntl.flock',side_effect=lambda *a:events.append('lock')))
|
|
stack.enter_context(patch.object(game,'assert_prefix_idle',side_effect=lambda *a:events.append('idle')))
|
|
cleaner=stack.enter_context(patch.object(game,'cleanup_owned_run',side_effect=lambda *a:(events.append('cleanup') or cleanup_report)))
|
|
stack.enter_context(patch.object(game.shutil,'which',return_value='/synthetic/installed'))
|
|
stack.enter_context(patch.object(game.subprocess,'Popen',side_effect=spawn))
|
|
stack.enter_context(patch.object(game.subprocess,'check_output',side_effect=audio))
|
|
stack.enter_context(patch.object(game.subprocess,'run',return_value=SimpleNamespace(returncode=0)))
|
|
native_report=stack.enter_context(patch.object(game.subprocess,'call',return_value=0))
|
|
stack.enter_context(patch.object(Path,'iterdir',iterdir))
|
|
stack.enter_context(patch.object(game.time,'time',side_effect=now))
|
|
stack.enter_context(patch.object(game.time,'sleep'))
|
|
stack.enter_context(patch.object(native_provenance,'snapshot',return_value={'synthetic':True}))
|
|
stack.enter_context(patch.object(native_provenance,'bindings',return_value={'launch_binding':'retained'}))
|
|
stack.enter_context(patch.object(native_provenance,'persist',side_effect=persist))
|
|
seal=stack.enter_context(patch.object(native_provenance,'seal_run',return_value={'source_unchanged_after_run':True}))
|
|
stdout=stack.enter_context(contextlib.redirect_stdout(io.StringIO()))
|
|
with self.assertRaises(SystemExit) as ended:runpy.run_path(str(WB),run_name='__main__')
|
|
summary=json.loads(stdout.getvalue())
|
|
return ended.exception.code,summary,events,cleaner.call_count,seal.call_count,native_report.call_count
|
|
def test_workbench_success_uses_cleanup_before_sealing_and_retains_report(self):
|
|
proof={'cleanup_confirmed':True,'idle_verified':True,'errors':[],'orphan_cleanup':{'attempted':False}}
|
|
code,summary,events,cleaned,sealed,native=self.execute(proof)
|
|
self.assertEqual(code,0);self.assertTrue(summary['cleanup_confirmed']);self.assertEqual(summary['cleanup_report'],proof)
|
|
self.assertEqual(events[:3],['lock','idle','launch']);self.assertEqual(events[-1],'cleanup')
|
|
self.assertEqual((cleaned,sealed,native),(1,1,1))
|
|
self.assertTrue((self.profile/'logs/logs_fixture/launch-source.json').is_file())
|
|
def test_workbench_cleanup_failure_preserves_native_and_launch_evidence_without_seal(self):
|
|
proof={'cleanup_confirmed':False,'idle_verified':False,'errors':['idle verification: still busy'],
|
|
'orphan_cleanup':{'attempted':True,'report':{'cleared':False,'term_sent':[12]}}}
|
|
code,summary,events,cleaned,sealed,native=self.execute(proof)
|
|
self.assertEqual(code,1);self.assertEqual(summary['result'],'cleanup-failed')
|
|
self.assertEqual(summary['result_before_cleanup'],'smoke-completed');self.assertFalse(summary['cleanup_confirmed'])
|
|
self.assertEqual((cleaned,sealed,native),(1,0,0));self.assertEqual(summary['launch_binding'],'retained')
|
|
session=self.profile/'logs/logs_fixture'
|
|
self.assertIn('ERR SMOKE COMPLETE:',(session/'console.log').read_text())
|
|
self.assertTrue((session/'launch-source.json').is_file());self.assertTrue((session/'runner.json').is_file())
|
|
def test_workbench_native_failure_stays_failed_despite_successful_cleanup(self):
|
|
proof={'cleanup_confirmed':True,'idle_verified':True,'errors':[]}
|
|
code,summary,*_=self.execute(proof,native_failure=True)
|
|
self.assertEqual(code,1);self.assertEqual(summary['result'],'test-failed')
|
|
def test_workbench_launch_failure_does_not_invoke_post_run_cleanup(self):
|
|
code,summary,events,cleaned,sealed,native=self.execute({},launch_error=True)
|
|
self.assertEqual(code,1);self.assertEqual(summary['result'],'runner-failed')
|
|
self.assertEqual((cleaned,sealed,native),(0,0,0));self.assertEqual(summary['runner_error'],'synthetic launch failure')
|
|
def test_audio_fallback_accepts_only_trailing_slash_prefix_spelling(self):
|
|
tree=ast.parse(WB.read_text());function=next(node for node in tree.body if isinstance(node,ast.FunctionDef) and node.name=='silence_test_streams')
|
|
namespace={'json':json,'subprocess':game.subprocess,'Path':Path,'compat':self.compat,'silent_index':7,
|
|
'silent_sink':'owned-fixture-sink','muted_streams':set()}
|
|
exec(compile(ast.Module(body=[function],type_ignores=[]),str(WB),'exec'),namespace)
|
|
stream={'index':8,'sink':0,'mute':False,'properties':{'application.process.id':'555'}}
|
|
for value,expected in ((str(self.compat/'pfx')+'/',2),(str(self.compat/'pfx')+'///',2),(str(self.compat/'pfx')+'-foreign/',0)):
|
|
with patch.object(game.subprocess,'check_output',return_value=json.dumps([stream])),\
|
|
patch.object(Path,'read_bytes',return_value=('WINEPREFIX='+value).encode()+b'\0'),\
|
|
patch.object(game.subprocess,'run') as audio,contextlib.redirect_stdout(io.StringIO()):
|
|
namespace['silence_test_streams']()
|
|
self.assertEqual(audio.call_count,expected)
|
|
|
|
|
|
if __name__=='__main__':unittest.main()
|