1#!/usr/bin/env python3
2import json
3import os
4from pathlib import Path
5import pty
6import select
7import shlex
8import struct
9import subprocess
10import tempfile
11import time
12import tty
13
14source = Path(__file__).resolve().parent
15subprocess.run(['python3', str(source / 'build.py'), '--platform', 'linux'], check=True)
16schema = json.loads((source / 'protocol.json').read_text())
17ids = {key: value['id'] for key, value in schema['packets'].items()}
18flags = shlex.split(subprocess.check_output(['pkg-config', '--cflags', '--libs', 'gio-unix-2.0', 'libpipewire-0.3', 'x11', 'xfixes', 'xext'], text=True))
19with tempfile.TemporaryDirectory() as temporary:
20 binary = Path(temporary) / 'linux-tests'
21 subprocess.run(['cc', '-std=gnu99', '-Wall', '-Wextra', '-Werror', '-g', '-fsanitize=address,undefined', '-no-pie', '-I', str(source / 'build'), str(source / 'linux-tests.c'), str(source / 'cursor.c'), *flags, '-o', str(binary)], check=True)
22 subprocess.run([str(binary)], check=True)
23 fixture = Path(temporary) / 'fixture.c'
24 fixture.write_text('''#include <X11/Xlib.h>
25#include <X11/cursorfont.h>
26#include <unistd.h>
27int main(void) {
28Display *d = XOpenDisplay(0); if (!d) return 1;
29Window root = DefaultRootWindow(d);
30XSetWindowBackground(d, root, 0x314159); XClearWindow(d, root);
31XDefineCursor(d, root, XCreateFontCursor(d, XC_left_ptr));
32XWarpPointer(d, None, root, 0, 0, 0, 0, 40, 40); XSync(d, 0);
33sleep(2); XDefineCursor(d, root, XCreateFontCursor(d, XC_xterm)); XSync(d, 0);
34sleep(10); XCloseDisplay(d); return 0;
35}''')
36 helper = Path(temporary) / 'fixture'
37 subprocess.run(['cc', str(fixture), '-lX11', '-o', str(helper)], check=True)
38 xvfb = subprocess.Popen(['Xvfb', '-displayfd', '1', '-screen', '0', '160x100x24', '-nolisten', 'tcp'], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
39 display = ':' + xvfb.stdout.readline().strip()
40 env = {key: value for key, value in os.environ.items() if key not in ('WAYLAND_DISPLAY', 'XDG_SESSION_TYPE', 'DBUS_SESSION_BUS_ADDRESS')}
41 env['DISPLAY'] = display
42 fixture_process = subprocess.Popen([str(helper)], env=env)
43 master, slave = pty.openpty(); tty.setraw(slave)
44 agent = subprocess.Popen([str(source / 'build/snowglobe-guest'), os.ttyname(slave)], env=env, stderr=subprocess.PIPE)
45 pending = bytearray()
46 def send(kind, data=b''):
47 os.write(master, struct.pack('!4sII', schema['magic'].encode(), ids[kind], len(data)) + data)
48 def packet(timeout=5):
49 deadline = time.monotonic() + timeout
50 while True:
51 if len(pending) >= 12:
52 magic, kind, length = struct.unpack('!4sII', pending[:12])
53 assert magic == schema['magic'].encode()
54 if len(pending) >= 12 + length:
55 payload = bytes(pending[12:12 + length]); del pending[:12 + length]
56 return kind, payload
57 remaining = deadline - time.monotonic()
58 if remaining <= 0 or not select.select([master], [], [], remaining)[0]: raise TimeoutError((agent.poll(), agent.stderr.read() if agent.poll() is not None else b''))
59 pending.extend(os.read(master, 65536))
60 try:
61 kind, payload = packet(); assert kind == ids['capabilities'] and struct.unpack('!II', payload) == (0, 1)
62 send('stream', struct.pack('!I', 1)); send('refresh')
63 cursors = set(); first_frame = None
64 deadline = time.monotonic() + 6
65 while time.monotonic() < deadline and (len(cursors) < 2 or first_frame is None):
66 kind, payload = packet()
67 if kind == ids['cursor']:
68 cursor_id, w, h, hot_x, hot_y, frames = struct.unpack('!6I', payload[:24])
69 assert w and h and hot_x < w and hot_y < h and frames == 1
70 cursors.add(payload[28:28 + w * h * 4])
71 if kind == ids['screen']:
72 w, h, x, y, rw, rh = struct.unpack('!6I', payload[:24]); assert (w, h, x, y, rw, rh) == (160, 100, 0, 0, 160, 100)
73 first_frame = payload[24:]
74 assert all(first_frame[i:i + 3] == bytes.fromhex('594131') for i in range(0, len(first_frame), 4)), 'Cursor leaked into screen pixels'
75 assert kind not in (ids['windows'], ids['surface'])
76 assert len(cursors) >= 2 and first_frame is not None
77 send('stop')
78 while packet()[0] != ids['stop']: pass
79 send('refresh'); send('stream', struct.pack('!I', 1))
80 seen = set()
81 deadline = time.monotonic() + 5
82 while time.monotonic() < deadline and not {ids['cursor'], ids['screen']} <= seen: seen.add(packet()[0])
83 assert {ids['cursor'], ids['screen']} <= seen
84 finally:
85 for process in (agent, fixture_process, xvfb):
86 process.terminate(); process.wait(timeout=5)
87 os.close(master); os.close(slave)
88print('Linux cursor metadata bounds, alpha, dirty frames, fragmented stop, X11 shape changes, cursor-free video, and reconnect passed')