#!/usr/bin/env python3 import contextlib import importlib.util import io import re from pathlib import Path import sys import tempfile import unittest from unittest.mock import patch spec = importlib.util.spec_from_file_location('dedicated_native', Path(__file__).parents[1] / 'tools/dedicated_native.py') runner = importlib.util.module_from_spec(spec) spec.loader.exec_module(runner) class DedicatedNativeTests(unittest.TestCase): def probe(self, namespace='isolated', interfaces=None, routes=None, ipv6=''): interfaces = interfaces if interfaces is not None else [(1, 'lo')] routes = routes if routes is not None else 'Iface Destination Gateway\n' def content(path, *args): return io.StringIO(ipv6 if path.endswith('ipv6_route') else routes) with patch('os.readlink', return_value=namespace), patch('socket.if_nameindex', return_value=interfaces), patch('builtins.open', side_effect=content), patch.object(sys, 'argv', ['probe', 'host', '/server']), patch('os.execv') as launch, contextlib.redirect_stdout(io.StringIO()): try: exec(runner.PROBE, {}) except AssertionError: self.assertFalse(launch.called) return False launch.assert_called_once_with('/server', ['/server']) return True def test_isolation_gate_executes_only_private_loopback(self): self.assertTrue(self.probe()) self.assertFalse(self.probe(namespace='host')) self.assertFalse(self.probe(interfaces=[(1, 'lo'), (2, 'eth0')])) self.assertFalse(self.probe(routes='Iface Destination\neth0 00000000\n')) self.assertFalse(self.probe(ipv6='0000 eth0\n')) self.assertTrue(self.probe(ipv6='0000 lo\n')) def test_vanilla_control_has_no_successor_or_custom_systems(self): args=runner.engine_args('vanilla-everon') self.assertNotIn('-addons',args) self.assertNotIn('-addonsDir',args) self.assertNotIn('-worldSystemsConfig',args) self.assertEqual(args[-2:],['-server','Missions/21_GM_Eden.conf']) self.assertEqual(args[0],'/tmp/runtime/ArmaReforgerServer') def test_exact_core_and_chat_header_and_diagnostic_selection(self): self.assertEqual(runner.engine_args('header'), runner.engine_args('header', 'core')) expected = { 'core': ('2528466DE220D91C', '{C1E20667DCE324ED}Missions/ERR_Everon.conf'), 'chat': ('2528466DE220D91C', '{C1E20667DCE324ED}Missions/ERR_Everon.conf'), } root = Path(__file__).parents[1] for project, (guid, header) in expected.items(): # Keep the CLI identities tied to the actual authored resources. gproj = (root/'addons/core'/'addon.gproj').read_text() self.assertIn('GUID "'+guid+'"', gproj) resource = header.split('}', 1)[1] self.assertIn('Name "'+header+'"', (root/'addons/core'/(resource+'.meta')).read_text()) for route in ('header', 'diag-world'): args = runner.engine_args(route, project) self.assertEqual(args[args.index('-addonsDir')+1], '/tmp/source/addons') self.assertEqual(args[args.index('-addons')+1], guid) self.assertEqual(args.count('-addons'), 1) self.assertEqual(args[args.index('-profile')+1], '/tmp/work/profile') self.assertEqual(args[args.index('-bindIP')+1], '127.0.0.1') self.assertIn('-noBackend', args) self.assertIn('-noSound', args) if route == 'header': self.assertEqual(args[0], '/tmp/runtime/ArmaReforgerServer') self.assertEqual(args[args.index('-server')+1], header) self.assertNotIn('-worldSystemsConfig', args) else: self.assertEqual(args[0], '/tmp/runtime/ArmaReforgerServerDiag') self.assertEqual(args[args.index('-server')+1], '{B6C516F86302C679}Worlds/ERR_Everon.ent') self.assertEqual(args[args.index('-worldSystemsConfig')+1], '{C50579A2EC48E8EF}Configs/Systems/MissionSystems.conf') definition = (root/'addons/core/addon.gproj').read_text() dependencies = set(re.findall(r'"([A-F0-9]{16})"', re.search(r'Dependencies\s*\{([^}]+)', definition)[1])) self.assertEqual(dependencies, {'58D0FB3206B6F859', '595F2BF2F44836FB', '629B2BA37EFFD577', '5E524E4FEECCA92B', '65930CB4CD0237B2'}) def test_invalid_project_route_combinations_fail_before_files_or_processes(self): for route, project in [('vanilla-everon', 'chat'), ('unknown', 'core'), ('header', 'unknown')]: with self.subTest(route=route, project=project), self.assertRaises(ValueError): runner.engine_args(route, project) argv = ['dedicated_native.py', '--source', '/tmp/not-read-source', '--runtime', '/tmp/not-read-runtime', '--archive', '/tmp/not-created', '--route', 'vanilla-everon', '--project', 'chat'] stderr = io.StringIO() with patch.object(sys, 'argv', argv), patch.object(Path, 'resolve') as resolve, patch.object(Path, 'mkdir') as mkdir, patch.object(runner.subprocess, 'run') as run, patch.object(runner.subprocess, 'Popen') as launch, contextlib.redirect_stderr(stderr): with self.assertRaises(SystemExit) as error: runner.main() self.assertEqual(error.exception.code, 2) resolve.assert_not_called() mkdir.assert_not_called() run.assert_not_called() launch.assert_not_called() self.assertIn('vanilla-everon cannot be combined with --project chat', stderr.getvalue()) def addon_input(self, root, names=('core',)): repository = Path(__file__).parents[1] for addon in names: folder = root/'addons'/addon folder.mkdir(parents=True) (folder/'addon.gproj').write_bytes((repository/'addons'/addon/'addon.gproj').read_bytes()) # Synthetic presence-only files, never supplied to a real engine. (folder/'resourceDatabase.rdb').write_bytes(b'nonempty-presence-test') (folder/'data.pak').write_bytes(b'nonempty-presence-test') def test_preflight_accepts_packed_input_without_loose_scripts(self): with tempfile.TemporaryDirectory() as directory: source = Path(directory) self.addon_input(source) for route in ('header', 'diag-world'): for project in ('core', 'chat'): runner.preflight_addons(source, route, project) self.assertFalse(list(source.rglob('*.c'))) # Core-only means no companion RDB is required. with tempfile.TemporaryDirectory() as directory: source = Path(directory) self.addon_input(source, ('core',)) runner.preflight_addons(source, 'header') runner.preflight_addons(source, 'header', 'chat') runner.preflight_addons(Path('/not-a-real-addon-input'), 'vanilla-everon') def test_missing_core_dependency_and_empty_database_refuse_before_archive_or_process(self): for project, relative, empty in [ ('core', 'core/resourceDatabase.rdb', False), ('chat', 'core/addon.gproj', False), ('chat', 'core/resourceDatabase.rdb', False), ('chat', 'core/resourceDatabase.rdb', True), ]: with self.subTest(project=project, relative=relative, empty=empty), tempfile.TemporaryDirectory() as directory: root = Path(directory); source=root/'source'; archive=root/'archive' self.addon_input(source) target = source/'addons'/relative if empty: target.write_bytes(b'') else: target.unlink() argv = ['dedicated_native.py', '--source', str(source), '--archive', str(archive), '--runtime', str(root/'runtime'), '--project', project] stderr = io.StringIO() with patch.object(sys, 'argv', argv), patch.object(runner, 'manifest') as manifest, patch.object(runner.subprocess, 'run') as preflight, patch.object(runner.subprocess, 'Popen') as launch, contextlib.redirect_stderr(stderr): with self.assertRaises(SystemExit) as error: runner.main() self.assertEqual(error.exception.code, 2) manifest.assert_not_called(); preflight.assert_not_called(); launch.assert_not_called() self.assertFalse(archive.exists()) self.assertIn(relative, stderr.getvalue()) def test_preflight_refuses_linked_directory_linked_database_and_wrong_identity(self): with tempfile.TemporaryDirectory() as directory: source = Path(directory)/'source'; self.addon_input(source) database = source/'addons/core/resourceDatabase.rdb' database.unlink(); database.symlink_to(source/'addons/core/data.pak') with self.assertRaisesRegex(ValueError, 'regular file'): runner.preflight_addons(source, 'header') database.unlink(); database.mkdir() with self.assertRaisesRegex(ValueError, 'regular file'): runner.preflight_addons(source, 'header') database.rmdir(); database.write_bytes(b'presence-only') project = source/'addons/core/addon.gproj' project.write_text(project.read_text().replace('2528466DE220D91C', '1111111111111111')) with self.assertRaisesRegex(ValueError, 'identity mismatch'): runner.preflight_addons(source, 'header') with tempfile.TemporaryDirectory() as directory: root = Path(directory); self.addon_input(root/'real') source=root/'source'; source.mkdir(); (source/'addons').symlink_to(root/'real/addons') with self.assertRaisesRegex(ValueError, 'regular addons directory'): runner.preflight_addons(source, 'header') def test_manifest_detects_mutation_and_refuses_links(self): with tempfile.TemporaryDirectory() as directory: root=Path(directory); target=root/'a.c'; target.write_text('first') old=runner.manifest(root); target.write_text('second') self.assertNotEqual(old,runner.manifest(root)) (root/'linked').symlink_to(target) with self.assertRaises(ValueError): runner.manifest(root) def test_routes_do_not_mix_public_config_or_client_flags(self): for route in ['header','diag-world','vanilla-everon']: args=runner.engine_args(route) self.assertNotIn('-config',args) self.assertNotIn('-client',args) self.assertIn('-noBackend',args) self.assertNotIn('-rpl-validation-rdb-disable',args) normal=runner.engine_args('header') self.assertEqual(normal[0],'/tmp/runtime/ArmaReforgerServer') self.assertTrue(normal[normal.index('-server')+1].endswith('ERR_Everon.conf')) self.assertNotIn('-worldSystemsConfig',normal) self.assertIn('-worldSystemsConfig',runner.engine_args('diag-world')) if __name__ == '__main__': unittest.main()