From b105ea8ef903a4bff56da369126c41feccdb383b Mon Sep 17 00:00:00 2001 From: Samuel Onoja Date: Thu, 1 Oct 2026 07:44:36 +0100 Subject: [PATCH] Cover portal appearance with a theme IPC owner in contract tests The Settings portal reads appearance exclusively from the compositor's theme control socket and falls back to defaults when no compositor is reachable, so the contract test could no longer assert appearance from an on-disk KDL document. Add a fixtures/theme_owner.py peer that speaks the theme IPC framing and publishes schema-v2 snapshots, point the test at a private runtime socket, and clear WAYLAND_DISPLAY and WAYLAND_SOCKET so the backend cannot reach a host compositor. Assert the initial values, the SettingChanged payload after a republish, and that ReadAll matches the published snapshot, then confirm an invalid local document does not disturb portal-owned appearance. --- scripts/tests/fixtures/theme_owner.py | 127 +++++++++++++++++++++++++ scripts/tests/test_portal_contracts.py | 28 ++++-- 2 files changed, 147 insertions(+), 8 deletions(-) create mode 100644 scripts/tests/fixtures/theme_owner.py diff --git a/scripts/tests/fixtures/theme_owner.py b/scripts/tests/fixtures/theme_owner.py new file mode 100644 index 0000000..ea2e41e --- /dev/null +++ b/scripts/tests/fixtures/theme_owner.py @@ -0,0 +1,127 @@ +"""Minimal compositor theme IPC peer for portal wire-contract tests.""" +import json +import socket +import struct +import threading + + +def snapshot(appearance, accent, reduced_motion, contrast): + # A complete schema-v2 snapshot, with deliberately different effective appearance + # and surface brightness so consumers cannot infer the scheme from color tokens. + theme = { + "appearance": appearance, "requested_accent": accent, + "accessibility": {"increase_contrast": contrast, "reduce_transparency": False}, + "reduced_motion": reduced_motion, + "tokens": { + "colors": { + "surface_base": "#111111" if appearance == "light" else "#ffffff", + "surface_raised": "#1E2530", "application_background": "#171E27", + "text_primary": "#F4F7FB", "text_muted": "#8793A2", "accent": accent, + "on_accent": "#FFFFFF", "border": "#FFFFFF18", "shadow": "#00000055", + }, + "material": {"style": "solid", "opacity": 1.0, "blur_radius": 0.0, "tint_strength": 1.0}, + "geometry": { + "border_width": 1.0, "focus_ring_width": 2.0, "window_radius": 14.0, + "shell_radius": 14.0, "top_bar_height": 28.0, "top_bar_margin_top": 0, + "top_bar_margin_horizontal": 0, "top_bar_window_gap": 0, + "panel_padding": 12.0, "control_gap": 12.0, + }, + "typography": {"font_family": "Inter"}, + "background": {"path": "", "lock_path": None, "mode": "fill"}, + "surface": {"bar": {"background": "#111111", "text_primary": "#F4F7FB", "text_muted": "#8793A2"}}, + "shadow": {"soft": {"offset_y": 4.0, "blur": 18.0, "opacity": 0.2}}, + "border": {"gradient": None}, "focus_ring": {"gradient": None}, + }, + } + return {"version": 2, "revision": 1, "mode": appearance, "theme": theme, + "presented": theme, "warnings": [], "error": None, "families": [], "fallback_note": None} + + +class ThemeOwner: + def __init__(self, runtime): + path = runtime / "ferese/control.sock" + path.parent.mkdir() + self.condition = threading.Condition() + self.current = snapshot("light", "#ff8000", True, False) + self.closed = False + self.errors = [] + self.clients = [] + self.workers = [] + self.listener = socket.socket(socket.AF_UNIX) + self.listener.bind(str(path)) + self.listener.listen() + self.listener.settimeout(0.1) + self.thread = threading.Thread(target=self.accept, daemon=True) + self.thread.start() + + def publish(self, appearance, accent, reduced_motion, contrast): + with self.condition: + next_snapshot = snapshot(appearance, accent, reduced_motion, contrast) + next_snapshot["revision"] = self.current["revision"] + 1 + self.current = next_snapshot + self.condition.notify_all() + + def accept(self): + while not self.closed: + try: + client, _ = self.listener.accept() + except socket.timeout: + continue + except OSError: + if self.closed: + return + raise + self.clients.append(client) + worker = threading.Thread(target=self.serve, args=(client,), daemon=True) + self.workers.append(worker) + worker.start() + + def serve(self, client): + try: + with client, client.makefile("rwb") as stream: + while not self.closed: + header = stream.read(4) + if not header: + return + assert len(header) == 4 + size, = struct.unpack("!I", header) + assert size <= 1024 * 1024 + request = json.loads(stream.read(size)) + assert request["version"] == 1 and request["type"] == "command" + if request["command"] not in ("theme-get", "theme-watch"): + # Other portal components share this socket. Never implement + # desktop operations in the appearance-only test peer. + response = {"version": 1, "id": request["id"], "error": { + "code": "unsupported_command", "message": "Only theme IPC is available in this test", + }} + else: + with self.condition: + if request["command"] == "theme-watch": + self.condition.wait_for(lambda: self.closed or request["args"]["since"] != self.current["revision"]) + if self.closed: + return + response = {"version": 1, "id": request["id"], "result": self.current} + payload = json.dumps(response).encode() + stream.write(struct.pack("!I", len(payload)) + payload) + stream.flush() + except (BrokenPipeError, ConnectionResetError): + pass # The portal cancels its outstanding watch when it exits. + except Exception as error: + if not self.closed: + self.errors.append(error) + + def close(self): + with self.condition: + self.closed = True + self.condition.notify_all() + self.thread.join(timeout=2) + self.listener.close() + for client in self.clients: + try: + client.shutdown(socket.SHUT_RDWR) + except OSError: + pass + for worker in self.workers: + worker.join(timeout=2) + assert not self.thread.is_alive() and not any(worker.is_alive() for worker in self.workers) + assert not self.errors, self.errors diff --git a/scripts/tests/test_portal_contracts.py b/scripts/tests/test_portal_contracts.py index 2de7d3e..1b23165 100644 --- a/scripts/tests/test_portal_contracts.py +++ b/scripts/tests/test_portal_contracts.py @@ -10,6 +10,8 @@ import time import unittest import xml.etree.ElementTree as ET +from fixtures.theme_owner import ThemeOwner + REPO = Path(__file__).resolve().parents[2] BINARY = REPO / os.environ.get("FERESE_TEST_PORTAL_BINARY", "target/debug/xdg-desktop-portal-ferese") NAME = "org.freedesktop.impl.portal.desktop.ferese" @@ -39,10 +41,16 @@ def private_bus_checks(): root = Path(directory) config = root / "ferese/config.kdl" config.parent.mkdir() - config.write_text('theme { colors { surface-base "#ffffff"; accent "#ff8000"; }; }; animations { reduced-motion #true; }\n') + # Lockdown still uses this document; appearance comes exclusively from theme IPC. + config.write_text('theme { mode "dark"; }; animations { reduced-motion #false; }\n') + runtime = root / "runtime" + runtime.mkdir(mode=0o700) + theme_owner = ThemeOwner(runtime) log = (root / "backend.log").open("w") - process = subprocess.Popen([str(BINARY)], env=dict(os.environ, XDG_CONFIG_HOME=directory), - stdout=log, stderr=log, start_new_session=True) + env = dict(os.environ, XDG_CONFIG_HOME=directory, XDG_RUNTIME_DIR=str(runtime), + WAYLAND_DISPLAY="wayland-unavailable") + env.pop("WAYLAND_SOCKET", None) + process = subprocess.Popen([str(BINARY)], env=env, stdout=log, stderr=log, start_new_session=True) try: deadline = time.monotonic() + 15 while True: @@ -221,15 +229,18 @@ def private_bus_checks(): signals = [] subscription = bus.signal_subscribe(NAME, "org.freedesktop.impl.portal.Settings", "SettingChanged", PATH, None, Gio.DBusSignalFlags.NONE, lambda *args: signals.append(args[-1].unpack())) - temporary = config.with_suffix(".tmp") - temporary.write_text('theme { colors { surface-base "#111111"; accent "#0080ff"; }; }; animations { enabled #false; }\n') - temporary.replace(config) + theme_owner.publish("dark", "#0080ff", False, True) deadline = time.monotonic() + 5 - while time.monotonic() < deadline and not any(signal[1] == "color-scheme" for signal in signals): + while time.monotonic() < deadline and not {"color-scheme", "accent-color", "reduced-motion", "contrast"}.issubset({signal[1] for signal in signals}): GLib.MainContext.default().iteration(False) time.sleep(0.02) assert any(signal[1] == "color-scheme" and signal[2] == 1 for signal in signals), signals - assert any(signal[1] == "accent-color" for signal in signals), signals + changed = {event[1]: event[2] for event in signals} + assert changed["accent-color"] == (0., 128 / 255, 1.), changed + assert changed["reduced-motion"] == 0 and changed["contrast"] == 1, changed + current, = call("org.freedesktop.impl.portal.Settings", "ReadAll", GLib.Variant("(as)", (["org.freedesktop.*"],)), "(a{sa{sv}})") + assert current["org.freedesktop.appearance"] == changed, current + # Portal appearance stays with the owner even when the local document is invalid. config.write_text('theme { broken') time.sleep(1.2) scheme, = call("org.freedesktop.impl.portal.Settings", "Read", GLib.Variant("(ss)", ("org.freedesktop.appearance", "color-scheme"))) @@ -239,6 +250,7 @@ def private_bus_checks(): os.killpg(process.pid, signal.SIGTERM) process.wait(timeout=5) log.close() + theme_owner.close() if __name__ == "__main__":