import importlib.util import io import json from pathlib import Path import sys import tarfile import tempfile import unittest from unittest.mock import patch ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(ROOT / 'tools/deploy')) import deploy import receiver def site(): return dict(schema=1, operator_token='operator_token_12345678', controller_url='http://192.168.77.1:8766', cabinets=[dict(id='cabinet_01', host='192.168.77.11', user='arcade', port=22, token='cabinet1_token_12345678', gpio=False, pins=[], gpio_controller='cabinet_01_gpio')]) def archive(extra=None, missing=False, wrong_hash=False): import hashlib files = {name: b'example' for name in receiver.FILES} manifest = dict(release='test-1', files={name: hashlib.sha256(data).hexdigest() for name, data in files.items()}) if wrong_hash: manifest['files']['ArcadeEscape.pck'] = '0' * 64 stream = io.BytesIO() with tarfile.open(fileobj=stream, mode='w:gz') as tar: entries = [('manifest.json', json.dumps(manifest).encode())] + list(files.items()) if missing: entries.pop() for name, data in entries: item = tarfile.TarInfo(name) item.size = len(data) tar.addfile(item, io.BytesIO(data)) if extra: tar.addfile(extra) stream.seek(0) return stream class DeployTests(unittest.TestCase): def test_config_separates_credentials_and_keeps_gpio_off(self): with tempfile.TemporaryDirectory() as temp: folder = Path(temp) settings = site() deploy.generate(folder, settings) cabinet = (folder / 'cabinets/cabinet_01.ini').read_text() self.assertNotIn(settings['operator_token'], cabinet) self.assertIn('gpio = false', cabinet) self.assertIn('~/.local/state/arcade_gpio/requests.sqlite3', cabinet) self.assertIn(settings['operator_token'], (folder / 'controller.ini').read_text()) def test_duplicate_identity_host_and_tokens_rejected(self): for change in ('id', 'host', 'token'): settings = site() other = dict(settings['cabinets'][0], id='cabinet_02', host='192.168.77.12', token='cabinet2_token_12345678') other[change] = settings['cabinets'][0][change] settings['cabinets'].append(other) with self.subTest(change=change), self.assertRaises(ValueError): deploy.validate(settings) def test_unknown_profile_and_command_injection_rejected(self): for key, value in [('id', 'unknown'), ('host', 'pi; reboot'), ('user', '-root'), ('port', True)]: settings = site() settings['cabinets'][0][key] = value with self.subTest(key=key), self.assertRaises(ValueError): deploy.validate(settings) def test_bad_url_and_gpio_allowlists_rejected(self): for url in ('http://127.0.0.1:8766', 'http://user:secret@192.168.77.1:8766', 'http://192.168.77.1:8766/path'): settings = site() settings['controller_url'] = url with self.assertRaises(ValueError): deploy.validate(settings) for pins in ([], [17, 17], [28], [True]): with self.assertRaises(ValueError): receiver.render_config(dict(site()['cabinets'][0], url=site()['controller_url'], gpio=True, pins=pins)) def test_archive_success_and_retry(self): with tempfile.TemporaryDirectory() as temp: root = Path(temp) receiver.stage(root, archive()) receiver.stage(root, archive()) self.assertEqual(receiver.verify_release(root / 'releases/test-1')['release'], 'test-1') self.assertFalse((root / 'active.json').exists()) def test_archive_traversal_symlinks_duplicates_and_corruption_rejected(self): cases = [] cases.append(archive(extra=tarfile.TarInfo('../escape'))) link = tarfile.TarInfo('unexpected') link.type = tarfile.SYMTYPE link.linkname = '/etc/passwd' cases.append(archive(extra=link)) cases.append(archive(extra=tarfile.TarInfo('ArcadeEscape.pck'))) cases.append(archive(missing=True)) cases.append(archive(wrong_hash=True)) for stream in cases: with tempfile.TemporaryDirectory() as temp, self.assertRaises(ValueError): receiver.unpack(stream, Path(temp)) def test_ssh_is_strict_and_no_token_in_command(self): with tempfile.TemporaryDirectory() as temp: folder = Path(temp) (folder / 'ssh').mkdir() (folder / 'ssh/arcade_cabinets').touch() command = deploy.ssh_args(folder, site()['cabinets'][0]) self.assertIn('StrictHostKeyChecking=yes', command) self.assertIn('BatchMode=yes', command) self.assertNotIn(site()['cabinets'][0]['token'], str(command)) def test_activation_refuses_running_game_before_writing_config(self): with tempfile.TemporaryDirectory() as temp: root = Path(temp) with patch.object(receiver, 'verify_release'), patch.object(receiver, 'render_config', return_value='config'), \ patch.object(receiver.Path, 'read_text', return_value=json.dumps({'cabinets': {'cabinet_01': {}}})), \ patch.object(receiver.subprocess, 'run') as run: run.return_value.returncode = 0 with self.assertRaisesRegex(ValueError, 'running'): receiver.activate(root, 'test-1', site()['cabinets'][0]) self.assertFalse((root / 'releases/test-1/tools/launch/launcher.ini').exists()) def test_setup_rerun_keeps_credentials(self): with tempfile.TemporaryDirectory() as temp: folder = Path(temp) settings = site() receiver.private_write(folder / 'inventory.json', json.dumps(settings)) with patch('builtins.input', side_effect=['', '']): deploy.setup(folder) self.assertEqual(deploy.load(folder), settings) def test_activation_failure_restores_previous_service_and_config(self): with tempfile.TemporaryDirectory() as temp: root = Path(temp) / 'deploy' home = Path(temp) / 'home' folder = root / 'releases/test-1' (folder / 'tools/room_controller').mkdir(parents=True) (folder / 'tools/room_controller/catalog.json').write_text(json.dumps({'cabinets': {'cabinet_01': {}}})) config = folder / 'tools/launch/launcher.ini' receiver.private_write(config, 'old configuration') unit = home / '.config/systemd/user/arcade-cabinet.service' receiver.private_write(unit, receiver.MARKER + '\nold unit') settings = dict(site()['cabinets'][0], url=site()['controller_url']) def run(command, **kwargs): import subprocess if command[0] == sys.executable: receiver.private_write(unit, receiver.MARKER + '\nchanged unit') raise subprocess.CalledProcessError(1, command) return subprocess.CompletedProcess(command, 3 if 'is-active' in command else 0) import subprocess with patch.object(receiver, 'verify_release'), patch.object(receiver.Path, 'home', return_value=home), \ patch.object(receiver.subprocess, 'run', side_effect=run), self.assertRaises(subprocess.CalledProcessError): receiver.activate(root, 'test-1', settings) self.assertEqual(config.read_text(), 'old configuration') self.assertEqual(unit.read_text(), receiver.MARKER + '\nold unit') self.assertFalse((root / 'active.json').exists()) def test_master_package_excludes_live_site_data(self): with tempfile.TemporaryDirectory() as temp: target = Path(temp) / 'master' deploy.master_package(target) names = [str(p.relative_to(target)) for p in target.rglob('*') if p.is_file()] self.assertTrue((target / 'tools/deploy/deploy.py').exists()) self.assertFalse(any(n.endswith(('.ini', '.sqlite3', '.cfg')) for n in names)) if __name__ == '__main__': unittest.main()