soubarnak
ea7be5f111
The fork takes its own name everywhere a user can see it: binaries (voculinux, voculinux-gui), desktop entry, autostart, tray, D-Bus namespace (com.voculinux.Voculinux, no alias for the old name), config/data/cache dirs, single-instance lock, IBus engine identity, tray icons, resource prefixes, VOCULINUX_* environment names, and the update-checker user agent. The internal Python module stays vocalinux and the distribution metadata keeps the upstream URLs as provenance. On first run a previous Vocalinux install's config.json, custom dictionary, and portal restore token are copied once into the new locations (new config always wins, no merge, no write-back); models re-resolve through the pinned-checksum download path. An offline naming-sweep test proves no old-name strings remain on user-visible surfaces outside the allowlist. Refs #18
1173 lines
48 KiB
Python
1173 lines
48 KiB
Python
"""Tests for the RemoteDesktop portal injection backend and its selection."""
|
|
|
|
import logging
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import threading
|
|
from types import SimpleNamespace
|
|
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,
|
|
errors: Optional[Dict[str, int]] = None,
|
|
returned_paths: Optional[Dict[str, str]] = None,
|
|
) -> None:
|
|
self.version = version
|
|
self.results = dict(results or {})
|
|
self.cancels = dict(cancels or {}) # method -> remaining cancelled replies
|
|
self.errors = dict(errors or {}) # method -> remaining raised calls
|
|
self.returned_paths = dict(returned_paths or {})
|
|
# Raise inside NotifyKeyboardKeysym once this many notifies landed.
|
|
self.notify_fail_at = notify_fail_at
|
|
self.calls: List[Tuple[str, str, Any, Any]] = []
|
|
self.notifies: List[Tuple[int, int]] = []
|
|
self.closed_sessions: List[Any] = []
|
|
self.subscriptions: Dict[str, Any] = {}
|
|
self.subscription_paths: Dict[int, str] = {}
|
|
self.unsubscribed: List[int] = []
|
|
self.unsubscribed_paths: List[str] = []
|
|
self.pending: List[Tuple[str, int, Dict[str, Any]]] = []
|
|
self.exported_request_paths: set[str] = set()
|
|
self.request_subscriptions_at_call: List[Tuple[str, str, bool]] = []
|
|
self._next_subscription = 0
|
|
|
|
def get_unique_name(self) -> str:
|
|
return ":1.571"
|
|
|
|
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"):
|
|
options = params.value[-1]
|
|
if isinstance(options, _FakeVariant):
|
|
options = options.value
|
|
token_variant = options.get("handle_token")
|
|
token = token_variant.get_string() if token_variant is not None else "t"
|
|
predicted_path = rdp._predicted_request_path(self.get_unique_name(), token)
|
|
request_path = self.returned_paths.get(method, predicted_path)
|
|
self.request_subscriptions_at_call.append(
|
|
(method, predicted_path, predicted_path in self.subscriptions)
|
|
)
|
|
remaining_errors = self.errors.get(method, 0)
|
|
if remaining_errors:
|
|
self.errors[method] = remaining_errors - 1
|
|
raise RuntimeError(f"call failed: {method}")
|
|
|
|
collision = request_path in self.exported_request_paths
|
|
self.exported_request_paths.add(request_path)
|
|
remaining = self.cancels.get(method, 0)
|
|
code = rdp._RESPONSE_CANCELLED if remaining else rdp._RESPONSE_OK
|
|
if remaining:
|
|
self.cancels[method] = remaining - 1
|
|
if not collision:
|
|
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._next_subscription += 1
|
|
subscription_id = self._next_subscription
|
|
self.subscriptions[object_path] = callback
|
|
self.subscription_paths[subscription_id] = object_path
|
|
return subscription_id
|
|
|
|
def signal_unsubscribe(self, subscription_id: int) -> None:
|
|
self.unsubscribed.append(subscription_id)
|
|
path = self.subscription_paths.pop(subscription_id)
|
|
self.unsubscribed_paths.append(path)
|
|
self.subscriptions.pop(path, None)
|
|
|
|
|
|
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",
|
|
}
|
|
|
|
|
|
def _request_options(call: Tuple[str, str, Any, Any]) -> Dict[str, _FakeVariant]:
|
|
"""Return the options dictionary passed to a portal request method."""
|
|
options = call[3].value[-1]
|
|
return options.value if isinstance(options, _FakeVariant) else options
|
|
|
|
|
|
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_reset_session_closes_old_path_and_allows_session_recreation(
|
|
self, portal_client: PortalFactory
|
|
) -> None:
|
|
conn = _FakePortalConn(version=2, results=_SESSION_RESULTS)
|
|
portal = portal_client(conn)
|
|
assert portal.inject_text("a") is True
|
|
old_session_path = portal._session_path
|
|
assert old_session_path is not None
|
|
first_create = next(call for call in conn.calls if call[1] == "CreateSession")
|
|
first_token = _request_options(first_create)["session_handle_token"].get_string()
|
|
|
|
portal.reset_session()
|
|
|
|
assert portal._session_path is None
|
|
assert conn.closed_sessions == [old_session_path]
|
|
assert portal._closed is False
|
|
assert portal.inject_text("b") is True
|
|
|
|
creates = [call for call in conn.calls if call[1] == "CreateSession"]
|
|
second_token = _request_options(creates[1])["session_handle_token"].get_string()
|
|
assert len(creates) == 2
|
|
assert first_token != second_token
|
|
assert portal._closed is False
|
|
|
|
def test_reset_session_without_session_is_a_noop(self, portal_client: PortalFactory) -> None:
|
|
conn = _FakePortalConn()
|
|
portal = portal_client(conn)
|
|
|
|
portal.reset_session()
|
|
|
|
assert conn.closed_sessions == []
|
|
assert portal._session_path is None
|
|
assert portal._closed is False
|
|
|
|
def test_requests_use_distinct_tokens_and_subscribe_before_call(
|
|
self, portal_client: PortalFactory
|
|
) -> None:
|
|
conn = _FakePortalConn(version=2, results=_SESSION_RESULTS)
|
|
portal = portal_client(conn)
|
|
portal.inject_text("a")
|
|
|
|
calls = [
|
|
call for call in conn.calls if call[1] in ("CreateSession", "SelectDevices", "Start")
|
|
]
|
|
options = [_request_options(call) for call in calls]
|
|
tokens = [entry["handle_token"].get_string() for entry in options]
|
|
assert len(tokens) == 3
|
|
assert len(set(tokens)) == 3
|
|
assert all(re.fullmatch(r"[A-Za-z0-9_]+", token) for token in tokens)
|
|
assert [method for method, _, _ in conn.request_subscriptions_at_call] == [
|
|
"CreateSession",
|
|
"SelectDevices",
|
|
"Start",
|
|
]
|
|
assert all(subscribed for _, _, subscribed in conn.request_subscriptions_at_call)
|
|
assert conn.subscriptions == {}
|
|
assert len(conn.unsubscribed) == len(calls)
|
|
|
|
def test_session_handle_tokens_are_unique_between_instances(
|
|
self, portal_client: PortalFactory
|
|
) -> None:
|
|
session_tokens: List[str] = []
|
|
for _ in range(2):
|
|
conn = _FakePortalConn(version=1, results=_SESSION_RESULTS)
|
|
portal = portal_client(conn)
|
|
portal.inject_text("a")
|
|
create_call = next(call for call in conn.calls if call[1] == "CreateSession")
|
|
session_tokens.append(
|
|
_request_options(create_call)["session_handle_token"].get_string()
|
|
)
|
|
assert session_tokens[0] != session_tokens[1]
|
|
|
|
def test_subscribes_to_returned_path_when_it_differs(
|
|
self, portal_client: PortalFactory
|
|
) -> None:
|
|
actual_path = "/org/freedesktop/portal/request/legacy/request_42"
|
|
conn = _FakePortalConn(
|
|
version=1,
|
|
results=_SESSION_RESULTS,
|
|
returned_paths={"Start": actual_path},
|
|
)
|
|
portal = portal_client(conn)
|
|
assert portal.inject_text("a") is True
|
|
assert actual_path in conn.unsubscribed_paths
|
|
assert conn.subscriptions == {}
|
|
|
|
def test_retry_after_cancelled_start_uses_fresh_request_paths(
|
|
self, portal_client: PortalFactory
|
|
) -> None:
|
|
conn = _FakePortalConn(
|
|
version=2,
|
|
results=_SESSION_RESULTS,
|
|
cancels={"Start": 1},
|
|
)
|
|
portal = portal_client(conn)
|
|
with pytest.raises(RemoteDesktopPortalError):
|
|
portal.inject_text("a")
|
|
assert portal.inject_text("a") is True
|
|
|
|
request_paths = [path for _, path, _ in conn.request_subscriptions_at_call]
|
|
assert len(request_paths) == 6
|
|
assert len(set(request_paths)) == len(request_paths)
|
|
|
|
def test_request_subscription_is_removed_when_call_raises(
|
|
self, portal_client: PortalFactory
|
|
) -> None:
|
|
conn = _FakePortalConn(errors={"CreateSession": 1})
|
|
portal = portal_client(conn)
|
|
with pytest.raises(RemoteDesktopPortalError, match="call failed: CreateSession"):
|
|
portal.inject_text("a")
|
|
assert conn.subscriptions == {}
|
|
assert conn.unsubscribed == [1]
|
|
|
|
def test_long_wait_logs_debug_progress(
|
|
self,
|
|
portal_client: PortalFactory,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
caplog: pytest.LogCaptureFixture,
|
|
) -> None:
|
|
conn = _FakePortalConn()
|
|
portal = portal_client(conn)
|
|
elapsed = [0.0]
|
|
|
|
def monotonic() -> float:
|
|
return elapsed[0]
|
|
|
|
def sleep(_seconds: float) -> None:
|
|
elapsed[0] += 5.1
|
|
|
|
monkeypatch.setattr(rdp, "time", SimpleNamespace(monotonic=monotonic, sleep=sleep))
|
|
path = "/org/freedesktop/portal/desktop/request/1_571/waiting"
|
|
with caplog.at_level(logging.DEBUG, logger=rdp.__name__):
|
|
with pytest.raises(RemoteDesktopPortalError, match=path):
|
|
portal._await_response("Start", path, {}, 10.0)
|
|
assert "still waiting for Start Response on" in caplog.text
|
|
assert path in caplog.text
|
|
|
|
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["VOCULINUX_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.voculinux.Voculinux",
|
|
"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
|
|
|
|
|
|
def _injector_for_portal_failure(portal: Any) -> TextInjector:
|
|
"""Build a TextInjector with the collaborators used by inject_text()."""
|
|
injector = _bare_injector(environment=DesktopEnvironment.WAYLAND, wayland_tool="portal")
|
|
injector._abort_injections = threading.Event()
|
|
injector._portal = portal
|
|
injector._log_current_window_info = MagicMock()
|
|
injector._wait_for_modifiers_released = MagicMock()
|
|
injector._copy_to_clipboard = MagicMock(return_value=True)
|
|
injector._show_clipboard_fallback_notification = MagicMock()
|
|
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_flatpak_keeps_portal_when_ydotoold_is_unavailable(
|
|
self, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
injector = _bare_injector(environment=DesktopEnvironment.WAYLAND, wayland_tool="portal")
|
|
portal = MagicMock()
|
|
portal.inject_text.side_effect = RemoteDesktopPortalError("boom")
|
|
injector._portal = portal
|
|
injector._ensure_ydotoold = MagicMock(return_value=False)
|
|
monkeypatch.setattr(
|
|
"shutil.which", lambda cmd: "/usr/bin/ydotool" if cmd == "ydotool" else None
|
|
)
|
|
|
|
with (
|
|
patch.dict(
|
|
os.environ,
|
|
{"FLATPAK_ID": "com.voculinux.Voculinux"},
|
|
clear=True,
|
|
),
|
|
patch.object(injector, "_wait_for_modifiers_released"),
|
|
):
|
|
with pytest.raises(RuntimeError):
|
|
injector._inject_with_wayland_tool("hello")
|
|
with pytest.raises(RuntimeError):
|
|
injector._inject_with_wayland_tool("again")
|
|
|
|
assert injector.wayland_tool == "portal"
|
|
assert injector._portal is portal
|
|
portal.close.assert_not_called()
|
|
assert portal.inject_text.call_count == 2
|
|
assert portal.reset_session.call_count == 2
|
|
assert injector._ensure_ydotoold.call_count == 2
|
|
|
|
def test_flatpak_keep_branch_resets_portal_session(
|
|
self, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
injector = _bare_injector(environment=DesktopEnvironment.WAYLAND, wayland_tool="portal")
|
|
portal = MagicMock()
|
|
injector._portal = portal
|
|
monkeypatch.setattr("shutil.which", lambda _cmd: None)
|
|
|
|
with patch.dict(
|
|
os.environ,
|
|
{"FLATPAK_ID": "com.voculinux.Voculinux"},
|
|
clear=True,
|
|
):
|
|
assert injector._demote_portal_backend() is False
|
|
|
|
assert injector.wayland_tool == "portal"
|
|
assert injector._portal is portal
|
|
portal.reset_session.assert_called_once()
|
|
|
|
def test_flatpak_demotes_to_ydotool_only_when_ydotoold_is_ready(
|
|
self, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
injector = _bare_injector(environment=DesktopEnvironment.WAYLAND, wayland_tool="portal")
|
|
portal = MagicMock()
|
|
injector._portal = portal
|
|
injector._ensure_ydotoold = MagicMock(return_value=True)
|
|
monkeypatch.setattr(
|
|
"shutil.which", lambda cmd: "/usr/bin/ydotool" if cmd == "ydotool" else None
|
|
)
|
|
|
|
with patch.dict(
|
|
os.environ,
|
|
{"FLATPAK_ID": "com.voculinux.Voculinux"},
|
|
clear=True,
|
|
):
|
|
assert injector._demote_portal_backend() is True
|
|
|
|
assert injector.wayland_tool == "ydotool"
|
|
assert injector._portal is None
|
|
portal.close.assert_called_once()
|
|
injector._ensure_ydotoold.assert_called_once()
|
|
|
|
def test_flatpak_backspace_keeps_portal_when_ydotoold_is_unavailable(
|
|
self, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
injector = _bare_injector(environment=DesktopEnvironment.WAYLAND, wayland_tool="portal")
|
|
injector._abort_injections = threading.Event()
|
|
portal = MagicMock()
|
|
portal.tap_keysym.side_effect = RemoteDesktopPortalError("boom")
|
|
injector._portal = portal
|
|
injector._ensure_ydotoold = MagicMock(return_value=False)
|
|
monkeypatch.setattr(
|
|
"shutil.which", lambda cmd: "/usr/bin/ydotool" if cmd == "ydotool" else None
|
|
)
|
|
|
|
with (
|
|
patch.dict(
|
|
os.environ,
|
|
{"FLATPAK_ID": "com.voculinux.Voculinux"},
|
|
clear=True,
|
|
),
|
|
patch.object(injector, "_wait_for_modifiers_released"),
|
|
):
|
|
assert injector.press_backspace(1) is False
|
|
assert injector.press_backspace(1) is False
|
|
|
|
assert injector.wayland_tool == "portal"
|
|
assert injector._portal is portal
|
|
portal.close.assert_not_called()
|
|
assert portal.tap_keysym.call_count == 2
|
|
|
|
def test_flatpak_keep_portal_copies_only_undelivered_text(
|
|
self, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
portal = MagicMock()
|
|
portal.inject_text.side_effect = RemoteDesktopPortalError("boom", delivered=3)
|
|
injector = _injector_for_portal_failure(portal)
|
|
monkeypatch.setattr("shutil.which", lambda _cmd: None)
|
|
|
|
with patch.dict(
|
|
os.environ,
|
|
{"FLATPAK_ID": "com.voculinux.Voculinux"},
|
|
clear=True,
|
|
):
|
|
assert injector.inject_text("hello world") is False
|
|
|
|
assert injector.last_typed_count == 3
|
|
injector._copy_to_clipboard.assert_called_once_with("lo world")
|
|
portal.reset_session.assert_called_once()
|
|
|
|
def test_flatpak_keep_portal_does_not_copy_when_delivery_count_is_unknown(
|
|
self, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
portal = MagicMock()
|
|
portal.inject_text.side_effect = RemoteDesktopPortalError("boom", delivered=None)
|
|
injector = _injector_for_portal_failure(portal)
|
|
monkeypatch.setattr("shutil.which", lambda _cmd: None)
|
|
|
|
with patch.dict(
|
|
os.environ,
|
|
{"FLATPAK_ID": "com.voculinux.Voculinux"},
|
|
clear=True,
|
|
):
|
|
assert injector.inject_text("hello world") is False
|
|
|
|
assert injector.last_typed_count == -1
|
|
injector._copy_to_clipboard.assert_not_called()
|
|
portal.reset_session.assert_called_once()
|
|
|
|
def test_flatpak_keep_portal_copies_full_text_when_nothing_was_delivered(
|
|
self, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
portal = MagicMock()
|
|
portal.inject_text.side_effect = RemoteDesktopPortalError("boom", delivered=0)
|
|
injector = _injector_for_portal_failure(portal)
|
|
monkeypatch.setattr("shutil.which", lambda _cmd: None)
|
|
|
|
with patch.dict(
|
|
os.environ,
|
|
{"FLATPAK_ID": "com.voculinux.Voculinux"},
|
|
clear=True,
|
|
):
|
|
assert injector.inject_text("hello world") is True
|
|
|
|
injector._copy_to_clipboard.assert_called_once_with("hello world")
|
|
portal.reset_session.assert_called_once()
|
|
|
|
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_partial_type_rebases_onto_portal_prefix(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""A fallback's typed count must include what the portal delivered."""
|
|
injector = _bare_injector(environment=DesktopEnvironment.WAYLAND, wayland_tool="portal")
|
|
injector._portal = MagicMock()
|
|
# The portal typed 2 chars before the error; the fallback gets "x"*248.
|
|
injector._portal.inject_text.side_effect = RemoteDesktopPortalError("boom", delivered=2)
|
|
monkeypatch.setattr("shutil.which", lambda c: "/usr/bin/wtype" if c == "wtype" else None)
|
|
|
|
calls = []
|
|
|
|
def wtype_side_effect(cmd: list, **kwargs: object) -> MagicMock:
|
|
calls.append(cmd)
|
|
if len(calls) == 2:
|
|
raise subprocess.CalledProcessError(1, cmd, stderr="wedged")
|
|
return MagicMock(returncode=0, stderr="")
|
|
|
|
with (
|
|
patch.object(injector, "_wait_for_modifiers_released"),
|
|
patch("subprocess.run", side_effect=wtype_side_effect),
|
|
):
|
|
with pytest.raises(ti._PartiallyTyped) as exc_info:
|
|
injector._inject_with_wayland_tool("x" * 250)
|
|
# The error count is against the original 250-char text, so the
|
|
# caller clips the typed prefix instead of replaying it.
|
|
assert exc_info.value.typed == 2 + 200
|
|
|
|
def test_inject_untracked_timeout_replays_nothing(
|
|
self, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""A portal timeout with no delivered count must not retype anything."""
|
|
injector = _bare_injector(environment=DesktopEnvironment.WAYLAND, wayland_tool="portal")
|
|
injector._portal = MagicMock()
|
|
injector._portal.inject_text.side_effect = RemoteDesktopPortalError(
|
|
"Timed out waiting for the portal", delivered=None
|
|
)
|
|
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,
|
|
):
|
|
with pytest.raises(subprocess.TimeoutExpired):
|
|
injector._inject_with_wayland_tool("hello")
|
|
# The broken portal is demoted, but no fallback retypes the text.
|
|
assert injector.wayland_tool == "wtype"
|
|
mock_run.assert_not_called()
|
|
|
|
def test_backspace_untracked_timeout_replays_nothing(
|
|
self, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
injector = _bare_injector(environment=DesktopEnvironment.WAYLAND, wayland_tool="portal")
|
|
injector._portal = MagicMock()
|
|
injector._portal.tap_keysym.side_effect = RemoteDesktopPortalError(
|
|
"Timed out waiting for the portal", delivered=None
|
|
)
|
|
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 False
|
|
assert injector.wayland_tool == "wtype"
|
|
mock_run.assert_not_called()
|
|
|
|
def test_shortcut_untracked_timeout_replays_nothing(
|
|
self, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
injector = _bare_injector(environment=DesktopEnvironment.WAYLAND, wayland_tool="portal")
|
|
injector._portal = MagicMock()
|
|
injector._portal.send_shortcut.side_effect = RemoteDesktopPortalError(
|
|
"Timed out waiting for the portal", delivered=None
|
|
)
|
|
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._inject_shortcut_with_wayland_tool("ctrl+a") is False
|
|
mock_run.assert_not_called()
|
|
|
|
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
|