mirror of
https://github.com/saphid/frame-control.git
synced 2026-10-25 09:01:43 +01:00
237 lines
11 KiB
Python
237 lines
11 KiB
Python
"""The OBS relay must preserve transport packets and keep its output local."""
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import signal
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
|
|
SCRIPT = Path(__file__).resolve().parents[1] / 'scripts' / 'obs-headset-source.py'
|
|
spec = importlib.util.spec_from_file_location('obs_headset_source', SCRIPT)
|
|
obs = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(obs)
|
|
|
|
|
|
class Receiver:
|
|
def __init__(self):
|
|
self.chunks = []
|
|
|
|
def send(self, data):
|
|
self.chunks.append(data)
|
|
|
|
|
|
class ObsSourceTests(unittest.TestCase):
|
|
def test_desktop_packages_the_helper_as_a_required_resource(self):
|
|
root = SCRIPT.parents[1]
|
|
build = json.loads((root / 'app' / 'package.json').read_text())['build']
|
|
scripts = next(item for item in build['extraResources'] if item['to'] == 'scripts')
|
|
self.assertIn(SCRIPT.name, scripts['filter'])
|
|
self.assertIn('scripts/' + SCRIPT.name, (root / 'app' / 'build' / 'check-resources.js').read_text())
|
|
|
|
def test_relay_keeps_whole_transport_packets(self):
|
|
stream = bytes(range(188)) * 101
|
|
receiver = Receiver()
|
|
self.assertEqual(obs.relay(io.BytesIO(stream), receiver), len(stream))
|
|
self.assertEqual(b''.join(receiver.chunks), stream)
|
|
self.assertTrue(all(len(chunk) % 188 == 0 and len(chunk) <= 1316
|
|
for chunk in receiver.chunks))
|
|
|
|
def test_incomplete_packet_fails(self):
|
|
with self.assertRaisesRegex(RuntimeError, 'incomplete'):
|
|
obs.relay(io.BytesIO(b'x' * 189), Receiver())
|
|
|
|
def test_remote_command_uses_recording_sources(self):
|
|
with_audio = obs.remote_command()
|
|
self.assertIn('/dev/video99', with_audio)
|
|
self.assertIn('@DEFAULT_MONITOR@', with_audio)
|
|
self.assertIn('-map 1:a:0', with_audio)
|
|
self.assertIn('-f mpegts', with_audio)
|
|
self.assertNotIn('@DEFAULT_MONITOR@', obs.remote_command(audio=False))
|
|
|
|
def test_watchdog_ends_a_stalled_child_on_every_host_platform(self):
|
|
class Child:
|
|
def __init__(self):
|
|
self.stopped = threading.Event()
|
|
|
|
def poll(self):
|
|
return 255 if self.stopped.is_set() else None
|
|
|
|
def terminate(self):
|
|
self.stopped.set()
|
|
|
|
def wait(self, timeout=None):
|
|
return 255
|
|
|
|
for ready in (False, True):
|
|
with self.subTest(ready=ready):
|
|
child = Child()
|
|
watchdog = obs.StreamWatchdog(child, startup=0.1, stall=0.1)
|
|
if ready:
|
|
watchdog.packet()
|
|
try:
|
|
self.assertTrue(child.stopped.wait(1), 'a stalled SSH child was not terminated')
|
|
self.assertIn('stopped' if ready else 'no OBS video', watchdog.reason)
|
|
finally:
|
|
watchdog.close()
|
|
|
|
def test_shutdown_kills_an_unresponsive_ssh_child(self):
|
|
proc = mock.Mock()
|
|
proc.poll.return_value = None
|
|
proc.wait.side_effect = [subprocess.TimeoutExpired('ssh', 5), 137]
|
|
obs.stop_child(proc)
|
|
proc.terminate.assert_called_once()
|
|
proc.kill.assert_called_once()
|
|
self.assertEqual(proc.wait.call_count, 2)
|
|
|
|
def test_windows_sender_ignores_absent_receiver_and_hides_ssh_console(self):
|
|
proc = mock.Mock()
|
|
proc.stdout = io.BytesIO(bytes(range(188)) * 7)
|
|
proc.poll.return_value = proc.wait.return_value = 0
|
|
# These are the actual Python Winsock control and subprocess constants.
|
|
with mock.patch.object(obs.sys, 'platform', 'win32'), \
|
|
mock.patch.object(obs.subprocess, 'CREATE_NO_WINDOW', 0x08000000, create=True), \
|
|
mock.patch.object(obs.socket, 'socket') as socket_factory, \
|
|
mock.patch.object(obs.subprocess, 'Popen', return_value=proc) as popen:
|
|
self.assertEqual(obs.main(['--frame', 'test-frame']), 0)
|
|
udp = socket_factory.return_value.__enter__.return_value
|
|
self.assertEqual(popen.call_args.kwargs['creationflags'], 0x08000000)
|
|
udp.sendto.assert_called_once()
|
|
|
|
def test_windows_recovers_from_an_unreachable_udp_receiver(self):
|
|
udp = mock.Mock()
|
|
udp.sendto.side_effect = [ConnectionResetError(10054, 'receiver closed'), 188]
|
|
sender = obs._Sender(udp, ('127.0.0.1', 15099))
|
|
with mock.patch.object(obs.sys, 'platform', 'win32'):
|
|
sender.send(b'G' * 188)
|
|
self.assertFalse(sender.ready)
|
|
sender.send(b'G' * 188)
|
|
self.assertTrue(sender.ready)
|
|
self.assertEqual(udp.sendto.call_count, 2)
|
|
|
|
@unittest.skipIf(sys.platform == 'win32', 'the fake SSH executable uses a POSIX shebang')
|
|
def test_stall_exits_cleanly_even_while_app_control_pipe_stays_open(self):
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
directory = Path(folder)
|
|
fake = directory / 'ssh'
|
|
fake.write_text("#!/usr/bin/env python3\n"
|
|
"import signal,sys,time\n"
|
|
"signal.signal(signal.SIGTERM, signal.SIG_IGN)\n"
|
|
"sys.stdout.buffer.write(b'G'*1316);sys.stdout.buffer.flush()\n"
|
|
"while True: time.sleep(1)\n")
|
|
fake.chmod(0o755)
|
|
wrapper = ("import importlib.util; "
|
|
f"s=importlib.util.spec_from_file_location('obs', {str(SCRIPT)!r}); "
|
|
"o=importlib.util.module_from_spec(s);s.loader.exec_module(o); "
|
|
"o.STARTUP_TIMEOUT=o.STREAM_STALL=o.CHILD_STOP_TIMEOUT=0.15; "
|
|
"o.main(['--frame','fake-frame','--control-stdin'])")
|
|
env = {**os.environ, 'PATH': str(directory) + os.pathsep + os.environ.get('PATH', '')}
|
|
process = subprocess.Popen([sys.executable, '-c', wrapper], stdin=subprocess.PIPE,
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, env=env)
|
|
try:
|
|
# Keep stdin OPEN: real Electron does this until the helper exits.
|
|
self.assertEqual(process.wait(timeout=4), 1)
|
|
error = process.stderr.read().decode(errors='replace')
|
|
self.assertIn('video stopped', error)
|
|
self.assertNotIn('Fatal Python error', error)
|
|
finally:
|
|
if process.poll() is None:
|
|
process.kill(); process.wait(timeout=5)
|
|
process.stdin.close(); process.stderr.close()
|
|
|
|
@unittest.skipIf(sys.platform == 'win32', 'the fake SSH executable uses a POSIX shebang')
|
|
def test_relay_stops_its_ssh_child_on_app_signal_or_control_pipe(self):
|
|
for mode in ('signal', 'control-pipe'):
|
|
with self.subTest(mode=mode):
|
|
self._check_stop_mode(mode)
|
|
|
|
def _check_stop_mode(self, mode):
|
|
with tempfile.TemporaryDirectory() as folder, socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as receiver:
|
|
directory = Path(folder)
|
|
stopped = directory / 'stopped'
|
|
fake = directory / 'ssh'
|
|
fake.write_text("#!/usr/bin/env python3\n"
|
|
"import os, signal, sys, time\n"
|
|
"def stop(*_):\n"
|
|
" open(os.environ['FAKE_SSH_STOPPED'], 'w').close()\n"
|
|
" sys.exit(255)\n"
|
|
"signal.signal(signal.SIGTERM, stop)\n"
|
|
"sys.stdout.buffer.write(b'G' * 1316)\n"
|
|
"sys.stdout.buffer.flush()\n"
|
|
"while True: time.sleep(1)\n")
|
|
fake.chmod(0o755)
|
|
receiver.bind(('127.0.0.1', 0))
|
|
receiver.settimeout(8)
|
|
env = {**os.environ, 'PATH': str(directory) + os.pathsep + os.environ.get('PATH', ''),
|
|
'FAKE_SSH_STOPPED': str(stopped)}
|
|
arguments = [sys.executable, str(SCRIPT), '--frame', 'fake-frame',
|
|
'--port', str(receiver.getsockname()[1])]
|
|
if mode == 'control-pipe':
|
|
arguments.append('--control-stdin')
|
|
process = subprocess.Popen(arguments,
|
|
stdin=subprocess.PIPE if mode == 'control-pipe' else subprocess.DEVNULL,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE, env=env)
|
|
try:
|
|
packet, _ = receiver.recvfrom(1500)
|
|
self.assertEqual(packet, b'G' * 1316)
|
|
if mode == 'control-pipe':
|
|
process.stdin.close()
|
|
process.wait(timeout=8)
|
|
errors = process.stderr.read()
|
|
else:
|
|
process.send_signal(signal.SIGTERM)
|
|
_, errors = process.communicate(timeout=8)
|
|
self.assertEqual(process.returncode, 0, errors.decode())
|
|
self.assertIn(b'Frame Control OBS stream is ready', errors)
|
|
for _ in range(30):
|
|
if stopped.exists():
|
|
break
|
|
time.sleep(0.1)
|
|
self.assertTrue(stopped.exists(), 'the SSH child survived relay shutdown')
|
|
finally:
|
|
if process.poll() is None:
|
|
process.kill()
|
|
process.communicate(timeout=5)
|
|
process.stdout.close()
|
|
process.stderr.close()
|
|
|
|
@unittest.skipIf(sys.platform == 'win32', 'the stand-in SSH script uses a POSIX shebang')
|
|
def test_startup_and_stream_stalls_end_ssh_with_a_clear_error(self):
|
|
for mode in ('startup', 'stream'):
|
|
with self.subTest(mode=mode), tempfile.TemporaryDirectory() as folder:
|
|
directory = Path(folder)
|
|
fake = directory / 'ssh'
|
|
fake.write_text("#!/usr/bin/env python3\n"
|
|
"import os, signal, sys, time\n"
|
|
"def stop(*_):\n"
|
|
" open(os.environ['FAKE_SSH_STOPPED'], 'w').close()\n"
|
|
" sys.exit(255)\n"
|
|
"signal.signal(signal.SIGTERM, stop)\n"
|
|
"if os.environ['FAKE_SSH_MODE'] == 'stream':\n"
|
|
" sys.stdout.buffer.write(b'G' * 1316)\n"
|
|
" sys.stdout.buffer.flush()\n"
|
|
"while True: time.sleep(1)\n")
|
|
fake.chmod(0o755)
|
|
stopped = directory / 'stopped'
|
|
env = {'PATH': str(directory) + os.pathsep + os.environ.get('PATH', ''),
|
|
'FAKE_SSH_STOPPED': str(stopped), 'FAKE_SSH_MODE': mode}
|
|
with mock.patch.dict(os.environ, env), \
|
|
mock.patch.object(obs, 'STARTUP_TIMEOUT', 0.3), \
|
|
mock.patch.object(obs, 'STREAM_STALL', 0.3):
|
|
with self.assertRaisesRegex(RuntimeError, 'no OBS video' if mode == 'startup' else 'video stopped'):
|
|
obs.main(['--frame', 'fake-frame'])
|
|
self.assertTrue(stopped.exists(), 'SSH survived the relay timeout')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|