fix(ibus): probe engine readiness at startup and harden NO_ENGINE retries (#391)
* 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>
This commit is contained in:
Jatin K Malik
co-authored by
Sisyphus
Sisyphus
parent
653f0e3d90
commit
9939ce6b63
@@ -666,6 +666,13 @@ class VocalinuxEngine(IBus.Engine if IBUS_AVAILABLE else object):
|
||||
|
||||
data = conn.recv(65536) # Max text size
|
||||
if data:
|
||||
if data == b"\x00PING":
|
||||
if cls._active_instance:
|
||||
conn.sendall(b"OK")
|
||||
else:
|
||||
conn.sendall(b"NO_ENGINE")
|
||||
continue
|
||||
|
||||
text = data.decode("utf-8")
|
||||
# Schedule injection on main thread
|
||||
if cls._active_instance:
|
||||
@@ -832,20 +839,6 @@ class IBusTextInjector:
|
||||
if not start_engine_process():
|
||||
raise IBusSetupError("Failed to start IBus engine process. Check logs for details.")
|
||||
|
||||
# Verify the engine is fully ready before proceeding.
|
||||
# start_engine_process() only confirms the subprocess is alive —
|
||||
# register_component() and socket setup may still be in progress.
|
||||
for _attempt in range(15):
|
||||
if SOCKET_PATH.exists():
|
||||
logger.debug("Engine socket is ready")
|
||||
break
|
||||
time.sleep(0.2)
|
||||
else:
|
||||
logger.warning(
|
||||
"Engine process started but socket not ready after retries; "
|
||||
"proceeding with activation attempt"
|
||||
)
|
||||
|
||||
# Capture current XKB layout before switching engines
|
||||
# This is critical for preserving the user's keyboard layout
|
||||
# when IBus engine switching might override it
|
||||
@@ -870,6 +863,8 @@ class IBusTextInjector:
|
||||
"Try manually: ibus engine vocalinux"
|
||||
)
|
||||
|
||||
self._wait_for_engine_ready()
|
||||
|
||||
# Restore the user's XKB layout immediately after engine activation.
|
||||
# Switching to the Vocalinux IBus engine can override the system
|
||||
# keyboard layout (e.g. Spanish, French AZERTY) with the engine's
|
||||
@@ -880,6 +875,38 @@ class IBusTextInjector:
|
||||
layout, variant, option = self._previous_xkb_layout
|
||||
restore_xkb_layout(layout, variant, option)
|
||||
|
||||
def _wait_for_engine_ready(self, max_attempts: int = 8) -> None:
|
||||
delay = 0.25
|
||||
|
||||
for attempt in range(max_attempts):
|
||||
try:
|
||||
if not SOCKET_PATH.exists():
|
||||
raise FileNotFoundError("IBus engine socket not found")
|
||||
|
||||
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock:
|
||||
sock.settimeout(2.0)
|
||||
sock.connect(str(SOCKET_PATH))
|
||||
sock.sendall(b"\x00PING")
|
||||
response = sock.recv(64).decode("utf-8")
|
||||
|
||||
if response == "OK":
|
||||
logger.debug(f"IBus engine readiness probe succeeded on attempt {attempt + 1}")
|
||||
return
|
||||
|
||||
if response != "NO_ENGINE":
|
||||
logger.warning(f"Unexpected readiness probe response: {response}")
|
||||
except (socket.timeout, FileNotFoundError, ConnectionRefusedError, OSError) as e:
|
||||
logger.debug(f"IBus readiness probe attempt {attempt + 1} failed: {e}")
|
||||
|
||||
if attempt < max_attempts - 1:
|
||||
time.sleep(delay)
|
||||
delay = min(delay * 2, 2.0)
|
||||
|
||||
raise IBusSetupError(
|
||||
"Vocalinux IBus engine did not become ready after startup. "
|
||||
"Try manually: ibus engine vocalinux"
|
||||
)
|
||||
|
||||
def stop(self) -> None:
|
||||
"""
|
||||
Stop the IBus text injector and restore previous engine and XKB layout.
|
||||
|
||||
@@ -52,11 +52,19 @@ class TextInjector:
|
||||
"""
|
||||
self._ibus_injector: Optional[IBusTextInjector] = None
|
||||
self.environment = self._detect_environment()
|
||||
self._session_environment = self.environment
|
||||
self._ibus_ready = False
|
||||
self._ibus_init_failed = False
|
||||
self._ibus_init_thread: Optional[threading.Thread] = None
|
||||
self._state_lock = threading.Lock()
|
||||
self._clipboard_tool_health = {}
|
||||
self._clipboard_timeout = 0.35
|
||||
|
||||
# Force Wayland mode if requested
|
||||
if wayland_mode and self.environment == DesktopEnvironment.X11:
|
||||
logger.info("Forcing Wayland compatibility mode")
|
||||
self.environment = DesktopEnvironment.WAYLAND
|
||||
self._session_environment = self.environment
|
||||
|
||||
logger.info(f"Using text injection for {self.environment.value} environment")
|
||||
|
||||
@@ -109,6 +117,7 @@ class TextInjector:
|
||||
logger.info("Stopping IBus text injector")
|
||||
self._ibus_injector.stop()
|
||||
self._ibus_injector = None
|
||||
self._ibus_ready = False
|
||||
|
||||
def _detect_environment(self) -> DesktopEnvironment:
|
||||
"""
|
||||
@@ -134,6 +143,8 @@ class TextInjector:
|
||||
|
||||
def _check_dependencies(self):
|
||||
"""Check for the required tools for text injection."""
|
||||
ibus_requested = False
|
||||
|
||||
# Prefer IBus on both X11 and Wayland - it sends Unicode directly,
|
||||
# bypassing keyboard layout issues entirely
|
||||
if is_ibus_available():
|
||||
@@ -154,21 +165,16 @@ class TextInjector:
|
||||
)
|
||||
else:
|
||||
try:
|
||||
self._ibus_injector = IBusTextInjector(auto_activate=True)
|
||||
if self.environment == DesktopEnvironment.X11:
|
||||
self.environment = DesktopEnvironment.X11_IBUS
|
||||
else:
|
||||
self.environment = DesktopEnvironment.WAYLAND_IBUS
|
||||
logger.info(
|
||||
"Using IBus for "
|
||||
f"{self.environment.value} text injection (best compatibility)"
|
||||
)
|
||||
return
|
||||
self._ibus_injector = IBusTextInjector(auto_activate=False)
|
||||
ibus_requested = True
|
||||
except Exception as e:
|
||||
logger.warning(f"IBus initialization failed: {e}, trying alternatives")
|
||||
if self.environment == DesktopEnvironment.X11:
|
||||
# Check for xdotool
|
||||
if not shutil.which("xdotool"):
|
||||
if ibus_requested:
|
||||
self._start_ibus_initialization()
|
||||
return
|
||||
logger.error("xdotool not found. Please install it with: sudo apt install xdotool")
|
||||
raise RuntimeError("Missing required dependency: xdotool")
|
||||
else:
|
||||
@@ -212,6 +218,9 @@ class TextInjector:
|
||||
"No native Wayland tools found. Using xdotool with XWayland as fallback"
|
||||
)
|
||||
else:
|
||||
if ibus_requested:
|
||||
self._start_ibus_initialization()
|
||||
return
|
||||
logger.error(
|
||||
"No text injection tools found. Please install one of:\n"
|
||||
"- IBus (recommended, usually pre-installed)\n"
|
||||
@@ -227,6 +236,96 @@ class TextInjector:
|
||||
)
|
||||
raise RuntimeError("Missing required dependencies for text injection")
|
||||
|
||||
if ibus_requested:
|
||||
self._start_ibus_initialization()
|
||||
|
||||
def _start_ibus_initialization(self) -> None:
|
||||
if self._ibus_injector is None or self._ibus_init_thread is not None:
|
||||
return
|
||||
|
||||
if self.environment == DesktopEnvironment.WAYLAND and not hasattr(self, "wayland_tool"):
|
||||
if shutil.which("wtype"):
|
||||
self.wayland_tool = "wtype"
|
||||
elif shutil.which("ydotool"):
|
||||
self.wayland_tool = "ydotool"
|
||||
|
||||
self._ibus_init_failed = False
|
||||
self._ibus_init_thread = threading.Thread(
|
||||
target=self._initialize_ibus_in_background,
|
||||
daemon=True,
|
||||
)
|
||||
self._ibus_init_thread.start()
|
||||
logger.info("Starting IBus warmup in background")
|
||||
|
||||
def _initialize_ibus_in_background(self) -> None:
|
||||
if self._ibus_injector is None:
|
||||
return
|
||||
|
||||
try:
|
||||
self._ibus_injector._setup_engine()
|
||||
with self._state_lock:
|
||||
self._ibus_ready = True
|
||||
self._ibus_init_failed = False
|
||||
if self._session_environment == DesktopEnvironment.X11:
|
||||
self.environment = DesktopEnvironment.X11_IBUS
|
||||
else:
|
||||
self.environment = DesktopEnvironment.WAYLAND_IBUS
|
||||
logger.info(
|
||||
f"Using IBus for {self.environment.value} text injection (best compatibility)"
|
||||
)
|
||||
except Exception as e:
|
||||
with self._state_lock:
|
||||
self._ibus_ready = False
|
||||
self._ibus_init_failed = True
|
||||
logger.warning(f"IBus initialization failed: {e}, continuing with fallback")
|
||||
|
||||
def _get_clipboard_tools(self):
|
||||
tools = []
|
||||
if self._session_environment == DesktopEnvironment.WAYLAND and shutil.which("wl-copy"):
|
||||
tools.append("wl-copy")
|
||||
if shutil.which("xclip"):
|
||||
tools.append("xclip")
|
||||
if shutil.which("xsel"):
|
||||
tools.append("xsel")
|
||||
if self._session_environment != DesktopEnvironment.WAYLAND and shutil.which("wl-copy"):
|
||||
tools.append("wl-copy")
|
||||
return tools
|
||||
|
||||
def _run_clipboard_command(self, tool: str, text: str) -> bool:
|
||||
if tool == "wl-copy":
|
||||
subprocess.run(
|
||||
["wl-copy", text],
|
||||
check=True,
|
||||
stderr=subprocess.PIPE,
|
||||
text=True,
|
||||
timeout=self._clipboard_timeout,
|
||||
)
|
||||
return True
|
||||
|
||||
if tool == "xclip":
|
||||
subprocess.run(
|
||||
["xclip", "-selection", "clipboard"],
|
||||
input=text,
|
||||
check=True,
|
||||
stderr=subprocess.PIPE,
|
||||
text=True,
|
||||
timeout=self._clipboard_timeout,
|
||||
)
|
||||
return True
|
||||
|
||||
if tool == "xsel":
|
||||
subprocess.run(
|
||||
["xsel", "--clipboard", "--input"],
|
||||
input=text,
|
||||
check=True,
|
||||
stderr=subprocess.PIPE,
|
||||
text=True,
|
||||
timeout=self._clipboard_timeout,
|
||||
)
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
def _switch_to_non_ibus_backend(self) -> bool:
|
||||
"""Switch from IBus mode to a non-IBus backend for runtime fallback."""
|
||||
if self.environment == DesktopEnvironment.X11_IBUS:
|
||||
@@ -379,52 +478,22 @@ class TextInjector:
|
||||
"""
|
||||
logger.info("Copying text to clipboard")
|
||||
|
||||
# Try wl-copy first (Wayland native)
|
||||
if shutil.which("wl-copy"):
|
||||
try:
|
||||
subprocess.run(
|
||||
["wl-copy", text],
|
||||
check=True,
|
||||
stderr=subprocess.PIPE,
|
||||
text=True,
|
||||
timeout=5,
|
||||
)
|
||||
logger.info("Text copied to Wayland clipboard using wl-copy")
|
||||
return True
|
||||
except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as e:
|
||||
logger.warning(f"wl-copy failed: {e}")
|
||||
for tool in self._get_clipboard_tools():
|
||||
if self._clipboard_tool_health.get(tool) is False:
|
||||
continue
|
||||
|
||||
# Try xclip (X11 / XWayland)
|
||||
if shutil.which("xclip"):
|
||||
try:
|
||||
subprocess.run(
|
||||
["xclip", "-selection", "clipboard"],
|
||||
input=text,
|
||||
check=True,
|
||||
stderr=subprocess.PIPE,
|
||||
text=True,
|
||||
timeout=5,
|
||||
)
|
||||
logger.info("Text copied to clipboard using xclip")
|
||||
return True
|
||||
except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as e:
|
||||
logger.warning(f"xclip failed: {e}")
|
||||
|
||||
# Try xsel as last resort
|
||||
if shutil.which("xsel"):
|
||||
try:
|
||||
subprocess.run(
|
||||
["xsel", "--clipboard", "--input"],
|
||||
input=text,
|
||||
check=True,
|
||||
stderr=subprocess.PIPE,
|
||||
text=True,
|
||||
timeout=5,
|
||||
)
|
||||
logger.info("Text copied to clipboard using xsel")
|
||||
return True
|
||||
except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as e:
|
||||
logger.warning(f"xsel failed: {e}")
|
||||
if self._run_clipboard_command(tool, text):
|
||||
self._clipboard_tool_health[tool] = True
|
||||
logger.info(f"Text copied to clipboard using {tool}")
|
||||
return True
|
||||
except (
|
||||
subprocess.CalledProcessError,
|
||||
subprocess.TimeoutExpired,
|
||||
FileNotFoundError,
|
||||
) as e:
|
||||
self._clipboard_tool_health[tool] = False
|
||||
logger.warning(f"{tool} failed: {e}")
|
||||
|
||||
logger.warning(
|
||||
"Clipboard copy failed. Install wl-copy (Wayland) or xclip/xsel "
|
||||
|
||||
+13
-12
@@ -98,8 +98,7 @@ class TestIBusTextInjectorSetupFailures(unittest.TestCase):
|
||||
|
||||
@patch("vocalinux.text_injection.ibus_engine.IBUS_AVAILABLE", True)
|
||||
@patch("vocalinux.text_injection.ibus_engine.ensure_ibus_dir")
|
||||
@patch("vocalinux.text_injection.ibus_engine.SOCKET_PATH")
|
||||
@patch("vocalinux.text_injection.ibus_engine.time")
|
||||
@patch("vocalinux.text_injection.ibus_engine.IBusTextInjector._wait_for_engine_ready")
|
||||
@patch("vocalinux.text_injection.ibus_engine.start_engine_process")
|
||||
@patch("vocalinux.text_injection.ibus_engine.get_current_xkb_layout")
|
||||
@patch("vocalinux.text_injection.ibus_engine.restore_xkb_layout")
|
||||
@@ -114,27 +113,22 @@ class TestIBusTextInjectorSetupFailures(unittest.TestCase):
|
||||
mock_restore_xkb,
|
||||
mock_get_xkb,
|
||||
mock_start_engine,
|
||||
mock_time,
|
||||
mock_socket_path,
|
||||
mock_wait_ready,
|
||||
mock_ensure_dir,
|
||||
):
|
||||
"""Covers the for/else warning branch: all 15 socket-readiness retries exhaust,
|
||||
warning is logged, and activation proceeds anyway (graceful degradation)."""
|
||||
"""Readiness probe is invoked as part of setup before returning."""
|
||||
mock_start_engine.return_value = True
|
||||
mock_socket_path.exists.return_value = False
|
||||
mock_is_active.return_value = False
|
||||
mock_get_engine.return_value = "xkb:us::eng"
|
||||
mock_switch.return_value = True
|
||||
mock_get_xkb.return_value = ("us", "", "")
|
||||
mock_wait_ready.return_value = None
|
||||
|
||||
from vocalinux.text_injection.ibus_engine import IBusTextInjector
|
||||
|
||||
with self.assertLogs("vocalinux.text_injection.ibus_engine", level="WARNING") as log:
|
||||
injector = IBusTextInjector(auto_activate=True)
|
||||
injector = IBusTextInjector(auto_activate=True)
|
||||
|
||||
self.assertTrue(any("socket not ready" in msg for msg in log.output))
|
||||
self.assertEqual(mock_socket_path.exists.call_count, 15)
|
||||
self.assertEqual(mock_time.sleep.call_count, 15)
|
||||
mock_wait_ready.assert_called_once_with()
|
||||
self.assertIsNotNone(injector)
|
||||
|
||||
|
||||
@@ -292,6 +286,7 @@ class TestIBusTextInjector(unittest.TestCase):
|
||||
@patch("vocalinux.text_injection.ibus_engine.IBUS_AVAILABLE", True)
|
||||
@patch("vocalinux.text_injection.ibus_engine.ensure_ibus_dir")
|
||||
@patch("vocalinux.text_injection.ibus_engine.SOCKET_PATH")
|
||||
@patch("vocalinux.text_injection.ibus_engine.IBusTextInjector._wait_for_engine_ready")
|
||||
@patch("vocalinux.text_injection.ibus_engine.is_engine_active")
|
||||
@patch("vocalinux.text_injection.ibus_engine.start_engine_process")
|
||||
@patch("vocalinux.text_injection.ibus_engine.get_current_engine")
|
||||
@@ -302,6 +297,7 @@ class TestIBusTextInjector(unittest.TestCase):
|
||||
mock_get_current,
|
||||
mock_start_engine,
|
||||
mock_is_active,
|
||||
mock_wait_ready,
|
||||
mock_socket_path,
|
||||
mock_ensure_dir,
|
||||
):
|
||||
@@ -311,6 +307,7 @@ class TestIBusTextInjector(unittest.TestCase):
|
||||
mock_start_engine.return_value = True
|
||||
mock_get_current.return_value = "xkb:us::eng"
|
||||
mock_switch.return_value = True
|
||||
mock_wait_ready.return_value = None
|
||||
|
||||
from vocalinux.text_injection.ibus_engine import IBusTextInjector
|
||||
|
||||
@@ -322,6 +319,7 @@ class TestIBusTextInjector(unittest.TestCase):
|
||||
mock_start_engine.assert_called_once()
|
||||
# Should switch to vocalinux engine
|
||||
mock_switch.assert_called_once_with("vocalinux")
|
||||
mock_wait_ready.assert_called_once_with()
|
||||
self.assertEqual(injector._previous_engine, "xkb:us::eng")
|
||||
|
||||
@patch("vocalinux.text_injection.ibus_engine.IBUS_AVAILABLE", True)
|
||||
@@ -409,6 +407,7 @@ class TestIBusTextInjector(unittest.TestCase):
|
||||
@patch("vocalinux.text_injection.ibus_engine.restore_xkb_layout")
|
||||
@patch("vocalinux.text_injection.ibus_engine.get_current_xkb_layout")
|
||||
@patch("vocalinux.text_injection.ibus_engine.SOCKET_PATH")
|
||||
@patch("vocalinux.text_injection.ibus_engine.IBusTextInjector._wait_for_engine_ready")
|
||||
@patch("vocalinux.text_injection.ibus_engine.start_engine_process", return_value=True)
|
||||
@patch("vocalinux.text_injection.ibus_engine.switch_engine", return_value=True)
|
||||
@patch("vocalinux.text_injection.ibus_engine.get_current_engine", return_value="xkb:us::eng")
|
||||
@@ -422,6 +421,7 @@ class TestIBusTextInjector(unittest.TestCase):
|
||||
mock_get_engine,
|
||||
mock_switch,
|
||||
mock_start_proc,
|
||||
mock_wait_ready,
|
||||
mock_socket_path,
|
||||
mock_get_xkb,
|
||||
mock_restore_xkb,
|
||||
@@ -431,6 +431,7 @@ class TestIBusTextInjector(unittest.TestCase):
|
||||
|
||||
mock_socket_path.exists.return_value = True
|
||||
mock_get_xkb.return_value = ("fr", "azerty", "")
|
||||
mock_wait_ready.return_value = None
|
||||
|
||||
injector = IBusTextInjector(auto_activate=False)
|
||||
injector._setup_engine()
|
||||
|
||||
@@ -481,15 +481,17 @@ class TestIBusTextInjector(unittest.TestCase):
|
||||
|
||||
@patch("vocalinux.text_injection.ibus_engine.ensure_ibus_dir")
|
||||
@patch("vocalinux.text_injection.ibus_engine.SOCKET_PATH")
|
||||
@patch("vocalinux.text_injection.ibus_engine.IBusTextInjector._wait_for_engine_ready")
|
||||
@patch("vocalinux.text_injection.ibus_engine.start_engine_process", return_value=True)
|
||||
@patch("vocalinux.text_injection.ibus_engine.is_engine_active", return_value=True)
|
||||
def test_ibus_text_injector_init_with_auto_activate(
|
||||
self, mock_active, mock_start, mock_socket_path, mock_ensure_dir
|
||||
self, mock_active, mock_start, mock_wait_ready, mock_socket_path, mock_ensure_dir
|
||||
):
|
||||
"""Test IBusTextInjector initialization with auto-activation."""
|
||||
from vocalinux.text_injection.ibus_engine import IBusTextInjector
|
||||
|
||||
injector = IBusTextInjector(auto_activate=True)
|
||||
mock_wait_ready.assert_called_once_with()
|
||||
self.assertIsNotNone(injector)
|
||||
|
||||
@patch("vocalinux.text_injection.ibus_engine.ensure_ibus_dir")
|
||||
@@ -820,5 +822,210 @@ class TestIBusEngineMainEntrypoint(unittest.TestCase):
|
||||
mock_app.run.assert_called_once_with()
|
||||
|
||||
|
||||
class TestIBusTextInjectorRetry(unittest.TestCase):
|
||||
"""Test IBusTextInjector.inject_text() retry logic on NO_ENGINE response."""
|
||||
|
||||
@patch("vocalinux.text_injection.ibus_engine.ensure_ibus_dir")
|
||||
@patch("vocalinux.text_injection.ibus_engine.switch_engine", return_value=True)
|
||||
@patch("vocalinux.text_injection.ibus_engine.is_engine_active", return_value=True)
|
||||
@patch("socket.socket")
|
||||
@patch("vocalinux.text_injection.ibus_engine.SOCKET_PATH")
|
||||
@patch("vocalinux.text_injection.ibus_engine.time")
|
||||
def test_retry_on_no_engine_then_succeeds(
|
||||
self, mock_time, mock_socket_path, mock_socket_cls, mock_active, mock_switch, mock_ensure
|
||||
):
|
||||
"""Two NO_ENGINE responses followed by OK returns True and retried twice."""
|
||||
from vocalinux.text_injection.ibus_engine import IBusTextInjector
|
||||
|
||||
mock_socket_path.exists.return_value = True
|
||||
mock_sock = MagicMock()
|
||||
mock_sock.__enter__.return_value = mock_sock
|
||||
mock_sock.__exit__.return_value = None
|
||||
mock_sock.recv.side_effect = [b"NO_ENGINE", b"NO_ENGINE", b"OK"]
|
||||
mock_socket_cls.return_value = mock_sock
|
||||
|
||||
injector = IBusTextInjector(auto_activate=False)
|
||||
result = injector.inject_text("hello")
|
||||
|
||||
self.assertTrue(result)
|
||||
self.assertEqual(mock_sock.recv.call_count, 3)
|
||||
|
||||
@patch("vocalinux.text_injection.ibus_engine.ensure_ibus_dir")
|
||||
@patch("vocalinux.text_injection.ibus_engine.switch_engine", return_value=True)
|
||||
@patch("vocalinux.text_injection.ibus_engine.is_engine_active", return_value=True)
|
||||
@patch("socket.socket")
|
||||
@patch("vocalinux.text_injection.ibus_engine.SOCKET_PATH")
|
||||
@patch("vocalinux.text_injection.ibus_engine.time")
|
||||
def test_all_retries_exhausted_returns_false(
|
||||
self, mock_time, mock_socket_path, mock_socket_cls, mock_active, mock_switch, mock_ensure
|
||||
):
|
||||
"""All NO_ENGINE responses after exhausting retries returns False."""
|
||||
from vocalinux.text_injection.ibus_engine import IBusTextInjector
|
||||
|
||||
mock_socket_path.exists.return_value = True
|
||||
mock_sock = MagicMock()
|
||||
mock_sock.__enter__.return_value = mock_sock
|
||||
mock_sock.__exit__.return_value = None
|
||||
mock_sock.recv.return_value = b"NO_ENGINE"
|
||||
mock_socket_cls.return_value = mock_sock
|
||||
|
||||
injector = IBusTextInjector(auto_activate=False)
|
||||
result = injector.inject_text("hello")
|
||||
|
||||
self.assertFalse(result)
|
||||
# 3 max_attempts: first 2 get NO_ENGINE and retry, 3rd gets NO_ENGINE and fails
|
||||
self.assertEqual(mock_sock.recv.call_count, 3)
|
||||
|
||||
|
||||
class TestIBusEngineStartupReadiness(unittest.TestCase):
|
||||
"""Test startup readiness probe behavior in setup flow."""
|
||||
|
||||
@patch("vocalinux.text_injection.ibus_engine.ensure_ibus_dir")
|
||||
@patch(
|
||||
"vocalinux.text_injection.ibus_engine.get_current_xkb_layout", return_value=("us", "", "")
|
||||
)
|
||||
@patch("vocalinux.text_injection.ibus_engine.restore_xkb_layout")
|
||||
@patch("vocalinux.text_injection.ibus_engine.switch_engine", return_value=True)
|
||||
@patch("vocalinux.text_injection.ibus_engine.get_current_engine", return_value="xkb:us::eng")
|
||||
@patch("vocalinux.text_injection.ibus_engine.is_engine_active", return_value=False)
|
||||
@patch("vocalinux.text_injection.ibus_engine.start_engine_process", return_value=True)
|
||||
@patch("vocalinux.text_injection.ibus_engine.IBusTextInjector._wait_for_engine_ready")
|
||||
def test_setup_calls_wait_for_engine_ready(
|
||||
self,
|
||||
mock_wait_ready,
|
||||
mock_start_engine,
|
||||
mock_is_active,
|
||||
mock_get_engine,
|
||||
mock_switch,
|
||||
mock_restore_xkb,
|
||||
mock_get_xkb,
|
||||
mock_ensure_dir,
|
||||
):
|
||||
from vocalinux.text_injection.ibus_engine import IBusTextInjector
|
||||
|
||||
IBusTextInjector(auto_activate=True)
|
||||
mock_wait_ready.assert_called_once_with()
|
||||
|
||||
@patch("vocalinux.text_injection.ibus_engine.ensure_ibus_dir")
|
||||
@patch(
|
||||
"vocalinux.text_injection.ibus_engine.get_current_xkb_layout", return_value=("us", "", "")
|
||||
)
|
||||
@patch("vocalinux.text_injection.ibus_engine.switch_engine", return_value=True)
|
||||
@patch("vocalinux.text_injection.ibus_engine.get_current_engine", return_value="xkb:us::eng")
|
||||
@patch("vocalinux.text_injection.ibus_engine.is_engine_active", return_value=False)
|
||||
@patch("vocalinux.text_injection.ibus_engine.start_engine_process", return_value=True)
|
||||
@patch("vocalinux.text_injection.ibus_engine.IBusTextInjector._wait_for_engine_ready")
|
||||
def test_setup_propagates_readiness_failure(
|
||||
self,
|
||||
mock_wait_ready,
|
||||
mock_start_engine,
|
||||
mock_is_active,
|
||||
mock_get_engine,
|
||||
mock_switch,
|
||||
mock_get_xkb,
|
||||
mock_ensure_dir,
|
||||
):
|
||||
from vocalinux.text_injection.ibus_engine import IBusSetupError, IBusTextInjector
|
||||
|
||||
mock_wait_ready.side_effect = IBusSetupError("engine not ready")
|
||||
|
||||
with self.assertRaises(IBusSetupError):
|
||||
IBusTextInjector(auto_activate=True)
|
||||
|
||||
@patch("vocalinux.text_injection.ibus_engine.ensure_ibus_dir")
|
||||
@patch("socket.socket")
|
||||
@patch("vocalinux.text_injection.ibus_engine.SOCKET_PATH")
|
||||
def test_wait_for_engine_ready_succeeds(
|
||||
self, mock_socket_path, mock_socket_cls, mock_ensure_dir
|
||||
):
|
||||
from vocalinux.text_injection.ibus_engine import IBusTextInjector
|
||||
|
||||
mock_socket_path.exists.return_value = True
|
||||
mock_sock = MagicMock()
|
||||
mock_sock.__enter__.return_value = mock_sock
|
||||
mock_sock.__exit__.return_value = None
|
||||
mock_sock.recv.return_value = b"OK"
|
||||
mock_socket_cls.return_value = mock_sock
|
||||
|
||||
injector = IBusTextInjector(auto_activate=False)
|
||||
injector._wait_for_engine_ready(max_attempts=1)
|
||||
|
||||
mock_sock.sendall.assert_called_once_with(b"\x00PING")
|
||||
|
||||
@patch("vocalinux.text_injection.ibus_engine.ensure_ibus_dir")
|
||||
@patch("vocalinux.text_injection.ibus_engine.time")
|
||||
@patch("socket.socket")
|
||||
@patch("vocalinux.text_injection.ibus_engine.SOCKET_PATH")
|
||||
def test_wait_for_engine_ready_retries_unexpected_response_then_fails(
|
||||
self, mock_socket_path, mock_socket_cls, mock_time, mock_ensure_dir
|
||||
):
|
||||
from vocalinux.text_injection.ibus_engine import IBusSetupError, IBusTextInjector
|
||||
|
||||
mock_socket_path.exists.return_value = True
|
||||
mock_sock = MagicMock()
|
||||
mock_sock.__enter__.return_value = mock_sock
|
||||
mock_sock.__exit__.return_value = None
|
||||
mock_sock.recv.return_value = b"MAYBE"
|
||||
mock_socket_cls.return_value = mock_sock
|
||||
|
||||
injector = IBusTextInjector(auto_activate=False)
|
||||
|
||||
with self.assertRaises(IBusSetupError):
|
||||
injector._wait_for_engine_ready(max_attempts=3)
|
||||
|
||||
mock_time.sleep.assert_has_calls([call(0.25), call(0.5)])
|
||||
|
||||
@patch("vocalinux.text_injection.ibus_engine.ensure_ibus_dir")
|
||||
@patch("vocalinux.text_injection.ibus_engine.time")
|
||||
@patch("socket.socket")
|
||||
@patch("vocalinux.text_injection.ibus_engine.SOCKET_PATH")
|
||||
def test_wait_for_engine_ready_retries_socket_failure_then_raises(
|
||||
self, mock_socket_path, mock_socket_cls, mock_time, mock_ensure_dir
|
||||
):
|
||||
from vocalinux.text_injection.ibus_engine import IBusSetupError, IBusTextInjector
|
||||
|
||||
mock_socket_path.exists.return_value = True
|
||||
mock_sock = MagicMock()
|
||||
mock_sock.__enter__.return_value = mock_sock
|
||||
mock_sock.__exit__.return_value = None
|
||||
mock_sock.connect.side_effect = ConnectionRefusedError("not ready")
|
||||
mock_socket_cls.return_value = mock_sock
|
||||
|
||||
injector = IBusTextInjector(auto_activate=False)
|
||||
|
||||
with self.assertRaises(IBusSetupError):
|
||||
injector._wait_for_engine_ready(max_attempts=2)
|
||||
|
||||
mock_time.sleep.assert_called_once_with(0.25)
|
||||
|
||||
@patch("vocalinux.text_injection.ibus_engine.ensure_ibus_dir")
|
||||
@patch("vocalinux.text_injection.ibus_engine.switch_engine")
|
||||
@patch("vocalinux.text_injection.ibus_engine.is_engine_active", return_value=False)
|
||||
@patch("socket.socket")
|
||||
@patch("vocalinux.text_injection.ibus_engine.SOCKET_PATH")
|
||||
def test_inject_text_reactivates_engine_before_retrying(
|
||||
self,
|
||||
mock_socket_path,
|
||||
mock_socket_cls,
|
||||
mock_is_active,
|
||||
mock_switch_engine,
|
||||
mock_ensure_dir,
|
||||
):
|
||||
from vocalinux.text_injection.ibus_engine import ENGINE_NAME, IBusTextInjector
|
||||
|
||||
mock_socket_path.exists.return_value = True
|
||||
mock_sock = MagicMock()
|
||||
mock_sock.__enter__.return_value = mock_sock
|
||||
mock_sock.__exit__.return_value = None
|
||||
mock_sock.recv.return_value = b"OK"
|
||||
mock_socket_cls.return_value = mock_sock
|
||||
|
||||
injector = IBusTextInjector(auto_activate=False)
|
||||
result = injector.inject_text("hello")
|
||||
|
||||
self.assertTrue(result)
|
||||
mock_switch_engine.assert_called_once_with(ENGINE_NAME)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
@@ -3,7 +3,9 @@
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import threading
|
||||
import unittest
|
||||
from typing import Any, cast
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
@@ -38,12 +40,19 @@ class TestDesktopEnvironmentEnum(unittest.TestCase):
|
||||
self.assertEqual(DesktopEnvironment.UNKNOWN.value, "unknown")
|
||||
|
||||
|
||||
def _make_injector(env):
|
||||
def _make_injector(env) -> Any:
|
||||
from vocalinux.text_injection.text_injector import TextInjector
|
||||
|
||||
obj = TextInjector.__new__(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
|
||||
|
||||
|
||||
@@ -288,6 +297,48 @@ class TestCopyToClipboard(unittest.TestCase):
|
||||
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):
|
||||
@@ -315,5 +366,63 @@ class TestStop(unittest.TestCase):
|
||||
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()
|
||||
|
||||
Reference in New Issue
Block a user