Files
EscapistsReintroduced/tests/test_native_package.py
T

393 lines
23 KiB
Python

"""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'):
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 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):
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')
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'
with patch.object(n, 'assert_no_game'), patch.object(n, 'assert_prefix_idle'), \
patch.object(n, 'cleanup') 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)
cleanup.assert_called_once()
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') 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_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)
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):
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)
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,
'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)
if __name__ == '__main__':
unittest.main()