diff --git a/src/vocalinux/text_injection/text_injector.py b/src/vocalinux/text_injection/text_injector.py index eb3429cd..fc19a81e 100644 --- a/src/vocalinux/text_injection/text_injector.py +++ b/src/vocalinux/text_injection/text_injector.py @@ -284,6 +284,29 @@ def _kde_virtual_keyboard_enabled(self) -> bool: ) return False + @staticmethod + def _ibus_wayland_bridge_running() -> bool: + """Whether IBus' zwp_input_method_v2 bridge (``ibus-wayland``) is running. + + The ``_IBUS_UNBRIDGED_COMPOSITORS`` denylist assumes nothing sits between + the compositor and ibus-daemon. Since IBus 1.5.32 the ``ibus-wayland`` + helper implements ``zwp_input_method_v2``, so on a compositor that + exposes ``zwp_input_method_manager_v2`` (the wlroots/smithay ones on the + denylist all do) it relays commits to native Wayland clients speaking + text-input-v3. When it is running, those compositors are bridged. + + Mirrors ``is_ibus_daemon_running()`` in ibus_engine.py. + """ + try: + result = subprocess.run( + ["pgrep", "-x", "ibus-wayland"], + capture_output=True, + timeout=2, + ) + return result.returncode == 0 + except (subprocess.SubprocessError, FileNotFoundError): + return False + def _wayland_compositor_bridges_ibus(self) -> bool: """Whether native Wayland clients receive IBus commits on this compositor. @@ -294,6 +317,12 @@ def _wayland_compositor_bridges_ibus(self) -> bool: a denylist rather than an allowlist so that unrecognised desktops keep the previous IBus-preferred behaviour. + A denylisted compositor is still bridged when ``ibus-wayland`` is running, + since that supplies exactly the input-method-v2 relay those compositors + lack (issue #607). The check is a live probe rather than configuration, + so this degrades back to the denylist on its own if the bridge dies or + was never started. + On KDE Plasma Wayland, bridging also requires KWin VirtualKeyboard to be enabled (issue #574); otherwise commit_text succeeds at the IBus layer but never reaches apps. @@ -306,6 +335,14 @@ def _wayland_compositor_bridges_ibus(self) -> bool: for var in ("XDG_CURRENT_DESKTOP", "XDG_SESSION_DESKTOP", "DESKTOP_SESSION") ).lower() if any(name in desktop for name in self._IBUS_UNBRIDGED_COMPOSITORS): + if self._ibus_wayland_bridge_running(): + logger.info( + "Compositor '%s' is on the unbridged list, but the ibus-wayland " + "input-method-v2 bridge is running; IBus can reach native " + "Wayland clients.", + os.environ.get("XDG_CURRENT_DESKTOP", "unknown"), + ) + return True return False if _is_kde_plasma_session(): return self._kde_virtual_keyboard_enabled() @@ -415,13 +452,40 @@ def _ensure_ydotoold(self) -> bool: logger.warning("ydotoold did not become ready in time") return False + @staticmethod + def _forced_backend() -> str: + """Backend pinned via ``VOCALINUX_FORCE_BACKEND``, or ``"auto"``. + + Autodetection has to infer whether IBus commits actually reach the + focused app, and it cannot verify that: ``commit_text()`` reports + success even when the text is dropped. This gives users an escape hatch + when the inference is wrong, and makes the two paths A/B-testable + without editing code. + + Accepts ``ibus``, ``wtype``, ``ydotool`` or ``auto``. Anything else is + ignored with a warning, so a typo cannot silently pin a backend. + """ + value = os.environ.get("VOCALINUX_FORCE_BACKEND", "").strip().lower() + if not value or value == "auto": + return "auto" + if value in ("ibus", "wtype", "ydotool"): + return value + logger.warning( + "Ignoring unknown VOCALINUX_FORCE_BACKEND=%r (expected ibus/wtype/ydotool/auto)", + value, + ) + return "auto" + def _check_dependencies(self): """Check for the required tools for text injection.""" ibus_requested = False + forced = self._forced_backend() + if forced != "auto": + logger.info("VOCALINUX_FORCE_BACKEND=%s: overriding backend autodetection", forced) # Prefer IBus on both X11 and Wayland - it sends Unicode directly, # bypassing keyboard layout issues entirely - if is_ibus_available(): + if is_ibus_available() and forced in ("auto", "ibus"): ibus_active = is_ibus_active_input_method() gtk_im = os.environ.get("GTK_IM_MODULE", "").lower() qt_im = os.environ.get("QT_IM_MODULE", "").lower() @@ -445,13 +509,16 @@ def _check_dependencies(self): # Check if IBus is the active input method (not just installed) # This is important because IBus may be installed but not being used, # e.g., when the user has configured ydotool or Fcitx instead. - if not ibus_active and not wayland_scoped_ibus: + # VOCALINUX_FORCE_BACKEND=ibus bypasses the reachability guards below + # and goes straight to setup. + force_ibus = forced == "ibus" + if not force_ibus and not ibus_active and not wayland_scoped_ibus: logger.info( "IBus is installed but not the active input method. " "Falling back to alternative text injection method." ) # Check if ibus-daemon is running before attempting setup - elif not is_ibus_daemon_running(): + elif not force_ibus and not is_ibus_daemon_running(): logger.info( "IBus daemon not running. This is normal on some desktop environments " "(e.g., KDE Plasma). Using alternative text injection method. " @@ -460,7 +527,7 @@ def _check_dependencies(self): # Some Wayland compositors (COSMIC, Sway, Hyprland, ...) do not deliver # IBus commits to native Wayland apps, so IBus would silently drop the # text even though commit_text() reports success. - elif not self._wayland_compositor_bridges_ibus(): + elif not force_ibus and not self._wayland_compositor_bridges_ibus(): logger.info( "Compositor '%s' does not bridge IBus to native Wayland apps; " "using virtual-keyboard injection (wtype/ydotool) instead.", @@ -496,7 +563,14 @@ def _check_dependencies(self): # Prefer ydotool when the daemon is (or can be) ready. Flatpak ships # ydotool for native Wayland typing; wtype needs a Wayland socket. - if ydotool_available and self._ensure_ydotoold(): + if forced == "wtype" and wtype_available: + self.wayland_tool = "wtype" + logger.info("VOCALINUX_FORCE_BACKEND=wtype: using wtype for Wayland injection") + elif forced == "ydotool" and ydotool_available: + self._ensure_ydotoold() + self.wayland_tool = "ydotool" + logger.info("VOCALINUX_FORCE_BACKEND=ydotool: using ydotool for Wayland injection") + elif ydotool_available and self._ensure_ydotoold(): self.wayland_tool = "ydotool" logger.info("Using ydotool for Wayland text injection") elif ydotool_available and not wtype_available: diff --git a/tests/test_text_injector.py b/tests/test_text_injector.py index 4dfa6b4f..240f10ad 100644 --- a/tests/test_text_injector.py +++ b/tests/test_text_injector.py @@ -1828,7 +1828,13 @@ def test_bridged_desktops_prefer_ibus(self): self.assertTrue(injector._wayland_compositor_bridges_ibus(), desktop) def test_unbridged_compositors_skip_ibus(self): - """COSMIC and wlroots compositors do not deliver IBus commits to native apps.""" + """COSMIC and wlroots compositors do not deliver IBus commits to native apps. + + The ibus-wayland bridge is explicitly absent here; see + ``test_unbridged_compositors_use_ibus_when_bridge_running`` for the + opposite case. Patching it keeps the result independent of whether the + machine running the suite happens to have the bridge up. + """ injector = self._bare_injector() for desktop in ("COSMIC", "sway", "Hyprland", "wayfire", "niri", "river"): with patch.dict( @@ -1839,7 +1845,55 @@ def test_unbridged_compositors_skip_ibus(self): "DESKTOP_SESSION": desktop, }, ): - self.assertFalse(injector._wayland_compositor_bridges_ibus(), desktop) + with patch.object(TextInjector, "_ibus_wayland_bridge_running", return_value=False): + self.assertFalse(injector._wayland_compositor_bridges_ibus(), desktop) + + def test_unbridged_compositors_use_ibus_when_bridge_running(self): + """ibus-wayland supplies the input-method-v2 relay these compositors lack (#607).""" + injector = self._bare_injector() + for desktop in ("COSMIC", "sway", "Hyprland", "wayfire", "niri", "river"): + with patch.dict( + "os.environ", + { + "XDG_CURRENT_DESKTOP": desktop, + "XDG_SESSION_DESKTOP": desktop, + "DESKTOP_SESSION": desktop, + }, + ): + with patch.object(TextInjector, "_ibus_wayland_bridge_running", return_value=True): + self.assertTrue(injector._wayland_compositor_bridges_ibus(), desktop) + + def test_bridge_probe_not_consulted_for_bridged_desktops(self): + """GNOME and friends never reach the probe; behaviour there is unchanged.""" + injector = self._bare_injector() + with patch.dict( + "os.environ", + { + "XDG_CURRENT_DESKTOP": "GNOME", + "XDG_SESSION_DESKTOP": "GNOME", + "DESKTOP_SESSION": "GNOME", + "KDE_FULL_SESSION": "", + }, + clear=False, + ): + with patch.object(TextInjector, "_ibus_wayland_bridge_running") as probe: + self.assertTrue(injector._wayland_compositor_bridges_ibus()) + probe.assert_not_called() + + def test_bridge_probe_detects_running_process(self): + """The probe shells out to pgrep -x ibus-wayland.""" + with patch("subprocess.run", return_value=MagicMock(returncode=0)) as run: + self.assertTrue(TextInjector._ibus_wayland_bridge_running()) + self.assertEqual(run.call_args[0][0], ["pgrep", "-x", "ibus-wayland"]) + + with patch("subprocess.run", return_value=MagicMock(returncode=1)): + self.assertFalse(TextInjector._ibus_wayland_bridge_running()) + + def test_bridge_probe_survives_missing_pgrep(self): + """No pgrep (or a hung one) must not raise -- just report 'no bridge'.""" + for boom in (FileNotFoundError(), subprocess.TimeoutExpired("pgrep", 2)): + with patch("subprocess.run", side_effect=boom): + self.assertFalse(TextInjector._ibus_wayland_bridge_running()) def test_non_wayland_always_bridges(self): """On X11/XWayland IBus works via XIM regardless of desktop.""" @@ -2116,7 +2170,11 @@ def test_cosmic_skips_ibus_and_uses_wtype( "DESKTOP_SESSION": "cosmic", }, ): - injector = TextInjector() + # No ibus-wayland relay here, so COSMIC stays unbridged. Stated + # explicitly because mock_run returns returncode=0 for every + # subprocess, which the bridge probe would otherwise read as a hit. + with patch.object(TextInjector, "_ibus_wayland_bridge_running", return_value=False): + injector = TextInjector() self.assertEqual(injector.environment, DesktopEnvironment.WAYLAND) self.assertEqual(injector.wayland_tool, "wtype") @@ -2142,3 +2200,26 @@ def test_ydotool_direct_mode_when_no_daemon_and_no_wtype(self, mock_which, _mock ): injector = TextInjector() self.assertEqual(injector.wayland_tool, "ydotool") + + +class TestForcedBackend(unittest.TestCase): + """VOCALINUX_FORCE_BACKEND overrides backend autodetection.""" + + def test_unset_or_auto_means_auto(self): + for value in ("", "auto", " AUTO "): + with patch.dict("os.environ", {"VOCALINUX_FORCE_BACKEND": value}): + self.assertEqual(TextInjector._forced_backend(), "auto") + + def test_recognised_backends(self): + for value in ("ibus", "wtype", "ydotool"): + with patch.dict("os.environ", {"VOCALINUX_FORCE_BACKEND": value.upper()}): + self.assertEqual(TextInjector._forced_backend(), value) + + def test_unknown_value_falls_back_to_auto(self): + """A typo must not silently pin the wrong backend.""" + with patch.dict("os.environ", {"VOCALINUX_FORCE_BACKEND": "ibsu"}): + self.assertEqual(TextInjector._forced_backend(), "auto") + + def test_missing_variable_means_auto(self): + with patch.dict("os.environ", {}, clear=True): + self.assertEqual(TextInjector._forced_backend(), "auto") diff --git a/tests/test_text_injector_ext.py b/tests/test_text_injector_ext.py index cd92301f..c70c2851 100644 --- a/tests/test_text_injector_ext.py +++ b/tests/test_text_injector_ext.py @@ -277,7 +277,7 @@ def test_gnome_wayland_uses_ibus_when_daemon_runs_with_xkb_engine(self): self.assertEqual(obj.wayland_tool, "wtype") def test_unbridged_wayland_skips_ibus_when_engine_is_xkb(self): - from vocalinux.text_injection.text_injector import DesktopEnvironment + from vocalinux.text_injection.text_injector import DesktopEnvironment, TextInjector obj = _make_injector(DesktopEnvironment.WAYLAND) @@ -301,12 +301,137 @@ def test_unbridged_wayland_skips_ibus_when_engine_is_xkb(self): "shutil.which", side_effect=lambda cmd: "/usr/bin/wtype" if cmd == "wtype" else None, ), + # No ibus-wayland relay, so sway stays unbridged. Stated explicitly so + # the result does not depend on whether the machine running the suite + # happens to have the bridge up. + patch.object(TextInjector, "_ibus_wayland_bridge_running", return_value=False), ): obj._check_dependencies() mock_ibus_class.assert_not_called() self.assertEqual(obj.wayland_tool, "wtype") + def test_unbridged_wayland_uses_ibus_when_bridge_running(self): + """ibus-wayland makes an otherwise-unbridged compositor usable (#607).""" + from vocalinux.text_injection.text_injector import DesktopEnvironment, TextInjector + + obj = _make_injector(DesktopEnvironment.WAYLAND) + + with ( + patch.dict( + os.environ, + {"XDG_SESSION_TYPE": "wayland", "XDG_CURRENT_DESKTOP": "Hyprland"}, + clear=True, + ), + patch("vocalinux.text_injection.text_injector.is_ibus_available", return_value=True), + patch( + "vocalinux.text_injection.text_injector.is_ibus_active_input_method", + return_value=False, + ), + patch( + "vocalinux.text_injection.text_injector.is_ibus_daemon_running", + return_value=True, + ), + patch("vocalinux.text_injection.text_injector.IBusTextInjector") as mock_ibus_class, + patch.object(obj, "_start_ibus_initialization"), + patch( + "shutil.which", + side_effect=lambda cmd: "/usr/bin/wtype" if cmd == "wtype" else None, + ), + patch.object(TextInjector, "_ibus_wayland_bridge_running", return_value=True), + ): + obj._check_dependencies() + + mock_ibus_class.assert_called_once() + + def test_force_backend_wtype_skips_ibus(self): + """VOCALINUX_FORCE_BACKEND=wtype pins wtype even where IBus would be chosen.""" + from vocalinux.text_injection.text_injector import DesktopEnvironment + + obj = _make_injector(DesktopEnvironment.WAYLAND) + + with ( + patch.dict( + os.environ, + {"XDG_SESSION_TYPE": "wayland", "VOCALINUX_FORCE_BACKEND": "wtype"}, + clear=True, + ), + patch("vocalinux.text_injection.text_injector.is_ibus_available", return_value=True), + patch("vocalinux.text_injection.text_injector.IBusTextInjector") as mock_ibus_class, + patch( + "shutil.which", + side_effect=lambda cmd: f"/usr/bin/{cmd}" if cmd in ("wtype", "ydotool") else None, + ), + ): + obj._check_dependencies() + + self.assertEqual(obj.wayland_tool, "wtype") + mock_ibus_class.assert_not_called() + + def test_force_backend_ydotool_skips_ibus_and_wtype(self): + """VOCALINUX_FORCE_BACKEND=ydotool pins ydotool even when wtype is available.""" + from vocalinux.text_injection.text_injector import DesktopEnvironment + + obj = _make_injector(DesktopEnvironment.WAYLAND) + + with ( + patch.dict( + os.environ, + {"XDG_SESSION_TYPE": "wayland", "VOCALINUX_FORCE_BACKEND": "ydotool"}, + clear=True, + ), + patch("vocalinux.text_injection.text_injector.is_ibus_available", return_value=True), + patch("vocalinux.text_injection.text_injector.IBusTextInjector") as mock_ibus_class, + patch.object(obj, "_ensure_ydotoold", return_value=True) as mock_ensure, + patch( + "shutil.which", + side_effect=lambda cmd: f"/usr/bin/{cmd}" if cmd in ("wtype", "ydotool") else None, + ), + ): + obj._check_dependencies() + + self.assertEqual(obj.wayland_tool, "ydotool") + mock_ensure.assert_called_once() + mock_ibus_class.assert_not_called() + + def test_force_backend_ibus_bypasses_reachability_guards(self): + """VOCALINUX_FORCE_BACKEND=ibus selects IBus even on an unbridged compositor.""" + from vocalinux.text_injection.text_injector import DesktopEnvironment, TextInjector + + obj = _make_injector(DesktopEnvironment.WAYLAND) + + with ( + patch.dict( + os.environ, + { + "XDG_SESSION_TYPE": "wayland", + "XDG_CURRENT_DESKTOP": "Hyprland", + "VOCALINUX_FORCE_BACKEND": "ibus", + }, + clear=True, + ), + patch("vocalinux.text_injection.text_injector.is_ibus_available", return_value=True), + patch( + "vocalinux.text_injection.text_injector.is_ibus_active_input_method", + return_value=False, + ), + patch( + "vocalinux.text_injection.text_injector.is_ibus_daemon_running", + return_value=False, + ), + patch("vocalinux.text_injection.text_injector.IBusTextInjector") as mock_ibus_class, + patch.object(obj, "_start_ibus_initialization"), + # Even with no bridge and no daemon, the explicit override wins. + patch.object(TextInjector, "_ibus_wayland_bridge_running", return_value=False), + patch( + "shutil.which", + side_effect=lambda cmd: "/usr/bin/wtype" if cmd == "wtype" else None, + ), + ): + obj._check_dependencies() + + mock_ibus_class.assert_called_once() + def test_kde_wayland_respects_explicit_non_ibus_input_method(self): from vocalinux.text_injection.text_injector import DesktopEnvironment