165 lines
9.9 KiB
Python
165 lines
9.9 KiB
Python
"""Exercise live isolated window sharing in a disposable nested compositor.
|
|
|
|
FERESE_TEST_WINDOW_STREAM=1 python3 scripts/tests/test_window_stream_isolated.py
|
|
Requires a Wayland host, PipeWire, GStreamer Python bindings and a C compiler.
|
|
Only this test's compositor, windows and streams are stopped.
|
|
"""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
from session_socket import ipc_environment, ipc_socket
|
|
import select
|
|
import signal
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
|
|
from test_inhibit_isolated import IPC
|
|
|
|
REPO = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
@unittest.skipUnless(os.environ.get("FERESE_TEST_WINDOW_STREAM") == "1", "requires Wayland and PipeWire")
|
|
class WindowStream(unittest.TestCase):
|
|
def test_isolation_resize_workspace_and_close(self):
|
|
import gi
|
|
gi.require_version("Gst", "1.0")
|
|
from gi.repository import Gst
|
|
Gst.init(None)
|
|
processes = []
|
|
pipeline = None
|
|
admin = None
|
|
|
|
def wait_for(check, message):
|
|
deadline = time.monotonic() + 10
|
|
while time.monotonic() < deadline:
|
|
value = check()
|
|
if value:
|
|
return value
|
|
time.sleep(0.02)
|
|
self.fail(message)
|
|
|
|
with tempfile.TemporaryDirectory(prefix="ferese-window-stream-") as directory:
|
|
root = Path(directory)
|
|
protocols = subprocess.check_output(["pkg-config", "--variable=pkgdatadir", "wayland-protocols"], text=True).strip()
|
|
xml = str(Path(protocols) / "stable/xdg-shell/xdg-shell.xml")
|
|
for kind, output in [("client-header", "xdg-shell-client-protocol.h"), ("private-code", "xdg-shell-protocol.c")]:
|
|
subprocess.run(["wayland-scanner", kind, xml, str(root / output)], check=True)
|
|
flags = subprocess.check_output(["pkg-config", "--cflags", "--libs", "wayland-client"], text=True).split()
|
|
subprocess.run(["cc", "-Wall", "-Wextra", "-I", str(root), str(REPO / "scripts/tests/fixtures/window-capture.c"), str(root / "xdg-shell-protocol.c"), "-o", str(root / "client"), *flags], check=True)
|
|
lock_xml = str(Path(protocols) / "staging/ext-session-lock/ext-session-lock-v1.xml")
|
|
for kind, output in [("client-header", "ext-session-lock-client-protocol.h"), ("private-code", "ext-session-lock-protocol.c")]:
|
|
subprocess.run(["wayland-scanner", kind, lock_xml, str(root / output)], check=True)
|
|
subprocess.run(["cc", "-Wall", "-Wextra", "-I", str(root), str(REPO / "scripts/tests/fixtures/session-lock.c"), str(root / "ext-session-lock-protocol.c"), "-o", str(root / "locker"), *flags], check=True)
|
|
runtime = root / "runtime"
|
|
runtime.mkdir(mode=0o700)
|
|
config = root / "config/ferese/config.kdl"
|
|
config.parent.mkdir(parents=True)
|
|
config.write_text('window-rule app-id="ferese.test.window-capture" floating=#true\nstatus { keybinding-guide #false; }\n')
|
|
host_runtime = os.environ["XDG_RUNTIME_DIR"]
|
|
display = Path(os.environ["WAYLAND_DISPLAY"])
|
|
if not display.is_absolute():
|
|
display = Path(host_runtime) / display
|
|
env = dict(os.environ, XDG_RUNTIME_DIR=str(runtime), XDG_CONFIG_HOME=str(root / "config"), XDG_DATA_HOME=str(root / "data"), WAYLAND_DISPLAY=str(display), PIPEWIRE_RUNTIME_DIR=host_runtime, FERESE_ENABLE_SCREENCOPY="1")
|
|
env.pop("WAYLAND_SOCKET", None)
|
|
env.pop("FERESE_SHELL_CONTROL_SOCKET", None)
|
|
log = (root / "test.log").open("w")
|
|
try:
|
|
compositor = subprocess.Popen([str(REPO / "target/debug/ferese"), "--backend=nested"], env=ipc_environment(env), stdout=log, stderr=log, start_new_session=True)
|
|
processes.append(compositor)
|
|
wait_for(lambda: ipc_socket(runtime).is_socket(), "Nested compositor did not start")
|
|
socket = ipc_socket(runtime)
|
|
admin = IPC(socket)
|
|
sockets = [path for path in runtime.glob("wayland-*") if not path.name.endswith(".lock")]
|
|
env["WAYLAND_DISPLAY"] = str(sockets[0])
|
|
target = subprocess.Popen([str(root / "client"), "resize"], env=ipc_environment(env), stdin=subprocess.PIPE, stdout=log, stderr=log, start_new_session=True)
|
|
processes.append(target)
|
|
windows = wait_for(lambda: [window for window in admin.call("get-windows") if window["capture_width"] == 640], "Window fixture did not map; private manager must be hidden from ordinary clients")
|
|
target_id = windows[0]["id"]
|
|
cover = subprocess.Popen([str(root / "client"), "cover"], env=ipc_environment(env), stdout=log, stderr=log, start_new_session=True)
|
|
processes.append(cover)
|
|
wait_for(lambda: len(admin.call("get-windows")) == 2, "Occluding window did not map")
|
|
helper = subprocess.Popen([str(REPO / "target/debug/xdg-desktop-portal-ferese"), "--stream-window", str(target_id), "hidden"], env=ipc_environment(env), stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=log, text=True, start_new_session=True)
|
|
processes.append(helper)
|
|
self.assertTrue(select.select([helper.stdout], [], [], 10)[0], "Stream did not announce a PipeWire node")
|
|
ready = json.loads(helper.stdout.readline())
|
|
self.assertEqual(ready["logical_size"], [640, 480])
|
|
pipeline = Gst.parse_launch(f'pipewiresrc path={ready["node"]} ! video/x-raw,format=BGRx ! appsink name=frames sync=false max-buffers=1 drop=true')
|
|
sink = pipeline.get_by_name("frames")
|
|
pipeline.set_state(Gst.State.PLAYING)
|
|
|
|
def frame(expected=None):
|
|
deadline = time.monotonic() + 8
|
|
while time.monotonic() < deadline:
|
|
sample = sink.emit("try-pull-sample", Gst.SECOND)
|
|
if sample is None:
|
|
continue
|
|
caps = sample.get_caps().get_structure(0)
|
|
size = (caps.get_value("width"), caps.get_value("height"))
|
|
if expected and size != expected:
|
|
continue
|
|
buffer = sample.get_buffer()
|
|
pixels = buffer.extract_dup(0, buffer.get_size())
|
|
width, height = size
|
|
self.assertEqual(tuple(pixels[((height // 4) * width + width // 2) * 4:][:3]), (0, 0, 255))
|
|
self.assertEqual(tuple(pixels[((height * 3 // 4) * width + width // 2) * 4:][:3]), (255, 0, 0))
|
|
return size
|
|
message = pipeline.get_bus().pop_filtered(Gst.MessageType.ERROR)
|
|
self.fail(f"Missing expected frame {expected}; pipeline error: {message.parse_error() if message else None}")
|
|
|
|
original = frame()
|
|
self.assertEqual(original, (ready["width"], ready["height"]))
|
|
admin.call("workspace", {"index": 2})
|
|
time.sleep(0.4)
|
|
frame(original)
|
|
self.assertIsNone(helper.poll(), "Inactive workspace stopped the stream")
|
|
target.stdin.write(b"r")
|
|
target.stdin.flush()
|
|
scale = ready["width"] / 640
|
|
resized = (round(480 * scale), round(320 * scale))
|
|
frame(resized)
|
|
self.assertIsNone(helper.poll(), "Resize stopped the stream")
|
|
target.terminate()
|
|
target.wait(timeout=3)
|
|
wait_for(lambda: helper.poll() is not None, "Closing the selected window left capture running")
|
|
self.assertIsNone(cover.poll(), "Unrelated window was closed")
|
|
pipeline.set_state(Gst.State.NULL)
|
|
pipeline = None
|
|
cover_id = next(window["id"] for window in admin.call("get-windows") if window["title"] == "Occluding window")
|
|
paused = subprocess.Popen([str(REPO / "target/debug/xdg-desktop-portal-ferese"), "--stream-window", str(cover_id), "hidden"], env=ipc_environment(env), stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=log, text=True, start_new_session=True)
|
|
processes.append(paused)
|
|
self.assertTrue(select.select([paused.stdout], [], [], 10)[0])
|
|
self.assertGreater(json.loads(paused.stdout.readline())["width"], 0)
|
|
locker = subprocess.Popen([str(root / "locker")], env=ipc_environment(env), stdout=subprocess.PIPE, stderr=log, text=True, start_new_session=True)
|
|
processes.append(locker)
|
|
self.assertTrue(select.select([locker.stdout], [], [], 5)[0], "Nested lock request did not complete")
|
|
self.assertIn(locker.stdout.readline().strip(), ("locked", "requested"))
|
|
wait_for(lambda: paused.poll() is not None, "Lock did not end a paused window stream")
|
|
print("PASS: isolated pixels, private manager, upright video, inactive workspace, resize renegotiation, closure and paused lock revocation")
|
|
except Exception:
|
|
print((root / "test.log").read_text())
|
|
raise
|
|
finally:
|
|
if pipeline:
|
|
pipeline.set_state(Gst.State.NULL)
|
|
if admin:
|
|
admin.close()
|
|
for process in reversed(processes):
|
|
if process.poll() is None:
|
|
os.killpg(process.pid, signal.SIGTERM)
|
|
try:
|
|
process.wait(timeout=3)
|
|
except subprocess.TimeoutExpired:
|
|
os.killpg(process.pid, signal.SIGKILL)
|
|
process.wait()
|
|
for process in processes:
|
|
for pipe in (process.stdin, process.stdout, process.stderr):
|
|
if pipe:
|
|
pipe.close()
|
|
log.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|