From 9939ce6b63516e77653edb359175ae3593bdb234 Mon Sep 17 00:00:00 2001 From: Jatin K Malik Date: Sat, 2 May 2026 19:33:31 -0700 Subject: [PATCH] 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 * 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 * 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 * test(ibus): update retry tests for merged NO_ENGINE recovery logic --------- Co-authored-by: Sisyphus --- src/vocalinux/text_injection/ibus_engine.py | 55 +++-- src/vocalinux/text_injection/text_injector.py | 177 ++++++++++----- tests/test_ibus_engine.py | 25 ++- tests/test_ibus_engine_core.py | 209 +++++++++++++++++- tests/test_text_injector_ext.py | 113 +++++++++- 5 files changed, 496 insertions(+), 83 deletions(-) diff --git a/src/vocalinux/text_injection/ibus_engine.py b/src/vocalinux/text_injection/ibus_engine.py index f791fb3..ac9ad50 100644 --- a/src/vocalinux/text_injection/ibus_engine.py +++ b/src/vocalinux/text_injection/ibus_engine.py @@ -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. diff --git a/src/vocalinux/text_injection/text_injector.py b/src/vocalinux/text_injection/text_injector.py index 72b7d73..2f67834 100644 --- a/src/vocalinux/text_injection/text_injector.py +++ b/src/vocalinux/text_injection/text_injector.py @@ -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 " diff --git a/tests/test_ibus_engine.py b/tests/test_ibus_engine.py index dffb328..3ba6bcc 100644 --- a/tests/test_ibus_engine.py +++ b/tests/test_ibus_engine.py @@ -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() diff --git a/tests/test_ibus_engine_core.py b/tests/test_ibus_engine_core.py index f80cd55..2682aa1 100644 --- a/tests/test_ibus_engine_core.py +++ b/tests/test_ibus_engine_core.py @@ -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() diff --git a/tests/test_text_injector_ext.py b/tests/test_text_injector_ext.py index cf618ab..8c80bf6 100644 --- a/tests/test_text_injector_ext.py +++ b/tests/test_text_injector_ext.py @@ -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()