mirror of
https://github.com/saphid/frame-control.git
synced 2026-10-24 15:01:37 +02:00
SAPHIDandClaude Opus 5.5
7a7dca70de
host_view now picks the PC host on Windows only; Linux and macOS keep the Mac view as on main, so #38's Linux adapter is never offered. The PC host starts nothing, loads no native library and lists no windows until the person ticks "Turn on PC in the headset" on the card (action "enable" on /api/macview). Turning it off ends streams, viewers and the helper. The choice is a client preference, like mvSeparate. Tools lists "PC in the headset" on Windows only, beside the macOS-only Mac link, with an Experimental tag and a short note on what hasn't been tried. The Mac card's wording and Linux-portal tweaks from #38 go back to main's, apart from the shared "your computer" hint. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
521 lines
25 KiB
Python
521 lines
25 KiB
Python
"""PC adapters: grants, input cleanup, launcher reuse and native codec contract.
|
|
Native tests require the bundled libraries; CI sets REQUIRE_NATIVE so a missing
|
|
bundle is a failure, never a silent skip. No desktop consent or GPU is faked as
|
|
successful real-host capture.
|
|
|
|
The encoding tests need x264, which only CI's own builds bundle (it's GPL, so
|
|
the installer leaves it out). FRAME_PC_REQUIRE_X264=1 makes its absence a
|
|
failure; FRAME_PC_INSTALLER_BUNDLE=1 checks the installer's bundle instead.
|
|
"""
|
|
import ctypes
|
|
import http.client
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shutil
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
import unittest
|
|
from unittest import mock
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
sys.path.insert(0, str(ROOT / 'ui'))
|
|
import frame_pc_agent as agent
|
|
import frame_pc_capture as capture
|
|
import frame_pcview
|
|
import frame_macview
|
|
from frame_stream_stats import Stats
|
|
from test_macview import WS
|
|
|
|
|
|
class GrantsTests(unittest.TestCase):
|
|
def test_source_bound_retry_ack_and_stop(self):
|
|
g = agent.Grants('master')
|
|
ticket = g.ticket('window:1')
|
|
self.assertIsNone(g.redeem('window:2', {'t': ticket}))
|
|
key = g.redeem('window:1', {'t': ticket})
|
|
self.assertEqual(key, g.redeem('window:1', {'t': ticket}))
|
|
g.ack(key)
|
|
self.assertIsNone(g.redeem('window:1', {'t': ticket}))
|
|
self.assertEqual(key, g.redeem('window:1', {'r': key}))
|
|
self.assertIsNone(g.redeem('window:2', {'r': key}))
|
|
g.revoke('window:1')
|
|
self.assertIsNone(g.redeem('window:1', {'r': key}))
|
|
|
|
def test_expiry_and_stop_before_redeem(self):
|
|
g = agent.Grants('master')
|
|
with mock.patch.object(agent.time, 'monotonic', return_value=1):
|
|
ticket = g.ticket('test')
|
|
with mock.patch.object(agent.time, 'monotonic', return_value=62):
|
|
self.assertIsNone(g.redeem('test', {'t': ticket}))
|
|
ticket = g.ticket('test')
|
|
g.revoke()
|
|
self.assertIsNone(g.redeem('test', {'t': ticket}))
|
|
|
|
|
|
class GatingTests(unittest.TestCase):
|
|
"""Experimental, Windows only, and off until the person turns it on."""
|
|
|
|
def view(self):
|
|
return frame_pcview.PCView(['ssh'], mock.Mock(), 'frame')
|
|
|
|
def test_only_windows_gets_the_pc_host(self):
|
|
for platform, cls in (('win32', frame_pcview.PCView), ('linux', frame_macview.MacView),
|
|
('darwin', frame_macview.MacView)):
|
|
with mock.patch.object(frame_pcview.sys, 'platform', platform):
|
|
view = frame_pcview.host_view(['ssh'], mock.Mock(), 'frame')
|
|
self.assertIs(type(view), cls, platform)
|
|
|
|
def test_off_by_default_starts_nothing(self):
|
|
view = self.view()
|
|
with mock.patch.object(frame_pcview.sys, 'platform', 'win32'), \
|
|
mock.patch.object(frame_macview.subprocess, 'Popen', side_effect=AssertionError('started while off')):
|
|
state = view.state()
|
|
self.assertEqual(state, {'available': False, 'reason': frame_pcview.OFF, 'host': view.host,
|
|
'experimental': True, 'enabled': False})
|
|
with self.assertRaises(frame_macview.MacViewError):
|
|
view.show('window:12')
|
|
view.run.assert_not_called()
|
|
|
|
def test_never_offered_off_windows_even_when_turned_on(self):
|
|
view = self.view()
|
|
view.enabled = True
|
|
with mock.patch.object(frame_pcview.sys, 'platform', 'linux'):
|
|
self.assertIn('only available on Windows', view.unavailable())
|
|
|
|
def test_turning_on_checks_for_the_bundled_libraries(self):
|
|
view = self.view()
|
|
missing = mock.Mock(is_file=mock.Mock(return_value=False))
|
|
with mock.patch.object(frame_pcview.sys, 'platform', 'win32'), \
|
|
mock.patch.object(frame_pcview, 'LIBRARY', missing):
|
|
state = view.set_enabled(True)
|
|
self.assertTrue(state['enabled'])
|
|
self.assertFalse(state['available'])
|
|
self.assertIn('libraries are missing', state['reason'])
|
|
|
|
def test_turning_off_stops_streams_and_the_helper(self):
|
|
view = self.view()
|
|
view.enabled = True
|
|
agent = mock.Mock()
|
|
agent.poll.return_value = None
|
|
view.agent, view.tunnel = agent, mock.Mock()
|
|
tunnel = view.tunnel
|
|
tunnel.poll.return_value = None
|
|
view.stop = mock.Mock()
|
|
with mock.patch.object(frame_pcview.sys, 'platform', 'win32'):
|
|
state = view.set_enabled(False)
|
|
view.stop.assert_called_once_with()
|
|
agent.stdin.close.assert_called_once()
|
|
agent.wait.assert_called()
|
|
tunnel.terminate.assert_called_once()
|
|
self.assertIsNone(view.agent)
|
|
self.assertEqual((state['enabled'], state['reason']), (False, frame_pcview.OFF))
|
|
|
|
def test_server_enable_action(self):
|
|
import sandbox # noqa: F401 (keeps the server off real data and services)
|
|
import server
|
|
with mock.patch.object(server, 'macview', frame_macview.MacView(['ssh'], mock.Mock(), 'frame')):
|
|
with self.assertRaises(server.Failure) as e:
|
|
server.macview_action({'action': 'enable', 'on': True})
|
|
self.assertIn('only available on Windows', str(e.exception))
|
|
view = self.view()
|
|
view.set_enabled = mock.Mock(return_value={'enabled': True})
|
|
with mock.patch.object(server, 'macview', view):
|
|
self.assertEqual(server.macview_action({'action': 'enable', 'on': True}), {'enabled': True})
|
|
server.macview_action({'action': 'enable', 'on': 'yes'}) # anything but true is off
|
|
self.assertEqual([c.args for c in view.set_enabled.call_args_list], [(True,), (False,)])
|
|
|
|
|
|
class PageTests(unittest.TestCase):
|
|
page = (ROOT / 'ui' / 'index.html').read_text(encoding='utf-8')
|
|
|
|
def test_tool_link_is_windows_only_and_marked_experimental(self):
|
|
self.assertIn('<a href="#macview" id="pcToolLink" hidden><span>PC in the headset</span>'
|
|
'<small>Experimental:', self.page)
|
|
self.assertIn('$("pcToolLink").hidden = h.mobile || h.os !== "Windows";', self.page)
|
|
self.assertIn('$("macToolLink").hidden = h.mobile || h.os !== "macOS";', self.page)
|
|
|
|
def test_card_says_experimental_and_what_is_unverified(self):
|
|
self.assertIn('<span class="tag" id="mvTag" hidden>Experimental</span>', self.page)
|
|
self.assertIn('<input type="checkbox" id="mvPcOn">', self.page)
|
|
self.assertIn("hasn't been tried on a real\n Windows desktop or with a headset yet", self.page)
|
|
self.assertIn('docs/pc-in-headset.md', self.page)
|
|
|
|
def test_opt_in_is_remembered_with_the_client_preferences(self):
|
|
self.assertIn("const keys = ['mvSeparate', 'padUsed', 'pcView']", self.page)
|
|
prefs = (ROOT / 'app' / 'preferences.js').read_text(encoding='utf-8')
|
|
self.assertIn('const KEYS = ["mvSeparate", "padUsed", "pcView"];', prefs)
|
|
|
|
|
|
class AdapterTests(unittest.TestCase):
|
|
def test_windows_uses_wgc_and_media_foundation(self):
|
|
for kind in ('window', 'display'):
|
|
pipeline = capture.pipeline(dict(src=kind+':42'), 'win32', 'mfh264enc', 640, 360, 30, 1000000)
|
|
self.assertIn('capture-api=wgc', pipeline)
|
|
self.assertIn('window-handle=42' if kind == 'window' else 'monitor-handle=42', pipeline)
|
|
self.assertIn('mfh264enc name=enc low-latency=true bframes=0', pipeline)
|
|
self.assertNotIn('drop=true', pipeline)
|
|
with self.assertRaises(ValueError):
|
|
capture.pipeline(dict(src='window:42 ! fakesink'), 'win32', 'mfh264enc', 640, 360, 30, 1000000)
|
|
|
|
def test_linux_uses_only_portal_fd_and_node(self):
|
|
p = capture.pipeline(dict(src='window:consented', fd=19, node=47), 'linux', 'x264enc', 640, 360, 30, 1000000)
|
|
self.assertIn('pipewiresrc fd=19 path=47', p)
|
|
self.assertIn('tune=zerolatency', p)
|
|
self.assertEqual(capture.dimensions(1920, 1080, 1280), (1280, 720))
|
|
|
|
def test_portal_input_clamps_and_releases_held_state(self):
|
|
native = mock.Mock()
|
|
native.lib.fc_portal_input.return_value = 1
|
|
inp = capture.PortalInput(native, dict(portal=123, w=100, h=50))
|
|
inp.handle(dict(t='m', e='down', b=0, x=2, y=-1))
|
|
inp.handle(dict(t='k', e='down', key='Control'))
|
|
inp.release()
|
|
calls = native.lib.fc_portal_input.call_args_list
|
|
self.assertEqual(calls[0].args, (123, 0, 99, 0, 0, 0))
|
|
self.assertIn(mock.call(123, 1, 0, 0, 0x110, 0), calls)
|
|
self.assertIn(mock.call(123, 3, 0, 0, 0xffe3, 0), calls)
|
|
self.assertFalse(inp.buttons or inp.keys)
|
|
|
|
def test_windows_release_and_negative_monitor_origin(self):
|
|
host = mock.Mock()
|
|
host.user.GetSystemMetrics.side_effect = [-1920, 0, 3840, 1080]
|
|
inp = capture.WindowsInput(host, dict(src='display:4', x=-1920, y=0, w=1920, h=1080))
|
|
inp.handle(dict(t='m', e='down', b=2, x=0, y=0))
|
|
self.assertEqual(host.send.call_args_list[0], mock.call(mouse=(0, 0, 0, 0xc001)))
|
|
inp.handle(dict(t='k', e='down', code='ControlLeft', key='Control'))
|
|
inp.release()
|
|
self.assertIn(mock.call(mouse=(0, 0, 0, 0x10)), host.send.call_args_list)
|
|
self.assertIn(mock.call(key=(162, 0, 2)), host.send.call_args_list)
|
|
|
|
def test_refused_windows_focus_never_clicks_the_covering_app(self):
|
|
host = mock.Mock()
|
|
host.rect.return_value = (0, 0, 800, 600)
|
|
host.user.GetForegroundWindow.return_value = 99
|
|
host.user.SetForegroundWindow.return_value = False
|
|
inp = capture.WindowsInput(host, dict(src='window:12', w=800, h=600))
|
|
with self.assertRaisesRegex(RuntimeError, 'could not focus'):
|
|
inp.handle(dict(t='m', e='down', b=0, x=.5, y=.5))
|
|
host.send.assert_not_called()
|
|
|
|
def test_pc_reuses_launcher_and_namespaces_panel_ids(self):
|
|
view = frame_pcview.PCView(['ssh'], mock.Mock(return_value='panel created'), 'frame')
|
|
view.call = mock.Mock(side_effect=lambda path, **kw: {'screen': True, 'ticket': 'one-use', 'streams': []})
|
|
view.ensure_tunnel = mock.Mock()
|
|
view.remote_port = 47900
|
|
view.show('window:12')
|
|
self.assertEqual(view.run.call_args.kwargs['stdin'], frame_macview.LAUNCH)
|
|
self.assertNotEqual(frame_macview.panel_id('window:12'), frame_macview.panel_id('window:12', 'windows'))
|
|
self.assertFalse(view.prefer_usb)
|
|
with self.assertRaises(frame_macview.MacViewError):
|
|
view.show('separate:12')
|
|
|
|
def test_session_end_releases_native_input_only_once(self):
|
|
session = object.__new__(agent.Session)
|
|
session.lock, session.input_lock = threading.RLock(), threading.RLock()
|
|
session.stop_event = threading.Event()
|
|
session.ws, session.input = mock.Mock(), mock.Mock()
|
|
session.released = False
|
|
session.end()
|
|
session.end()
|
|
session.input.release.assert_called_once()
|
|
self.assertTrue(session.stop_event.is_set())
|
|
|
|
def test_idle_source_is_not_a_stall(self):
|
|
# PipeWire/WGC deliver frames only on damage: a static desktop sends
|
|
# its first frame and then nothing. That must not end the session.
|
|
session = object.__new__(agent.Session)
|
|
session.lock, session.pending = threading.RLock(), {}
|
|
session.stats = Stats(lambda: 0)
|
|
session.stats.captured = 1
|
|
self.assertIsNone(session.stalled(60000000, 0, 0))
|
|
# A frame that entered the encoder and never came out is a fault.
|
|
session.pending[5] = dict(cap=0, arr=0, e0=1000000, tier=0, br=0)
|
|
self.assertIn('encoder', session.stalled(12000000, 0, 0))
|
|
self.assertIsNone(session.stalled(10000000, 0, 0))
|
|
# No initial frame at all after the pipeline opened is a fault.
|
|
session.pending.clear()
|
|
self.assertIn('capture', session.stalled(12000000, 0, 1))
|
|
self.assertIsNone(session.stalled(9000000, 0, 1))
|
|
|
|
def test_controller_is_inert_after_close(self):
|
|
lib = mock.Mock()
|
|
lib.fc_new.return_value = 1234
|
|
lib.fc_value.return_value = 100
|
|
controller = capture.Controller(lib, 60, 5000000)
|
|
controller.close()
|
|
lib.fc_free.assert_called_once_with(1234)
|
|
lib.reset_mock()
|
|
state = controller.state()
|
|
self.assertEqual((state['fps'], state['target'], state['scale']), (60, 0, 1))
|
|
self.assertEqual(controller.call('gate', 1, 1), 0)
|
|
self.assertEqual(controller.update(1), 0)
|
|
controller.close()
|
|
self.assertEqual(lib.method_calls, []) # nothing touched the freed pointer
|
|
|
|
def test_library_path_prepend_adds_no_empty_entry(self):
|
|
env = {}
|
|
frame_pcview.prepend(env, 'LD_LIBRARY_PATH', '/opt/fc/lib')
|
|
self.assertEqual(env['LD_LIBRARY_PATH'], '/opt/fc/lib')
|
|
frame_pcview.prepend(env, 'LD_LIBRARY_PATH', '/x')
|
|
self.assertEqual(env['LD_LIBRARY_PATH'], '/x' + os.pathsep + '/opt/fc/lib')
|
|
with mock.patch.dict(os.environ, clear=True):
|
|
env = frame_pcview.PCView([], lambda *a: None, 'frame').agent_environment()
|
|
for name in ('PATH', 'LD_LIBRARY_PATH'):
|
|
parts = env.get(name, 'x').split(os.pathsep)
|
|
self.assertNotIn('', parts, name)
|
|
|
|
def test_malformed_viewer_input_is_dropped_not_fatal(self):
|
|
session = object.__new__(agent.Session)
|
|
session.lock, session.input_lock = threading.RLock(), threading.RLock()
|
|
session.stop_event, session.key_event = threading.Event(), threading.Event()
|
|
session.src, session.codec, session.key, session.w, session.h = 'display:1', 'h264', 'r', 100, 100
|
|
session.source = dict(src='display:1', w=100, h=100, portal=1)
|
|
session.input_enabled, session.released = True, False
|
|
session.native = mock.Mock()
|
|
session.native.now.return_value = 0
|
|
session.controller, session.agent = mock.Mock(), mock.Mock()
|
|
session.stats = Stats(lambda: 0)
|
|
session.produce = lambda: None
|
|
lib = mock.Mock()
|
|
lib.fc_portal_input.return_value = 1
|
|
session.input = capture.PortalInput(mock.Mock(lib=lib), session.source)
|
|
session.ws = mock.Mock()
|
|
session.ws.receive.side_effect = [
|
|
dict(t='m', x='left', y=0), dict(t='wheel', dx=[1]), dict(t='k', e='down', key=None, i='x'),
|
|
dict(t='fd', drop='many'), dict(t='rx', s=[1]), dict(t='m', x=.5, y=.5), ConnectionError('closed')]
|
|
with self.assertRaises(ConnectionError):
|
|
session.start()
|
|
# Every malformed message was skipped; the good move after them still ran.
|
|
moves = [c.args for c in lib.fc_portal_input.call_args_list if c.args[1] == 0]
|
|
self.assertAlmostEqual(moves[-1][2], .5*99)
|
|
self.assertEqual(session.ws.receive.call_count, 7)
|
|
|
|
def test_stats_keep_capture_time_and_bound_records(self):
|
|
stats = Stats(lambda: 10000000)
|
|
stats.input(dict(t='m', i=1, tv=9999990))
|
|
f = stats.add(cap=10000001, arr=10000002, e0=10000003, e1=10000004)
|
|
self.assertEqual(f['echo'], 1)
|
|
stats.report(dict(t='rx', s=f['s'], r=10000010))
|
|
stats.report(dict(t='fd', f=[[f['s'], 10000011, 10000012, 10000013]]))
|
|
self.assertEqual(stats.snapshot(settle=0)['inputs'][0]['frame'], f['s'])
|
|
for _ in range(4100):
|
|
stats.add(cap=10000001)
|
|
self.assertEqual(len(stats.frames), 4096)
|
|
|
|
|
|
def x264_bundled():
|
|
return any((capture.NATIVE / 'lib' / 'gstreamer-1.0').glob('*gstx264*'))
|
|
|
|
|
|
def needs_x264(test):
|
|
def run(self):
|
|
if not x264_bundled():
|
|
if os.environ.get('FRAME_PC_REQUIRE_X264') == '1':
|
|
raise AssertionError('CI required x264 in the native PC bundle (desktop/build.py --with-x264)')
|
|
raise unittest.SkipTest('x264 is only in CI test bundles, not the installer')
|
|
return test(self)
|
|
run.__name__, run.__doc__ = test.__name__, test.__doc__
|
|
return run
|
|
|
|
|
|
class NativeTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
if not capture.LIBRARY.exists():
|
|
if os.environ.get('FRAME_PC_REQUIRE_NATIVE') == '1':
|
|
raise AssertionError('CI required native PC libraries but they were not built')
|
|
raise unittest.SkipTest('native PC libraries not built on this host')
|
|
cls.token = 'test-' + os.urandom(8).hex()
|
|
env = frame_pcview.PCView([], lambda *a: None, 'frame').agent_environment()
|
|
env['FRAME_MAC_VIEW_TOKEN'] = cls.token
|
|
cls.proc = subprocess.Popen([sys.executable, str(ROOT/'ui/frame_pc_agent.py'), 'serve', '--port', '0',
|
|
'--page', str(ROOT/'ui/mac-view.html'), '--exit-on-eof'], stdin=subprocess.PIPE, stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE, text=True, env=env)
|
|
line = cls.proc.stdout.readline()
|
|
if 'listening' not in line:
|
|
raise AssertionError('Native agent failed: ' + cls.proc.stderr.read())
|
|
cls.port = int(line.rsplit(':', 1)[1])
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
cls.proc.stdin.close()
|
|
try:
|
|
rc = cls.proc.wait(10)
|
|
except subprocess.TimeoutExpired:
|
|
cls.proc.kill()
|
|
raise AssertionError('Native agent did not stop after EOF')
|
|
error = cls.proc.stderr.read()
|
|
cls.proc.stdout.close()
|
|
cls.proc.stderr.close()
|
|
if rc:
|
|
raise AssertionError('Native agent exited %s: %s' % (rc, error))
|
|
|
|
def request(self, path, method='GET'):
|
|
conn = http.client.HTTPConnection('127.0.0.1', self.port, timeout=10)
|
|
try:
|
|
conn.request(method, path)
|
|
response = conn.getresponse()
|
|
return response.status, response.read()
|
|
finally:
|
|
conn.close()
|
|
|
|
def tearDown(self):
|
|
self.request('/close?k='+self.token, 'POST')
|
|
|
|
@unittest.skipUnless(sys.platform.startswith('linux'), 'Linux PipeWire client libraries')
|
|
def test_bundled_pipewire_context_needs_no_system_modules(self):
|
|
# No portal request, remote connection or desktop capture: create and
|
|
# destroy a private client context to test the dynamically loaded SPA
|
|
# and protocol modules that ldd cannot discover.
|
|
code = """
|
|
import ctypes as C, os
|
|
lib = C.CDLL(os.path.join(os.environ['FRAME_PC_NATIVE'], 'lib', 'libpipewire-0.3.so.0'))
|
|
for name, ret, args in [
|
|
('pw_init', None, [C.c_void_p, C.c_void_p]),
|
|
('pw_main_loop_new', C.c_void_p, [C.c_void_p]),
|
|
('pw_main_loop_get_loop', C.c_void_p, [C.c_void_p]),
|
|
('pw_context_new', C.c_void_p, [C.c_void_p, C.c_void_p, C.c_size_t]),
|
|
('pw_context_find_factory', C.c_void_p, [C.c_void_p, C.c_char_p]),
|
|
('pw_context_find_spa_lib', C.c_char_p, [C.c_void_p, C.c_char_p]),
|
|
('pw_context_destroy', None, [C.c_void_p]),
|
|
('pw_main_loop_destroy', None, [C.c_void_p])]:
|
|
fn = getattr(lib, name)
|
|
fn.restype, fn.argtypes = ret, args
|
|
lib.pw_init(None, None)
|
|
loop = lib.pw_main_loop_new(None)
|
|
assert loop, 'bundled SPA loop support did not load'
|
|
context = lib.pw_context_new(lib.pw_main_loop_get_loop(loop), None, 0)
|
|
assert context, 'bundled PipeWire client modules did not load'
|
|
assert lib.pw_context_find_factory(context, b'adapter'), 'video adapter factory is missing'
|
|
# The exported factory is video.adapt, which needs a live follower node to
|
|
# instantiate. Check its configured library and exported factory without
|
|
# pretending a headless context is a consented video stream.
|
|
plugin_name = lib.pw_context_find_spa_lib(context, b'video.adapt')
|
|
assert plugin_name, 'video.adapt has no bundled library mapping'
|
|
plugin = C.CDLL(os.path.join(os.environ['SPA_PLUGIN_DIR'], plugin_name.decode() + '.so'))
|
|
class Factory(C.Structure):
|
|
_fields_ = [('version', C.c_uint32), ('name', C.c_char_p)]
|
|
plugin.spa_handle_factory_enum.argtypes = [C.POINTER(C.POINTER(Factory)), C.POINTER(C.c_uint32)]
|
|
plugin.spa_handle_factory_enum.restype = C.c_int
|
|
factory, index, names = C.POINTER(Factory)(), C.c_uint32(), []
|
|
while plugin.spa_handle_factory_enum(C.byref(factory), C.byref(index)) > 0:
|
|
names.append(factory.contents.name)
|
|
assert b'video.adapt' in names, 'bundled SPA video adapter factory is missing'
|
|
lib.pw_context_destroy(context)
|
|
lib.pw_main_loop_destroy(loop)
|
|
"""
|
|
env = frame_pcview.PCView([], lambda *a: None, 'frame').agent_environment()
|
|
env['FRAME_PC_NATIVE'] = str(capture.NATIVE)
|
|
result = subprocess.run([sys.executable, '-c', code], env=env, capture_output=True, text=True, timeout=15)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
|
|
def test_agent_reports_its_host_and_encoder(self):
|
|
self.assertEqual(self.request('/status')[0], 403)
|
|
status, body = self.request('/status?k='+self.token)
|
|
self.assertEqual(status, 200)
|
|
data = json.loads(body)
|
|
if sys.platform == 'win32':
|
|
self.assertEqual((data['host'], data['encoder']), ('windows', 'mfh264enc'))
|
|
self.assertEqual(data['streams'], [])
|
|
|
|
@unittest.skipUnless(os.environ.get('FRAME_PC_INSTALLER_BUNDLE') == '1', 'checks the installer\'s bundle')
|
|
def test_installer_bundle_leaves_out_gpl_x264(self):
|
|
self.assertFalse(x264_bundled())
|
|
self.assertFalse(list((capture.NATIVE / 'bin').glob('libx264*')))
|
|
plugins = [p.name for p in (capture.NATIVE / 'lib' / 'gstreamer-1.0').iterdir()]
|
|
for name in ('mediafoundation', 'd3d11', 'jpeg'):
|
|
self.assertTrue(any(p.endswith('gst%s.dll' % name) for p in plugins), (name, plugins))
|
|
|
|
@needs_x264
|
|
def test_last_raw_picture_survives_a_closed_gate(self):
|
|
# A source sends ONE changed frame, then becomes idle. It must reach
|
|
# the encoder after congestion clears, without needing another update.
|
|
code = """
|
|
import ctypes as C, sys, time
|
|
from frame_pc_capture import Native, GATE, Encoded, pipeline
|
|
native = Native()
|
|
start = None
|
|
closing = False
|
|
phases = []
|
|
def gate(stage, pts, capture, arrived):
|
|
global start
|
|
phases.append(stage)
|
|
if closing: return -1
|
|
if start is None: start = time.monotonic()
|
|
return int(stage == 1 or time.monotonic()-start >= .15)
|
|
callback = GATE(gate)
|
|
p = pipeline(dict(src='test'), sys.platform, 'x264enc', 320, 180, 30, 300000)
|
|
p = p.replace('is-live=true', 'is-live=true num-buffers=1')
|
|
error = C.create_string_buffer(1024)
|
|
handle = native.lib.fc_capture_open(p.encode(), callback, error, len(error))
|
|
assert handle, error.value
|
|
try:
|
|
frame = Encoded()
|
|
result = 0
|
|
deadline = time.monotonic()+5
|
|
while not result and time.monotonic() < deadline:
|
|
result = native.lib.fc_capture_pull(handle, C.byref(frame))
|
|
assert result == 1, native.lib.fc_capture_error(handle)
|
|
assert frame.key and frame.size > 17
|
|
assert phases.count(0) == 1 and 2 in phases, phases
|
|
finally:
|
|
closing = True
|
|
native.lib.fc_capture_close(handle)
|
|
"""
|
|
env = frame_pcview.PCView([], lambda *a: None, 'frame').agent_environment()
|
|
env['PYTHONPATH'] = str(ROOT / 'ui')
|
|
result = subprocess.run([sys.executable, '-c', code], env=env, capture_output=True, text=True, timeout=10)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
|
|
@needs_x264
|
|
def test_auth_and_real_h264_stats(self):
|
|
self.assertEqual(self.request('/status')[0], 403)
|
|
status, body = self.request('/ticket?src=test&k='+self.token, 'POST')
|
|
self.assertEqual(status, 200)
|
|
ticket = json.loads(body)['ticket']
|
|
ws = WS(self.port, '/stream?src=test&fps=30&max=640&t='+ticket)
|
|
try:
|
|
self.assertEqual(ws.status, 101)
|
|
key, got, echo = None, 0, False
|
|
ws.send_text(json.dumps(dict(t='m', e='down', i=7, tv=1, x=.2, y=.3)))
|
|
deadline = time.monotonic()+10
|
|
while time.monotonic() < deadline and (got < 5 or not echo):
|
|
op, data = ws.recv()
|
|
if op == 1:
|
|
m = json.loads(data)
|
|
self.assertNotEqual(m.get('t'), 'error', m)
|
|
if m.get('t') == 'hello':
|
|
key = m['r']
|
|
ws.send_text(json.dumps(dict(t='ack')))
|
|
elif op == 2:
|
|
got += 1
|
|
seq, echoed = struct.unpack('!II', data[9:17])
|
|
now = struct.unpack('!Q', data[1:9])[0]
|
|
echo |= echoed == 7
|
|
self.assertIn(b'\x00\x00\x00\x01', data[17:])
|
|
ws.send_text(json.dumps(dict(t='rx', s=seq, r=now+100)))
|
|
ws.send_text(json.dumps(dict(t='fd', f=[[seq, now+200, now+300, now+400]])))
|
|
self.assertGreaterEqual(got, 5)
|
|
self.assertTrue(echo)
|
|
time.sleep(.1)
|
|
stats = json.loads(self.request('/stats?settle=0&k='+self.token)[1])['streams'][0]
|
|
self.assertGreater(stats['frames'][0]['e1'], stats['frames'][0]['e0'])
|
|
self.assertIn('controller', stats)
|
|
used = WS(self.port, '/stream?src=test&t='+ticket)
|
|
self.assertEqual(used.status, 403)
|
|
used.close()
|
|
self.request('/close?src=test&k='+self.token, 'POST')
|
|
revoked = WS(self.port, '/stream?src=test&r='+key)
|
|
self.assertEqual(revoked.status, 403)
|
|
revoked.close()
|
|
finally:
|
|
ws.close()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|