mirror of
https://github.com/saphid/frame-control.git
synced 2026-10-24 21:01:43 +02:00
583 lines
28 KiB
Python
583 lines
28 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('Tested in a Windows VM; physical PCs, hardware encoding and laser clicks remain unverified.', 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_dxgi_only_bundle_captures_displays_without_optional_wgc_properties(self):
|
|
p = capture.pipeline(dict(src='display:65537', capture_api='dxgi'), 'win32', 'mfh264enc', 640, 360, 30, 1000000)
|
|
self.assertIn('monitor-handle=65537', p)
|
|
self.assertNotIn('capture-api=', p)
|
|
self.assertNotIn('show-border=', p)
|
|
self.assertIn('mfh264enc', p)
|
|
for source in (dict(src='window:42', capture_api='dxgi'),
|
|
dict(src='display:42', capture_api='dxgi ! fakesink')):
|
|
with self.assertRaises(ValueError):
|
|
capture.pipeline(source, 'win32', 'mfh264enc', 640, 360, 30, 1000000)
|
|
|
|
def test_windows_sources_follow_actual_wgc_capabilities(self):
|
|
for wgc in (False, True):
|
|
with self.subTest(wgc=wgc):
|
|
native, windows = mock.Mock(), mock.Mock()
|
|
native.has_property.return_value = wgc
|
|
windows.sources.side_effect = lambda: ([dict(src='window:12', w=800, h=600)],
|
|
[dict(src='display:4', w=1920, h=1080)])
|
|
with mock.patch.object(agent.sys, 'platform', 'win32'), \
|
|
mock.patch.object(agent, 'Windows', return_value=windows):
|
|
host = agent.Agent(native, 'test', ROOT/'ui/mac-view.html')
|
|
wins, displays = host.lists()
|
|
self.assertEqual(len(wins), int(wgc))
|
|
self.assertEqual(displays[0]['capture_api'], 'wgc' if wgc else 'dxgi')
|
|
self.assertEqual(host.source('display:4')['capture_api'], displays[0]['capture_api'])
|
|
if wgc:
|
|
self.assertEqual(host.source('window:12')['capture_api'], 'wgc')
|
|
self.assertEqual(host.window_capture_reason, '')
|
|
else:
|
|
with self.assertRaisesRegex(ValueError, 'Individual windows are unavailable'):
|
|
host.source('window:12')
|
|
self.assertIn('whole screen', host.window_capture_reason)
|
|
# Synthetic encoder/protocol checks remain available without WGC.
|
|
self.assertEqual(host.source('test')['src'], 'test')
|
|
|
|
def test_pc_card_receives_window_capture_limitation(self):
|
|
view = frame_pcview.PCView([], lambda *a: None, 'frame')
|
|
view.enabled = True
|
|
view.call = mock.Mock(side_effect=lambda path: {
|
|
'/status': dict(screen=True, accessibility=True, windowCaptureReason=agent.WINDOW_CAPTURE_UNAVAILABLE),
|
|
'/windows': dict(windows=[]), '/displays': dict(displays=[dict(src='display:4')])}[path])
|
|
with mock.patch.object(frame_pcview.sys, 'platform', 'win32'), \
|
|
mock.patch.object(frame_pcview, 'LIBRARY', mock.Mock(is_file=lambda: True)):
|
|
state = view.state()
|
|
self.assertTrue(state['available'])
|
|
self.assertEqual(state['windowCaptureReason'], agent.WINDOW_CAPTURE_UNAVAILABLE)
|
|
self.assertEqual(state['windows'], [])
|
|
self.assertEqual(state['displays'], [dict(src='display:4')])
|
|
|
|
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')
|
|
|
|
def test_native_property_probe_checks_the_instantiated_element(self):
|
|
code = """
|
|
from frame_pc_capture import Native
|
|
n = Native()
|
|
assert n.has_property('fakesink', 'sync')
|
|
assert not n.has_property('fakesink', 'not-a-gstreamer-property')
|
|
assert not n.has_property('not-a-gstreamer-element', 'sync')
|
|
"""
|
|
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=15)
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
|
|
@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()
|