jakub
cab998fb64
Merge owned systems into one addon while preserving resource identities. Gate player admission on the replicated prison scene, restore the original framed door, and hold enemy AI until escape. Correct scenario branding and author metadata, include publishing assets, and retain regression and native verification evidence.
182 lines
12 KiB
Python
182 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
|
|
import test_external_addons
|
|
self.external = test_external_addons.ExternalAddonsTests()
|
|
self.external.setUp(); self.addCleanup(self.external.doCleanups)
|
|
self.argv=[str(WB),'--steam',str(self.steam),'--compat',str(self.compat),'--profile',str(self.profile),'--project','core','--rounds','1','--timeout','60','--external-addons',str(self.external.source)]
|
|
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('external_addons.policy',return_value=self.external.value))
|
|
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()
|