9939ce6b63
* fix(ibus): wait for engine readiness before first injection Add a startup readiness probe that pings the IBus engine after activation so first-use dictation no longer races do_enable(). Also switch NO_ENGINE retry behavior to exponential backoff and cover the new flow in tests. Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai> * test(ibus): cover readiness probe edge cases Add focused tests for the new startup readiness path, including direct PING success, retry exhaustion, unexpected responses, and engine re-activation before injection. This raises coverage on vocalinux.text_injection.ibus_engine from 68% to 73% in the targeted suite. Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai> * fix(text-injection): avoid blocking IBus and clipboard fallbacks Warm IBus in the background so startup no longer stalls while readiness probing, and make clipboard helpers fail fast with per-session health caching so non-ASCII ydotool fallback does not lock dictation. Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai> * test(ibus): update retry tests for merged NO_ENGINE recovery logic --------- Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>
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._setup_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()
|