"""Offline packaging orchestration tests. Native child/audio are mocked, never launched.""" import argparse import hashlib import json from pathlib import Path import shutil import subprocess import sys import tempfile import unittest from unittest.mock import Mock, patch ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT/'tools')) import native_package as n class NativePackageTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory(prefix='erri-package-test-') self.addCleanup(self.temp.cleanup) self.home = Path(self.temp.name) self.source = self.home/'original' shutil.copytree(ROOT/'addons', self.source/'addons') (self.source/'docs').mkdir() for name in ('addons.json', 'vanilla-resources.json', 'acceptance.json', 'external-addons.json'): shutil.copy2(ROOT/'docs'/name, self.source/'docs'/name) self.session = self.home/'session' self.compat = self.home/'compat' (self.compat/'pfx').mkdir(parents=True) (self.compat/'pfx/system.reg').write_text('test only') self.steam = self.home/'steam' command = n.command_for(self.steam, self.session, 'core') i = command.index('--') for path in (Path(command[i+1]), Path(command[i+3]), self.steam/'steamapps/common/Arma Reforger/addons/data/ArmaReforger.gproj'): path.parent.mkdir(parents=True, exist_ok=True) path.write_text('synthetic installed-file double; never executed') def external_fixture(self): # Synthetic exact closure for orchestration doubles; never an installed addon or engine. import test_external_addons fixture = test_external_addons.ExternalAddonsTests() fixture.setUp() self.addCleanup(fixture.doCleanups) (self.source/'docs/external-addons.json').write_text(json.dumps(fixture.value)) return fixture.source def test_closure_order(self): self.assertEqual(n.project_order('core'), ['core']) self.assertEqual(n.project_order('ai'), ['core', 'ai']) self.assertEqual(n.project_order('chat'), ['core', 'ai', 'intel', 'map', 'chat']) with self.assertRaises(ValueError): n.project_order('original') def test_supported_command_is_private_and_never_publishes_or_plays(self): command = n.command_for(self.steam, self.session, 'core') self.assertIn('-packAddon', command) self.assertIn('-wbModule=ResourceManager', command) self.assertIn('-noBackend', command) self.assertIn('-disableCrashReporter', command) self.assertEqual(command[command.index('--backend')+1], 'headless') for flag in ('-gproj', '-addonsDir', '-profile', '-packAddonDir', '-packAddonCacheDir'): self.assertTrue(command[command.index(flag)+1].startswith('Z:'+str(self.session))) for flag in ('-publishAddon', '-wbSilent', '-plugin', '-errPlay', '-load', '-exitAfter'): self.assertNotIn(flag, command) def test_copy_validated_payload_and_detect_changed_authored_bits(self): before = n.inventory(self.source) (self.source/'addons/core/resourceDatabase.rdb').write_text('generated') authored = n.copy_payload(self.source, self.session) self.assertEqual(authored, before) self.assertFalse((self.session/'source/addons/core/resourceDatabase.rdb').exists()) self.assertFalse((self.session/'source/docs').exists()) self.assertTrue(n.copied_authored_unchanged(self.session/'source', authored)) # Generated native cache changes do not excuse authored changes. (self.session/'source/addons/core/resourceDatabase.rdb').write_text('new native cache') self.assertTrue(n.copied_authored_unchanged(self.session/'source', authored)) name = next(iter(authored)) (self.session/'source'/name).write_text('changed') self.assertFalse(n.copied_authored_unchanged(self.session/'source', authored)) self.assertEqual(n.inventory(self.source), before) def test_invalid_source_refused_before_session_creation(self): path = self.source/'addons/core/addon.gproj' path.write_text(path.read_text().replace('ERR_Core', 'Original')) with self.assertRaises(ValueError): n.copy_payload(self.source, self.session) self.assertFalse(self.session.exists()) def test_session_never_reused_or_outside_tmp(self): fresh = Path('/tmp')/('erri-fresh-'+self.home.name) n.validate_session(self.source, self.steam, self.compat, fresh) self.session.mkdir() with self.assertRaises(ValueError): n.validate_session(self.source, self.steam, self.compat, self.session) with self.assertRaises(ValueError): n.validate_session(self.source, self.steam, self.compat, Path('/home/not-allowed')) with self.assertRaises(FileExistsError): n.copy_payload(self.source, self.session) link = self.home/'link' link.symlink_to(self.session, target_is_directory=True) with self.assertRaises(ValueError): n.validate_session(self.source, self.steam, self.compat, link) def test_raw_native_failures_never_become_success(self): path = self.home/'console.log' for content in ['SCRIPT (E): compile failed', 'ENGINE (F): crash', 'Virtual Machine Exception', 'Packing failed', 'Can\'t compile "Game"']: path.write_text(content) logs, errors = n.log_evidence([path]) self.assertTrue(errors, content) self.assertFalse(n.candidate_ok(0, 'process-exited', True, True, logs, errors, {'data.pak': {'bytes': 50}})) path = self.home/'error.log' path.write_text('unexpected failure without a category prefix') self.assertTrue(n.log_evidence([path])[1]) def test_success_requires_clean_exit_logs_cleanup_hashes_and_real_nonempty_file(self): folder = self.home/'packages'; folder.mkdir() self.assertFalse(n.candidate_ok(0, 'process-exited', True, True, ['log'], [], n.package_inventory(folder))) (folder/'data.pak').write_bytes(b'') self.assertFalse(n.candidate_ok(0, 'process-exited', True, True, ['log'], [], n.package_inventory(folder))) (folder/'data.pak').write_bytes(b'SYNTHETIC TEST PACKAGE BYTES, NOT NATIVE') packages = n.package_inventory(folder) self.assertTrue(n.candidate_ok(0, 'process-exited', True, True, ['log'], [], packages)) for args in [(1,'process-exited',True,True,['log']), (0,'timeout',True,True,['log']), (0,'process-exited',False,True,['log']), (0,'process-exited',True,False,['log']), (0,'process-exited',True,True,[])]: self.assertFalse(n.candidate_ok(*args, [], packages)) self.assertEqual(packages['data.pak']['sha256'], hashlib.sha256((folder/'data.pak').read_bytes()).hexdigest()) (folder/'external.pak').symlink_to(self.source/'addons/core/addon.gproj') with self.assertRaises(ValueError): n.package_inventory(folder) def native_double(self, authored, timeout=30, mutate=False, failure=None, native_code=0, write_status=True, launcher_code=0, silent_workbench=False, exit_after_init=False, cleanup_proof=None, prefix_remains_active=False): if cleanup_proof is None: cleanup_proof = {'cleanup_confirmed': True, 'idle_verified': True, 'errors': [], 'orphan_cleanup': {'attempted': False}} child = Mock(); child.poll.return_value = launcher_code def spawn(*args, **kwargs): self.assertNotIn('WINEPREFIX', kwargs['env']) self.assertNotIn('STEAM_COMPAT_DATA_PATH', kwargs['env']) # Explicitly fake outputs: never evidence for a real native build. folder = self.session/'profiles/core/logs/logs_test'; folder.mkdir(parents=True) (folder/'console.log').write_text('SYNTHETIC NATIVE PROCESS DOUBLE\n'+(failure or '')) (self.session/'packages/core/data.pak').write_bytes(b'SYNTHETIC TEST ONLY') if write_status: native_command = n.command_for(self.steam, self.session, 'core', silent_workbench, exit_after_init) native_command = native_command[native_command.index('--')+1:] (self.session/'evidence/core/native-exit.json').write_text(json.dumps({'command': native_command, 'exit_code': native_code})) if mutate: (self.session/'source'/next(iter(authored))).write_text('changed authored byte') return child def check_output(args, **kwargs): return '[]' if 'sink-inputs' in args else '987' probes = 0 def idle(*args): nonlocal probes probes += 1 if prefix_remains_active and probes > 2: raise ValueError('synthetic adopted helper remains after native exit') with patch.object(n, 'assert_no_game'), patch.object(n, 'assert_prefix_idle', side_effect=idle), \ patch.object(n, 'cleanup_owned_run', return_value=cleanup_proof) as cleanup, patch.object(n.subprocess, 'Popen', side_effect=spawn), \ patch.object(n.subprocess, 'check_output', side_effect=check_output), patch.object(n.subprocess, 'run') as process: result = n.run_project(self.steam, self.compat, self.session, 'core', timeout, authored, silent_workbench, exit_after_init) cleanup.assert_called_once_with(child, Path(n.command_for(self.steam, self.session, 'core')[n.command_for(self.steam, self.session, 'core').index('--')+1]), unittest.mock.ANY, self.compat) self.assertTrue(any(call.args[0] == ['pactl', 'unload-module', '987'] for call in process.call_args_list)) return result def test_mock_process_candidate_keeps_all_qualification_pending(self): authored = n.copy_payload(self.source, self.session) report = self.native_double(authored) self.assertTrue(report['candidate_created']) for key in ('release_qualified', 'package_contents_verified', 'native_consumption_verified', 'tool_completion_semantics_verified'): self.assertFalse(report[key]) self.assertIn('sha256', next(iter(report['logs'].values()))) self.assertTrue((self.session/'evidence/core/launch.json').is_file()) def test_mock_process_authored_mutation_fails_after_clean_exit(self): authored = n.copy_payload(self.source, self.session) report = self.native_double(authored, mutate=True) self.assertFalse(report['candidate_created']) self.assertFalse(report['copied_authored_unchanged']) def test_mock_process_timeout_cleans_only_owned_child_and_cannot_qualify(self): authored = n.copy_payload(self.source, self.session) report = self.native_double(authored, timeout=0) self.assertEqual(report['result'], 'timeout') self.assertFalse(report['candidate_created']) self.assertIsNone(report['launcher_exit_code']) def test_mock_process_native_error_with_output_still_fails(self): authored = n.copy_payload(self.source, self.session) report = self.native_double(authored, failure='ENGINE (E): synthetic failure') self.assertFalse(report['candidate_created']) self.assertTrue(report['errors']) def test_orphaned_prefix_after_launcher_exit_cannot_qualify(self): import itertools authored = n.copy_payload(self.source, self.session) child = Mock(); child.poll.return_value = 0 probes = 0 def idle(*args): nonlocal probes probes += 1 if probes > 2: raise ValueError('synthetic owned prefix child remains') with patch.object(n, 'assert_no_game'), patch.object(n, 'assert_prefix_idle', side_effect=idle), \ patch.object(n, 'cleanup_owned_run', return_value={'cleanup_confirmed': False, 'idle_verified': False, 'errors': ['synthetic owned prefix child remains']}) as cleanup, patch.object(n.subprocess, 'Popen', return_value=child), \ patch.object(n.subprocess, 'check_output', side_effect=lambda args, **kw: '[]' if 'sink-inputs' in args else '987'), \ patch.object(n.subprocess, 'run'), patch.object(n.time, 'monotonic', side_effect=itertools.count(0, 6)), patch.object(n.time, 'sleep'): report = n.run_project(self.steam, self.compat, self.session, 'core', 100, authored) cleanup.assert_called_once() self.assertEqual(report['result'], 'native-or-launcher-exited-prefix-active') self.assertFalse(report['candidate_created']) def test_audio_changes_only_fresh_application_or_owned_prefix(self): streams = [{'index': 1, 'properties': {'application.id': 'own'}}, {'index': 2, 'properties': {'application.id': 'user-music'}}] with patch.object(n.subprocess, 'check_output', return_value=json.dumps(streams)), patch.object(n.subprocess, 'run') as run: n.quiet_streams('own', 'private-sink', self.compat) self.assertEqual([call.args[0] for call in run.call_args_list], [['pactl', 'move-sink-input', '1', 'private-sink'], ['pactl', 'set-sink-input-mute', '1', '1']]) def test_audio_disappearing_move_skips_mute_and_continues_owned_streams(self): streams = [{'index': 1, 'properties': {'application.id': 'own'}}, {'index': 2, 'properties': {'application.id': 'user-music'}}, {'index': 3, 'properties': {'application.id': 'own'}}] error = subprocess.CalledProcessError(1, ['pactl', 'move-sink-input', '1', 'private-sink']) # A newly observed stream is not authority to modify it in this pass. fresh = streams[1:] + [{'index': 4, 'properties': {'application.id': 'user-music'}}] with patch.object(n.subprocess, 'check_output', side_effect=[json.dumps(streams), json.dumps(fresh)]) as listing, patch.object(n.subprocess, 'run', side_effect=[error, None, None]) as run: n.quiet_streams('own', 'private-sink', self.compat) self.assertEqual(listing.call_count, 2) self.assertEqual([call.args[0] for call in run.call_args_list], [ ['pactl', 'move-sink-input', '1', 'private-sink'], ['pactl', 'move-sink-input', '3', 'private-sink'], ['pactl', 'set-sink-input-mute', '3', '1']]) def test_audio_disappearing_mute_is_confirmed_by_fresh_listing(self): stream = [{'index': 1, 'properties': {'application.id': 'own'}}] error = subprocess.CalledProcessError(1, ['pactl', 'set-sink-input-mute', '1', '1']) with patch.object(n.subprocess, 'check_output', side_effect=[json.dumps(stream), '[]']) as listing, patch.object(n.subprocess, 'run', side_effect=[None, error]) as run: n.quiet_streams('own', 'private-sink', self.compat) self.assertEqual(listing.call_count, 2) self.assertEqual(run.call_count, 2) def test_audio_operation_failure_remains_when_same_index_is_present(self): stream = [{'index': 1, 'properties': {'application.id': 'own'}}] # Even changed ownership at the same index cannot excuse the failure or # permit another mute/move against that now unrelated stream. for fresh in (stream, [{'index': 1, 'properties': {'application.id': 'user-music'}}]): for position in (0, 1): with self.subTest(fresh=fresh, operation=position): error = subprocess.CalledProcessError(1, ['pactl', 'failed-operation']) with patch.object(n.subprocess, 'check_output', side_effect=[json.dumps(stream), json.dumps(fresh)]), patch.object(n.subprocess, 'run', side_effect=[None]*position+[error]) as run: with self.assertRaises(subprocess.CalledProcessError) as raised: n.quiet_streams('own', 'private-sink', self.compat) self.assertIs(raised.exception, error) self.assertEqual(run.call_count, position+1) def test_audio_failed_or_malformed_recheck_cannot_prove_absence(self): stream = [{'index': 1, 'properties': {'application.id': 'own'}}] reads = [OSError('pactl unavailable'), subprocess.CalledProcessError(1, ['pactl', 'list']), subprocess.TimeoutExpired(['pactl', 'list'], 5), 'invalid json', '{}', 'null', '[{}]', '[null]', '[{"index":"1"}]', '[{"index":true}]', '[{"index":-1}]'] for fresh in reads: with self.subTest(read=fresh): error = subprocess.CalledProcessError(1, ['pactl', 'move-sink-input']) with patch.object(n.subprocess, 'check_output', side_effect=[json.dumps(stream), fresh]), patch.object(n.subprocess, 'run', side_effect=error) as run: with self.assertRaises(subprocess.CalledProcessError) as raised: n.quiet_streams('own', 'private-sink', self.compat) self.assertIs(raised.exception, error) run.assert_called_once() def test_audio_operation_timeout_is_not_a_disappeared_stream_success(self): stream = [{'index': 1, 'properties': {'application.id': 'own'}}] with patch.object(n.subprocess, 'check_output', return_value=json.dumps(stream)) as listing, patch.object(n.subprocess, 'run', side_effect=subprocess.TimeoutExpired(['pactl', 'move-sink-input'], 5)): with self.assertRaises(subprocess.TimeoutExpired): n.quiet_streams('own', 'private-sink', self.compat) listing.assert_called_once() def test_orchestration_stops_after_first_failed_dependency_without_launch(self): import uuid import contextlib import io fresh = Path('/tmp')/('erri-package-orchestration-'+uuid.uuid4().hex) self.addCleanup(lambda: shutil.rmtree(fresh, ignore_errors=True)) args = argparse.Namespace(steam=self.steam, compat=self.compat, session=fresh, project='chat', timeout=30, external_addons=self.external_fixture()) with patch.object(n, 'assert_no_game'), patch.object(n, 'assert_prefix_idle'), patch.object(n.shutil, 'which', return_value='/mock'), \ patch.object(n, 'run_project', return_value={'candidate_created':False}) as run, patch.object(n.subprocess, 'Popen') as child, contextlib.redirect_stdout(io.StringIO()): self.assertEqual(n.run(args, self.source), 1) run.assert_called_once() self.assertEqual(run.call_args.args[3], 'core') child.assert_not_called() summary = json.loads((fresh/'package-report.json').read_text()) self.assertFalse(summary['all_candidates_created']) self.assertFalse(summary['published']) self.assertTrue(summary['source_repository_unchanged']) def test_same_prefix_lock_refuses_without_creating_session(self): import fcntl import uuid fresh = Path('/tmp')/('erri-package-lock-'+uuid.uuid4().hex) args = argparse.Namespace(steam=self.steam, compat=self.compat, session=fresh, project='core', timeout=30) with (self.compat/'.erri-workbench.lock').open('a') as lock: fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) with patch.object(n, 'assert_no_game'), patch.object(n.shutil, 'which', return_value='/mock'), patch.object(n, 'copy_payload') as copy: with self.assertRaisesRegex(ValueError, 'owns the disposable prefix'): n.run(args, self.source) copy.assert_not_called() self.assertFalse(fresh.exists()) def orchestration_double(self, reports, continuation=True, idle=None, exit_after_init=False): import contextlib import io import uuid fresh = Path('/tmp')/('erri-package-orchestration-'+uuid.uuid4().hex) self.addCleanup(lambda: shutil.rmtree(fresh, ignore_errors=True)) args = argparse.Namespace(steam=self.steam, compat=self.compat, session=fresh, project='chat', timeout=30, continue_on_failure=continuation, exit_after_init=exit_after_init, external_addons=self.external_fixture()) with patch.object(n, 'assert_no_game'), patch.object(n, 'assert_prefix_idle', side_effect=idle), \ patch.object(n.shutil, 'which', return_value='/mock'), \ patch.object(n, 'run_project', side_effect=reports) as run, \ patch.object(n.subprocess, 'Popen') as child, contextlib.redirect_stdout(io.StringIO()): code = n.run(args, self.source) child.assert_not_called() return code, json.loads((fresh/'package-report.json').read_text()), run.call_args_list @staticmethod def completed_report(success=False, **changes): report = {'candidate_created': success, 'cleanup_confirmed': True, 'runner_error': None, 'copied_authored_unchanged': True, 'profile_addons_unchanged': True, 'external_addons_unchanged_after_run': True, 'completion_observed': 'native-child-exited', 'errors': [] if success else ['RESOURCES (E): synthetic shutdown leak']} report.update(changes) return report def test_diagnostic_continuation_runs_full_closure_keeps_failures_and_nonzero(self): reports = [self.completed_report(success=i not in (0, 2)) for i in range(5)] code, summary, calls = self.orchestration_double(reports) self.assertEqual([call.args[3] for call in calls], ['core', 'ai', 'intel', 'map', 'chat']) self.assertEqual(len({str(call.args[2]) for call in calls}), 1) # one private copy/session self.assertEqual(summary['projects'], reports) # raw failures retained self.assertEqual(code, 1) self.assertFalse(summary['all_candidates_created']) self.assertFalse(summary['release_qualified']) self.assertEqual(summary['skipped_projects'], []) self.assertIsNone(summary['stopped_reason']) def test_default_stops_even_after_safely_completed_validation_failure(self): code, summary, calls = self.orchestration_double([self.completed_report()], False) self.assertEqual(code, 1) self.assertEqual(len(calls), 1) self.assertFalse(summary['continue_on_failure']) self.assertEqual(summary['skipped_projects'], ['ai', 'intel', 'map', 'chat']) self.assertIn('continuation not requested', summary['stopped_reason']) def test_continuation_cannot_override_cleanup_runner_source_or_completion_failure(self): for change in [{'cleanup_confirmed': False}, {'runner_error': 'synthetic failure'}, {'copied_authored_unchanged': False}, {'completion_observed': 'timeout'}, {'completion_observed': 'interrupted'}, {'completion_observed': 'native-or-launcher-exited-prefix-active'}]: with self.subTest(change=change): code, summary, calls = self.orchestration_double([self.completed_report(**change)]) self.assertEqual(code, 1) self.assertEqual(len(calls), 1) self.assertTrue(summary['stopped_reason']) self.assertEqual(len(summary['skipped_projects']), 4) def test_prefix_becoming_busy_between_addons_stops_before_next_launch(self): # First probe owns idle prefix before copy; second is between addons. code, summary, calls = self.orchestration_double( [self.completed_report()], idle=[None, ValueError('synthetic prefix still active')]) self.assertEqual(code, 1) self.assertEqual(len(calls), 1) self.assertIn('prefix not verifiably idle', summary['stopped_reason']) def test_all_clean_candidates_still_unqualified_and_flag_defaults_false(self): code, summary, calls = self.orchestration_double([self.completed_report(True) for _ in range(5)]) self.assertEqual(code, 0) self.assertTrue(summary['all_candidates_created']) self.assertFalse(summary['release_qualified']) with patch.object(n, 'run', return_value=1) as run: self.assertEqual(n.main(['--session', '/tmp/not-launched', '--compat', str(self.compat)]), 1) self.assertFalse(run.call_args.args[0].continue_on_failure) n.main(['--session', '/tmp/not-launched', '--compat', str(self.compat), '--continue-on-failure']) self.assertTrue(run.call_args.args[0].continue_on_failure) def test_prefix_symlink_and_traversal_refused(self): fresh = Path('/tmp')/('erri-unused-'+self.home.name) shutil.rmtree(self.compat/'pfx') outside = self.home/'installed-prefix'; outside.mkdir() (outside/'system.reg').write_text('must not mutate') (self.compat/'pfx').symlink_to(outside, target_is_directory=True) with self.assertRaisesRegex(ValueError, 'real pfx'): n.validate_session(self.source, self.steam, self.compat, fresh) (self.compat/'pfx').unlink(); (self.compat/'pfx').mkdir() (self.compat/'pfx/system.reg').write_text('private') (self.compat/'.erri-workbench.lock').symlink_to(outside/'system.reg') with self.assertRaisesRegex(ValueError, 'lock must not'): n.validate_session(self.source, self.steam, self.compat, fresh) (self.compat/'.erri-workbench.lock').unlink() with self.assertRaises(ValueError): n.validate_session(self.source, self.steam, self.compat, self.home/'..'/fresh.name) def test_added_authored_file_is_not_mistaken_for_generated_cache(self): authored = n.copy_payload(self.source, self.session) (self.session/'source/addons/core/Scripts/Game/ERR/Injected.c').write_text('// synthetic new authored content') self.assertFalse(n.copied_authored_unchanged(self.session/'source', authored)) def test_native_failure_not_masked_by_successful_gamescope_exit(self): authored = n.copy_payload(self.source, self.session) report = self.native_double(authored, native_code=7) self.assertEqual(report['launcher_exit_code'], 0) self.assertEqual(report['native_exit_code'], 7) self.assertFalse(report['candidate_created']) def test_missing_direct_native_exit_cannot_qualify(self): authored = n.copy_payload(self.source, self.session) report = self.native_double(authored, write_status=False) self.assertIsNone(report['native_exit_code']) self.assertFalse(report['candidate_created']) def test_private_recorder_captures_actual_subprocess_status_without_shell(self): script = self.home/'recorder.py'; script.write_text(n.CHILD_RECORDER) output = self.home/'exit.json' # Ordinary Python test child only. No Proton, Wine, game or engine. command = [sys.executable, '-c', 'raise SystemExit(7)'] environment = self.home/'native-env.json'; environment.write_text('{}') completed = subprocess.run([sys.executable, str(script), str(output), str(environment), *command]) self.assertEqual(completed.returncode, 7) self.assertEqual(n.child_status(output, command), 7) self.assertIsNone(n.child_status(output, ['different-command'])) output.write_text('[]'); self.assertIsNone(n.child_status(output, command)) output.write_text('{bad'); self.assertIsNone(n.child_status(output, command)) def test_slow_observation_cannot_turn_expired_attempt_into_success(self): authored = n.copy_payload(self.source, self.session) clock = [0] def slow_audio(*args): clock[0] = 31 with patch.object(n.time, 'monotonic', side_effect=lambda: clock[0]), patch.object(n, 'quiet_streams', side_effect=slow_audio): report = self.native_double(authored, timeout=30) self.assertEqual(report['result'], 'timeout') self.assertFalse(report['candidate_created']) def test_naturally_exited_native_child_can_close_lingering_owned_display(self): authored = n.copy_payload(self.source, self.session) report = self.native_double(authored, launcher_code=None) self.assertEqual(report['native_exit_code'], 0) self.assertIsNone(report['launcher_exit_code']) self.assertTrue(report['display_cleanup_after_native_exit']) self.assertEqual(report['completion_observed'], 'native-child-exited') self.assertTrue(report['candidate_created']) self.assertFalse(report['release_qualified']) env = json.loads((self.session/'evidence/core/native-prefix-environment.json').read_text()) self.assertEqual(env['WINEPREFIX'], str(self.compat/'pfx')) def test_native_child_completion_does_not_hide_shutdown_errors(self): authored = n.copy_payload(self.source, self.session) report = self.native_double(authored, launcher_code=None, failure='RESOURCES (E): synthetic Resource leaks') self.assertTrue(report['display_cleanup_after_native_exit']) self.assertFalse(report['candidate_created']) self.assertTrue(report['errors']) def test_wb_silent_is_explicit_and_preserves_packaging_and_error_policy(self): ordinary = n.command_for(self.steam, self.session, 'core') silent = n.command_for(self.steam, self.session, 'core', True) self.assertEqual(silent, ordinary+['-wbSilent']) self.assertIn('-packAddon', silent) self.assertIn('-noBackend', silent) def test_exit_after_init_changes_only_documented_lifecycle_flag(self): ordinary = n.command_for(self.steam, self.session, 'core') experiment = n.command_for(self.steam, self.session, 'core', exit_after_init=True) self.assertEqual(experiment, ordinary + ['-exitAfterInit']) self.assertNotIn('-wbSilent', experiment) self.assertNotIn('-run', experiment) with self.assertRaisesRegex(ValueError, 'only one'): n.command_for(self.steam, self.session, 'core', True, True) with patch.object(n, 'run', return_value=1) as run: base = ['--session', '/tmp/not-launched', '--compat', str(self.compat)] n.main(base) self.assertFalse(run.call_args.args[0].silent_workbench) self.assertFalse(run.call_args.args[0].exit_after_init) n.main(base + ['--silent-workbench']) self.assertTrue(run.call_args.args[0].silent_workbench) self.assertFalse(run.call_args.args[0].exit_after_init) n.main(base + ['--exit-after-init']) self.assertTrue(run.call_args.args[0].exit_after_init) self.assertFalse(run.call_args.args[0].silent_workbench) run.reset_mock() with self.assertRaises(SystemExit) as refused: n.main(base + ['--silent-workbench', '--exit-after-init']) self.assertEqual(refused.exception.code, 2) run.assert_not_called() def test_exit_after_init_forwards_across_dependency_closure(self): code, summary, calls = self.orchestration_double( [self.completed_report(True) for _ in range(5)], exit_after_init=True) self.assertEqual(code, 0) self.assertEqual(len(calls), 5) self.assertTrue(all(call.args[-2:] == (False, True) for call in calls)) self.assertFalse(summary['release_qualified']) def test_exit_after_init_records_exact_command_and_retains_shutdown_errors(self): authored = n.copy_payload(self.source, self.session) report = self.native_double(authored, launcher_code=None, exit_after_init=True, failure='RESOURCES (E): synthetic shutdown UI leak') launch = json.loads((self.session/'evidence/core/launch.json').read_text()) self.assertEqual(launch['native_command'][-1], '-exitAfterInit') self.assertTrue(launch['exit_after_init_experiment']) self.assertTrue(report['exit_after_init_experiment']) self.assertFalse(report['silent_workbench_experiment']) status = json.loads((self.session/'evidence/core/native-exit.json').read_text()) self.assertEqual(status['command'], launch['native_command']) self.assertEqual(report['native_exit_code'], 0) self.assertEqual(report['completion_observed'], 'native-child-exited') self.assertFalse(report['candidate_created']) self.assertTrue(report['errors']) self.assertFalse(report['release_qualified']) console = self.session/'profiles/core/logs/logs_test/console.log' self.assertIn('synthetic shutdown UI leak', console.read_text()) self.assertEqual(report['logs'][str(console)]['sha256'], n.digest(console)) def test_exit_after_init_without_recorded_native_exit_still_refuses(self): authored = n.copy_payload(self.source, self.session) report = self.native_double(authored, exit_after_init=True, write_status=False) self.assertFalse(report['candidate_created']) self.assertIsNone(report['native_exit_code']) def test_exit_after_init_timeout_keeps_error_and_cleanup_rules(self): authored = n.copy_payload(self.source, self.session) report = self.native_double(authored, exit_after_init=True, timeout=0) self.assertEqual(report['completion_observed'], 'timeout') self.assertFalse(report['candidate_created']) self.assertTrue(report['cleanup_confirmed']) def test_shared_cleanup_report_preserves_partial_refusal_and_native_errors(self): authored = n.copy_payload(self.source, self.session) proof = {'cleanup_confirmed': False, 'idle_verified': False, 'errors': ['private helper cleanup: ambiguous live process'], 'orphan_cleanup': {'attempted': True, 'report': {'cleared': False, 'term_sent': [123]}}} report = self.native_double(authored, launcher_code=None, failure='RESOURCES (E): retained UI leak', cleanup_proof=proof) self.assertEqual(report['cleanup_report'], proof) self.assertFalse(report['cleanup_confirmed']) self.assertFalse(report['candidate_created']) self.assertIn('ambiguous live process', report['runner_error']) self.assertTrue(any('retained UI leak' in error for error in report['errors'])) self.assertEqual(report['native_exit_code'], 0) self.assertEqual(report['completion_observed'], 'native-child-exited') def test_successful_adopted_cleanup_cannot_upgrade_timeout(self): authored = n.copy_payload(self.source, self.session) proof = {'cleanup_confirmed': True, 'idle_verified': True, 'errors': [], 'orphan_cleanup': {'attempted': True, 'report': {'cleared': True, 'term_sent': [123], 'kill_sent': []}}} report = self.native_double(authored, timeout=0, cleanup_proof=proof) self.assertEqual(report['cleanup_report'], proof) self.assertTrue(report['cleanup_confirmed']) self.assertEqual(report['completion_observed'], 'timeout') self.assertFalse(report['candidate_created']) def test_post_run_cleanup_does_not_reclassify_prefix_active_native_exit(self): import itertools authored = n.copy_payload(self.source, self.session) proof = {'cleanup_confirmed': True, 'idle_verified': True, 'errors': [], 'orphan_cleanup': {'attempted': True, 'report': {'cleared': True, 'term_sent': [123]}}} with patch.object(n.time, 'monotonic', side_effect=itertools.count(0, 2)), patch.object(n.time, 'sleep'): report = self.native_double(authored, timeout=100, launcher_code=None, cleanup_proof=proof, prefix_remains_active=True, failure='RESOURCES (E): synthetic shutdown leak') self.assertEqual(report['native_exit_code'], 0) self.assertEqual(report['completion_observed'], 'native-or-launcher-exited-prefix-active') self.assertTrue(report['cleanup_confirmed']) self.assertFalse(report['candidate_created']) self.assertTrue(report['errors']) self.assertEqual(report['cleanup_report'], proof) def test_packaging_uses_exact_shared_cleanup_adapter_and_scoping(self): import headless_game as game import private_prefix_cleanup as private self.assertIs(n.cleanup_owned_run, game.cleanup_owned_run) for compat, expected in ((private.COMPAT, ['normal', 'adopted', 'idle']), (self.compat, ['normal', 'idle'])): events = [] with patch.object(game, 'cleanup', side_effect=lambda *args: events.append('normal')), \ patch.object(private, 'cleanup_private_prefix', side_effect=lambda *args: events.append('adopted') or {'cleared': True}), \ patch.object(game, 'assert_prefix_idle', side_effect=lambda *args: events.append('idle')): report = n.cleanup_owned_run(Mock(), private.INSTALLED_PROTON, {}, compat) self.assertEqual(events, expected) self.assertTrue(report['cleanup_confirmed']) self.assertEqual(report['orphan_cleanup']['attempted'], compat == private.COMPAT) def test_failed_native_launch_does_not_call_post_run_cleanup(self): authored = n.copy_payload(self.source, self.session) with patch.object(n, 'assert_no_game'), patch.object(n, 'assert_prefix_idle'), \ patch.object(n.subprocess, 'check_output', return_value='987'), \ patch.object(n.subprocess, 'run'), \ patch.object(n.subprocess, 'Popen', side_effect=OSError('synthetic launch refusal')), \ patch.object(n, 'cleanup_owned_run') as cleanup: report = n.run_project(self.steam, self.compat, self.session, 'core', 30, authored) cleanup.assert_not_called() self.assertIsNone(report['cleanup_report']) self.assertEqual(report['completion_observed'], 'runner-failed') self.assertFalse(report['candidate_created']) if __name__ == '__main__': unittest.main()