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__":