99 lines
5.3 KiB
Python
99 lines
5.3 KiB
Python
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import socket
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
spec = importlib.util.spec_from_file_location('arcade_launch', ROOT / 'tools/launch/launch.py')
|
|
launch = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(launch)
|
|
|
|
|
|
class LaunchTests(unittest.TestCase):
|
|
def config(self, folder, gpio='false'):
|
|
filename = Path(folder) / 'launcher.ini'
|
|
filename.write_text(f'''[controller]
|
|
operator_token = operator-test-token-1234
|
|
cabinet_tokens = {json.dumps({'cabinet_01':'cabinet-test-token-5678'})}
|
|
catalog = {ROOT / 'tools/room_controller/catalog.json'}
|
|
database = state/test.sqlite3
|
|
[cabinet]
|
|
executable = {sys.executable}
|
|
id = cabinet_01
|
|
gpio = {gpio}
|
|
''')
|
|
return filename
|
|
|
|
def test_combined_plan_and_relative_paths(self):
|
|
with tempfile.TemporaryDirectory() as folder, patch.dict(os.environ, {}, clear=True):
|
|
jobs, logs = launch.plan(self.config(folder), 'all')
|
|
self.assertEqual([j['name'] for j in jobs], ['controller', 'cabinet'])
|
|
self.assertEqual(jobs[1]['env']['ARCADE_CABINET_TOKEN'], 'cabinet-test-token-5678')
|
|
self.assertEqual(jobs[1]['env']['ARCADE_ROOM_URL'], 'http://127.0.0.1:8766')
|
|
self.assertEqual(logs, Path(folder) / 'logs')
|
|
self.assertNotIn('cabinet-test-token-5678', ' '.join(jobs[1]['command']))
|
|
self.assertNotIn('--gpio=real', jobs[1]['command'])
|
|
|
|
def test_cabinet_preserves_godot_config_fallback(self):
|
|
with tempfile.TemporaryDirectory() as folder, patch.dict(os.environ, {}, clear=True):
|
|
jobs, _ = launch.plan(self.config(folder), 'cabinet')
|
|
self.assertNotIn('ARCADE_CABINET_TOKEN', jobs[0]['env'])
|
|
self.assertNotIn('ARCADE_ROOM_URL', jobs[0]['env'])
|
|
|
|
def test_mismatched_combined_credentials_rejected(self):
|
|
with tempfile.TemporaryDirectory() as folder, patch.dict(os.environ, {'ARCADE_CABINET_TOKEN':'different-token-1234'}, clear=True):
|
|
with self.assertRaisesRegex(ValueError, 'does not match'):
|
|
launch.plan(self.config(folder), 'all')
|
|
|
|
def test_real_gpio_rejected_on_windows(self):
|
|
with tempfile.TemporaryDirectory() as folder, patch.dict(os.environ, {}, clear=True), patch.object(sys, 'platform', 'win32'):
|
|
with self.assertRaisesRegex(ValueError, 'Linux Pi'):
|
|
launch.plan(self.config(folder, 'true'), 'cabinet')
|
|
|
|
def test_pi_plan_starts_bridge_before_real_game(self):
|
|
with tempfile.TemporaryDirectory() as folder, patch.dict(os.environ, {}, clear=True), patch.object(sys, 'platform', 'linux'), patch.object(launch.importlib.util, 'find_spec', return_value=object()):
|
|
jobs, _ = launch.plan(self.config(folder, 'true'), 'cabinet')
|
|
self.assertEqual([j['name'] for j in jobs], ['gpio', 'cabinet'])
|
|
self.assertEqual(jobs[0]['env']['GPIOZERO_PIN_FACTORY'], 'lgpio')
|
|
self.assertIn('cabinet_01_gpio', jobs[0]['command'])
|
|
self.assertIn('--gpio=real', jobs[1]['command'])
|
|
self.assertEqual(jobs[0]['port'], ('127.0.0.1', 8765))
|
|
|
|
def test_occupied_port_does_not_get_reused_or_killed(self):
|
|
with socket.socket() as existing:
|
|
existing.bind(('127.0.0.1', 0))
|
|
existing.listen()
|
|
with self.assertRaisesRegex(ValueError, 'unavailable'):
|
|
launch.check_ports([dict(name='test', port=existing.getsockname())])
|
|
self.assertGreater(existing.fileno(), -1)
|
|
|
|
def test_game_exit_stops_owned_server(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
with socket.socket() as probe:
|
|
probe.bind(('127.0.0.1', 0))
|
|
port = probe.getsockname()[1]
|
|
service = dict(name='controller', command=[sys.executable, '-u', '-m', 'http.server', str(port), '--bind', '127.0.0.1'], env=dict(os.environ), directory=Path(folder), port=('127.0.0.1',port), ready=f'http://127.0.0.1:{port}/')
|
|
game = dict(name='cabinet', command=[sys.executable,'-c','import time; time.sleep(.3)'], env=dict(os.environ), directory=Path(folder), port=None, ready=None)
|
|
self.assertEqual(launch.run([service,game], Path(folder)/'logs'), 0)
|
|
with socket.socket() as probe:
|
|
self.assertNotEqual(probe.connect_ex(('127.0.0.1',port)), 0)
|
|
|
|
def test_startup_failure_cleans_up_previous_child(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
with socket.socket() as probe:
|
|
probe.bind(('127.0.0.1', 0))
|
|
port = probe.getsockname()[1]
|
|
service = dict(name='controller', command=[sys.executable, '-u', '-m', 'http.server', str(port), '--bind', '127.0.0.1'], env=dict(os.environ), directory=Path(folder), port=('127.0.0.1',port), ready=f'http://127.0.0.1:{port}/')
|
|
broken = dict(name='cabinet', command=[str(Path(folder)/'missing-executable')], env=dict(os.environ), directory=Path(folder), port=None, ready=None)
|
|
with self.assertRaises(OSError): launch.run([service,broken], Path(folder)/'logs')
|
|
with socket.socket() as probe:
|
|
self.assertNotEqual(probe.connect_ex(('127.0.0.1',port)), 0)
|
|
|
|
|
|
if __name__ == '__main__': unittest.main()
|