room_glitch/tests/room_network/test_deploy.py

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()