From f36c3f6ae53bb38156f5055af44d7b3cdf338b65 Mon Sep 17 00:00:00 2001 From: "devin-ai-integration[bot]" <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Tue, 29 Sep 2026 23:55:27 -0700 Subject: [PATCH] feat(text_injection): add RemoteDesktop portal injection backend (#885) * feat(text_injection): add RemoteDesktop portal injection backend Wayland offers exactly one sanctioned key-injection path: org.freedesktop.portal.RemoteDesktop's NotifyKeyboardKeysym. It reaches native clients on compositors with no input-method-v2 bridge, works from inside the Flatpak sandbox without /dev/uinput or a helper daemon, and delivers keysyms so non-US layouts survive -- unlike ydotool's evdev keycodes. RemoteDesktopPortal is a synchronous client over a dedicated worker thread holding a private bus connection and GLib context, which keeps Request::Response delivery deterministic. Sessions are created lazily on first use so startup probing only reads the interface version; v2 portal sessions persist a restore token (mode 0600 under data_dir) so the permission prompt appears once. Backend selection now prefers the portal on Wayland: it is probed at runtime ahead of ydotool/wtype/xdotool in autodetection, is pinnable via VOCALINUX_FORCE_BACKEND=portal or text_injection.backend, and demotes to the previous order on runtime failure. Flatpak still needs --device=all for evdev hotkeys, but uinput is only a fallback path now. Closes #750 * fix(text_injection): resolve Greptile findings on RemoteDesktop portal injection - Move persist_mode/restore_token to SelectDevices options (portal v2 sends them there, not CreateSession); a bad token is ignored rather than sinking the session, so drop the stale-token retry paths that could issue a second Start after a user cancel - Track delivered units on RemoteDesktopPortalError so Wayland fallbacks retry only the undelivered tail instead of retyping the prefix (text chars, backspace taps, shortcut steps) - Give queued portal jobs a single deadline budget across CreateSession/SelectDevices/Start waits; on submit timeout the job is cancelled via a Gio.Cancellable plus abort checkpoints so the worker cannot keep emitting keys after the caller gave up - Close sessions whose Start never succeeded - Release held shortcut modifiers on mid-step failure - Skip the portal probe when an installed wtype/ydotool backend is explicitly pinned - Type annotations on the new signatures and touched tests --------- Co-authored-by: Jatin K Malik --- docs/USER_GUIDE.md | 20 +- packaging/flatpak/README.md | 16 +- packaging/flatpak/com.vocalinux.Vocalinux.yml | 8 +- .../text_injection/remote_desktop_portal.py | 721 +++++++++++++++++ src/vocalinux/text_injection/text_injector.py | 242 +++++- tests/conftest.py | 11 + tests/test_remote_desktop_portal.py | 740 ++++++++++++++++++ 7 files changed, 1733 insertions(+), 25 deletions(-) create mode 100644 src/vocalinux/text_injection/remote_desktop_portal.py create mode 100644 tests/test_remote_desktop_portal.py diff --git a/docs/USER_GUIDE.md b/docs/USER_GUIDE.md index f1a7d83..3feb689 100644 --- a/docs/USER_GUIDE.md +++ b/docs/USER_GUIDE.md @@ -244,13 +244,19 @@ Pin the backend explicitly in `~/.config/vocalinux/config.json`: |---|---| | `auto` | Autodetect (default; autodetection may select IBus) | | `ibus` | IBus input method; on Wayland, bypasses compositor checks and may silently do nothing in native Wayland apps | +| `portal` | RemoteDesktop portal (Wayland; the sandboxed path, asks for permission once, works under Flatpak) | | `wtype` | wtype virtual keyboard (Wayland) | | `ydotool` | ydotool uinput (Wayland; needs `ydotoold`) | | `xdotool` | xdotool (X11). On Wayland it only turns IBus off -- the Wayland tool is still picked automatically | The setting takes effect on the next start. `auto` leaves normal autodetection -in place and may select IBus. An explicit non-IBus pin (`wtype`, `ydotool`, or -`xdotool`) skips IBus selection. +in place and may select IBus. An explicit non-IBus pin (`portal`, `wtype`, +`ydotool`, or `xdotool`) skips IBus selection. + +On Wayland, when IBus is not selected, autodetection tries the RemoteDesktop +portal first -- it is the only injection path Wayland officially supports and +it needs no uinput device or helper daemon -- then `ydotool`, `wtype`, and +finally `xdotool` under XWayland. On X11 the injection tool is `xdotool` regardless of which non-`ibus` value you pin, so `xdotool` is the name to use there when IBus is unreliable in a @@ -271,7 +277,9 @@ it does not prove the backend was available or that text reached the focused application. Later logs identify a fallback when a pin was not applied. If a pinned tool is unavailable, Vocalinux warns and continues with its normal -fallback selection. `ydotool` also needs a usable `/dev/uinput` and a working -`ydotoold` setup. `xdotool` types into X11/XWayland windows, not native Wayland -windows. A live test in the target application is still the final confirmation -that text is delivered. +fallback selection. `portal` needs a desktop implementing the +`org.freedesktop.portal.RemoteDesktop` interface (GNOME and KDE Plasma do; +most wlroots compositors do not). `ydotool` also needs a usable `/dev/uinput` +and a working `ydotoold` setup. `xdotool` types into X11/XWayland windows, +not native Wayland windows. A live test in the target application is still the +final confirmation that text is delivered. diff --git a/packaging/flatpak/README.md b/packaging/flatpak/README.md index ac64f92..7f9728f 100644 --- a/packaging/flatpak/README.md +++ b/packaging/flatpak/README.md @@ -47,12 +47,16 @@ the PyPI package is wheel-only (no sdist), so a Flathub-ready VOSK build would need to compile VOSK and its native deps from source. Global keyboard shortcuts use **evdev** (`/dev/input`). -Text injection uses **wl-copy** + **ydotool Ctrl+V** (instant paste into native -Wayland apps). Character-by-character `ydotool type` is only a fallback. -`xdotool` remains for pure X11/XWayland clients. +Text injection prefers the **RemoteDesktop portal** +(`org.freedesktop.portal.RemoteDesktop` `NotifyKeyboardKeysym` — keysyms, so +non-US layouts survive) and falls back to **wl-copy** + **ydotool Ctrl+V** +(instant paste into native Wayland apps) when the portal is unavailable. +Character-by-character `ydotool type` is only a fallback. `xdotool` remains +for pure X11/XWayland clients. Permissions: `--socket=wayland` (clipboard), `--socket=x11` (xdotool fallback), -`--device=all` (evdev hotkeys + uinput; Flatpak has no narrower uinput flag). +`--device=all` (evdev hotkeys + uinput for the ydotool fallback; Flatpak has +no narrower uinput flag). ## Local Build @@ -142,8 +146,8 @@ Release `.flatpak` assets are the supported paths. - Runtime / SDK: `org.gnome.Platform//50`, `org.gnome.Sdk//50` - Mic: `--socket=pulseaudio` · GPU: `--device=dri` · models: `--share=network` -- Input: `--device=all` (evdev hotkeys + ydotool/`uinput`) -- Injection: packaged `wl-copy`, `ydotool`/`ydotoold`, plus `xdotool`/`xsel` fallback +- Input: `--device=all` (evdev hotkeys + ydotool/`uinput` fallback) +- Injection: RemoteDesktop portal first; packaged `wl-copy`, `ydotool`/`ydotoold`, plus `xdotool`/`xsel` fallback - Display: `--socket=wayland` (clipboard) and `--socket=x11` (xdotool fallback) - IBus: `--talk-name=org.freedesktop.IBus` - Tray: `--talk-name=org.kde.StatusNotifierWatcher` diff --git a/packaging/flatpak/com.vocalinux.Vocalinux.yml b/packaging/flatpak/com.vocalinux.Vocalinux.yml index 3c34c85..a593cda 100644 --- a/packaging/flatpak/com.vocalinux.Vocalinux.yml +++ b/packaging/flatpak/com.vocalinux.Vocalinux.yml @@ -16,15 +16,17 @@ finish-args: # Shared memory for X11 / XWayland. - --share=ipc # X11 for xdotool fallback; Wayland for wl-copy (instant paste into native apps). - # Injection itself uses ydotool/uinput (not the Wayland virtual-keyboard protocol). + # Injection prefers the RemoteDesktop portal (DBus, keysyms) and falls back + # to ydotool/uinput -- not the Wayland virtual-keyboard protocol. - --socket=x11 - --socket=wayland # Microphone capture (works against PulseAudio and PipeWire-pulse). - --socket=pulseaudio # GPU render node for the Vulkan-accelerated whisper.cpp build. - --device=dri - # /dev/input for global hotkeys (evdev) and /dev/uinput for ydotool (Ctrl+V / - # type). Flatpak has no narrower flag than "all" for uinput. + # /dev/input for global hotkeys (evdev) and /dev/uinput for the ydotool + # injection fallback (Ctrl+V / type) when the RemoteDesktop portal is + # unavailable. Flatpak has no narrower flag than "all" for uinput. - --device=all # IBus Unicode injection backend. - --talk-name=org.freedesktop.IBus diff --git a/src/vocalinux/text_injection/remote_desktop_portal.py b/src/vocalinux/text_injection/remote_desktop_portal.py new file mode 100644 index 0000000..47deafa --- /dev/null +++ b/src/vocalinux/text_injection/remote_desktop_portal.py @@ -0,0 +1,721 @@ +""" +RemoteDesktop portal text injection for Vocalinux. + +``org.freedesktop.portal.RemoteDesktop`` is the only key-injection path +Wayland itself sanctions: the compositor's portal backend prompts the user +once, hands back a session, and after that ``NotifyKeyboardKeysym`` injects +keysyms without ``/dev/uinput``, a helper daemon, or compositor support for +``zwp_input_method_v2``. It also works from inside a Flatpak sandbox, where +the Wayland socket may not exist at all. + +The D-Bus conversation is asynchronous: each request method returns a +``org.freedesktop.portal.Request`` object that later emits ``Response``. To +keep that deterministic this client runs a dedicated worker thread holding a +private bus connection whose signals dispatch on a private +``GLib.MainContext``; callers submit jobs and block on the result. + +Portal versions: version 2 of the interface adds ``persist_mode`` and the +``restore_token`` option, which lets a later session resume without the +permission prompt. When unavailable (version 1) the session simply has to be +re-authorized on each app start. +""" + +import logging +import os +import queue +import threading +import time +from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple + +from ..utils.paths import data_dir + +logger = logging.getLogger(__name__) + +try: + import gi + + gi.require_version("Gio", "2.0") + from gi.repository import Gio, GLib + + PORTAL_AVAILABLE = True +except (ImportError, ValueError) as e: # pragma: no cover - needs host PyGObject + logger.debug(f"Gio not available for portal injection: {e}") + PORTAL_AVAILABLE = False + Gio = None + GLib = None + + +class RemoteDesktopPortalError(RuntimeError): + """Raised when the portal cannot deliver an injection request. + + ``delivered`` counts the units that reached the compositor before the + failure -- characters for ``inject_text``, taps for ``tap_keysym`` and + steps for ``send_shortcut`` -- so the caller can retry only the + remainder through a fallback backend instead of replaying the whole + request. + """ + + def __init__(self, message: str = "", delivered: int = 0) -> None: + super().__init__(message) + self.delivered = delivered + + +_DESKTOP_BUS_NAME = "org.freedesktop.portal.Desktop" +_DESKTOP_OBJECT_PATH = "/org/freedesktop/portal/desktop" +_REMOTE_DESKTOP_IFACE = "org.freedesktop.portal.RemoteDesktop" +_REQUEST_IFACE = "org.freedesktop.portal.Request" +_SESSION_IFACE = "org.freedesktop.portal.Session" +_PROPERTIES_IFACE = "org.freedesktop.DBus.Properties" + +# org.freedesktop.portal.RemoteDesktop device types bitmask. +_DEVICE_TYPE_KEYBOARD = 1 + +# Keyboard state values for NotifyKeyboard* calls. +_KEY_RELEASED = 0 +_KEY_PRESSED = 1 + +# org.freedesktop.portal.Request::Response codes. +_RESPONSE_OK = 0 +_RESPONSE_CANCELLED = 1 +_RESPONSE_ENDED = 2 + +# persist_mode values (RemoteDesktop interface version >= 2). +_PERSIST_UNTIL_REVOKED = 2 + +_RESPONSE_POLL_INTERVAL_S = 0.005 +_REQUEST_TIMEOUT_S = 15.0 +# Start() blocks on the user clicking through the compositor's permission +# dialog, so its budget is measured in minutes rather than seconds. +_START_TIMEOUT_S = 180.0 +_SUBMIT_TIMEOUT_S = _START_TIMEOUT_S + _REQUEST_TIMEOUT_S +# A job's internal deadline lands a beat before the caller's wait expires, +# leaving room for the worker to report what it already delivered. +_JOB_DEADLINE_HEADROOM_S = 1.0 + +# XKB keysym values (X11/keysymdef.h minus the XK_ prefix). Only names used by +# ActionHandler._SHORTCUT_ACTIONS or the injector's own key paths are listed. +KEYSYM_BACKSPACE = 0xFF08 +_KEYSYMS = { + "space": 0x0020, + "tab": 0xFF09, + "return": 0xFF0D, + "enter": 0xFF0D, + "escape": 0xFF1B, + "backspace": KEYSYM_BACKSPACE, + "delete": 0xFFFF, + "home": 0xFF50, + "end": 0xFF57, + "left": 0xFF51, + "up": 0xFF52, + "right": 0xFF53, + "down": 0xFF54, +} + +_MODIFIER_KEYSYMS = { + "ctrl": 0xFFE3, # Control_L + "control": 0xFFE3, + "shift": 0xFFE1, # Shift_L + "alt": 0xFFE9, # Alt_L + "altgr": 0xFE03, # ISO_Level3_Shift + "super": 0xFFEB, # Super_L + "meta": 0xFFEB, + "win": 0xFFEB, + "logo": 0xFFEB, +} + + +def portal_keysym_for_name(name: str) -> Optional[int]: + """Return the keysym for a shortcut key/modifier name, or None. + + Accepts the same spelling as ``TextInjector._parse_shortcut`` tokens: + modifier aliases, the named keys in ``_KEYSYMS`` and single characters + (letters are keysym-identical to their codepoint). + """ + lowered = name.lower() + if lowered in _MODIFIER_KEYSYMS: + return _MODIFIER_KEYSYMS[lowered] + if lowered in _KEYSYMS: + return _KEYSYMS[lowered] + if len(name) == 1: + return char_to_keysym(name) + return None + + +def char_to_keysym(char: str) -> Optional[int]: + """Map one Unicode character to the XKB keysym for ``NotifyKeyboardKeysym``. + + XKB keysyms coincide with the codepoint for all of Latin-1, and any higher + character has a well-defined "Unicode keysym" ``0x01000000 | codepoint`` + that compositors translate back. Characters with no keysym (control codes + other than newline/tab) return None so callers can skip them. + """ + if char == "\n": + return 0xFF0D # Return + if char == "\r": + return None + if char == "\t": + return 0xFF09 # Tab + codepoint = ord(char) + if codepoint < 0x20 or 0x7F <= codepoint < 0xA0 or 0xD800 <= codepoint <= 0xDFFF: + return None + if codepoint < 0x100: + return codepoint + return 0x01000000 | codepoint + + +def _restore_token_path() -> str: + """File holding the persisted RemoteDesktop restore token, if any.""" + return os.path.join(data_dir(), "remote_desktop_restore_token") + + +def _load_restore_token() -> Optional[str]: + """Read a previously persisted restore token (empty/missing -> None).""" + try: + with open(_restore_token_path(), "r") as f: + token = f.read().strip() + except OSError: + return None + return token or None + + +def _save_restore_token(token: str) -> None: + """Persist the restore token owner-only; it is a capability.""" + try: + os.makedirs(data_dir(), exist_ok=True) + fd = os.open(_restore_token_path(), os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) + with os.fdopen(fd, "w") as f: + f.write(token + "\n") + os.chmod(_restore_token_path(), 0o600) + except OSError as e: + logger.debug(f"Could not persist RemoteDesktop restore token: {e}") + + +class RemoteDesktopPortal: + """Synchronous client for ``org.freedesktop.portal.RemoteDesktop``. + + Construction is cheap: the worker thread, private bus connection and + portal session are only created on first use (``probe`` or an injection + call). After ``close()`` the instance must not be used again. + """ + + def __init__(self) -> None: + self._lock = threading.Lock() + self._jobs: "queue.Queue" = queue.Queue() + self._worker: Optional[threading.Thread] = None + self._ready = threading.Event() + self._context: Any = None + self._conn: Any = None + self._connect_error: Optional[BaseException] = None + self._available: Optional[bool] = None + self._version = 0 + self._session_path: Optional[str] = None + self._closed = False + # Abort plumbing for the job the worker is currently running; set by + # _worker_main from the job's box and cleared when the job ends. + self._job_deadline: Optional[float] = None + self._job_cancel: Optional[threading.Event] = None + self._job_cancellable: Any = None + + @staticmethod + def supported() -> bool: + """Whether PyGObject/Gio can be imported at all on this install.""" + return PORTAL_AVAILABLE + + def probe(self) -> bool: + """Whether the RemoteDesktop portal can inject keyboard events. + + Reads the interface ``version`` property, which both confirms the + portal backend implements the interface (compositors whose portal + lacks RemoteDesktop fail this call) and tells us whether persistence + (``persist_mode``/``restore_token``) is supported. Result is cached. + """ + with self._lock: + if self._available is not None: + return self._available + try: + self._version = self._submit(self._read_version, timeout=_REQUEST_TIMEOUT_S) + self._available = True + except Exception as e: + logger.info(f"RemoteDesktop portal is not usable here: {e}") + self._available = False + self._shutdown_locked() + return self._available + + def inject_text(self, text: str) -> bool: + """Type ``text`` as keysyms. Raises RemoteDesktopPortalError on failure.""" + if not any(char_to_keysym(c) is not None for c in text): + return True + with self._lock: + self._submit(lambda: self._inject_text(text), timeout=_SUBMIT_TIMEOUT_S) + return True + + def tap_keysym(self, keysym: int, count: int = 1) -> None: + """Press and release ``keysym`` ``count`` times.""" + if count <= 0: + return + with self._lock: + self._submit(lambda: self._tap(keysym, count), timeout=_SUBMIT_TIMEOUT_S) + + def notify_keysym(self, keysym: int, pressed: bool) -> None: + """Hold (``pressed=True``) or release one keysym.""" + state = _KEY_PRESSED if pressed else _KEY_RELEASED + with self._lock: + self._submit(lambda: self._notify(keysym, state), timeout=_SUBMIT_TIMEOUT_S) + + def send_shortcut(self, steps: Sequence[Tuple[Sequence[str], str]]) -> None: + """Send shortcut steps parsed by ``TextInjector._parse_shortcut``. + + Each step presses its modifier names, taps the key, then releases the + modifiers, so ``Home+shift+End`` works as two sequential steps. + + Raises: + RemoteDesktopPortalError: when a name has no keysym or the portal + cannot deliver the events. + """ + with self._lock: + self._submit(lambda: self._shortcut(steps), timeout=_SUBMIT_TIMEOUT_S) + + def close(self) -> None: + """Close the portal session and stop the worker thread.""" + with self._lock: + session_path = self._session_path + self._session_path = None + if session_path and self._conn is not None: + try: + self._submit(lambda: self._close_session(session_path), timeout=5) + except Exception as e: + logger.debug(f"Could not close RemoteDesktop session: {e}") + self._closed = True + self._shutdown_locked() + + # ------------------------------------------------------------------ + # Worker plumbing. Everything below _submit() runs on the worker thread. + # ------------------------------------------------------------------ + + def _ensure_worker_locked(self) -> None: + if self._worker is not None and self._worker.is_alive(): + return + self._ready.clear() + self._connect_error = None + self._worker = threading.Thread( + target=self._worker_main, name="vocalinux-portal", daemon=True + ) + self._worker.start() + if not self._ready.wait(10): + raise RemoteDesktopPortalError("Timed out connecting to the session bus") + if self._connect_error is not None or self._conn is None: + raise RemoteDesktopPortalError( + f"Could not connect to the session bus: {self._connect_error}" + ) + + def _worker_main(self) -> None: + """Worker thread body: own the bus connection and run submitted jobs.""" + try: + self._context = GLib.MainContext() + self._context.push_thread_default() + address = Gio.dbus_address_get_for_bus_sync(Gio.BusType.SESSION, None) + flags = ( + Gio.DBusConnectionFlags.AUTHENTICATION_CLIENT + | Gio.DBusConnectionFlags.MESSAGE_BUS_CONNECTION + ) + self._conn = Gio.DBusConnection.new_for_address_sync(address, flags, None, None) + except Exception as e: + self._connect_error = e + finally: + self._ready.set() + + while True: + job = self._jobs.get() + if job is None: + break + func, done, box = job + self._job_deadline = box.get("deadline") + self._job_cancel = box.get("cancel") + self._job_cancellable = box.get("cancellable") + try: + box["result"] = func() + except Exception as e: + box["error"] = e + finally: + self._job_deadline = None + self._job_cancel = None + self._job_cancellable = None + done.set() + + if self._conn is not None: + try: + self._conn.close_sync(None) + except Exception: + pass + self._conn = None + self._context = None + + def _submit(self, func: Callable[[], Any], timeout: float) -> Any: + """Run ``func`` on the worker thread and return its result. + + ``func`` executes with ``self._conn``/``self._context`` available; it + may only be entered through here so bus work stays single-threaded. + """ + if self._closed: + raise RemoteDesktopPortalError("RemoteDesktop portal client is closed") + self._ensure_worker_locked() + done = threading.Event() + cancel = threading.Event() + cancellable = Gio.Cancellable.new() if Gio is not None else None + box: Dict[str, Any] = { + # The job's own budget ends slightly before this wait does, so the + # worker normally reports before the caller would ever time out. + "deadline": time.monotonic() + max(0.0, timeout - _JOB_DEADLINE_HEADROOM_S), + "cancel": cancel, + "cancellable": cancellable, + } + self._jobs.put((func, done, box)) + if not done.wait(timeout): + # The caller gives up: cancel the job so it cannot keep emitting + # portal key events after the fallback backend has started typing. + cancel.set() + if cancellable is not None: + cancellable.cancel() + # Let the aborted job report what it delivered -- or land a late + # success -- before the timeout is declared. + done.wait(_JOB_DEADLINE_HEADROOM_S) + if "result" in box: + return box["result"] + error = box.get("error") + delivered = getattr(error, "delivered", 0) if error is not None else 0 + raise RemoteDesktopPortalError("Timed out waiting for the portal", delivered=delivered) + if "error" in box: + error = box["error"] + if isinstance(error, RemoteDesktopPortalError): + raise error + raise RemoteDesktopPortalError(str(error)) from error + return box.get("result") + + def _shutdown_locked(self) -> None: + """Stop the worker thread; safe when it never started.""" + if self._worker is None: + return + # Abort a job still in flight so the worker reaches the sentinel + # instead of finishing a request nobody is waiting for. + if self._job_cancel is not None: + self._job_cancel.set() + if self._job_cancellable is not None: + try: + self._job_cancellable.cancel() + except Exception: + pass + self._jobs.put(None) + self._worker.join(10) + self._worker = None + + # --------------------------------------------------------------- + # Portal protocol. All of these run on the worker thread via _submit. + # --------------------------------------------------------------- + + def _check_job_aborted(self) -> None: + """Raise when the running job's caller gave up or its deadline passed.""" + if self._job_cancel is not None and self._job_cancel.is_set(): + raise RemoteDesktopPortalError("Portal job cancelled after its deadline") + if self._job_deadline is not None and time.monotonic() >= self._job_deadline: + raise RemoteDesktopPortalError("Portal job exceeded the submit deadline") + + def _job_time_left(self, budget: float) -> float: + """Seconds left in the running job's deadline, capped at ``budget``.""" + if self._job_deadline is None: + return budget + return min(budget, max(0.0, self._job_deadline - time.monotonic())) + + def _call_sync( + self, + interface: str, + method: str, + params: Optional[Any], + reply_type: Optional[str], + timeout_ms: int = 10000, + ) -> Any: + self._check_job_aborted() + return self._conn.call_sync( + _DESKTOP_BUS_NAME, + _DESKTOP_OBJECT_PATH, + interface, + method, + params, + GLib.VariantType(reply_type) if reply_type else None, + Gio.DBusCallFlags.NONE, + timeout_ms, + self._job_cancellable, + ) + + def _read_version(self) -> int: + """Read the RemoteDesktop interface version property.""" + result = self._call_sync( + _PROPERTIES_IFACE, + "Get", + GLib.Variant("(ss)", (_REMOTE_DESKTOP_IFACE, "version")), + "(v)", + ) + return int(result.get_child_value(0).get_variant().get_uint32()) + + def _await_response(self, request_path: str, timeout_s: float) -> Dict[str, Any]: + """Wait for the Request::Response signal on ``request_path``. + + Signals dispatch on this thread's private main context; iterating it + here is what delivers the callback, so there is no race between the + method returning the request path and the subscription being added. + """ + outcome: Dict[str, Any] = {} + + def _on_response( + _conn: Any, + _sender: str, + _path: str, + _iface: str, + _signal: str, + params: Any, + ) -> None: + outcome["response"] = params.get_child_value(0).get_uint32() + outcome["results"] = params.get_child_value(1) + + subscription = self._conn.signal_subscribe( + _DESKTOP_BUS_NAME, + _REQUEST_IFACE, + "Response", + request_path, + None, + Gio.DBusSignalFlags.NONE, + _on_response, + ) + try: + deadline = time.monotonic() + timeout_s + if self._job_deadline is not None: + deadline = min(deadline, self._job_deadline) + while "response" not in outcome: + self._check_job_aborted() + if time.monotonic() >= deadline: + raise RemoteDesktopPortalError( + f"Portal request timed out after {timeout_s:.0f}s ({request_path})" + ) + self._context.iteration(False) + time.sleep(_RESPONSE_POLL_INTERVAL_S) + finally: + self._conn.signal_unsubscribe(subscription) + + response = outcome["response"] + if response == _RESPONSE_OK: + return outcome + if response == _RESPONSE_CANCELLED: + raise RemoteDesktopPortalError("Portal request was cancelled by the user") + raise RemoteDesktopPortalError(f"Portal request ended (response={response})") + + def _request( + self, + method: str, + params: Any, + timeout_s: float = _REQUEST_TIMEOUT_S, + ) -> Any: + """Issue a portal request method and return its Response results. + + Both the method call and the Response wait draw from the running + job's deadline, so ``CreateSession`` + ``SelectDevices`` + ``Start`` + share one budget instead of each claiming its own. + """ + budget = self._job_time_left(timeout_s) + if budget <= 0: + raise RemoteDesktopPortalError(f"Portal request {method} exceeded the deadline") + result = self._call_sync( + _REMOTE_DESKTOP_IFACE, + method, + params, + "(o)", + timeout_ms=int(max(1.0, min(_REQUEST_TIMEOUT_S, budget)) * 1000), + ) + request_path = result.get_child_value(0).get_string() + outcome = self._await_response(request_path, budget) + return outcome["results"] + + @staticmethod + def _sv_options(pairs: Dict[str, Any]) -> Dict[str, Any]: + """Build an ``a{sv}`` body: strings as ``s``, everything else as ``u``.""" + options: Dict[str, Any] = {} + for key, value in pairs.items(): + if isinstance(value, str): + options[key] = GLib.Variant("s", value) + else: + options[key] = GLib.Variant("u", value) + return options + + def _ensure_session(self) -> None: + """Create/restore the RemoteDesktop session (may prompt the user).""" + if self._session_path: + return + + if self._available is None: + # An injection path can reach here without a prior probe() -- read + # the version on demand so persistence support is still detected. + self._version = self._read_version() + self._available = True + + persist = self._version >= 2 + session_token = f"vocalinux_{os.getpid()}_{int(time.time())}" + results = self._request( + "CreateSession", + GLib.Variant( + "(a{sv})", + (self._sv_options({"session_handle_token": session_token}),), + ), + ) + + session_handle = results.lookup_value( + "session_handle", GLib.VariantType("s") + ) or results.lookup_value("session_handle", GLib.VariantType("o")) + if session_handle is None: + raise RemoteDesktopPortalError("CreateSession returned no session_handle") + session_path = session_handle.get_string() + + # persist_mode and restore_token are SelectDevices options (interface + # v2). The portal consumes the token there; an invalid one is simply + # ignored and the user gets the normal prompt. + select_options: Dict[str, Any] = {"types": _DEVICE_TYPE_KEYBOARD} + if persist: + select_options["persist_mode"] = _PERSIST_UNTIL_REVOKED + restore_token = _load_restore_token() + if restore_token: + select_options["restore_token"] = restore_token + + try: + self._request( + "SelectDevices", + GLib.Variant("(oa{sv})", (session_path, self._sv_options(select_options))), + ) + start_results = self._request( + "Start", + GLib.Variant("(osa{sv})", (session_path, "", {})), + timeout_s=_START_TIMEOUT_S, + ) + except Exception: + # A session that never reached a successful Start must not linger: + # the compositor keeps it (and any grant) alive until Close. + try: + self._close_session(session_path) + except Exception as close_error: + logger.debug(f"Could not close the unfinished portal session: {close_error}") + raise + + token_variant = start_results.lookup_value("restore_token", GLib.VariantType("s")) + if persist and token_variant is not None: + token = token_variant.get_string() + if token: + _save_restore_token(token) + + self._session_path = session_path + logger.info( + "RemoteDesktop portal session started (interface v%s%s)", + self._version, + ", persisted" if persist and token_variant else "", + ) + + def _notify(self, keysym: int, state: int) -> None: + self._ensure_session() + self._notify_key_state(keysym, state) + + def _tap(self, keysym: int, count: int) -> None: + self._ensure_session() + delivered = 0 + try: + for _ in range(count): + self._check_job_aborted() + self._notify_key_state(keysym, _KEY_PRESSED) + delivered += 1 + self._notify_key_state(keysym, _KEY_RELEASED) + except RemoteDesktopPortalError as e: + e.delivered = max(e.delivered, delivered) + raise + except Exception as e: + raise RemoteDesktopPortalError(str(e), delivered=delivered) from e + + def _notify_key_state(self, keysym: int, state: int) -> None: + self._call_sync( + _REMOTE_DESKTOP_IFACE, + "NotifyKeyboardKeysym", + GLib.Variant("(oa{sv}iu)", (self._session_path, {}, keysym, state)), + None, + ) + + def _inject_text(self, text: str) -> None: + self._ensure_session() + delivered = 0 + try: + for index, char in enumerate(text): + keysym = char_to_keysym(char) + if keysym is None: + continue + self._check_job_aborted() + self._notify_key_state(keysym, _KEY_PRESSED) + delivered = index + 1 + self._notify_key_state(keysym, _KEY_RELEASED) + except RemoteDesktopPortalError as e: + e.delivered = max(e.delivered, delivered) + raise + except Exception as e: + raise RemoteDesktopPortalError(str(e), delivered=delivered) from e + + def _shortcut(self, steps: Sequence[Tuple[Sequence[str], str]]) -> None: + """Worker-side shortcut delivery; names are mapped once up front.""" + resolved: List[Tuple[List[int], int]] = [] + for modifiers, key in steps: + modifier_keysyms = [] + for name in modifiers: + keysym = _MODIFIER_KEYSYMS.get(name.lower()) + if keysym is None: + raise ValueError(f"No portal keysym for modifier '{name}'") + modifier_keysyms.append(keysym) + keysym = portal_keysym_for_name(key) + if keysym is None: + raise ValueError(f"No portal keysym for key '{key}'") + resolved.append((modifier_keysyms, keysym)) + + self._ensure_session() + delivered = 0 + for modifier_keysyms, keysym in resolved: + try: + self._check_job_aborted() + for modifier in modifier_keysyms: + self._notify_key_state(modifier, _KEY_PRESSED) + self._notify_key_state(keysym, _KEY_PRESSED) + self._notify_key_state(keysym, _KEY_RELEASED) + delivered += 1 + for modifier in reversed(modifier_keysyms): + self._notify_key_state(modifier, _KEY_RELEASED) + except RemoteDesktopPortalError as e: + e.delivered = max(e.delivered, delivered) + self._release_held_modifiers(modifier_keysyms) + raise + except Exception as e: + self._release_held_modifiers(modifier_keysyms) + raise RemoteDesktopPortalError(str(e), delivered=delivered) from e + + def _release_held_modifiers(self, modifier_keysyms: List[int]) -> None: + """Best-effort release of a step's modifiers after a mid-step failure. + + A modifier left held on the portal session would corrupt whatever the + fallback backend types next. Releases stop at the first error -- the + connection is usually already gone by then. + """ + for modifier in reversed(modifier_keysyms): + try: + self._notify_key_state(modifier, _KEY_RELEASED) + except Exception: + break + + def _close_session(self, session_path: str) -> None: + self._conn.call_sync( + _DESKTOP_BUS_NAME, + session_path, + _SESSION_IFACE, + "Close", + None, + None, + Gio.DBusCallFlags.NONE, + 5000, + None, + ) diff --git a/src/vocalinux/text_injection/text_injector.py b/src/vocalinux/text_injection/text_injector.py index ce64702..a3d437e 100644 --- a/src/vocalinux/text_injection/text_injector.py +++ b/src/vocalinux/text_injection/text_injector.py @@ -14,7 +14,7 @@ import subprocess import threading import time from enum import Enum -from typing import Optional, Tuple +from typing import List, Optional, Tuple from ..utils.host_process import host_env from ..utils.paths import config_dir @@ -25,6 +25,7 @@ from .ibus_engine import ( is_ibus_available, is_ibus_daemon_running, ) +from .remote_desktop_portal import KEYSYM_BACKSPACE, RemoteDesktopPortal, RemoteDesktopPortalError logger = logging.getLogger(__name__) @@ -106,6 +107,7 @@ class TextInjector: wayland_mode: Force Wayland compatibility mode """ self._ibus_injector: Optional[IBusTextInjector] = None + self._portal: Optional[RemoteDesktopPortal] = None self._backend_pin: Tuple[str, Optional[str]] = ("auto", None) self.environment = self._detect_environment() self._session_environment = self.environment @@ -194,6 +196,11 @@ class TextInjector: logger.info("Stopping IBus text injector") self._ibus_injector.stop() self._ibus_injector = None + portal = getattr(self, "_portal", None) + if portal is not None: + logger.info("Closing RemoteDesktop portal session") + portal.close() + self._portal = None self._ibus_ready = False def _probe_wtype_support(self) -> subprocess.CompletedProcess: @@ -258,7 +265,11 @@ class TextInjector: # Backends a user may pin explicitly, via VOCALINUX_FORCE_BACKEND or the # text_injection.backend setting. Named once so the two readers and the two # "expected ..." messages cannot drift apart as values are added. - _SELECTABLE_BACKENDS = ("ibus", "wtype", "ydotool", "xdotool") + _SELECTABLE_BACKENDS = ("ibus", "portal", "wtype", "ydotool", "xdotool") + + # The chosen Wayland injection tool; None until _check_dependencies picks + # one (readers go through getattr because __new__ tests skip __init__). + wayland_tool: Optional[str] @staticmethod def _accepted_backends_help() -> str: @@ -502,7 +513,8 @@ class TextInjector: variable exists for. Collapsing it into "nothing was set" would let the saved pin win and make the variable useless for that. - Accepts ``ibus``, ``wtype``, ``ydotool``, ``xdotool`` or ``auto``. + Accepts ``ibus``, ``portal``, ``wtype``, ``ydotool``, ``xdotool`` or + ``auto``. """ raw = os.environ.get("VOCALINUX_FORCE_BACKEND") if raw is None: @@ -567,8 +579,9 @@ class TextInjector: when the inference is wrong, and makes the two paths A/B-testable without editing code. - Accepts ``ibus``, ``wtype``, ``ydotool``, ``xdotool`` or ``auto``. Anything else is - ignored with a warning, so a typo cannot silently pin a backend. + Accepts ``ibus``, ``portal``, ``wtype``, ``ydotool``, ``xdotool`` or + ``auto``. Anything else is ignored with a warning, so a typo cannot + silently pin a backend. This covers the environment variable only. ``_backend_preference()`` combines it with the persistent ``text_injection.backend`` setting, and @@ -788,10 +801,13 @@ class TextInjector: if resolved is None or resolved == pinned: return - if pinned in self._SELECTABLE_BACKENDS and pinned != "ibus": + if pinned == "ibus": + reason = " (IBus support is not available)" if not is_ibus_available() else "" + elif pinned == "portal": + # The portal is not a binary; probe it rather than PATH. + reason = "" if self._portal_probe() else " (RemoteDesktop portal is not available)" + elif pinned in self._SELECTABLE_BACKENDS: reason = f" ({pinned} is not installed)" if not shutil.which(pinned) else "" - elif pinned == "ibus" and not is_ibus_available(): - reason = " (IBus support is not available)" else: reason = "" logger.warning( @@ -904,19 +920,39 @@ class TextInjector: wtype_available = shutil.which("wtype") is not None ydotool_available = shutil.which("ydotool") is not None xdotool_available = shutil.which("xdotool") is not None + # A pinned wtype/ydotool that is installed is honoured without + # autodetection, so probing the portal -- a session-bus round + # trip on a private worker -- is skipped for those users. + pinned_tool_present = (forced == "wtype" and wtype_available) or ( + forced == "ydotool" and ydotool_available + ) + portal_available = not pinned_tool_present and self._portal_probe() if ydotool_available: _warn_if_ydotool_globally_enabled() # Prefer ydotool when the daemon is (or can be) ready. Flatpak ships # ydotool for native Wayland typing; wtype needs a Wayland socket. - if forced == "wtype" and wtype_available: + # The RemoteDesktop portal sits ahead of both in autodetection: it + # is the only injection path Wayland sanctions, works inside the + # Flatpak sandbox without /dev/uinput, and reaches native clients + # on compositors with no input-method-v2 bridge. + if forced == "portal" and portal_available: + self._select_portal_backend( + "%s=portal: using RemoteDesktop portal for Wayland injection", + pin_source, + ) + elif forced == "wtype" and wtype_available: self.wayland_tool = "wtype" logger.info("%s=wtype: using wtype for Wayland injection", pin_source) elif forced == "ydotool" and ydotool_available: self._ensure_ydotoold() self.wayland_tool = "ydotool" logger.info("%s=ydotool: using ydotool for Wayland injection", pin_source) + elif portal_available: + self._select_portal_backend( + "Using the RemoteDesktop portal for Wayland text injection" + ) elif ydotool_available and self._ensure_ydotoold(): self.wayland_tool = "ydotool" logger.info("Using ydotool for Wayland text injection") @@ -956,6 +992,144 @@ class TextInjector: if ibus_requested: self._start_ibus_initialization() + def _portal_probe(self) -> bool: + """Check whether the RemoteDesktop portal can inject keyboard events. + + The client is created lazily and cached on first success so + ``wayland_tool == "portal"`` reuses the same connection/worker. The + probe itself only asks the portal for its interface version; the + session handshake (and its one-time permission prompt) is deferred + to the first actual injection. + """ + # getattr: __new__-constructed injectors (tests) lack __init__ attrs. + portal = getattr(self, "_portal", None) + if portal is None: + if not RemoteDesktopPortal.supported(): + return False + try: + portal = RemoteDesktopPortal() + except Exception as e: + logger.debug(f"Could not create the RemoteDesktop portal client: {e}") + return False + try: + if portal.probe(): + self._portal = portal + return True + except Exception as e: + logger.debug(f"RemoteDesktop portal probe failed: {e}") + return False + + def _select_portal_backend(self, log_message: str, *args: object) -> None: + """Select the RemoteDesktop portal as the Wayland injection tool.""" + self.wayland_tool = "portal" + if self.environment == DesktopEnvironment.WAYLAND_XDOTOOL: + # The portal reaches native Wayland clients, not just XWayland, so + # a Flatpak session that demoted to XWayland upgrades back. + self.environment = DesktopEnvironment.WAYLAND + logger.info(log_message, *args) + + def _try_inject_with_portal(self, text: str) -> Tuple[bool, str]: + """Send text through the RemoteDesktop portal. + + Returns ``(True, "")`` on success. On failure the second element is + the part of ``text`` the portal had not typed yet -- the whole string + when it never got started, the tail when it failed mid-string -- so + the fallback backend does not duplicate delivered characters. + """ + portal = getattr(self, "_portal", None) + if portal is None: + return False, text + try: + return bool(portal.inject_text(text)), "" + except RemoteDesktopPortalError as e: + logger.warning(f"RemoteDesktop portal injection failed: {e}") + return False, text[e.delivered :] + except Exception as e: + logger.warning(f"RemoteDesktop portal injection failed: {e}") + return False, text + + def _try_portal_keypress(self, keysym: int, count: int = 1) -> Tuple[bool, int]: + """Tap one keysym ``count`` times through the portal. + + Returns ``(True, 0)`` on success; on failure the taps still owed, so + the fallback does not re-send presses the portal already delivered. + """ + portal = getattr(self, "_portal", None) + if portal is None: + return False, count + try: + portal.tap_keysym(keysym, count) + return True, 0 + except RemoteDesktopPortalError as e: + logger.warning(f"RemoteDesktop portal key event failed: {e}") + return False, max(0, count - e.delivered) + except Exception as e: + logger.warning(f"RemoteDesktop portal key event failed: {e}") + return False, count + + def _try_portal_shortcut( + self, steps: List[Tuple[List[str], str]] + ) -> Tuple[bool, List[Tuple[List[str], str]]]: + """Send parsed shortcut steps through the portal. + + Returns ``(True, [])`` on success; on failure the steps still to + send, so the fallback does not re-run steps that already fired. + """ + portal = getattr(self, "_portal", None) + if portal is None: + return False, steps + try: + portal.send_shortcut(steps) + return True, [] + except RemoteDesktopPortalError as e: + logger.warning(f"RemoteDesktop portal shortcut failed: {e}") + return False, list(steps[e.delivered :]) + except Exception as e: + logger.warning(f"RemoteDesktop portal shortcut failed: {e}") + return False, list(steps) + + def _demote_portal_backend(self) -> bool: + """Drop the portal for this run and pick the next Wayland tool. + + Mirrors the startup preference: ydotool with a ready daemon, then + ydotool alone when wtype is absent, then wtype, then the XWayland + fallback. Returns False when nothing else exists, leaving the + clipboard fallback to report the failure. + """ + portal = getattr(self, "_portal", None) + if portal is not None: + try: + portal.close() + except Exception as e: + logger.debug(f"Could not close RemoteDesktop portal client: {e}") + self._portal = None + with self._state_lock: + self.wayland_tool = None + + ydotool_available = shutil.which("ydotool") is not None + wtype_available = shutil.which("wtype") is not None + xdotool_available = shutil.which("xdotool") is not None + + tool = None + if ydotool_available and self._ensure_ydotoold(): + tool = "ydotool" + elif ydotool_available and not wtype_available: + tool = "ydotool" + elif wtype_available: + tool = "wtype" + elif xdotool_available: + with self._state_lock: + self.environment = DesktopEnvironment.WAYLAND_XDOTOOL + logger.warning("RemoteDesktop portal failed; using xdotool/XWayland fallback") + return True + else: + return False + + with self._state_lock: + self.wayland_tool = tool + logger.warning(f"RemoteDesktop portal failed; falling back to {tool}") + return True + def _start_ibus_initialization(self) -> None: if self._ibus_injector is None or self._ibus_init_thread is not None: return @@ -2274,9 +2448,25 @@ class TextInjector: """ # Wait for the shortcut modifier(s) to be released first, so a held Alt # doesn't turn the Ctrl+V paste into Ctrl+Alt+V (nothing pastes) or - # modify typed keys. + # modify typed keys. The portal's keysyms are equally affected by a + # still-held physical modifier, so the wait stays ahead of every tool. self._wait_for_modifiers_released() + if self.wayland_tool == "portal": + portal_ok, remaining_text = self._try_inject_with_portal(text) + if portal_ok: + return + if not self._demote_portal_backend(): + raise RuntimeError( + "RemoteDesktop portal injection failed and no Wayland fallback is available" + ) + if not remaining_text: + return + if self.environment == DesktopEnvironment.WAYLAND_XDOTOOL: + self._inject_with_xdotool(remaining_text) + return + text = remaining_text + # Prefer clipboard + paste chord for ydotool: one chord instead of # per-character evdev keycodes (those follow physical US key positions # and scramble text on other layouts). The paste chord is not @@ -2399,6 +2589,19 @@ class TextInjector: # wait it out first -- same reason and placement as _inject_with_wayland_tool. self._wait_for_modifiers_released() + if self.wayland_tool == "portal": + portal_ok, remaining_steps = self._try_portal_shortcut(steps) + if portal_ok: + return True + if not self._demote_portal_backend(): + return False + if not remaining_steps: + return True + steps = remaining_steps + if self.environment == DesktopEnvironment.WAYLAND_XDOTOOL: + return self._inject_shortcut_with_xdotool(self._steps_to_shortcut(steps)) + # Fall through to the demoted wtype/ydotool. + if self.wayland_tool == "wtype": # wtype does support combinations: -M presses a modifier, -k types a # key, -m releases it. Modifiers must be released explicitly -- wtype @@ -2593,6 +2796,15 @@ class TextInjector: raise ValueError("no key found") return steps + @staticmethod + def _steps_to_shortcut(steps: List[Tuple[List[str], str]]) -> str: + """Render parsed ``_parse_shortcut`` steps back as a ``+``-joined string. + + Used to hand only the unsent tail of a partially delivered shortcut + to the xdotool fallback, which takes a shortcut string, not steps. + """ + return "+".join("+".join(modifiers + [key]) for modifiers, key in steps) + def press_backspace(self, count: int) -> bool: """Send ``count`` real BackSpace key events. @@ -2651,6 +2863,16 @@ class TextInjector: logger.error("No virtual-keyboard tool available to send backspaces") return False + if tool == "portal": + portal_ok, remaining = self._try_portal_keypress(KEYSYM_BACKSPACE, count) + if portal_ok: + return True + if not self._demote_portal_backend(): + return False + if remaining <= 0: + return True + return self.press_backspace(remaining) + if tool == "ydotool" and not self._ensure_ydotoold(): # Checked on every call, not just when the tool is first resolved: # wayland_tool is cached from startup, so a daemon that died since diff --git a/tests/conftest.py b/tests/conftest.py index 0e6963f..0d81a20 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -140,6 +140,17 @@ def _suppress_desktop_notifications(request): yield +@pytest.fixture(autouse=True) +def _no_remote_desktop_portal(monkeypatch): + """Keep the RemoteDesktop portal probe off in tests. + + ``gi`` is a MagicMock here, so the module's import-time availability flag + is True and an unpatched probe would spawn its worker thread against a + mocked D-Bus. Tests that exercise the portal re-patch this themselves. + """ + monkeypatch.setattr("vocalinux.text_injection.remote_desktop_portal.PORTAL_AVAILABLE", False) + + @pytest.fixture(autouse=True) def _reset_shared_config_manager(): """Drop the process-wide ConfigManager between tests. diff --git a/tests/test_remote_desktop_portal.py b/tests/test_remote_desktop_portal.py new file mode 100644 index 0000000..9f3b3ed --- /dev/null +++ b/tests/test_remote_desktop_portal.py @@ -0,0 +1,740 @@ +"""Tests for the RemoteDesktop portal injection backend and its selection.""" + +import threading +from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple +from unittest.mock import MagicMock, patch + +import pytest + +from vocalinux.text_injection import remote_desktop_portal as rdp +from vocalinux.text_injection import text_injector as ti +from vocalinux.text_injection.remote_desktop_portal import ( + KEYSYM_BACKSPACE, + RemoteDesktopPortal, + RemoteDesktopPortalError, + char_to_keysym, + portal_keysym_for_name, +) +from vocalinux.text_injection.text_injector import DesktopEnvironment, TextInjector + +# --------------------------------------------------------------------------- +# Minimal GLib/Gio stand-ins. conftest replaces ``gi`` with a MagicMock, so the +# portal client runs against these fakes: a Variant carrying a real Python +# value tree, a connection that answers portal Requests synchronously, and a +# MainContext that delivers them on iteration(). +# --------------------------------------------------------------------------- + + +class _FakeVariantType: + def __init__(self, signature: str) -> None: + self.signature = signature + + +class _FakeVariant: + """Tiny GLib.Variant stand-in wrapping a Python value.""" + + def __init__(self, fmt: Optional[str], value: Any) -> None: + self.format = fmt + self.value = value + + def get_child_value(self, index: int) -> "_FakeVariant": + child = self.value[index] + if isinstance(child, _FakeVariant): + return child + return _FakeVariant(None, child) + + def get_variant(self) -> "_FakeVariant": + return ( + self.value if isinstance(self.value, _FakeVariant) else _FakeVariant(None, self.value) + ) + + def get_string(self) -> str: + return str(self.value) + + def get_uint32(self) -> int: + return int(self.value) + + def get_int32(self) -> int: + return int(self.value) + + def lookup_value( + self, key: str, variant_type: Optional[_FakeVariantType] + ) -> Optional["_FakeVariant"]: + child = self.value.get(key) if isinstance(self.value, dict) else None + if child is None: + return None + sig = variant_type.signature if variant_type is not None else None + if isinstance(child, _FakeVariant): + if sig is not None and child.format != sig: + return None + return child + if sig in ("s", "o") and not isinstance(child, str): + return None + if sig in ("u", "i", "b", "d", "t", "x") and not isinstance(child, (int, float)): + return None + return _FakeVariant(sig, child) + + +class _FakeGLib: + """Namespace covering the GLib surface the portal client touches.""" + + Variant = _FakeVariant + VariantType = _FakeVariantType + + +class _FakePortalConn: + """Session-bus double: records calls and answers portal Requests. + + Request methods (CreateSession/SelectDevices/Start) queue a Response that + ``_FakeContext.iteration`` dispatches once the client has subscribed — + exactly how the real portal behaves. + """ + + def __init__( + self, + version: Optional[int] = 2, + results: Optional[Dict[str, Any]] = None, + cancels: Optional[Dict[str, int]] = None, + notify_fail_at: Optional[int] = None, + ) -> None: + self.version = version + self.results = dict(results or {}) + self.cancels = dict(cancels or {}) # method -> remaining cancelled replies + # Raise inside NotifyKeyboardKeysym once this many notifies landed. + self.notify_fail_at = notify_fail_at + self.calls: List[Tuple[str, str, Any]] = [] + self.notifies: List[Tuple[int, int]] = [] + self.closed_sessions: List[Any] = [] + self.subscriptions: Dict[str, Any] = {} + self.pending: List[Tuple[str, int, Dict[str, Any]]] = [] + self._next_request = 0 + + def call_sync( + self, + bus_name: Any, + object_path: Any, + interface: str, + method: str, + params: Any, + reply_type: Any, + flags: Any, + timeout_ms: int, + cancellable: Any, + ) -> Any: + self.calls.append((interface, method, object_path, params)) + if interface == rdp._PROPERTIES_IFACE: + if self.version is None: + raise RuntimeError("RemoteDesktop interface not implemented") + return _FakeVariant("(v)", (_FakeVariant("u", self.version),)) + if interface == rdp._SESSION_IFACE: + self.closed_sessions.append(object_path) + return None + if method == "NotifyKeyboardKeysym": + if self.notify_fail_at is not None and len(self.notifies) >= self.notify_fail_at: + raise RuntimeError("session-bus connection lost mid-injection") + self.notifies.append( + ( + params.get_child_value(2).get_int32(), + params.get_child_value(3).get_uint32(), + ) + ) + return None + if method in ("CreateSession", "SelectDevices", "Start"): + self._next_request += 1 + request_path = f"/org/freedesktop/portal/request/{self._next_request}" + remaining = self.cancels.get(method, 0) + code = rdp._RESPONSE_CANCELLED if remaining else rdp._RESPONSE_OK + if remaining: + self.cancels[method] = remaining - 1 + self.pending.append((request_path, code, dict(self.results))) + return _FakeVariant("(o)", (request_path,)) + raise AssertionError(f"Unexpected portal call: {interface}.{method}") + + def signal_subscribe( + self, + sender: Any, + interface: Any, + member: Any, + object_path: Any, + arg0: Any, + flags: Any, + callback: Any, + ) -> int: + self.subscriptions[object_path] = callback + return len(self.subscriptions) + + def signal_unsubscribe(self, subscription_id: int) -> None: + pass + + +class _FakeContext: + """MainContext double: delivers queued Responses on iteration.""" + + def __init__(self, conn: _FakePortalConn) -> None: + self._conn = conn + + def iteration(self, may_block: bool) -> bool: + pending, self._conn.pending = self._conn.pending, [] + for request_path, code, results in pending: + callback = self._conn.subscriptions.get(request_path) + if callback is not None: + callback( + self._conn, + "org.freedesktop.portal.Desktop", + request_path, + rdp._REQUEST_IFACE, + "Response", + _FakeVariant("(ua{sv})", (code, results)), + ) + return True + + +@pytest.fixture() +def portal_client( + monkeypatch: pytest.MonkeyPatch, +) -> Callable[[_FakePortalConn], RemoteDesktopPortal]: + """Build a RemoteDesktopPortal wired to a fake bus; returns (portal, conn).""" + + def _run_submit(self: RemoteDesktopPortal, func: Callable[[], Any], timeout: float) -> Any: + try: + return func() + except RemoteDesktopPortalError: + raise + except Exception as e: + raise RemoteDesktopPortalError(str(e)) from e + + def _make(conn: _FakePortalConn) -> RemoteDesktopPortal: + monkeypatch.setattr(rdp, "GLib", _FakeGLib) + monkeypatch.setattr(RemoteDesktopPortal, "_submit", _run_submit) + portal = RemoteDesktopPortal() + portal._conn = conn + portal._context = _FakeContext(conn) + return portal + + return _make + + +_SESSION_RESULTS = { + "session_handle": "/org/freedesktop/portal/session/1", + "restore_token": "token-abc", +} + + +class TestKeysymMapping: + """Character/shortcut-name to keysym conversion.""" + + def test_latin1_and_ascii(self) -> None: + assert char_to_keysym("a") == 0x61 + assert char_to_keysym("A") == 0x41 + assert char_to_keysym("é") == 0xE9 + assert char_to_keysym("1") == 0x31 + + def test_unicode_above_latin1_uses_unicode_keysym(self) -> None: + assert char_to_keysym("€") == 0x01000000 | 0x20AC + assert char_to_keysym("\u4e2d") == 0x01000000 | 0x4E2D + + def test_whitespace_maps_to_named_keysyms(self) -> None: + assert char_to_keysym("\n") == 0xFF0D + assert char_to_keysym("\t") == 0xFF09 + assert char_to_keysym("\r") is None + + def test_unrepresentable_chars_return_none(self) -> None: + assert char_to_keysym("\x00") is None + assert char_to_keysym("\x7f") is None + assert char_to_keysym("\ud800") is None + + def test_portal_keysym_for_name(self) -> None: + assert portal_keysym_for_name("ctrl") == 0xFFE3 + assert portal_keysym_for_name("SUPER") == 0xFFEB + assert portal_keysym_for_name("backspace") == KEYSYM_BACKSPACE + assert portal_keysym_for_name("z") == 0x7A + assert portal_keysym_for_name("not-a-key") is None + + +PortalFactory = Callable[[_FakePortalConn], RemoteDesktopPortal] + + +class TestPortalClient: + """The synchronous client against the faked portal protocol.""" + + def test_supported_tracks_gi_availability(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(rdp, "PORTAL_AVAILABLE", False) + assert RemoteDesktopPortal.supported() is False + monkeypatch.setattr(rdp, "PORTAL_AVAILABLE", True) + assert RemoteDesktopPortal.supported() is True + + def test_probe_reads_interface_version_and_caches(self, portal_client: PortalFactory) -> None: + conn = _FakePortalConn(version=2) + portal = portal_client(conn) + assert portal.probe() is True + assert portal.probe() is True + get_calls = [c for c in conn.calls if c[0] == rdp._PROPERTIES_IFACE] + assert len(get_calls) == 1 + + def test_probe_fails_when_interface_missing(self, portal_client: PortalFactory) -> None: + conn = _FakePortalConn(version=None) + portal = portal_client(conn) + assert portal.probe() is False + + def test_inject_text_emits_keysyms_after_one_session( + self, portal_client: PortalFactory + ) -> None: + conn = _FakePortalConn(version=2, results=_SESSION_RESULTS) + portal = portal_client(conn) + assert portal.inject_text("hé") is True + + methods = [c[1] for c in conn.calls] + assert methods[:4] == [ + "Get", + "CreateSession", + "SelectDevices", + "Start", + ] + assert conn.notifies == [ + (0x68, 1), + (0x68, 0), + (0xE9, 1), + (0xE9, 0), + ] + + # A second injection reuses the session rather than re-prompting. + conn.calls.clear() + assert portal.inject_text("a") is True + assert [c[1] for c in conn.calls] == ["NotifyKeyboardKeysym"] * 2 + + def test_inject_text_persists_restore_token(self, portal_client: PortalFactory) -> None: + conn = _FakePortalConn(version=2, results=_SESSION_RESULTS) + portal = portal_client(conn) + portal.inject_text("a") + assert rdp._load_restore_token() == "token-abc" + + def test_version1_session_skips_persistence( + self, portal_client: PortalFactory, monkeypatch: pytest.MonkeyPatch + ) -> None: + conn = _FakePortalConn(version=1, results=_SESSION_RESULTS) + saved: List[str] = [] + monkeypatch.setattr(rdp, "_save_restore_token", saved.append) + portal = portal_client(conn) + portal.inject_text("a") + create = [c for c in conn.calls if c[1] == "CreateSession"][0] + assert "persist_mode" not in create[3].get_child_value(0).value + select = [c for c in conn.calls if c[1] == "SelectDevices"][0] + select_options = select[3].get_child_value(1).value + assert "persist_mode" not in select_options + assert "restore_token" not in select_options + assert saved == [] + + def test_restore_token_is_sent_to_select_devices( + self, portal_client: PortalFactory, monkeypatch: pytest.MonkeyPatch + ) -> None: + # restore_token is a SelectDevices option on interface v2; sending it + # in CreateSession leaves it unread and the user is prompted again. + monkeypatch.setattr(rdp, "_load_restore_token", lambda: "old-token") + conn = _FakePortalConn(version=2, results=_SESSION_RESULTS) + portal = portal_client(conn) + portal.inject_text("a") + create = [c for c in conn.calls if c[1] == "CreateSession"][0] + create_options = create[3].get_child_value(0).value + assert "restore_token" not in create_options + select = [c for c in conn.calls if c[1] == "SelectDevices"][0] + select_options = select[3].get_child_value(1).value + assert select_options["restore_token"].value == "old-token" + assert select_options["persist_mode"].value == 2 + + def test_cancelled_start_is_not_retried( + self, portal_client: PortalFactory, monkeypatch: pytest.MonkeyPatch + ) -> None: + # Dismissing the permission dialog must be respected: no second Start + # on the same session, and the saved restore token is kept. + monkeypatch.setattr(rdp, "_load_restore_token", lambda: "old-token") + saved: List[str] = [] + monkeypatch.setattr(rdp, "_save_restore_token", saved.append) + conn = _FakePortalConn(version=2, results=_SESSION_RESULTS, cancels={"Start": 5}) + portal = portal_client(conn) + with pytest.raises(RemoteDesktopPortalError): + portal.inject_text("a") + starts = [c for c in conn.calls if c[1] == "Start"] + assert len(starts) == 1 + assert saved == [] + + def test_failed_session_is_closed( + self, portal_client: PortalFactory, monkeypatch: pytest.MonkeyPatch + ) -> None: + # A session whose Start never succeeds is closed so the compositor + # does not keep a half-granted session alive. + monkeypatch.setattr(rdp, "_load_restore_token", lambda: "old-token") + conn = _FakePortalConn(version=2, results=_SESSION_RESULTS, cancels={"Start": 1}) + portal = portal_client(conn) + with pytest.raises(RemoteDesktopPortalError): + portal.inject_text("a") + assert conn.closed_sessions == ["/org/freedesktop/portal/session/1"] + + def test_inject_text_reports_delivered_count(self, portal_client: PortalFactory) -> None: + # 'b' press lands but its release fails: the character was typed, so + # the error reports it and the caller retries only the tail. + conn = _FakePortalConn(version=1, results=_SESSION_RESULTS, notify_fail_at=3) + portal = portal_client(conn) + with pytest.raises(RemoteDesktopPortalError) as exc_info: + portal.inject_text("abc") + assert exc_info.value.delivered == 2 + + def test_user_cancel_raises(self, portal_client: PortalFactory) -> None: + conn = _FakePortalConn(version=2, cancels={"Start": 5}) + portal = portal_client(conn) + with pytest.raises(RemoteDesktopPortalError): + portal.inject_text("a") + + def test_session_handle_as_object_path_is_accepted(self, portal_client: PortalFactory) -> None: + conn = _FakePortalConn( + version=1, + results={"session_handle": _FakeVariant("o", "/session/o")}, + ) + portal = portal_client(conn) + assert portal.inject_text("a") is True + + def test_tap_keysym_press_releases_count_times(self, portal_client: PortalFactory) -> None: + conn = _FakePortalConn(version=1, results=_SESSION_RESULTS) + portal = portal_client(conn) + portal.tap_keysym(KEYSYM_BACKSPACE, 3) + assert ( + conn.notifies + == [ + (KEYSYM_BACKSPACE, 1), + (KEYSYM_BACKSPACE, 0), + ] + * 3 + ) + + def test_tap_keysym_reports_delivered_count(self, portal_client: PortalFactory) -> None: + conn = _FakePortalConn(version=1, results=_SESSION_RESULTS, notify_fail_at=2) + portal = portal_client(conn) + with pytest.raises(RemoteDesktopPortalError) as exc_info: + portal.tap_keysym(KEYSYM_BACKSPACE, 5) + # One full tap (press+release) landed; the second press failed. + assert exc_info.value.delivered == 1 + + def test_send_shortcut_wraps_key_with_modifiers(self, portal_client: PortalFactory) -> None: + conn = _FakePortalConn(version=1, results=_SESSION_RESULTS) + portal = portal_client(conn) + portal.send_shortcut([(["ctrl"], "a"), ([], "b")]) + assert conn.notifies == [ + (0xFFE3, 1), # ctrl down + (0x61, 1), # a down + (0x61, 0), # a up + (0xFFE3, 0), # ctrl up + (0x62, 1), # b down + (0x62, 0), # b up + ] + + def test_send_shortcut_unknown_key_raises(self, portal_client: PortalFactory) -> None: + conn = _FakePortalConn(version=1, results=_SESSION_RESULTS) + portal = portal_client(conn) + with pytest.raises(RemoteDesktopPortalError): + portal.send_shortcut([([], "flying-car")]) + + def test_close_closes_portal_session(self, portal_client: PortalFactory) -> None: + conn = _FakePortalConn(version=1, results=_SESSION_RESULTS) + portal = portal_client(conn) + portal.inject_text("a") + portal.close() + assert conn.closed_sessions == ["/org/freedesktop/portal/session/1"] + + +# --------------------------------------------------------------------------- +# Backend selection inside TextInjector. +# --------------------------------------------------------------------------- + +_TI = "vocalinux.text_injection.text_injector" + + +def _stub_portal(available: bool = True) -> Tuple[MagicMock, MagicMock]: + """A RemoteDesktopPortal stand-in: (class mock, shared instance mock).""" + instance = MagicMock() + instance.probe.return_value = available + cls = MagicMock() + cls.supported.return_value = True + cls.return_value = instance + return cls, instance + + +def _wayland_injector( + monkeypatch: pytest.MonkeyPatch, + *, + tools: Dict[str, Optional[str]], + portal: MagicMock, + forced: Optional[str] = None, + extra_env: Optional[Dict[str, Optional[str]]] = None, +) -> TextInjector: + """Construct a TextInjector on a mocked Wayland session. + + ``extra_env`` values of None remove the variable instead of setting it. + """ + env: Dict[str, Optional[str]] = { + "XDG_SESSION_TYPE": "wayland", + "WAYLAND_DISPLAY": "wayland-0", + } + env.update(extra_env or {}) + if forced: + env["VOCALINUX_FORCE_BACKEND"] = forced + for key, value in env.items(): + if value is None: + monkeypatch.delenv(key, raising=False) + else: + monkeypatch.setenv(key, value) + + monkeypatch.setattr("shutil.which", lambda cmd: tools.get(cmd)) + monkeypatch.setattr(f"{_TI}.is_ibus_available", lambda: False) + monkeypatch.setattr(f"{_TI}.RemoteDesktopPortal", portal) + run = MagicMock() + run.return_value.returncode = 0 + run.return_value.stderr = "" + monkeypatch.setattr("subprocess.run", run) + return TextInjector() + + +class TestPortalBackendSelection: + """Where the portal lands in Wayland backend selection.""" + + def test_auto_prefers_portal_over_ydotool_wtype(self, monkeypatch: pytest.MonkeyPatch) -> None: + cls, _ = _stub_portal(available=True) + injector = _wayland_injector( + monkeypatch, + tools={"wtype": "/usr/bin/wtype", "ydotool": "/usr/bin/ydotool"}, + portal=cls, + ) + assert injector.wayland_tool == "portal" + assert injector.environment == DesktopEnvironment.WAYLAND + + def test_pin_portal_selects_it(self, monkeypatch: pytest.MonkeyPatch) -> None: + cls, _ = _stub_portal(available=True) + injector = _wayland_injector( + monkeypatch, + tools={"wtype": "/usr/bin/wtype"}, + portal=cls, + forced="portal", + ) + assert injector.wayland_tool == "portal" + + def test_pin_wtype_beats_portal(self, monkeypatch: pytest.MonkeyPatch) -> None: + cls, _ = _stub_portal(available=True) + injector = _wayland_injector( + monkeypatch, + tools={"wtype": "/usr/bin/wtype"}, + portal=cls, + forced="wtype", + ) + assert injector.wayland_tool == "wtype" + + def test_pinned_tool_skips_portal_probe(self, monkeypatch: pytest.MonkeyPatch) -> None: + # With an installed backend explicitly pinned, the portal must not + # even be probed -- no worker thread, no session-bus round trip. + cls, _ = _stub_portal(available=True) + injector = _wayland_injector( + monkeypatch, + tools={"wtype": "/usr/bin/wtype"}, + portal=cls, + forced="wtype", + ) + assert injector.wayland_tool == "wtype" + cls.assert_not_called() + + def test_uninstalled_pin_still_probes_portal(self, monkeypatch: pytest.MonkeyPatch) -> None: + # A pin the system cannot honour falls back to autodetection, which + # still includes the portal. + cls, _ = _stub_portal(available=True) + injector = _wayland_injector( + monkeypatch, + tools={"ydotool": "/usr/bin/ydotool"}, + portal=cls, + forced="wtype", + ) + assert injector.wayland_tool == "portal" + + def test_portal_unavailable_falls_back_to_ydotool( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + cls, _ = _stub_portal(available=False) + with ( + patch.object(TextInjector, "_is_ydotoold_running", return_value=True), + patch.object(TextInjector, "_uinput_usable", return_value=True), + ): + injector = _wayland_injector( + monkeypatch, tools={"ydotool": "/usr/bin/ydotool"}, portal=cls + ) + assert injector.wayland_tool == "ydotool" + + def test_pin_portal_unavailable_falls_back(self, monkeypatch: pytest.MonkeyPatch) -> None: + cls, _ = _stub_portal(available=False) + injector = _wayland_injector( + monkeypatch, + tools={"wtype": "/usr/bin/wtype"}, + portal=cls, + forced="portal", + ) + assert injector.wayland_tool == "wtype" + + def test_flatpak_xdotool_session_upgrades_to_portal( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Flatpak without a Wayland socket starts WAYLAND_XDOTOOL; the portal + still reaches native clients, so selection upgrades the environment.""" + cls, _ = _stub_portal(available=True) + injector = _wayland_injector( + monkeypatch, + tools={}, + portal=cls, + extra_env={ + "WAYLAND_DISPLAY": None, + "FLATPAK_ID": "com.vocalinux.Vocalinux", + "DISPLAY": ":99", + }, + ) + assert injector.wayland_tool == "portal" + assert injector.environment == DesktopEnvironment.WAYLAND + + def test_x11_never_probes_the_portal(self, monkeypatch: pytest.MonkeyPatch) -> None: + cls, instance = _stub_portal(available=True) + injector = _wayland_injector( + monkeypatch, + tools={"xdotool": "/usr/bin/xdotool"}, + portal=cls, + extra_env={ + "XDG_SESSION_TYPE": "x11", + "WAYLAND_DISPLAY": None, + "DISPLAY": ":0", + }, + ) + assert injector.environment == DesktopEnvironment.X11 + cls.assert_not_called() + instance.probe.assert_not_called() + + +def _bare_injector(**attrs: Any) -> TextInjector: + """A TextInjector without __init__, carrying only what the test needs.""" + injector = TextInjector.__new__(TextInjector) + injector._state_lock = threading.Lock() + injector._ibus_injector = None + injector._ibus_ready = False + injector._ibus_init_failed = False + injector._ibus_init_thread = None + injector._portal = None + for key, value in attrs.items(): + setattr(injector, key, value) + return injector + + +class TestPortalInjectionPaths: + """Portal dispatch inside the Wayland injection paths.""" + + def test_inject_text_routes_to_portal(self) -> None: + injector = _bare_injector(environment=DesktopEnvironment.WAYLAND, wayland_tool="portal") + injector._portal = MagicMock() + injector._portal.inject_text.return_value = True + with ( + patch.object(injector, "_wait_for_modifiers_released"), + patch("subprocess.run") as mock_run, + ): + injector._inject_with_wayland_tool("hello") + injector._portal.inject_text.assert_called_once_with("hello") + mock_run.assert_not_called() + + def test_inject_demotes_to_wtype_on_portal_failure( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + injector = _bare_injector(environment=DesktopEnvironment.WAYLAND, wayland_tool="portal") + injector._portal = MagicMock() + injector._portal.inject_text.side_effect = RemoteDesktopPortalError("boom") + monkeypatch.setattr("shutil.which", lambda c: "/usr/bin/wtype" if c == "wtype" else None) + with ( + patch.object(injector, "_wait_for_modifiers_released"), + patch("subprocess.run") as mock_run, + ): + injector._inject_with_wayland_tool("hello") + assert injector.wayland_tool == "wtype" + mock_run.assert_called_once() + assert mock_run.call_args[0][0][:1] == ["wtype"] + + def test_inject_without_any_fallback_raises(self, monkeypatch: pytest.MonkeyPatch) -> None: + injector = _bare_injector(environment=DesktopEnvironment.WAYLAND, wayland_tool="portal") + injector._portal = MagicMock() + injector._portal.inject_text.side_effect = RemoteDesktopPortalError("boom") + monkeypatch.setattr("shutil.which", lambda c: None) + with patch.object(injector, "_wait_for_modifiers_released"): + with pytest.raises(RuntimeError): + injector._inject_with_wayland_tool("hello") + + def test_press_backspace_uses_portal_tap(self) -> None: + injector = _bare_injector(environment=DesktopEnvironment.WAYLAND, wayland_tool="portal") + injector._portal = MagicMock() + with patch.object(injector, "_wait_for_modifiers_released"): + assert injector.press_backspace(2) is True + injector._portal.tap_keysym.assert_called_once_with(KEYSYM_BACKSPACE, 2) + + def test_press_backspace_demotes_and_recurses(self, monkeypatch: pytest.MonkeyPatch) -> None: + injector = _bare_injector(environment=DesktopEnvironment.WAYLAND, wayland_tool="portal") + injector._portal = MagicMock() + injector._portal.tap_keysym.side_effect = RemoteDesktopPortalError("boom") + monkeypatch.setattr("shutil.which", lambda c: "/usr/bin/wtype" if c == "wtype" else None) + with ( + patch.object(injector, "_wait_for_modifiers_released"), + patch("subprocess.run") as mock_run, + ): + assert injector.press_backspace(2) is True + assert injector.wayland_tool == "wtype" + assert mock_run.call_args[0][0] == ["wtype", "-k", "BackSpace", "-k", "BackSpace"] + + def test_inject_retries_only_undelivered_text(self, monkeypatch: pytest.MonkeyPatch) -> None: + injector = _bare_injector(environment=DesktopEnvironment.WAYLAND, wayland_tool="portal") + injector._portal = MagicMock() + injector._portal.inject_text.side_effect = RemoteDesktopPortalError("boom", delivered=2) + monkeypatch.setattr("shutil.which", lambda c: "/usr/bin/wtype" if c == "wtype" else None) + with ( + patch.object(injector, "_wait_for_modifiers_released"), + patch("subprocess.run") as mock_run, + ): + injector._inject_with_wayland_tool("hello") + # The portal typed "he" before failing; the fallback gets only "llo". + assert mock_run.call_args[0][0] == ["wtype", "llo"] + + def test_backspace_retries_only_remaining_taps(self, monkeypatch: pytest.MonkeyPatch) -> None: + injector = _bare_injector(environment=DesktopEnvironment.WAYLAND, wayland_tool="portal") + injector._portal = MagicMock() + injector._portal.tap_keysym.side_effect = RemoteDesktopPortalError("boom", delivered=1) + monkeypatch.setattr("shutil.which", lambda c: "/usr/bin/wtype" if c == "wtype" else None) + with ( + patch.object(injector, "_wait_for_modifiers_released"), + patch("subprocess.run") as mock_run, + ): + assert injector.press_backspace(3) is True + # One press was already delivered through the portal. + assert mock_run.call_args[0][0] == ["wtype", "-k", "BackSpace", "-k", "BackSpace"] + + def test_shortcut_retries_only_remaining_steps(self, monkeypatch: pytest.MonkeyPatch) -> None: + injector = _bare_injector(environment=DesktopEnvironment.WAYLAND, wayland_tool="portal") + injector._portal = MagicMock() + injector._portal.send_shortcut.side_effect = RemoteDesktopPortalError("boom", delivered=1) + monkeypatch.setattr("shutil.which", lambda c: "/usr/bin/wtype" if c == "wtype" else None) + with ( + patch.object(injector, "_wait_for_modifiers_released"), + patch("subprocess.run") as mock_run, + ): + # Two steps: [([], "Home"), (["shift"], "End")]; the first fired. + assert injector._inject_shortcut_with_wayland_tool("Home+shift+End") is True + assert mock_run.call_args[0][0] == ["wtype", "-M", "shift", "-k", "End", "-m", "shift"] + + def test_shortcut_routes_to_portal(self) -> None: + injector = _bare_injector(environment=DesktopEnvironment.WAYLAND, wayland_tool="portal") + injector._portal = MagicMock() + with patch.object(injector, "_wait_for_modifiers_released"): + assert injector._inject_shortcut_with_wayland_tool("ctrl+a") is True + injector._portal.send_shortcut.assert_called_once() + steps = injector._portal.send_shortcut.call_args[0][0] + assert steps == [(["ctrl"], "a")] + + def test_stop_closes_portal(self) -> None: + injector = _bare_injector(environment=DesktopEnvironment.WAYLAND) + portal = MagicMock() + injector._portal = portal + injector.stop() + portal.close.assert_called_once() + assert injector._portal is None