Jatin K MalikandGitHub
e1384aad2e
* fix(ibus): preserve user engine for dead keys Keep the VocaLinux IBus engine warm without selecting it as the active keyboard engine. Temporarily switch to the VocaLinux engine only while committing dictated text, then restore the user's previous IBus engine. Wait for the IBus commit to complete before acknowledging socket injection, so engine restoration does not race the commit. This fixes dead-key composition (e.g. ^ + e → ê) on French AZERTY and other layouts that rely on the active IBus/XKB engine for compose state. Fixes #442 Supersedes #446 (rebased onto main after thread-safety hardening in #452) * test(ibus): add coverage for scoped activation error paths Cover previously-untested error paths in inject_text() scoped activation: - switch_engine failure during temporary Vocalinux activation - unknown current engine (get_current_engine returns None) - NO_ENGINE retry when previous engine could not be captured - ConnectionRefusedError exhausting all retries - FileNotFoundError retry-then-succeed and retry-then-fail - Generic exception catch-all
429 lines
16 KiB
Python
429 lines
16 KiB
Python
"""Extra tests for text_injector.py to improve branch coverage."""
|
|
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import unittest
|
|
from typing import Any, cast
|
|
from unittest.mock import MagicMock, 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")
|
|
|
|
|
|
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
|
|
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_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()
|
|
|
|
|
|
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_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)
|
|
|
|
|
|
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)
|
|
|
|
|
|
class TestShortcutWithWaylandTool(unittest.TestCase):
|
|
def test_wtype_not_supported(self):
|
|
from vocalinux.text_injection.text_injector import DesktopEnvironment
|
|
|
|
obj = _make_injector(DesktopEnvironment.WAYLAND)
|
|
obj.wayland_tool = "wtype"
|
|
result = obj._inject_shortcut_with_wayland_tool("ctrl+a")
|
|
self.assertFalse(result) # wtype doesn't support shortcuts
|
|
|
|
def test_ydotool_success(self):
|
|
from vocalinux.text_injection.text_injector import DesktopEnvironment
|
|
|
|
obj = _make_injector(DesktopEnvironment.WAYLAND)
|
|
obj.wayland_tool = "ydotool"
|
|
with patch("subprocess.run"):
|
|
result = obj._inject_shortcut_with_wayland_tool("ctrl+a")
|
|
self.assertTrue(result)
|
|
|
|
|
|
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_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()
|