mirror of
https://github.com/saphid/frame-control.git
synced 2026-10-06 07:00:37 +02:00
Add native PC capture adapters behind the shared panel viewer
This commit is contained in:
saphid committed
1 parent
559caa2dd0
commit
c5fc552c60
19 files changed
+1865
-36
No files matched your search
@@ -0,0 +1,210 @@
|
||||
"""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.
|
||||
"""
|
||||
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 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_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_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)
|
||||
|
||||
|
||||
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_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()
|
||||
Reference in new issue
Block a user