diff --git a/src/vocalinux/text_injection/focused_window.py b/src/vocalinux/text_injection/focused_window.py index d4d608f..b0cdacf 100644 --- a/src/vocalinux/text_injection/focused_window.py +++ b/src/vocalinux/text_injection/focused_window.py @@ -253,6 +253,120 @@ def _focused_window_x11() -> Optional[FocusedWindow]: ) +def _read_process_environ(pid: str) -> Optional[dict[str, str]]: + """Return ``/proc//environ`` as a dict, or None when it cannot be read.""" + if not pid.isdigit(): + return None + try: + with open(f"/proc/{pid}/environ", "rb") as handle: + raw = handle.read() + except OSError: + return None + environ: dict[str, str] = {} + for entry in raw.split(b"\0"): + key, sep, value = entry.partition(b"=") + if sep: + environ[key.decode("utf-8", "replace")] = value.decode("utf-8", "replace") + return environ + + +# Shared libraries that identify the toolkit an X11 client draws its text fields with. +_GTK_LIBRARY_MARKERS = ("libgtk-3.so", "libgtk-x11-2.0.so", "libgtk-4.so") +_QT_LIBRARY_MARKERS = ("libQt5Gui.so", "libQt6Gui.so", "libQtGui.so") +_IM_MODULE_VARIABLES = {"gtk": "GTK_IM_MODULE", "qt": "QT_IM_MODULE"} + + +def _process_toolkits(pid: str) -> set[str]: + """Return the GUI toolkits (``gtk``, ``qt``) mapped into a process, if readable.""" + if not pid.isdigit(): + return set() + try: + with open(f"/proc/{pid}/maps", "r", encoding="utf-8", errors="replace") as handle: + maps = handle.read() + except OSError: + return set() + toolkits: set[str] = set() + if any(marker in maps for marker in _GTK_LIBRARY_MARKERS): + toolkits.add("gtk") + if any(marker in maps for marker in _QT_LIBRARY_MARKERS): + toolkits.add("qt") + return toolkits + + +def _toolkit_uses_ibus(environ: dict[str, str], toolkit: str) -> bool: + """True when a GTK or Qt client with this environment loads the IBus module. + + The toolkit's own variable decides. ``xim`` hands the choice to XIM, which + reaches IBus only through ``XMODIFIERS=@im=ibus``. + """ + module = environ.get(_IM_MODULE_VARIABLES[toolkit], "").strip().lower() + if module == "xim": + return "@im=ibus" in environ.get("XMODIFIERS", "").lower() + return "ibus" in module + + +def environ_uses_ibus_im(environ: dict[str, str], toolkits: frozenset[str] = frozenset()) -> bool: + """True when a client with this environment and these toolkits talks to IBus. + + For a known GTK or Qt client only that toolkit's variable counts, so an + IBus setting meant for the other toolkit does not hide a missing one. When + the toolkit is unknown (plain Xlib, Tk, Motif or a custom toolkit), every + input method variable that is set must select IBus, and at least one must. + """ + known = toolkits & _IM_MODULE_VARIABLES.keys() + if known: + return all(_toolkit_uses_ibus(environ, toolkit) for toolkit in known) + values = [ + environ.get(name, "").strip().lower() + for name in ("GTK_IM_MODULE", "QT_IM_MODULE", "XMODIFIERS") + ] + selected = [value for value in values if value] + return bool(selected) and all("ibus" in value for value in selected) + + +def focused_xwayland_client_without_ibus() -> bool: + """True when a Wayland session has an XWayland client focused that IBus cannot reach. + + An X11 client only talks to IBus through an input method module that it + loads at startup from GTK_IM_MODULE (GTK), QT_IM_MODULE (Qt) or XMODIFIERS + (XIM). Its toolkit is read from the libraries it has mapped. Without + one, IBus has no input context for it: the Vocalinux engine still commits + and reports success, and the text goes nowhere (#972). The client's own + environment is read because it can differ from the one Vocalinux runs in. + + While a native Wayland client holds focus, XWayland's active window is + unset or a placeholder with no pid and no WM_CLASS, so this returns False. + A real client whose environment cannot be read counts as not connected, + since xdotool reaches every XWayland client and IBus does not. + """ + if not os.environ.get("WAYLAND_DISPLAY") or not os.environ.get("DISPLAY"): + return False + if not shutil.which("xdotool"): + return False + env = _x11_probe_env() + window_id = _run_text(["xdotool", "getactivewindow"], env) + if not window_id: + return False + pid = _run_text(["xdotool", "getwindowpid", window_id], env) + if not pid.isdigit(): + if not read_wm_class(window_id, env): + return False + logger.debug("Focused XWayland window %s has no pid; assuming no IBus", window_id) + return True + environ = _read_process_environ(pid) + if environ is None: + logger.debug("Cannot read the environment of XWayland pid %s; assuming no IBus", pid) + return True + if environ_uses_ibus_im(environ, frozenset(_process_toolkits(pid))): + return False + logger.debug( + "Focused XWayland client %s (pid %s) has no IBus input method module", + _process_name_for_pid(pid) or window_id, + pid, + ) + return True + + def _first_string(payload: dict[str, Any], *keys: str) -> str: """Return the first non-empty string value for the given keys.""" for key in keys: diff --git a/src/vocalinux/text_injection/text_injector.py b/src/vocalinux/text_injection/text_injector.py index 4f04ec0..ce7cbec 100644 --- a/src/vocalinux/text_injection/text_injector.py +++ b/src/vocalinux/text_injection/text_injector.py @@ -18,7 +18,11 @@ from typing import List, Optional, Tuple, cast from ..utils.host_process import host_env from ..utils.paths import config_dir -from .focused_window import is_focused_window_terminal, read_wm_class +from .focused_window import ( + focused_xwayland_client_without_ibus, + is_focused_window_terminal, + read_wm_class, +) from .ibus_engine import ( IBusTextInjector, is_ibus_active_input_method, @@ -1676,11 +1680,22 @@ class TextInjector: if self.environment == DesktopEnvironment.WAYLAND_XDOTOOL: self._try_recover_from_fallback() + xwayland_detour = False try: with self._state_lock: current_env = self.environment ibus_injector = self._ibus_injector + if current_env == DesktopEnvironment.WAYLAND_IBUS and self._should_bypass_ibus(): + # The IBus engine would commit into no input context and still + # report OK (#972); send this segment through XWayland instead. + logger.info("Focused XWayland client has no IBus input method; using xdotool") + with self._state_lock: + if self.environment == DesktopEnvironment.WAYLAND_IBUS: + self.environment = DesktopEnvironment.WAYLAND_XDOTOOL + xwayland_detour = True + current_env = self.environment + if ( current_env == DesktopEnvironment.WAYLAND_IBUS or current_env == DesktopEnvironment.X11_IBUS @@ -1823,6 +1838,25 @@ class TextInjector: except ImportError: logger.warning("Could not import audio feedback module") return False + finally: + if xwayland_detour: + with self._state_lock: + if self.environment == DesktopEnvironment.WAYLAND_XDOTOOL: + self.environment = DesktopEnvironment.WAYLAND_IBUS + + def _should_bypass_ibus(self) -> bool: + """True when IBus cannot reach the focused window but xdotool can (#972). + + Only applies when the backend was chosen automatically; a pinned IBus + backend is the user's call and is kept. + """ + if self._backend_pin[0] == "ibus": + return False + try: + return focused_xwayland_client_without_ibus() + except (OSError, subprocess.SubprocessError, ValueError) as exc: + logger.debug(f"XWayland focus probe failed, keeping IBus: {exc}") + return False def _inject_with_xdotool(self, text: str) -> None: """ @@ -1845,12 +1879,17 @@ class TextInjector: os.environ.setdefault("DISPLAY", env["DISPLAY"]) logger.debug(f"Using XWayland with DISPLAY={env['DISPLAY']}") self._wait_for_modifiers_released() - if self._inject_via_clipboard_paste(text, env=env): + if self._clipboard_holds_non_text(): + logger.info( + "Clipboard holds non-text content; typing instead of pasting to preserve it" + ) + elif self._inject_via_clipboard_paste(text, env=env): return - logger.warning( - "Clipboard paste failed, falling back to xdotool type " - "(character-by-character; text may be scrambled on non-US layouts)" - ) + else: + logger.warning( + "Clipboard paste failed, falling back to xdotool type " + "(character-by-character; text may be scrambled on non-US layouts)" + ) # Add a small delay to ensure text is injected properly time.sleep(0.3) # Increased delay for better reliability @@ -2038,7 +2077,7 @@ class TextInjector: except UnicodeEncodeError: return True - def _read_clipboard(self) -> Optional[str]: + def _read_clipboard(self, prefer_x11: Optional[bool] = None) -> Optional[str]: """ Read clipboard text only. @@ -2048,12 +2087,14 @@ class TextInjector: On X11 and WAYLAND_XDOTOOL this reads the X11 CLIPBOARD (xclip/xsel), matching the selection the xdotool paste wrote. wl-paste is a different - clipboard. + clipboard. ``prefer_x11`` overrides that choice so a caller can read the + clipboard it wrote even after the backend has changed. """ - prefer_x11 = self.environment in ( - DesktopEnvironment.X11, - DesktopEnvironment.WAYLAND_XDOTOOL, - ) + if prefer_x11 is None: + prefer_x11 = self.environment in ( + DesktopEnvironment.X11, + DesktopEnvironment.WAYLAND_XDOTOOL, + ) host_is_wayland = ( self._session_environment == DesktopEnvironment.WAYLAND or os.environ.get("XDG_SESSION_TYPE", "").lower() == "wayland" @@ -2096,8 +2137,12 @@ class TextInjector: return "" if saw_empty else None def _clipboard_holds_non_text(self) -> bool: - """Return whether the X11 clipboard has no readable text target.""" - if self.environment != DesktopEnvironment.X11: + """Return whether the X11 clipboard has no readable text target. + + Checked on X11 and WAYLAND_XDOTOOL, the two routes that paste through + the X11 clipboard and could not restore an image or file afterwards. + """ + if self.environment not in (DesktopEnvironment.X11, DesktopEnvironment.WAYLAND_XDOTOOL): return False if not shutil.which("xclip"): return self._read_clipboard() is None @@ -2302,7 +2347,7 @@ class TextInjector: if target is None: return # User copied something else during the delay — leave it alone. - if self._read_clipboard() != text: + if self._read_clipboard(prefer_x11=prefer_x11) != text: logger.debug("Clipboard changed during restore delay; skipping restore") return if target == "": diff --git a/tests/test_focused_window.py b/tests/test_focused_window.py index 4599843..be1a506 100644 --- a/tests/test_focused_window.py +++ b/tests/test_focused_window.py @@ -2,10 +2,14 @@ import json import unittest +from typing import Optional from unittest.mock import MagicMock, mock_open, patch from vocalinux.text_injection.focused_window import ( FocusedWindow, + _process_toolkits, + environ_uses_ibus_im, + focused_xwayland_client_without_ibus, get_focused_window, is_focused_window_terminal, looks_like_terminal, @@ -243,3 +247,151 @@ class TestGetFocusedWindow(unittest.TestCase): return_value=None, ): self.assertFalse(is_focused_window_terminal()) + + +class TestFocusedXWaylandClientWithoutIBus(unittest.TestCase): + """Detect XWayland clients that IBus has no input context for (#972).""" + + _MODULE = "vocalinux.text_injection.focused_window" + _WAYLAND_ENV = {"WAYLAND_DISPLAY": "wayland-0", "DISPLAY": ":0"} + + def _probe( + self, + outputs: dict[str, str], + *, + environ: Optional[bytes] = None, + wm_class: str = "", + toolkits: frozenset[str] = frozenset(), + ) -> bool: + def fake_run_text(cmd: list[str], env: Optional[dict[str, str]] = None) -> str: + return outputs.get(cmd[1], "") + + open_patch = ( + patch(f"{self._MODULE}.open", mock_open(read_data=environ)) + if environ is not None + else patch(f"{self._MODULE}.open", side_effect=PermissionError("denied")) + ) + with ( + patch.dict("os.environ", self._WAYLAND_ENV, clear=False), + patch(f"{self._MODULE}.shutil.which", return_value="/usr/bin/xdotool"), + patch(f"{self._MODULE}._run_text", side_effect=fake_run_text), + patch(f"{self._MODULE}.read_wm_class", return_value=wm_class), + patch(f"{self._MODULE}._process_toolkits", return_value=set(toolkits)), + open_patch, + ): + return focused_xwayland_client_without_ibus() + + def test_client_without_im_module(self) -> None: + outputs = {"getactivewindow": "4194311", "getwindowpid": "1234"} + self.assertTrue(self._probe(outputs, environ=b"HOME=/home/u\0LANG=C\0")) + + def test_client_with_qt_ibus_module(self) -> None: + outputs = {"getactivewindow": "4194311", "getwindowpid": "1234"} + self.assertFalse(self._probe(outputs, environ=b"QT_IM_MODULE=ibus\0")) + + def test_client_with_xim_ibus(self) -> None: + outputs = {"getactivewindow": "4194311", "getwindowpid": "1234"} + self.assertFalse(self._probe(outputs, environ=b"XMODIFIERS=@im=ibus\0")) + + def test_gtk_client_ignores_qt_ibus_setting(self) -> None: + outputs = {"getactivewindow": "4194311", "getwindowpid": "1234"} + environ = b"GTK_IM_MODULE=fcitx\0QT_IM_MODULE=ibus\0" + self.assertTrue(self._probe(outputs, environ=environ, toolkits=frozenset({"gtk"}))) + + def test_qt_client_with_qt_ibus_setting(self) -> None: + outputs = {"getactivewindow": "4194311", "getwindowpid": "1234"} + environ = b"GTK_IM_MODULE=fcitx\0QT_IM_MODULE=ibus\0" + self.assertFalse(self._probe(outputs, environ=environ, toolkits=frozenset({"qt"}))) + + def test_unknown_toolkit_with_mixed_settings(self) -> None: + outputs = {"getactivewindow": "4194311", "getwindowpid": "1234"} + environ = b"GTK_IM_MODULE=fcitx\0QT_IM_MODULE=ibus\0" + self.assertTrue(self._probe(outputs, environ=environ)) + + def test_unreadable_environment_counts_as_unconnected(self) -> None: + outputs = {"getactivewindow": "4194311", "getwindowpid": "1234"} + self.assertTrue(self._probe(outputs)) + + def test_kwin_placeholder_window_is_not_a_client(self) -> None: + self.assertFalse(self._probe({"getactivewindow": "2097152"})) + + def test_client_without_pid_but_with_class(self) -> None: + self.assertTrue(self._probe({"getactivewindow": "6291472"}, wm_class="tkprobe Tkprobe")) + + def test_no_active_x11_window(self) -> None: + self.assertFalse(self._probe({})) + + def test_not_a_wayland_session(self) -> None: + with ( + patch.dict("os.environ", {"DISPLAY": ":0"}, clear=True), + patch(f"{self._MODULE}._run_text") as run_text, + ): + self.assertFalse(focused_xwayland_client_without_ibus()) + run_text.assert_not_called() + + def test_no_xdotool(self) -> None: + with ( + patch.dict("os.environ", self._WAYLAND_ENV, clear=False), + patch(f"{self._MODULE}.shutil.which", return_value=None), + patch(f"{self._MODULE}._run_text") as run_text, + ): + self.assertFalse(focused_xwayland_client_without_ibus()) + run_text.assert_not_called() + + +class TestEnvironUsesIBusIm(unittest.TestCase): + """Decide from a client's IM variables and toolkit whether it reaches IBus.""" + + def test_variables(self) -> None: + self.assertTrue(environ_uses_ibus_im({"GTK_IM_MODULE": "ibus"})) + self.assertTrue(environ_uses_ibus_im({"QT_IM_MODULE": "IBus"})) + self.assertTrue(environ_uses_ibus_im({"XMODIFIERS": "@im=ibus"})) + self.assertFalse( + environ_uses_ibus_im({"GTK_IM_MODULE": "fcitx", "XMODIFIERS": "@im=fcitx"}) + ) + self.assertFalse(environ_uses_ibus_im({})) + + def test_unknown_toolkit_needs_every_set_variable(self) -> None: + self.assertFalse(environ_uses_ibus_im({"GTK_IM_MODULE": "fcitx", "QT_IM_MODULE": "ibus"})) + self.assertFalse(environ_uses_ibus_im({"QT_IM_MODULE": "ibus", "XMODIFIERS": "@im=none"})) + self.assertTrue(environ_uses_ibus_im({"GTK_IM_MODULE": "ibus", "QT_IM_MODULE": "ibus"})) + + def test_known_toolkit_uses_its_own_variable(self) -> None: + mixed = {"GTK_IM_MODULE": "fcitx", "QT_IM_MODULE": "ibus"} + self.assertFalse(environ_uses_ibus_im(mixed, frozenset({"gtk"}))) + self.assertTrue(environ_uses_ibus_im(mixed, frozenset({"qt"}))) + self.assertFalse(environ_uses_ibus_im(mixed, frozenset({"gtk", "qt"}))) + self.assertFalse(environ_uses_ibus_im({"XMODIFIERS": "@im=ibus"}, frozenset({"gtk"}))) + + def test_xim_module_defers_to_xmodifiers(self) -> None: + gtk = frozenset({"gtk"}) + self.assertTrue( + environ_uses_ibus_im({"GTK_IM_MODULE": "xim", "XMODIFIERS": "@im=ibus"}, gtk) + ) + self.assertFalse( + environ_uses_ibus_im({"GTK_IM_MODULE": "xim", "XMODIFIERS": "@im=fcitx"}, gtk) + ) + + +class TestProcessToolkits(unittest.TestCase): + """Read the GUI toolkit of a process from its mapped libraries.""" + + _MODULE = "vocalinux.text_injection.focused_window" + + def test_detects_gtk_and_qt(self) -> None: + maps = ( + "7f00 r-xp 0 08:01 1 /usr/lib/x86_64-linux-gnu/libgtk-3.so.0.2404.29\n" + "7f01 r-xp 0 08:01 2 /usr/lib/x86_64-linux-gnu/libQt5Gui.so.5.15.3\n" + ) + with patch(f"{self._MODULE}.open", mock_open(read_data=maps)): + self.assertEqual(_process_toolkits("1234"), {"gtk", "qt"}) + + def test_plain_xlib_client(self) -> None: + maps = "7f00 r-xp 0 08:01 1 /usr/lib/x86_64-linux-gnu/libX11.so.6.4.0\n" + with patch(f"{self._MODULE}.open", mock_open(read_data=maps)): + self.assertEqual(_process_toolkits("1234"), set()) + + def test_unreadable_maps(self) -> None: + with patch(f"{self._MODULE}.open", side_effect=PermissionError("denied")): + self.assertEqual(_process_toolkits("1234"), set()) + self.assertEqual(_process_toolkits("not-a-pid"), set()) diff --git a/tests/test_text_injector.py b/tests/test_text_injector.py index 450e173..bb7adab 100644 --- a/tests/test_text_injector.py +++ b/tests/test_text_injector.py @@ -11,7 +11,7 @@ import sys import threading import unittest from collections.abc import Iterator -from typing import Any +from typing import Any, Optional from unittest import mock from unittest.mock import MagicMock, mock_open, patch @@ -1713,7 +1713,10 @@ class TestTextInjectorEdgeCases(unittest.TestCase): with patch.dict("os.environ", {"XDG_SESSION_TYPE": "wayland", "WAYLAND_DISPLAY": "w-1"}): injector = TextInjector() injector.environment = DesktopEnvironment.WAYLAND_XDOTOOL - with patch.object(injector, "_should_use_terminal_paste", return_value=False): + with ( + patch.object(injector, "_should_use_terminal_paste", return_value=False), + patch.object(injector, "_clipboard_holds_non_text", return_value=False), + ): result = injector.inject_text("привет") self.assertTrue(result) @@ -1733,6 +1736,37 @@ class TestTextInjectorEdgeCases(unittest.TestCase): "must not fall through to the layout-dependent type path when the paste succeeds", ) + def test_xwayland_keeps_a_non_text_clipboard(self) -> None: + """An image on the X11 clipboard is not overwritten by an XWayland paste.""" + + def which_side_effect(cmd: str) -> Optional[str]: + if cmd in ("xdotool", "xclip"): + return f"/usr/bin/{cmd}" + return None + + self.mock_which.side_effect = which_side_effect + self.mock_subprocess.return_value = MagicMock(returncode=0, stdout="", stderr="") + + with patch.dict("os.environ", {"XDG_SESSION_TYPE": "wayland", "WAYLAND_DISPLAY": "w-1"}): + injector = TextInjector() + injector.environment = DesktopEnvironment.WAYLAND_XDOTOOL + with ( + patch.object(injector, "_clipboard_holds_non_text", return_value=True), + patch.object(injector, "_inject_via_clipboard_paste") as paste, + ): + injector._inject_with_xdotool("hi") + + paste.assert_not_called() + calls = [c.args[0] for c in self.mock_subprocess.call_args_list if c.args] + self.assertTrue(any(c[:2] == ["xdotool", "type"] for c in calls)) + + def test_xwayland_checks_x11_clipboard_targets(self) -> None: + injector = TextInjector() + injector.environment = DesktopEnvironment.WAYLAND_XDOTOOL + self._mock_x11_clipboard_targets("TARGETS\nimage/png\n") + + self.assertTrue(injector._clipboard_holds_non_text()) + def test_inject_with_xdotool_xwayland_uses_terminal_paste_chord(self) -> None: """XWayland xclip paste must still use Ctrl+Shift+V in terminals (#734).""" @@ -3683,3 +3717,141 @@ class TestKdeSkipsInactiveIbus(unittest.TestCase): injector._ibus_init_thread.join(timeout=5) mock_ibus_class.assert_called_once() self.assertEqual(injector.environment, DesktopEnvironment.WAYLAND_IBUS) + + +class TestXWaylandIBusBypass(unittest.TestCase): + """XWayland clients without an IBus IM module get xdotool instead (#972).""" + + def _make_injector(self, environment: DesktopEnvironment) -> TextInjector: + injector = TextInjector.__new__(TextInjector) + injector._state_lock = threading.Lock() + injector._abort_injections = threading.Event() + injector.environment = environment + injector.wayland_tool = None + injector.last_typed_count = 0 + injector._backend_pin = ("auto", None) + injector._ibus_injector = MagicMock() + injector._ibus_injector.inject_text.return_value = True + injector._log_current_window_info = MagicMock() # type: ignore[method-assign] + injector._should_copy_to_clipboard = MagicMock( # type: ignore[method-assign] + return_value=False + ) + return injector + + def test_unconnected_xwayland_client_uses_xdotool(self) -> None: + injector = self._make_injector(DesktopEnvironment.WAYLAND_IBUS) + seen: list[DesktopEnvironment] = [] + + def fake_xdotool(text: str) -> None: + seen.append(injector.environment) + + with ( + patch.object(injector, "_should_bypass_ibus", return_value=True), + patch.object(injector, "_inject_with_xdotool", side_effect=fake_xdotool) as xdo, + ): + self.assertTrue(injector.inject_text("hello")) + + xdo.assert_called_once_with("hello") + injector._ibus_injector.inject_text.assert_not_called() + self.assertEqual(seen, [DesktopEnvironment.WAYLAND_XDOTOOL]) + self.assertEqual(injector.environment, DesktopEnvironment.WAYLAND_IBUS) + self.assertEqual(injector.last_typed_count, 5) + + def test_ibus_connected_client_keeps_ibus(self) -> None: + injector = self._make_injector(DesktopEnvironment.WAYLAND_IBUS) + + with ( + patch.object(injector, "_should_bypass_ibus", return_value=False), + patch.object(injector, "_inject_with_xdotool") as xdo, + ): + self.assertTrue(injector.inject_text("hello")) + + injector._ibus_injector.inject_text.assert_called_once_with("hello") + xdo.assert_not_called() + self.assertEqual(injector.environment, DesktopEnvironment.WAYLAND_IBUS) + + def test_environment_restored_when_xdotool_fails(self) -> None: + injector = self._make_injector(DesktopEnvironment.WAYLAND_IBUS) + error = subprocess.CalledProcessError(1, ["xdotool"]) + + with ( + patch.object(injector, "_should_bypass_ibus", return_value=True), + patch.object(injector, "_inject_with_xdotool", side_effect=error), + patch.object(injector, "_copy_to_clipboard", return_value=False), + patch("vocalinux.ui.audio_feedback.play_error_sound"), + ): + self.assertFalse(injector.inject_text("hello")) + + self.assertEqual(injector.environment, DesktopEnvironment.WAYLAND_IBUS) + + def test_x11_ibus_is_not_probed(self) -> None: + injector = self._make_injector(DesktopEnvironment.X11_IBUS) + + with patch.object(injector, "_should_bypass_ibus") as probe: + self.assertTrue(injector.inject_text("hello")) + + probe.assert_not_called() + injector._ibus_injector.inject_text.assert_called_once_with("hello") + + def test_probe_error_keeps_ibus(self) -> None: + injector = self._make_injector(DesktopEnvironment.WAYLAND_IBUS) + + with patch( + "vocalinux.text_injection.text_injector.focused_xwayland_client_without_ibus", + side_effect=OSError("no display"), + ): + self.assertFalse(injector._should_bypass_ibus()) + + def test_pinned_ibus_is_not_bypassed(self) -> None: + injector = self._make_injector(DesktopEnvironment.WAYLAND_IBUS) + injector._backend_pin = ("ibus", "text_injection.backend") + + with patch( + "vocalinux.text_injection.text_injector.focused_xwayland_client_without_ibus", + return_value=True, + ) as probe: + self.assertFalse(injector._should_bypass_ibus()) + + probe.assert_not_called() + + def test_auto_backend_is_bypassed(self) -> None: + injector = self._make_injector(DesktopEnvironment.WAYLAND_IBUS) + + with patch( + "vocalinux.text_injection.text_injector.focused_xwayland_client_without_ibus", + return_value=True, + ): + self.assertTrue(injector._should_bypass_ibus()) + + def test_clipboard_restore_reads_the_x11_clipboard_after_detour(self) -> None: + injector = self._make_injector(DesktopEnvironment.WAYLAND_XDOTOOL) + injector._clipboard_restore_target = None + injector._clipboard_restore_generation = 0 + reads: list[Any] = [] + restore_targets: list[Any] = [] + + def fake_read(prefer_x11: Any = None) -> str: + reads.append(prefer_x11) + return "original" if len(reads) == 1 else "hello" + + def capture_thread(target: Any, daemon: bool = False) -> MagicMock: + restore_targets.append(target) + return MagicMock() + + module = "vocalinux.text_injection.text_injector" + with ( + patch.object(injector, "_read_clipboard", side_effect=fake_read), + patch.object(injector, "_x11_clipboard_tools", return_value=["xclip"]), + patch.object(injector, "_copy_to_clipboard", return_value=True) as copy, + patch.object(injector, "_should_use_terminal_paste", return_value=False), + patch.object(injector, "_clipboard_paste_command", return_value=["xdotool"]), + patch(f"{module}.subprocess.run"), + patch(f"{module}.threading.Thread", side_effect=capture_thread), + ): + self.assertTrue(injector._inject_via_clipboard_paste("hello")) + injector.environment = DesktopEnvironment.WAYLAND_IBUS + with patch(f"{module}.time.sleep"): + restore_targets[0]() + + self.assertEqual(reads[-1], True) + copy.assert_called_with("original", tools=["xclip"]) diff --git a/tests/test_text_injector_ext.py b/tests/test_text_injector_ext.py index e0cc881..f976947 100644 --- a/tests/test_text_injector_ext.py +++ b/tests/test_text_injector_ext.py @@ -74,6 +74,7 @@ def _make_injector(env) -> Any: obj = cast(Any, TextInjector.__new__(TextInjector)) obj._ibus_injector = None obj.environment = env + obj._backend_pin = ("auto", None) obj._session_environment = env obj._ibus_ready = False obj._ibus_init_failed = False