Samuel Onoja
baca852ee9
Render selected toplevels through a private portal-only capture protocol, preserve logical snapshot geometry, and renegotiate PipeWire video on window resizing. Bound outstanding snapshots and retry temporary contention; stop on source closure and session locking. Validation: portal unit tests (31), capture budget unit, private-bus portal contracts, and real nested/PipeWire isolation, workspace, resize, close and paused-lock tests pass. Independent review cleared. Monitor restore integration timed out on both this build and the unchanged previous compositor under the current nested host.
164 lines
9.8 KiB
Python
164 lines
9.8 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
|
|
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=env, stdout=log, stderr=log, start_new_session=True)
|
|
processes.append(compositor)
|
|
socket = runtime / "ferese/control.sock"
|
|
wait_for(socket.exists, "Nested compositor did not start")
|
|
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=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=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=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=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=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()
|