Files
voculinux/tests/test_text_injector_ext.py
T
d3ba4ee743 fix(injection): send newlines as Return on the xdotool backend (#962)
* fix(injection): send newlines as Return on the xdotool backend

xdotool type maps '\n' to the Linefeed keysym, which GTK targets
silently drop — "new line" produced nothing on X11 even though the
segment reached the injector.  Split each chunk on '\n' and press
Return between typed segments so a newline lands as a real keypress
in every app.

Verified against a live GtkTextView on Xvfb: the same "A\nB" input
that previously produced 'AB' now produces 'A\nB'.

Fixes #955

* fix(commands): consume trailing punctuation after mid-utterance commands

"New paragraph." inside a longer utterance substituted correctly but
left the engine's sentence-final period typed after the break
("\\n\\n. Does this work now?").  Text-command substitution now
consumes a run of trailing , . ! ? ; : … when it precedes whitespace
or the end of the segment — the reporter's comma-phrased case
("…, new paragraph, …") stops leaking the stray comma the same way.
A mark glued to following text ("newline.io") is untouched.

Fixes #955

* fix(injection,commands): resume mid-chunk on retry, don't eat produced marks

Greptile review follow-ups on #962:

- A failed Return mid-chunk left 'typed' at the chunk start, so the
  retry re-sent text xdotool had already delivered.  'typed' now
  advances per typed segment and per Return keypress, so the retry
  resumes exactly at the un-typed position — and the clipboard
  fallback reports the same progress if retries run out.
- _TRAILING_PUNCTUATION could consume a mark produced by an earlier
  command replacement ('hello new line exclamation mark' gave '\\n'
  instead of '\\n!').  The eater now only consumes marks before
  whitespace; the end-of-utterance case ('hello new line.') is a
  pre-pass on the raw segment before any substitution runs.

* style: black-format comment indent and test condition

* fix(commands): strip segment before the end-of-utterance cleanup

Trailing whitespace ('hello new line. ') hid the final mark from the
pre-pass: the raw segment ended in a space, and the stripped segment
left the dot at end-of-string where the mid-utterance eater cannot
touch it.  Strip first, then run the cleanup.

Also add the missing type hints on the new test helper.

---------

Co-authored-by: Jatin K Malik <jatinkrmalik@gmail.com>
2026-10-16 19:05:34 -07:00

2012 lines
87 KiB
Python

"""Extra tests for text_injector.py to improve branch coverage."""
import json
import os
import subprocess
import sys
import threading
import unittest
from typing import Any, Optional, cast
from unittest import mock
from unittest.mock import MagicMock, mock_open, patch
import pytest
if "gi" not in sys.modules:
sys.modules["gi"] = MagicMock()
if "gi.repository" not in sys.modules:
sys.modules["gi.repository"] = MagicMock()
@pytest.fixture(autouse=True)
def _restore_sys_modules():
saved = dict(sys.modules)
yield
added = set(sys.modules.keys()) - set(saved.keys())
for k in added:
del sys.modules[k]
for k, v in saved.items():
if k not in sys.modules or sys.modules[k] is not v:
sys.modules[k] = v
class TestDesktopEnvironmentEnum(unittest.TestCase):
def test_all_values(self):
from vocalinux.text_injection.text_injector import DesktopEnvironment
self.assertEqual(DesktopEnvironment.X11.value, "x11")
self.assertEqual(DesktopEnvironment.WAYLAND.value, "wayland")
self.assertEqual(DesktopEnvironment.X11_IBUS.value, "x11-ibus")
self.assertEqual(DesktopEnvironment.WAYLAND_XDOTOOL.value, "wayland-xdotool")
self.assertEqual(DesktopEnvironment.WAYLAND_IBUS.value, "wayland-ibus")
self.assertEqual(DesktopEnvironment.UNKNOWN.value, "unknown")
class TestKdePlasmaDetection(unittest.TestCase):
def test_detects_xdg_current_desktop_kde(self):
from vocalinux.text_injection.text_injector import _is_kde_plasma_session
with patch.dict(os.environ, {"XDG_CURRENT_DESKTOP": "KDE"}, clear=True):
self.assertTrue(_is_kde_plasma_session())
def test_detects_kde_full_session(self):
from vocalinux.text_injection.text_injector import _is_kde_plasma_session
with patch.dict(os.environ, {"KDE_FULL_SESSION": "true"}, clear=True):
self.assertTrue(_is_kde_plasma_session())
def test_detects_desktop_session_plasma(self):
from vocalinux.text_injection.text_injector import _is_kde_plasma_session
with patch.dict(os.environ, {"DESKTOP_SESSION": "plasma"}, clear=True):
self.assertTrue(_is_kde_plasma_session())
def test_ignores_non_kde_session(self):
from vocalinux.text_injection.text_injector import _is_kde_plasma_session
with patch.dict(os.environ, {"XDG_CURRENT_DESKTOP": "GNOME"}, clear=True):
self.assertFalse(_is_kde_plasma_session())
def _make_injector(env) -> Any:
from vocalinux.text_injection.text_injector import TextInjector
obj = cast(Any, TextInjector.__new__(TextInjector))
obj._ibus_injector = None
obj.environment = env
obj._session_environment = env
obj._ibus_ready = False
obj._ibus_init_failed = False
obj._ibus_init_thread = None
obj._state_lock = threading.Lock()
obj._clipboard_tool_health = {}
obj._clipboard_timeout = 0.35
obj._clipboard_restore_generation = 0
obj._clipboard_restore_target = None
return obj
class TestDetectEnvironment(unittest.TestCase):
def test_detect_wayland(self):
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.WAYLAND)
with patch.dict(os.environ, {"XDG_SESSION_TYPE": "wayland", "WAYLAND_DISPLAY": "w-0"}):
with patch(
"vocalinux.text_injection.text_injector.is_ibus_available", return_value=False
):
with patch(
"vocalinux.text_injection.text_injector.is_ibus_daemon_running",
return_value=False,
):
result = obj._detect_environment()
self.assertIn(
result, [DesktopEnvironment.WAYLAND, DesktopEnvironment.WAYLAND_IBUS]
)
def test_detect_flatpak_prefers_ydotool_when_available(self):
"""Flatpak prefers ydotool so injection reaches native Wayland apps."""
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.X11)
env = {
"FLATPAK_ID": "com.vocalinux.Vocalinux",
"XDG_SESSION_TYPE": "wayland",
"DISPLAY": ":0",
}
with patch.dict(os.environ, env, clear=True):
with patch("vocalinux.text_injection.text_injector.shutil.which") as which:
which.side_effect = lambda name: "/app/bin/ydotool" if name == "ydotool" else None
self.assertEqual(obj._detect_environment(), DesktopEnvironment.WAYLAND)
def test_detect_flatpak_without_ydotool_uses_xwayland(self):
"""Without ydotool, Flatpak falls back to xdotool/XWayland."""
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.X11)
env = {
"FLATPAK_ID": "com.vocalinux.Vocalinux",
"XDG_SESSION_TYPE": "wayland",
"DISPLAY": ":0",
}
with patch.dict(os.environ, env, clear=True):
with patch("vocalinux.text_injection.text_injector.shutil.which", return_value=None):
self.assertEqual(obj._detect_environment(), DesktopEnvironment.WAYLAND_XDOTOOL)
def test_detect_flatpak_with_wayland_socket_stays_wayland(self):
"""If the Wayland socket is exposed, normal Wayland detection still applies."""
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.WAYLAND)
env = {
"FLATPAK_ID": "com.vocalinux.Vocalinux",
"XDG_SESSION_TYPE": "wayland",
"WAYLAND_DISPLAY": "wayland-0",
"DISPLAY": ":0",
}
with patch.dict(os.environ, env, clear=True):
self.assertEqual(obj._detect_environment(), DesktopEnvironment.WAYLAND)
def test_detect_x11(self):
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.X11)
with patch.dict(os.environ, {"XDG_SESSION_TYPE": "x11"}):
with patch(
"vocalinux.text_injection.text_injector.is_ibus_available", return_value=False
):
with patch(
"vocalinux.text_injection.text_injector.is_ibus_daemon_running",
return_value=False,
):
with patch(
"vocalinux.text_injection.text_injector.is_ibus_active_input_method",
return_value=False,
):
result = obj._detect_environment()
self.assertIn(result, [DesktopEnvironment.X11, DesktopEnvironment.X11_IBUS])
# IBus detection tests removed due to test-ordering mock pollution issues
class TestCheckDependencies(unittest.TestCase):
def test_x11_xdotool_available(self):
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.X11)
with patch("shutil.which", return_value="/usr/bin/xdotool"):
obj._check_dependencies()
def test_wayland_wtype_available(self):
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.WAYLAND)
with patch(
"shutil.which", side_effect=lambda x: "/usr/bin/wtype" if x == "wtype" else None
):
obj._check_dependencies()
def test_wayland_no_tools(self):
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.WAYLAND)
with patch("shutil.which", return_value=None):
with patch(
"vocalinux.text_injection.text_injector.is_ibus_available", return_value=False
):
with self.assertRaises(RuntimeError):
obj._check_dependencies()
def test_kde_wayland_skips_leftover_ibus_when_daemon_runs_with_xkb_engine(self):
"""Leftover IBus on KDE is skipped so scoped inject cannot fake success (#752)."""
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.WAYLAND)
with (
patch.dict(
os.environ,
{
"XDG_SESSION_TYPE": "wayland",
"XDG_CURRENT_DESKTOP": "KDE",
"XMODIFIERS": "@im=none",
},
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") as mock_start,
patch(
"shutil.which",
side_effect=lambda cmd: "/usr/bin/wtype" if cmd == "wtype" else None,
),
):
obj._check_dependencies()
mock_ibus_class.assert_not_called()
self.assertIsNone(obj._ibus_injector)
mock_start.assert_not_called()
self.assertEqual(obj.wayland_tool, "wtype")
def test_gnome_wayland_uses_ibus_when_daemon_runs_with_xkb_engine(self):
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.WAYLAND)
mock_ibus = MagicMock()
with (
patch.dict(
os.environ,
{"XDG_SESSION_TYPE": "wayland", "XDG_CURRENT_DESKTOP": "GNOME"},
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",
return_value=mock_ibus,
),
patch.object(obj, "_start_ibus_initialization") as mock_start,
patch(
"shutil.which",
side_effect=lambda cmd: "/usr/bin/wtype" if cmd == "wtype" else None,
),
):
obj._check_dependencies()
self.assertIs(obj._ibus_injector, mock_ibus)
mock_start.assert_called_once_with()
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, TextInjector
obj = _make_injector(DesktopEnvironment.WAYLAND)
with (
patch.dict(
os.environ,
{"XDG_SESSION_TYPE": "wayland", "XDG_CURRENT_DESKTOP": "sway"},
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(
"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_config_backend_wtype_skips_ibus(self):
"""text_injection.backend=wtype pins wtype with no environment variable set.
The environment variable is a one-off; this is the persisted setting from
issue #476, so it has to survive a restart on its own.
"""
import json as _json
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.WAYLAND)
config = _json.dumps({"text_injection": {"backend": "wtype"}})
with (
patch.dict(os.environ, {"XDG_SESSION_TYPE": "wayland"}, clear=True),
patch("vocalinux.text_injection.text_injector.config_dir", return_value="/fake/config"),
patch("os.path.exists", return_value=True),
patch("builtins.open", mock_open(read_data=config)),
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_x11_config_pin_never_constructs_ibus(self):
"""On X11 the user guide promises any non-ibus pin turns the IBus path off.
The X11 reporter on #476 wanted exactly that. The pin does not choose the
tool there -- xdotool always does the typing -- so the only thing the pin
changes on X11 is whether IBus runs, and that is what this asserts.
"""
import json as _json
from vocalinux.text_injection.text_injector import DesktopEnvironment
def run(pinned: Optional[str]) -> MagicMock:
"""Drive _check_dependencies on X11 under conditions IBus would pass."""
obj = _make_injector(DesktopEnvironment.X11)
section = {"backend": pinned} if pinned else {}
config = _json.dumps({"text_injection": section})
with (
patch.dict(os.environ, {"XDG_SESSION_TYPE": "x11"}, clear=True),
patch(
"vocalinux.text_injection.text_injector.config_dir",
return_value="/fake/config",
),
patch("os.path.exists", return_value=True),
patch("builtins.open", mock_open(read_data=config)),
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=True,
),
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("shutil.which", side_effect=lambda c: f"/usr/bin/{c}"),
):
obj._check_dependencies()
return mock_ibus_class
# Control: with no pin these same conditions do construct IBus, so a
# not-called assertion below is the pin's doing and not the environment's.
self.assertTrue(run(None).called, "control failed: IBus was not constructed without a pin")
for pinned in ("xdotool", "wtype"):
with self.subTest(backend=pinned):
run(pinned).assert_not_called()
def test_explicit_auto_environment_ignores_the_saved_config_pin(self):
"""VOCALINUX_FORCE_BACKEND=auto is how you get one run of plain autodetection.
The resolver already knows an explicit auto from an unset variable; this
drives the whole of _check_dependencies to show the saved pin really is
not consulted, rather than only that the resolver returned "auto".
"""
import json as _json
from vocalinux.text_injection.text_injector import DesktopEnvironment, TextInjector
obj = _make_injector(DesktopEnvironment.WAYLAND)
obj._backend_pin = ("auto", None)
config = _json.dumps({"text_injection": {"backend": "wtype"}})
with (
patch.dict(
os.environ,
{"XDG_SESSION_TYPE": "wayland", "VOCALINUX_FORCE_BACKEND": "auto"},
clear=True,
),
patch("vocalinux.text_injection.text_injector.config_dir", return_value="/fake/config"),
patch("os.path.exists", return_value=True),
patch("builtins.open", mock_open(read_data=config)),
patch("vocalinux.text_injection.text_injector.is_ibus_available", return_value=False),
patch.object(TextInjector, "_ensure_ydotoold", lambda s: True),
patch("shutil.which", side_effect=lambda c: f"/usr/bin/{c}"),
self.assertLogs("vocalinux.text_injection.text_injector", level="DEBUG") as logs,
):
obj._check_dependencies()
obj._warn_if_pin_not_honoured(*obj._backend_pin)
# Autodetection prefers ydotool when its daemon is ready; the saved wtype
# pin would have won had the explicit auto not discarded it.
self.assertEqual(obj.wayland_tool, "ydotool")
self.assertFalse(
[line for line in logs.output if "backend pin requested" in line],
f"an explicit auto still reported a pin: {logs.output}",
)
self.assertFalse(self._not_applied(logs.output))
def _run_ibus_pin(
self,
bridges: bool,
desktop: str = "Hyprland",
env: Any = None,
config_pin: bool = False,
) -> tuple[list[str], int]:
"""Drive _check_dependencies with an ibus pin; return (logs, bridge_call_count)."""
from vocalinux.text_injection.text_injector import DesktopEnvironment, TextInjector
obj = _make_injector(env or DesktopEnvironment.WAYLAND)
calls = []
def fake_bridges(self):
calls.append(1)
return bridges
environ = {
"XDG_SESSION_TYPE": "wayland",
"XDG_CURRENT_DESKTOP": desktop,
}
if not config_pin:
environ["VOCALINUX_FORCE_BACKEND"] = "ibus"
with (
patch.dict(
os.environ,
environ,
clear=True,
),
patch("vocalinux.text_injection.text_injector.config_dir", return_value="/fake/config"),
patch("os.path.exists", return_value=True),
patch(
"builtins.open",
mock_open(read_data=json.dumps({"text_injection": {"backend": "ibus"}})),
),
patch.object(TextInjector, "_wayland_compositor_bridges_ibus", fake_bridges),
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=True,
),
patch(
"vocalinux.text_injection.text_injector.is_ibus_daemon_running", return_value=True
),
patch("vocalinux.text_injection.text_injector.IBusTextInjector"),
patch.object(TextInjector, "_start_ibus_initialization", lambda s: None),
# shutil.which is patched to claim every tool exists, so without this
# the real probe shells out to a ydotoold that may not be installed
# and logs a warning that has nothing to do with the pin.
patch.object(TextInjector, "_ensure_ydotoold", lambda s: True),
patch("shutil.which", side_effect=lambda c: f"/usr/bin/{c}"),
self.assertLogs("vocalinux.text_injection.text_injector", level="DEBUG") as logs,
):
obj._check_dependencies()
return logs.output, len(calls)
@staticmethod
def _silent_failure_warnings(output):
"""Only the denylist-bypass warning, not every warning the run emits."""
return [line for line in output if "may silently do nothing" in line]
def test_ibus_pin_on_unbridged_compositor_warns_about_silent_failure(self):
"""Pinning ibus can re-enable the silent drop the denylist exists to stop."""
output, _ = self._run_ibus_pin(bridges=False)
self.assertTrue(
self._silent_failure_warnings(output),
f"expected a silent-failure warning, got: {output}",
)
def test_ibus_pin_on_bridged_compositor_does_not_warn(self):
"""The guard must key off the compositor, not merely off the pin.
A bridged desktop -- including a denylisted compositor with the
ibus-wayland bridge running -- is not a bypass and must stay quiet.
"""
output, _ = self._run_ibus_pin(bridges=True)
self.assertFalse(
self._silent_failure_warnings(output),
f"expected no silent-failure warning on a bridged compositor, got: {output}",
)
def test_config_ibus_pin_checks_the_compositor_bridge(self):
"""A saved IBus pin follows the same denylist warning as an env pin."""
for bridges in (False, True):
with self.subTest(bridges=bridges):
output, _ = self._run_ibus_pin(bridges=bridges, config_pin=True)
if bridges:
self.assertFalse(self._silent_failure_warnings(output))
else:
self.assertTrue(self._silent_failure_warnings(output))
self.assertTrue([line for line in output if "text_injection.backend=ibus" in line])
def test_ibus_pin_never_double_checks_the_compositor(self):
"""The bridging check shells out (pgrep / gdbus), so it must run at most once.
The elif chain short-circuits on ``not force_ibus``, so a pinned run
reaches the new guard without having evaluated the check already. If a
future edit drops that short-circuit, this catches the extra subprocess.
"""
for bridges in (False, True):
_, call_count = self._run_ibus_pin(bridges=bridges)
self.assertEqual(call_count, 1, f"bridges={bridges}: expected 1 call, got {call_count}")
def test_ibus_pin_on_x11_does_not_warn(self):
"""The denylist is a Wayland concern; X11 reaches apps through XIM."""
from vocalinux.text_injection.text_injector import DesktopEnvironment
output, _ = self._run_ibus_pin(bridges=True, env=DesktopEnvironment.X11)
self.assertFalse(self._silent_failure_warnings(output))
def _run_with_pin(
self,
pin: Optional[str],
tools: tuple[str, ...],
ibus_available: bool = True,
bridges: bool = True,
env: Any = None,
ydotoold_ready: Optional[bool] = None,
ibus_ready: bool = False,
) -> tuple[Any, list[str]]:
"""Construct through _check_dependencies with a pin; return (injector, logs).
ydotoold_ready defaults to whether ydotool is installed, which is what
every caller wanted before the daemonless fallback needed exercising.
``ibus_ready`` promotes a pinned IBus path before reporting, modeling
the completed background initialization used by the honoured-pin cases.
Pass it explicitly to have ydotool present but its daemon unavailable.
"""
from vocalinux.text_injection.text_injector import DesktopEnvironment, TextInjector
obj = _make_injector(env or DesktopEnvironment.WAYLAND)
obj._backend_pin = ("auto", None)
environ = {
"XDG_SESSION_TYPE": "x11" if env == DesktopEnvironment.X11 else "wayland",
"XDG_CURRENT_DESKTOP": "Hyprland",
}
if pin:
environ["VOCALINUX_FORCE_BACKEND"] = pin
with (
patch.dict(os.environ, environ, clear=True),
patch(
"vocalinux.text_injection.text_injector.is_ibus_available",
return_value=ibus_available,
),
patch(
"vocalinux.text_injection.text_injector.is_ibus_active_input_method",
return_value=True,
),
patch(
"vocalinux.text_injection.text_injector.is_ibus_daemon_running", return_value=True
),
patch.object(TextInjector, "_wayland_compositor_bridges_ibus", lambda s: bridges),
patch("vocalinux.text_injection.text_injector.IBusTextInjector"),
patch.object(TextInjector, "_start_ibus_initialization", lambda s: None),
patch.object(
TextInjector,
"_ensure_ydotoold",
lambda s: ("ydotool" in tools) if ydotoold_ready is None else ydotoold_ready,
),
patch("shutil.which", side_effect=lambda c: f"/usr/bin/{c}" if c in tools else None),
self.assertLogs("vocalinux.text_injection.text_injector", level="DEBUG") as logs,
):
obj._check_dependencies()
if ibus_ready:
obj.environment = (
DesktopEnvironment.X11_IBUS
if env == DesktopEnvironment.X11
else DesktopEnvironment.WAYLAND_IBUS
)
obj._ibus_ready = True
obj._warn_if_pin_not_honoured(*obj._backend_pin)
return obj, logs.output
def _not_applied(self, output):
return [line for line in output if "was not applied" in line]
def test_pin_with_missing_binary_warns_and_names_the_reason(self):
"""A pin for a tool that is not installed must not resume autodetection silently."""
_, output = self._run_with_pin("wtype", tools=("ydotool",))
warned = self._not_applied(output)
self.assertTrue(warned, f"expected a not-applied warning, got: {output}")
self.assertIn("is not installed", warned[0])
self.assertIn("using ydotool instead", warned[0])
def test_pin_that_does_not_apply_to_this_session_warns_without_a_reason(self):
"""xdotool on Wayland resolves to whatever autodetection picks; say so plainly."""
_, output = self._run_with_pin("xdotool", tools=("wtype", "ydotool", "xdotool"))
warned = self._not_applied(output)
self.assertTrue(warned, f"expected a not-applied warning, got: {output}")
self.assertNotIn("is not installed", warned[0])
def test_ibus_pin_warns_when_ibus_support_is_unavailable(self):
_, output = self._run_with_pin("ibus", tools=("wtype", "ydotool"), ibus_available=False)
warned = self._not_applied(output)
self.assertTrue(warned, f"expected a not-applied warning, got: {output}")
self.assertIn("IBus support is not available", warned[0])
def test_honoured_pin_does_not_warn(self):
_, output = self._run_with_pin("wtype", tools=("wtype", "ydotool"))
self.assertFalse(self._not_applied(output))
def test_no_pin_never_warns(self):
_, output = self._run_with_pin(None, tools=("wtype", "ydotool"))
self.assertFalse(self._not_applied(output))
def test_honoured_ibus_pin_on_unbridged_compositor_warns_only_once(self):
"""The denylist-bypass warning and this one must not both fire.
The pin IS honoured there, so the backend in use matches what was
pinned; only the bypass warning is appropriate.
"""
_, output = self._run_with_pin(
"ibus", tools=("wtype", "ydotool"), bridges=False, ibus_ready=True
)
self.assertTrue([line for line in output if "overrides that check" in line])
self.assertFalse(self._not_applied(output))
def test_honoured_ibus_pin_on_bridged_compositor_warns_neither_way(self):
_, output = self._run_with_pin(
"ibus", tools=("wtype", "ydotool"), bridges=True, ibus_ready=True
)
self.assertFalse([line for line in output if "overrides that check" in line])
self.assertFalse(self._not_applied(output))
def test_pinned_log_line_names_the_config_source_not_the_variable(self):
"""A config-sourced pin must not be reported as VOCALINUX_FORCE_BACKEND."""
from vocalinux.text_injection.text_injector import DesktopEnvironment, TextInjector
obj = _make_injector(DesktopEnvironment.WAYLAND)
obj._backend_pin = ("auto", None)
config = json.dumps({"text_injection": {"backend": "wtype"}})
with (
patch.dict(os.environ, {"XDG_SESSION_TYPE": "wayland"}, clear=True),
patch("vocalinux.text_injection.text_injector.config_dir", return_value="/fake/config"),
patch("os.path.exists", return_value=True),
patch("builtins.open", mock_open(read_data=config)),
patch("vocalinux.text_injection.text_injector.is_ibus_available", return_value=False),
patch.object(TextInjector, "_ensure_ydotoold", lambda s: False),
patch("shutil.which", side_effect=lambda c: f"/usr/bin/{c}" if c == "wtype" else None),
self.assertLogs("vocalinux.text_injection.text_injector", level="DEBUG") as logs,
):
obj._check_dependencies()
pinned_lines = [line for line in logs.output if "wtype" in line and "=" in line]
self.assertTrue(pinned_lines)
self.assertFalse(
[line for line in pinned_lines if "VOCALINUX_FORCE_BACKEND" in line],
f"config-sourced pin reported as the env var: {pinned_lines}",
)
def test_x11_pin_resolving_to_xdotool_warns(self):
"""On X11 the tool is always xdotool, so a wtype pin cannot be honoured.
This is the X11 reporter's case on #476: they wanted IBus off, which a
non-ibus pin does, but the pin they set is not what ends up typing.
"""
from vocalinux.text_injection.text_injector import DesktopEnvironment
_, output = self._run_with_pin(
"wtype", tools=("xdotool", "wtype"), env=DesktopEnvironment.X11
)
warned = self._not_applied(output)
self.assertTrue(warned, f"expected a not-applied warning, got: {output}")
self.assertIn("using xdotool instead", warned[0])
def test_ibus_pin_that_fails_to_initialise_warns_without_a_reason(self):
"""IBus is available but construction blew up, so no cheap reason applies."""
from vocalinux.text_injection.text_injector import DesktopEnvironment, TextInjector
obj = _make_injector(DesktopEnvironment.WAYLAND)
obj._backend_pin = ("auto", None)
with (
patch.dict(
os.environ,
{"XDG_SESSION_TYPE": "wayland", "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=True,
),
patch(
"vocalinux.text_injection.text_injector.is_ibus_daemon_running", return_value=True
),
patch.object(TextInjector, "_wayland_compositor_bridges_ibus", lambda s: True),
patch(
"vocalinux.text_injection.text_injector.IBusTextInjector",
side_effect=RuntimeError("boom"),
),
patch.object(TextInjector, "_ensure_ydotoold", lambda s: True),
patch("shutil.which", side_effect=lambda c: f"/usr/bin/{c}"),
self.assertLogs("vocalinux.text_injection.text_injector", level="DEBUG") as logs,
):
obj._check_dependencies()
obj._warn_if_pin_not_honoured(*obj._backend_pin)
warned = self._not_applied(logs.output)
self.assertTrue(warned, f"expected a not-applied warning, got: {logs.output}")
self.assertNotIn("not available", warned[0])
self.assertNotIn("not installed", warned[0])
def test_ydotool_pin_with_missing_binary_warns_and_names_the_reason(self):
"""The wtype case has a counterpart: every selectable tool can be absent.
Contrast with test_pin_with_missing_binary_warns_and_names_the_reason,
which pins the tool the fallback then chooses; here the pin and the
fallback are swapped, so a check keyed on one name would still pass.
"""
_, output = self._run_with_pin("ydotool", tools=("wtype",))
warned = self._not_applied(output)
self.assertTrue(warned, f"expected a not-applied warning, got: {output}")
self.assertIn("ydotool is not installed", warned[0])
self.assertIn("using wtype instead", warned[0])
def test_wtype_pin_without_wtype_falls_back_to_daemonless_ydotool(self):
"""ydotool present but ydotoold not ready is its own branch, and a pin reaches it.
Both diagnostics have to survive together: the fallback says the daemon
is missing, and the pin check says the pin is not what ended up typing.
"""
obj, output = self._run_with_pin("wtype", tools=("ydotool",), ydotoold_ready=False)
self.assertEqual(obj.wayland_tool, "ydotool")
self.assertTrue(
[line for line in output if "ydotoold not ready" in line],
f"expected the daemonless-ydotool warning, got: {output}",
)
warned = self._not_applied(output)
self.assertTrue(warned, f"expected a not-applied warning, got: {output}")
self.assertIn("using ydotool instead", warned[0])
def test_ibus_pin_survives_the_x11_missing_xdotool_early_return(self):
"""_check_dependencies returns early here, before any tool is selected.
The pin is still honoured -- IBus is what types -- so the pin check must
not report a failure just because the usual selection chain was skipped.
"""
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj, output = self._run_with_pin("ibus", tools=(), env=DesktopEnvironment.X11)
self.assertIsNotNone(obj._ibus_injector)
self.assertFalse(
self._not_applied(output),
f"honoured ibus pin reported as not applied: {output}",
)
def test_ibus_pin_survives_the_no_tools_early_return(self):
"""The other early return: Wayland with nothing installed but IBus asked for.
Without the early return this path raises RuntimeError, so reaching the
pin check at all is part of what this asserts.
"""
obj, output = self._run_with_pin("ibus", tools=())
self.assertIsNotNone(obj._ibus_injector)
self.assertFalse(
self._not_applied(output),
f"honoured ibus pin reported as not applied: {output}",
)
def test_pin_log_states_the_request_without_claiming_the_outcome(self):
"""The pin line runs before anything is checked, so it may not claim success.
A pin that is not installed is reported by this line and then
contradicted by the not-applied warning, which is what made an
outcome-shaped word wrong here.
"""
_, output = self._run_with_pin("wtype", tools=("ydotool",))
requested = [line for line in output if "backend pin requested" in line]
self.assertTrue(requested, f"expected the pin-requested line, got: {output}")
self.assertFalse(
[line for line in output if "overriding" in line],
f"pin line still claims an outcome it cannot know: {output}",
)
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
obj = _make_injector(DesktopEnvironment.WAYLAND)
with (
patch.dict(
os.environ,
{
"XDG_SESSION_TYPE": "wayland",
"XDG_CURRENT_DESKTOP": "KDE",
"QT_IM_MODULE": "fcitx",
},
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(
"shutil.which",
side_effect=lambda cmd: "/usr/bin/ydotool" if cmd == "ydotool" else None,
),
patch.object(obj, "_is_ydotoold_running", return_value=True),
):
obj._check_dependencies()
mock_ibus_class.assert_not_called()
self.assertEqual(obj.wayland_tool, "ydotool")
def test_kde_wayland_wtype_probe_logs_ibus_hint(self):
from vocalinux.text_injection.text_injector import DesktopEnvironment, TextInjector
def which(cmd):
if cmd in {"wtype", "xdotool"}:
return f"/usr/bin/{cmd}"
return None
with patch.dict(
os.environ,
{"XDG_SESSION_TYPE": "wayland", "XDG_CURRENT_DESKTOP": "KDE"},
clear=True,
):
with patch(
"vocalinux.text_injection.text_injector.is_ibus_available",
return_value=False,
):
with patch("shutil.which", side_effect=which):
with patch("subprocess.run") as mock_run:
mock_run.return_value = MagicMock(
returncode=1,
stderr="compositor does not support virtual keyboard",
)
with patch.object(TextInjector, "_test_xdotool_fallback"):
with self.assertLogs(
"vocalinux.text_injection.text_injector",
level="WARNING",
) as logs:
injector = TextInjector()
log_output = "\n".join(logs.output)
self.assertEqual(injector.environment, DesktopEnvironment.WAYLAND_XDOTOOL)
self.assertIn("KDE Plasma Wayland detected", log_output)
self.assertIn("IBus Wayland", log_output)
class TestRecoverFromFallback(unittest.TestCase):
def test_wtype_recovery_probe_is_non_destructive(self):
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.WAYLAND_XDOTOOL)
with patch(
"vocalinux.text_injection.text_injector.shutil.which",
side_effect=lambda cmd: "/usr/bin/wtype" if cmd == "wtype" else None,
):
with patch("vocalinux.text_injection.text_injector.subprocess.run") as mock_run:
mock_run.return_value = MagicMock(returncode=0, stderr="")
self.assertTrue(obj._try_recover_from_fallback())
mock_run.assert_called_once_with(
["wtype", ""], stderr=subprocess.PIPE, text=True, check=False, timeout=2, env=mock.ANY
)
class TestInjectText(unittest.TestCase):
def test_inject_x11(self):
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.X11)
with patch("subprocess.run") as mock_run:
result = obj.inject_text("hello")
# Verify that subprocess.run was called (by _inject_with_xdotool)
self.assertTrue(mock_run.called)
self.assertTrue(result)
def test_inject_wayland(self):
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.WAYLAND)
obj.wayland_tool = "wtype"
with patch("subprocess.run") as mock_run:
result = obj.inject_text("hello")
# Verify that subprocess.run was called (by _inject_with_wayland_tool)
self.assertTrue(mock_run.called)
self.assertTrue(result)
def test_kde_wayland_wtype_failure_logs_ibus_hint_and_uses_xdotool(self):
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.WAYLAND)
obj.wayland_tool = "wtype"
error = subprocess.CalledProcessError(
1,
["wtype", "hello"],
stderr="compositor does not support virtual keyboard",
)
with patch.dict(os.environ, {"XDG_CURRENT_DESKTOP": "KDE"}, clear=True):
with patch.object(obj, "_inject_with_wayland_tool", side_effect=error):
with patch.object(obj, "_inject_with_xdotool") as mock_xdotool:
with patch("shutil.which", return_value="/usr/bin/xdotool"):
with self.assertLogs(
"vocalinux.text_injection.text_injector",
level="WARNING",
) as logs:
result = obj.inject_text("hello")
self.assertTrue(result)
self.assertEqual(obj.environment, DesktopEnvironment.WAYLAND_XDOTOOL)
mock_xdotool.assert_called_once_with("hello")
log_output = "\n".join(logs.output)
self.assertIn("KDE Plasma Wayland rejected virtual keyboard injection", log_output)
self.assertIn("IBus Wayland", log_output)
def test_inject_ibus(self):
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.X11_IBUS)
mock_ibus = MagicMock()
mock_ibus.inject_text.return_value = True
obj._ibus_injector = mock_ibus
result = obj.inject_text("hello")
mock_ibus.inject_text.assert_called_once_with("hello")
def test_inject_wayland_ibus(self):
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.WAYLAND_IBUS)
mock_ibus = MagicMock()
mock_ibus.inject_text.return_value = True
obj._ibus_injector = mock_ibus
result = obj.inject_text("hello")
mock_ibus.inject_text.assert_called_once_with("hello")
def test_inject_xwayland(self):
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.WAYLAND_XDOTOOL)
with patch("subprocess.run") as mock_run:
result = obj.inject_text("hello")
# Verify that subprocess.run was called (by _inject_with_xdotool)
self.assertTrue(mock_run.called)
self.assertTrue(result)
def test_inject_newline_only_segment_reaches_backend(self) -> None:
"""A bare "\n" from "new line" is typed, not skipped as empty (#955)."""
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.X11)
with patch.object(obj, "_inject_with_xdotool") as mock_xdotool:
self.assertTrue(obj.inject_text("\n"))
mock_xdotool.assert_called_once_with("\n")
self.assertEqual(obj.last_typed_count, 1)
def test_inject_paragraph_break_segment_reaches_backend(self) -> None:
"""A bare "\n\n" from "new paragraph" is typed, not skipped (#955)."""
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.X11)
with patch.object(obj, "_inject_with_xdotool") as mock_xdotool:
self.assertTrue(obj.inject_text("\n\n"))
mock_xdotool.assert_called_once_with("\n\n")
self.assertEqual(obj.last_typed_count, 2)
def test_inject_newline_segment_reaches_ibus(self) -> None:
"""The IBus backend also receives a bare newline segment (#955)."""
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.X11_IBUS)
mock_ibus = MagicMock()
mock_ibus.inject_text.return_value = True
obj._ibus_injector = mock_ibus
self.assertTrue(obj.inject_text("\n"))
mock_ibus.inject_text.assert_called_once_with("\n")
def test_inject_empty_string_still_skipped(self) -> None:
"""A truly empty segment is still a cheap no-op."""
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.X11)
with patch.object(obj, "_inject_with_xdotool") as mock_xdotool:
self.assertTrue(obj.inject_text(""))
mock_xdotool.assert_not_called()
self.assertEqual(obj.last_typed_count, 0)
@staticmethod
def _xdotool_calls(run_mock: MagicMock) -> list:
"""Return the xdotool type/key argv sequences run_mock received."""
return [
c.args[0]
for c in run_mock.call_args_list
if c.args and c.args[0][0] == "xdotool" and c.args[0][1] in ("type", "key")
]
def test_xdotool_newline_sent_as_return_key(self) -> None:
"""'\n' is `xdotool key Return`, never a Linefeed inside `type` (#955)."""
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.X11)
with patch("subprocess.run") as mock_run:
self.assertTrue(obj.inject_text("\n"))
self.assertEqual(
self._xdotool_calls(mock_run),
[["xdotool", "key", "--clearmodifiers", "Return"]],
)
def test_xdotool_newlines_split_chunk_into_type_and_return(self) -> None:
"""A mid-utterance '\n\n' splits the chunk: type, Return, Return (#955)."""
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.X11)
with patch("subprocess.run") as mock_run:
self.assertTrue(obj.inject_text("Testing 123, \n\n, works?"))
self.assertEqual(
self._xdotool_calls(mock_run),
[
["xdotool", "type", "--clearmodifiers", "--", "Testing 123, "],
["xdotool", "key", "--clearmodifiers", "Return"],
["xdotool", "key", "--clearmodifiers", "Return"],
["xdotool", "type", "--clearmodifiers", "--", ", wor"],
["xdotool", "type", "--clearmodifiers", "--", "ks?"],
],
)
def test_xdotool_paragraph_break_presses_return_twice(self) -> None:
"""'\n\n' from "new paragraph" is two Return presses, no type call (#955)."""
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.X11)
with patch("subprocess.run") as mock_run:
self.assertTrue(obj.inject_text("\n\n"))
self.assertEqual(
self._xdotool_calls(mock_run),
[
["xdotool", "key", "--clearmodifiers", "Return"],
["xdotool", "key", "--clearmodifiers", "Return"],
],
)
def test_xdotool_retry_resumes_after_delivered_segment(self) -> None:
"""A failed Return retries from the '\n', not from the top (#955)."""
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.X11)
failed_once = {"done": False}
def flaky_run(cmd: list, **kwargs: object) -> MagicMock:
if cmd[:2] == ["xdotool", "key"] and cmd[-1] == "Return" and not failed_once["done"]:
failed_once["done"] = True
raise subprocess.CalledProcessError(1, cmd)
return MagicMock(returncode=0)
with patch("subprocess.run", side_effect=flaky_run) as mock_run:
self.assertTrue(obj.inject_text("abc\ndef"))
self.assertEqual(
self._xdotool_calls(mock_run),
[
["xdotool", "type", "--clearmodifiers", "--", "abc"],
["xdotool", "key", "--clearmodifiers", "Return"],
["xdotool", "key", "--clearmodifiers", "Return"],
["xdotool", "type", "--clearmodifiers", "--", "def"],
],
)
class TestLogWindowInfo(unittest.TestCase):
def test_log_x11(self):
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.X11)
with patch.object(obj, "_log_x11_window_info") as mock_log:
obj._log_current_window_info()
# Verify that _log_x11_window_info was called for X11 environment
mock_log.assert_called_once()
def test_log_wayland(self):
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.WAYLAND)
# For pure Wayland, _log_current_window_info logs a debug message instead
# Just verify it doesn't raise
obj._log_current_window_info()
def test_log_xwayland(self):
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.WAYLAND_XDOTOOL)
with patch.object(obj, "_log_x11_window_info") as mock_log:
obj._log_current_window_info()
# Verify that _log_x11_window_info was called for WAYLAND_XDOTOOL environment
mock_log.assert_called_once()
def test_log_exception(self):
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.X11)
with patch.object(obj, "_log_x11_window_info", side_effect=Exception("err")):
obj._log_current_window_info() # Should not raise
class TestInjectKeyboardShortcut(unittest.TestCase):
def test_inject_shortcut_x11(self):
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.X11)
with patch.object(obj, "_inject_shortcut_with_xdotool", return_value=True):
result = obj._inject_keyboard_shortcut("ctrl+a")
self.assertTrue(result)
def test_inject_shortcut_wayland(self):
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.WAYLAND)
obj.wayland_tool = "wtype"
with patch.object(obj, "_inject_shortcut_with_wayland_tool", return_value=True):
result = obj._inject_keyboard_shortcut("ctrl+a")
self.assertTrue(result)
def test_inject_shortcut_xwayland(self):
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.WAYLAND_XDOTOOL)
with patch.object(obj, "_inject_shortcut_with_xdotool", return_value=True):
result = obj._inject_keyboard_shortcut("ctrl+a")
self.assertTrue(result)
class TestShortcutWithXdotool(unittest.TestCase):
def test_success(self):
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.X11)
with patch("subprocess.run"):
result = obj._inject_shortcut_with_xdotool("ctrl+a")
self.assertTrue(result)
def test_failure(self):
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.X11)
with patch("subprocess.run", side_effect=subprocess.CalledProcessError(1, "xdotool")):
result = obj._inject_shortcut_with_xdotool("ctrl+a")
self.assertFalse(result)
def _make_shortcut_injector(tool, legacy=False):
"""Injector wired for _inject_shortcut_with_wayland_tool.
``_make_injector`` sets neither ``wayland_tool`` nor the dialect cache, and
an unpinned dialect would run the real ``key --help`` probe.
"""
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.WAYLAND)
obj.wayland_tool = tool
obj._ydotool_legacy_named_keys = legacy
# _ensure_ydotoold spawns a real Popen, which patch("subprocess.run") misses.
obj._ensure_ydotoold = MagicMock(return_value=True)
obj._wait_for_modifiers_released = MagicMock()
return obj
class TestShortcutWithWaylandTool(unittest.TestCase):
def test_wtype_chords_modifiers(self):
obj = _make_shortcut_injector("wtype")
with patch("subprocess.run") as mock_run:
result = obj._inject_shortcut_with_wayland_tool("ctrl+a")
self.assertTrue(result)
self.assertEqual(
mock_run.call_args[0][0],
["wtype", "-M", "ctrl", "-k", "a", "-m", "ctrl"],
)
def test_ydotool_success(self):
obj = _make_shortcut_injector("ydotool", legacy=False)
with patch("subprocess.run") as mock_run:
result = obj._inject_shortcut_with_wayland_tool("ctrl+a")
self.assertTrue(result)
# Raw keycodes, not the literal string "ctrl+a": ctrl=29, a=30.
self.assertEqual(
mock_run.call_args[0][0],
["ydotool", "key", "29:1", "30:1", "30:0", "29:0"],
)
def test_ydotool_legacy_uses_named_chord(self):
"""0.1.x takes named sequences; raw 1.x codes would type digits there."""
obj = _make_shortcut_injector("ydotool", legacy=True)
with patch("subprocess.run") as mock_run:
self.assertTrue(obj._inject_shortcut_with_wayland_tool("ctrl+a"))
self.assertEqual(mock_run.call_args[0][0], ["ydotool", "key", "ctrl+a"])
def test_ydotool_legacy_multi_step_is_one_call(self):
"""0.1.x takes any number of sequences, one per step, in a single call."""
obj = _make_shortcut_injector("ydotool", legacy=True)
with patch("subprocess.run") as mock_run:
self.assertTrue(obj._inject_shortcut_with_wayland_tool("Home+shift+End"))
self.assertEqual(
mock_run.call_args[0][0],
["ydotool", "key", "home", "shift+end"],
)
self.assertEqual(mock_run.call_count, 1)
def test_ydotool_legacy_rejects_name_with_no_0_1_x_spelling(self):
"""0.1.x maps an unknown name to its first letter, so never send one."""
obj = _make_shortcut_injector("ydotool", legacy=True)
with patch("subprocess.run") as mock_run:
self.assertFalse(obj._inject_shortcut_with_wayland_tool("altgr+a"))
mock_run.assert_not_called()
def test_sequential_steps_not_treated_as_one_chord(self):
""" "Home+shift+End" is press Home, then Shift+End -- two steps."""
obj = _make_shortcut_injector("wtype")
with patch("subprocess.run") as mock_run:
self.assertTrue(obj._inject_shortcut_with_wayland_tool("Home+shift+End"))
self.assertEqual(
mock_run.call_args[0][0],
["wtype", "-k", "Home", "-M", "shift", "-k", "End", "-m", "shift"],
)
def test_unknown_ydotool_keycode_fails_loudly(self):
obj = _make_shortcut_injector("ydotool", legacy=False)
with patch("subprocess.run") as mock_run:
self.assertFalse(obj._inject_shortcut_with_wayland_tool("ctrl+F13"))
mock_run.assert_not_called()
def test_waits_for_modifiers_before_injecting(self):
"""A held PTT modifier would otherwise rewrite the chord."""
obj = _make_shortcut_injector("wtype")
with patch("subprocess.run"):
self.assertTrue(obj._inject_shortcut_with_wayland_tool("ctrl+a"))
obj._wait_for_modifiers_released.assert_called_once()
def test_ydotool_ensures_daemon_but_continues_when_not_ready(self):
"""0.1.x often has no daemon at all, so a warning must not abort."""
obj = _make_shortcut_injector("ydotool", legacy=False)
obj._ensure_ydotoold = MagicMock(return_value=False)
with patch("subprocess.run") as mock_run:
self.assertTrue(obj._inject_shortcut_with_wayland_tool("ctrl+a"))
obj._ensure_ydotoold.assert_called_once()
self.assertTrue(mock_run.called)
def test_timeout_returns_false(self):
obj = _make_shortcut_injector("wtype")
with patch("subprocess.run", side_effect=subprocess.TimeoutExpired("wtype", 3)):
self.assertFalse(obj._inject_shortcut_with_wayland_tool("ctrl+a"))
def test_runs_with_host_env_and_a_timeout(self):
obj = _make_shortcut_injector("wtype")
with patch("subprocess.run") as mock_run:
self.assertTrue(obj._inject_shortcut_with_wayland_tool("ctrl+a"))
kwargs = mock_run.call_args[1]
self.assertIn("env", kwargs)
self.assertGreaterEqual(kwargs["timeout"], 3)
class TestYdotoolLegacyToken(unittest.TestCase):
def test_matches_mains_own_paste_constants(self):
from vocalinux.text_injection.text_injector import TextInjector
self.assertEqual(TextInjector._ydotool_legacy_token(["ctrl"], "v"), "ctrl+v")
self.assertEqual(TextInjector._ydotool_legacy_token(["ctrl", "shift"], "v"), "ctrl+shift+v")
def test_super_is_not_canonicalised_to_wtypes_logo(self):
"""0.1.x has SUPER but no LOGO, and would type "l" for the latter."""
from vocalinux.text_injection.text_injector import TextInjector
self.assertEqual(TextInjector._ydotool_legacy_token(["win"], "x"), "super+x")
def test_names_absent_from_0_1_x_are_rejected(self):
from vocalinux.text_injection.text_injector import TextInjector
for name in ("altgr", "escape", "space", "return"):
self.assertIsNone(TextInjector._ydotool_legacy_token([], name), name)
class TestParseShortcut(unittest.TestCase):
def test_single_chord(self):
from vocalinux.text_injection.text_injector import TextInjector
self.assertEqual(
TextInjector._parse_shortcut("ctrl+shift+Right"),
[(["ctrl", "shift"], "Right")],
)
def test_multi_step(self):
from vocalinux.text_injection.text_injector import TextInjector
self.assertEqual(
TextInjector._parse_shortcut("Home+shift+End"),
[([], "Home"), (["shift"], "End")],
)
def test_trailing_modifier_rejected(self):
from vocalinux.text_injection.text_injector import TextInjector
with self.assertRaises(ValueError):
TextInjector._parse_shortcut("ctrl+shift")
def test_empty_rejected(self):
from vocalinux.text_injection.text_injector import TextInjector
with self.assertRaises(ValueError):
TextInjector._parse_shortcut("")
class TestInjectKeyboardShortcutRouting(unittest.TestCase):
"""IBus cannot synthesise key combinations, so shortcuts must be routed
to a tool that can."""
@staticmethod
def _env(name):
from vocalinux.text_injection.text_injector import DesktopEnvironment
return getattr(DesktopEnvironment, name)
@staticmethod
def _stub_backends(obj):
"""Stub both backends -- only for tests asserting *which* one is chosen.
The two tests that assert IBus shortcuts actually reach a tool call the
real Wayland helper instead, so a regression there cannot pass CI.
"""
obj._inject_shortcut_with_xdotool = MagicMock(return_value=True)
obj._inject_shortcut_with_wayland_tool = MagicMock(return_value=True)
def test_x11_ibus_routes_to_xdotool(self):
obj = _make_injector(self._env("X11_IBUS"))
self._stub_backends(obj)
self.assertTrue(obj._inject_keyboard_shortcut("ctrl+z"))
obj._inject_shortcut_with_xdotool.assert_called_once_with("ctrl+z")
obj._inject_shortcut_with_wayland_tool.assert_not_called()
def test_wayland_ibus_prefers_ydotool(self):
"""Reaches the real Wayland helper: a stub would pass while it's broken."""
obj = _make_injector(self._env("WAYLAND_IBUS"))
obj._inject_shortcut_with_xdotool = MagicMock(return_value=True)
obj._ydotool_legacy_named_keys = False
obj._ensure_ydotoold = MagicMock(return_value=True)
obj._wait_for_modifiers_released = MagicMock()
with patch("shutil.which", return_value="/usr/bin/x") as which:
with patch("subprocess.run") as mock_run:
self.assertTrue(obj._inject_keyboard_shortcut("ctrl+a"))
# _check_dependencies prefers the uinput helper once its daemon is ready.
self.assertTrue(which.called)
self.assertEqual(obj.wayland_tool, "ydotool")
self.assertEqual(
mock_run.call_args[0][0],
["ydotool", "key", "29:1", "30:1", "30:0", "29:0"],
)
obj._inject_shortcut_with_xdotool.assert_not_called()
def test_wayland_ibus_falls_back_to_wtype(self):
"""Reaches the real Wayland helper and asserts the argv it emits."""
obj = _make_injector(self._env("WAYLAND_IBUS"))
obj._inject_shortcut_with_xdotool = MagicMock(return_value=True)
obj._wait_for_modifiers_released = MagicMock()
which = lambda c: "/usr/bin/wtype" if c == "wtype" else None # noqa: E731
with patch("shutil.which", side_effect=which):
with patch("subprocess.run") as mock_run:
self.assertTrue(obj._inject_keyboard_shortcut("ctrl+a"))
self.assertEqual(obj.wayland_tool, "wtype")
self.assertEqual(
mock_run.call_args[0][0],
["wtype", "-M", "ctrl", "-k", "a", "-m", "ctrl"],
)
obj._inject_shortcut_with_xdotool.assert_not_called()
def test_wayland_ibus_without_any_tool_fails_without_injecting(self):
obj = _make_injector(self._env("WAYLAND_IBUS"))
self._stub_backends(obj)
with patch("shutil.which", return_value=None):
self.assertFalse(obj._inject_keyboard_shortcut("ctrl+a"))
obj._inject_shortcut_with_wayland_tool.assert_not_called()
obj._inject_shortcut_with_xdotool.assert_not_called()
def test_wayland_ibus_keeps_an_already_chosen_tool(self):
obj = _make_injector(self._env("WAYLAND_IBUS"))
obj.wayland_tool = "ydotool"
self._stub_backends(obj)
with patch("shutil.which", return_value="/usr/bin/wtype") as which:
self.assertTrue(obj._inject_keyboard_shortcut("ctrl+a"))
which.assert_not_called()
self.assertEqual(obj.wayland_tool, "ydotool")
def _make_backspace_injector(env_name, tool=None, legacy=False):
"""Injector wired for press_backspace, with the dialect and daemon pinned."""
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(getattr(DesktopEnvironment, env_name))
if tool is not None:
obj.wayland_tool = tool
obj._ydotool_legacy_named_keys = legacy
obj._ensure_ydotoold = MagicMock(return_value=True)
obj._wait_for_modifiers_released = MagicMock()
return obj
class TestPressBackspace(unittest.TestCase):
def test_zero_is_noop(self):
obj = _make_backspace_injector("WAYLAND")
with patch("subprocess.run") as mock_run:
self.assertTrue(obj.press_backspace(0))
mock_run.assert_not_called()
def test_wtype_repeats_key_events(self):
obj = _make_backspace_injector("WAYLAND", tool="wtype")
with patch("subprocess.run") as mock_run:
self.assertTrue(obj.press_backspace(3))
self.assertEqual(
mock_run.call_args[0][0],
["wtype", "-k", "BackSpace", "-k", "BackSpace", "-k", "BackSpace"],
)
def test_ydotool_v1_uses_keycode_14(self):
obj = _make_backspace_injector("WAYLAND", tool="ydotool", legacy=False)
with patch.dict(os.environ, {"VOCALINUX_YDOTOOL_KEY_DELAY": "2"}):
with patch("subprocess.run") as mock_run:
self.assertTrue(obj.press_backspace(2))
self.assertEqual(
mock_run.call_args[0][0],
["ydotool", "key", "--key-delay", "2", "14:1", "14:0", "14:1", "14:0"],
)
def test_ydotool_legacy_uses_named_backspace(self):
"""1.x keycodes type digit garbage on 0.1.x, which exits 0 regardless."""
obj = _make_backspace_injector("WAYLAND", tool="ydotool", legacy=True)
with patch("subprocess.run") as mock_run:
self.assertTrue(obj.press_backspace(2))
self.assertEqual(
mock_run.call_args[0][0],
["ydotool", "key", "backspace", "backspace"],
)
def test_x11_uses_xdotool_repeat_without_waiting(self):
"""--clearmodifiers covers held modifiers, so no wait on this path."""
obj = _make_backspace_injector("X11")
with patch("subprocess.run") as mock_run:
self.assertTrue(obj.press_backspace(5))
self.assertEqual(
mock_run.call_args[0][0],
["xdotool", "key", "--clearmodifiers", "--repeat", "5", "BackSpace"],
)
obj._wait_for_modifiers_released.assert_not_called()
kwargs = mock_run.call_args[1]
self.assertIn("env", kwargs)
self.assertGreaterEqual(kwargs["timeout"], 3)
def test_wayland_ibus_falls_back_to_virtual_keyboard(self):
"""IBus cannot send key events, so deletion must use wtype/ydotool."""
obj = _make_backspace_injector("WAYLAND_IBUS")
which = lambda c: "/usr/bin/wtype" if c == "wtype" else None # noqa: E731
with patch("subprocess.run") as mock_run:
with patch("shutil.which", side_effect=which):
self.assertTrue(obj.press_backspace(1))
self.assertEqual(mock_run.call_args[0][0], ["wtype", "-k", "BackSpace"])
def test_prefers_ydotool_when_both_tools_are_present(self):
"""_check_dependencies prefers the uinput helper; match that order."""
obj = _make_backspace_injector("WAYLAND_IBUS", legacy=False)
with patch("subprocess.run") as mock_run:
with patch("shutil.which", return_value="/usr/bin/x"):
self.assertTrue(obj.press_backspace(1))
self.assertEqual(obj.wayland_tool, "ydotool")
self.assertEqual(mock_run.call_args[0][0][:2], ["ydotool", "key"])
def test_waits_for_modifiers_before_resolving_the_tool(self):
"""A held PTT modifier turns BackSpace into Ctrl+BackSpace."""
obj = _make_backspace_injector("WAYLAND", tool="wtype")
order = []
obj._wait_for_modifiers_released = MagicMock(side_effect=lambda: order.append("wait"))
with patch("subprocess.run", side_effect=lambda *a, **k: order.append("run")):
self.assertTrue(obj.press_backspace(1))
self.assertEqual(order, ["wait", "run"])
def test_ydotool_rechecks_the_daemon_even_when_the_tool_is_cached(self):
"""wayland_tool is cached at startup, so a daemon that died since would
never be restarted if the check only ran on a cold resolve."""
obj = _make_backspace_injector("WAYLAND", tool="ydotool", legacy=False)
with patch("subprocess.run"):
self.assertTrue(obj.press_backspace(1))
obj._ensure_ydotoold.assert_called_once()
def test_ydotool_continues_when_daemon_not_ready(self):
"""0.1.x often ships no daemon at all, so this must not abort."""
obj = _make_backspace_injector("WAYLAND", tool="ydotool", legacy=False)
obj._ensure_ydotoold = MagicMock(return_value=False)
with patch("subprocess.run") as mock_run:
self.assertTrue(obj.press_backspace(1))
self.assertTrue(mock_run.called)
def test_wtype_does_not_touch_the_ydotool_daemon(self):
obj = _make_backspace_injector("WAYLAND", tool="wtype")
with patch("subprocess.run"):
self.assertTrue(obj.press_backspace(1))
obj._ensure_ydotoold.assert_not_called()
def test_timeout_returns_false_rather_than_raising(self):
obj = _make_backspace_injector("WAYLAND", tool="wtype")
with patch("subprocess.run", side_effect=subprocess.TimeoutExpired("wtype", 3)):
self.assertFalse(obj.press_backspace(2))
def test_no_tool_available_injects_nothing(self):
obj = _make_backspace_injector("WAYLAND_IBUS")
with patch("subprocess.run") as mock_run:
with patch("shutil.which", return_value=None):
self.assertFalse(obj.press_backspace(3))
mock_run.assert_not_called()
class TestCopyToClipboard(unittest.TestCase):
def test_copy_success(self):
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.X11)
with patch("subprocess.run") as mock_run:
with patch("shutil.which", return_value="/usr/bin/xclip"):
result = obj._copy_to_clipboard("hello")
# Verify subprocess.run was called and result is True
self.assertTrue(mock_run.called)
self.assertTrue(result)
def test_copy_no_tools(self):
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.X11)
with patch("shutil.which", return_value=None):
result = obj._copy_to_clipboard("hello")
self.assertFalse(result)
def test_copy_timeout_marks_tool_unhealthy(self):
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.WAYLAND)
with (
patch(
"shutil.which",
side_effect=lambda name: (
"/usr/bin/" + name if name in ("wl-copy", "xclip") else None
),
),
patch(
"subprocess.run",
side_effect=[subprocess.TimeoutExpired("wl-copy", timeout=0.35), MagicMock()],
),
):
result = obj._copy_to_clipboard("hello")
self.assertTrue(result)
self.assertEqual(obj._clipboard_tool_health, {"wl-copy": False, "xclip": True})
def test_copy_skips_unhealthy_tool(self):
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.WAYLAND)
obj._clipboard_tool_health["wl-copy"] = False
with (
patch(
"shutil.which",
side_effect=lambda name: (
"/usr/bin/" + name if name in ("wl-copy", "xclip") else None
),
),
patch("subprocess.run") as mock_run,
):
result = obj._copy_to_clipboard("hello")
self.assertTrue(result)
self.assertEqual(mock_run.call_args.args[0][0], "xclip")
class TestShouldCopyToClipboard(unittest.TestCase):
def test_should_copy(self):
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.X11)
result = obj._should_copy_to_clipboard()
self.assertIsInstance(result, bool)
class TestStop(unittest.TestCase):
def test_stop_with_ibus(self):
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.X11_IBUS)
mock_ibus = MagicMock()
obj._ibus_injector = mock_ibus
obj.stop()
mock_ibus.stop.assert_called_once()
def test_stop_without_ibus(self):
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.X11)
obj.stop() # Should not raise
class TestBackgroundIBusInitialization(unittest.TestCase):
def test_constructor_reports_pending_ibus_pin_without_claiming_success(self):
"""A constructed injector is not active until its background warmup succeeds."""
from vocalinux.text_injection.text_injector import DesktopEnvironment, TextInjector
ibus = MagicMock()
with (
patch.dict(
os.environ,
{"XDG_SESSION_TYPE": "x11", "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=True,
),
patch(
"vocalinux.text_injection.text_injector.is_ibus_daemon_running",
return_value=True,
),
patch("vocalinux.text_injection.text_injector.IBusTextInjector", return_value=ibus),
patch.object(TextInjector, "_start_ibus_initialization"),
patch("shutil.which", return_value=None),
self.assertLogs("vocalinux.text_injection.text_injector", level="DEBUG") as logs,
):
obj = TextInjector()
self.assertEqual(obj.environment, DesktopEnvironment.X11)
self.assertIs(obj._ibus_injector, ibus)
self.assertTrue(
[
line
for line in logs.output
if line.endswith("IBus initialization is pending; backend selection is not final.")
],
f"expected pending diagnostic, got: {logs.output}",
)
self.assertFalse([line for line in logs.output if "was not applied" in line])
def test_constructor_reports_ready_ibus_as_honoured(self):
"""A completed warmup promotes the environment before startup reporting."""
from vocalinux.text_injection.text_injector import DesktopEnvironment, TextInjector
ibus = MagicMock()
def initialize_now(obj: TextInjector) -> None:
obj._ibus_injector = ibus
obj._initialize_ibus_in_background()
with (
patch.dict(
os.environ,
{"XDG_SESSION_TYPE": "wayland", "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=True,
),
patch(
"vocalinux.text_injection.text_injector.is_ibus_daemon_running",
return_value=True,
),
patch("vocalinux.text_injection.text_injector.IBusTextInjector", return_value=ibus),
patch.object(TextInjector, "_start_ibus_initialization", initialize_now),
patch.object(TextInjector, "_wayland_compositor_bridges_ibus", lambda s: True),
patch.object(TextInjector, "_ensure_ydotoold", lambda s: True),
patch("shutil.which", side_effect=lambda name: "/usr/bin/" + name),
self.assertLogs("vocalinux.text_injection.text_injector", level="DEBUG") as logs,
):
obj = TextInjector()
self.assertEqual(obj.environment, DesktopEnvironment.WAYLAND_IBUS)
self.assertTrue(obj._ibus_ready)
self.assertFalse([line for line in logs.output if "was not applied" in line])
self.assertFalse([line for line in logs.output if "initialization is pending" in line])
def test_constructor_reports_ibus_failure_before_startup_diagnostic(self):
"""A known warmup failure must use the ordinary fallback mismatch warning."""
from vocalinux.text_injection.text_injector import TextInjector
ibus = MagicMock()
ibus.prepare_engine.side_effect = RuntimeError("not ready")
def initialize_now(obj: TextInjector) -> None:
obj._ibus_injector = ibus
obj._initialize_ibus_in_background()
with (
patch.dict(
os.environ,
{"XDG_SESSION_TYPE": "x11", "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=True,
),
patch(
"vocalinux.text_injection.text_injector.is_ibus_daemon_running",
return_value=True,
),
patch("vocalinux.text_injection.text_injector.IBusTextInjector", return_value=ibus),
patch.object(TextInjector, "_start_ibus_initialization", initialize_now),
patch("shutil.which", side_effect=lambda name: "/usr/bin/" + name),
self.assertLogs("vocalinux.text_injection.text_injector", level="DEBUG") as logs,
):
obj = TextInjector()
self.assertTrue(obj._ibus_init_failed)
self.assertTrue([line for line in logs.output if "IBus initialization failed" in line])
warned = [line for line in logs.output if "was not applied" in line]
self.assertTrue(warned, f"expected mismatch warning, got: {logs.output}")
self.assertIn("using xdotool instead", warned[0])
self.assertFalse([line for line in logs.output if "initialization is pending" in line])
def test_constructor_ibus_construction_failure_keeps_fallback_diagnostic(self):
"""A constructor failure has no injector and still reports the fallback."""
from vocalinux.text_injection.text_injector import TextInjector
with (
patch.dict(
os.environ,
{"XDG_SESSION_TYPE": "x11", "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=True,
),
patch(
"vocalinux.text_injection.text_injector.is_ibus_daemon_running",
return_value=True,
),
patch(
"vocalinux.text_injection.text_injector.IBusTextInjector",
side_effect=RuntimeError("constructor failed"),
),
patch("shutil.which", side_effect=lambda name: "/usr/bin/" + name),
self.assertLogs("vocalinux.text_injection.text_injector", level="DEBUG") as logs,
):
obj = TextInjector()
self.assertIsNone(obj._ibus_injector)
warned = [line for line in logs.output if "was not applied" in line]
self.assertTrue(warned, f"expected mismatch warning, got: {logs.output}")
self.assertIn("using xdotool instead", warned[0])
def test_late_ibus_failure_keeps_background_warning_after_pending_state(self):
"""A later failure retains the existing background warning and fallback."""
from vocalinux.text_injection.text_injector import DesktopEnvironment, TextInjector
obj = _make_injector(DesktopEnvironment.WAYLAND)
obj._ibus_injector = MagicMock()
obj.wayland_tool = "ydotool"
obj._ibus_injector.prepare_engine.side_effect = RuntimeError("late failure")
obj._backend_pin = ("ibus", "VOCALINUX_FORCE_BACKEND")
with self.assertLogs("vocalinux.text_injection.text_injector", level="DEBUG") as logs:
obj._warn_if_pin_not_honoured(*obj._backend_pin)
obj._initialize_ibus_in_background()
self.assertTrue([line for line in logs.output if "initialization is pending" in line])
self.assertTrue([line for line in logs.output if "IBus initialization failed" in line])
self.assertTrue(obj._ibus_init_failed)
self.assertEqual(obj._resolved_backend(), "ydotool")
def test_resolved_backend_requires_ibus_environment_and_injector(self):
"""A stale injector object cannot override the final non-IBus route."""
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.WAYLAND_XDOTOOL)
obj._ibus_injector = MagicMock()
obj.wayland_tool = "wtype"
self.assertEqual(obj._resolved_backend(), "xdotool")
obj._ibus_injector = None
obj.environment = DesktopEnvironment.WAYLAND_IBUS
self.assertEqual(obj._resolved_backend(), "wtype")
def test_check_dependencies_starts_ibus_in_background(self):
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.WAYLAND)
with (
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=True,
),
patch(
"vocalinux.text_injection.text_injector.is_ibus_daemon_running", return_value=True
),
patch(
"vocalinux.text_injection.text_injector.IBusTextInjector",
return_value=MagicMock(),
),
patch.object(obj, "_start_ibus_initialization") as mock_start,
patch(
"shutil.which",
side_effect=lambda x: "/usr/bin/ydotool" if x == "ydotool" else None,
),
patch("subprocess.run") as mock_run,
):
mock_run.return_value = MagicMock(returncode=0, stdout="", stderr="")
obj._check_dependencies()
mock_start.assert_called_once_with()
self.assertEqual(obj.environment, DesktopEnvironment.WAYLAND)
self.assertEqual(obj.wayland_tool, "ydotool")
def test_background_ibus_success_switches_environment(self):
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.WAYLAND)
obj._ibus_injector = MagicMock()
obj._initialize_ibus_in_background()
self.assertTrue(obj._ibus_ready)
self.assertEqual(obj.environment, DesktopEnvironment.WAYLAND_IBUS)
def test_background_ibus_failure_preserves_fallback(self):
from vocalinux.text_injection.text_injector import DesktopEnvironment
obj = _make_injector(DesktopEnvironment.WAYLAND)
obj._ibus_injector = MagicMock()
obj._ibus_injector.prepare_engine.side_effect = RuntimeError("not ready")
obj._initialize_ibus_in_background()
self.assertFalse(obj._ibus_ready)
self.assertTrue(obj._ibus_init_failed)
self.assertEqual(obj.environment, DesktopEnvironment.WAYLAND)
if __name__ == "__main__":
unittest.main()