From e1384aad2eff0ef17c4e0ee10aa2bb0366c5625b Mon Sep 17 00:00:00 2001 From: Jatin K Malik Date: Sat, 6 Jun 2026 19:57:25 -0700 Subject: [PATCH] fix(ibus): preserve user engine for dead keys (#457) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 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 --- src/vocalinux/text_injection/ibus_engine.py | 217 ++++++++----- src/vocalinux/text_injection/text_injector.py | 5 +- tests/test_ibus_engine.py | 297 +++++++++++++++++- tests/test_ibus_engine_core.py | 7 +- tests/test_text_injector_ext.py | 2 +- 5 files changed, 440 insertions(+), 88 deletions(-) diff --git a/src/vocalinux/text_injection/ibus_engine.py b/src/vocalinux/text_injection/ibus_engine.py index c643465..ddf2a85 100644 --- a/src/vocalinux/text_injection/ibus_engine.py +++ b/src/vocalinux/text_injection/ibus_engine.py @@ -677,13 +677,22 @@ class VocalinuxEngine(IBus.Engine if IBUS_AVAILABLE else object): # Snapshot active instance to avoid TOCTOU race with do_destroy instance = cls._active_instance if instance: + done = threading.Event() + injection_result = {"ok": False} def do_inject(t, inst=instance): - inst.inject_text(t) + try: + injection_result["ok"] = inst.inject_text(t) + finally: + done.set() return False # Run only once GLib.idle_add(do_inject, text) - conn.sendall(b"OK") + if done.wait(timeout=3.0): + conn.sendall(b"OK" if injection_result["ok"] else b"ERROR") + else: + logger.error("Timed out waiting for IBus text commit") + conn.sendall(b"TIMEOUT") else: logger.warning("No active engine instance") conn.sendall(b"NO_ENGINE") @@ -804,13 +813,13 @@ class IBusTextInjector: Text injector that uses IBus for text injection. This class connects to the Vocalinux IBus engine via Unix socket - and sends text to be injected. On initialization, it automatically: - 1. Starts the engine process (registers via D-Bus register_component) - 2. Saves the current engine and XKB layout - 3. Switches to the Vocalinux engine - 4. Restores the XKB layout + and sends text to be injected. The normal TextInjector warmup path starts + the engine process without selecting it as the user's active IBus engine. + Each injection temporarily switches to the Vocalinux engine, commits the + text, and restores the user's previous engine. - On cleanup (stop), it restores the previous engine and layout. + The auto_activate=True path still performs the older eager activation flow + for compatibility with direct callers and setup tests. """ def __init__(self, auto_activate: bool = True): @@ -830,6 +839,19 @@ class IBusTextInjector: if auto_activate: self._setup_engine() + def prepare_engine(self) -> None: + """ + Start the IBus engine process without selecting it as the active engine. + + The Vocalinux engine is only needed while committing dictated text. Keeping + the user's normal IBus/XKB engine active preserves dead-key composition and + layout-specific behavior during ordinary typing. + """ + if not start_engine_process(): + raise IBusSetupError("Failed to start IBus engine process. Check logs for details.") + + self._wait_for_engine_ready(require_active=False) + def _setup_engine(self) -> None: """Install and activate the IBus engine.""" # Start the engine process — it calls register_component() via @@ -876,7 +898,7 @@ 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: + def _wait_for_engine_ready(self, max_attempts: int = 8, require_active: bool = True) -> None: delay = 0.25 for attempt in range(max_attempts): @@ -894,6 +916,10 @@ class IBusTextInjector: logger.debug(f"IBus engine readiness probe succeeded on attempt {attempt + 1}") return + if response == "NO_ENGINE" and not require_active: + logger.debug("IBus engine socket is ready; active engine instance not required") + return + if response != "NO_ENGINE": logger.warning(f"Unexpected readiness probe response: {response}") except (socket.timeout, FileNotFoundError, ConnectionRefusedError, OSError) as e: @@ -947,10 +973,25 @@ class IBusTextInjector: logger.info(f"Starting IBus text injection: '{text[:20]}...' (length: {len(text)})") - # Ensure vocalinux engine is active (user may have switched keyboard) - if not is_engine_active(): - logger.debug("Vocalinux engine not active, re-activating...") - switch_engine(ENGINE_NAME) + restore_engine: Optional[str] = None + can_reactivate_engine = is_engine_active() + if not can_reactivate_engine: + current_engine = get_current_engine() + if current_engine: + restore_engine = current_engine + can_reactivate_engine = True + logger.debug( + "Temporarily activating Vocalinux IBus engine " + f"(will restore {current_engine})" + ) + if not switch_engine(ENGINE_NAME): + logger.error("Failed to activate Vocalinux IBus engine for injection") + return False + else: + logger.warning( + "Could not determine current IBus engine; trying existing " + "Vocalinux engine instance without switching" + ) # Try injection with bounded retries for transient socket/engine races. # This can happen if IBus re-created the engine instance or if the @@ -965,85 +1006,97 @@ class IBusTextInjector: time.sleep(0.3) return True - for attempt in range(max_attempts): - try: - if not SOCKET_PATH.exists(): - if not is_engine_process_running() and not restart_engine_process(): + try: + for attempt in range(max_attempts): + try: + if not SOCKET_PATH.exists(): + if not is_engine_process_running() and not restart_engine_process(): + return False + + if attempt < max_attempts - 1: + logger.warning( + "IBus engine socket not found on attempt " + f"{attempt + 1}/{max_attempts}; retrying..." + ) + time.sleep(0.2 * (attempt + 1)) + continue + + logger.error( + "IBus engine socket not found. " + "Make sure Vocalinux IBus engine is running." + ) return False + # Connect to engine socket and send text + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock: + sock.settimeout(5.0) + sock.connect(str(SOCKET_PATH)) + sock.sendall(text.encode("utf-8")) + + # Wait for response + response = sock.recv(64).decode("utf-8") + if response == "OK": + logger.debug("Text injection successful") + return True + elif response == "NO_ENGINE" and attempt < max_attempts - 1: + # Engine instance was destroyed (layout switch). + # Re-activate to create a new instance and retry. + if not can_reactivate_engine: + logger.error( + "IBus engine instance is not active and the previous " + "engine could not be captured safely" + ) + return False + + logger.info("Engine instance not active, re-activating and retrying...") + switch_engine(ENGINE_NAME) + time.sleep(0.3) + continue + else: + logger.error(f"Text injection failed: {response}") + return False + + except socket.timeout: if attempt < max_attempts - 1: logger.warning( - "IBus engine socket not found on attempt " + "Timeout connecting to IBus engine on attempt " f"{attempt + 1}/{max_attempts}; retrying..." ) time.sleep(0.2 * (attempt + 1)) continue - - logger.error( - "IBus engine socket not found. " - "Make sure Vocalinux IBus engine is running." - ) + logger.error("Timeout connecting to IBus engine") return False - - # Connect to engine socket and send text - with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock: - sock.settimeout(5.0) - sock.connect(str(SOCKET_PATH)) - sock.sendall(text.encode("utf-8")) - - # Wait for response - response = sock.recv(64).decode("utf-8") - if response == "OK": - logger.debug("Text injection successful") - return True - elif response == "NO_ENGINE" and attempt < max_attempts - 1: - # Engine instance was destroyed (layout switch). - # Re-activate to create a new instance and retry. - logger.info("Engine instance not active, re-activating and retrying...") - switch_engine(ENGINE_NAME) - time.sleep(0.3) + except ConnectionRefusedError as e: + if attempt < max_attempts - 1: + logger.warning( + "IBus engine refused connection on attempt " + f"{attempt + 1}/{max_attempts}: {e}. Retrying..." + ) + if not is_engine_process_running() and not restart_engine_process(): + return False + time.sleep(0.2 * (attempt + 1)) continue - else: - logger.error(f"Text injection failed: {response}") - return False - - except socket.timeout: - if attempt < max_attempts - 1: - logger.warning( - "Timeout connecting to IBus engine on attempt " - f"{attempt + 1}/{max_attempts}; retrying..." - ) - time.sleep(0.2 * (attempt + 1)) - continue - logger.error("Timeout connecting to IBus engine") - return False - except ConnectionRefusedError as e: - if attempt < max_attempts - 1: - logger.warning( - "IBus engine refused connection on attempt " - f"{attempt + 1}/{max_attempts}: {e}. Retrying..." - ) - if not is_engine_process_running() and not restart_engine_process(): - return False - time.sleep(0.2 * (attempt + 1)) - continue - logger.error(f"Failed to inject text via IBus: {e}") - return False - except FileNotFoundError: - if attempt < max_attempts - 1: - logger.warning( - "IBus engine socket disappeared on attempt " - f"{attempt + 1}/{max_attempts}; retrying..." - ) - if not is_engine_process_running() and not restart_engine_process(): - return False - time.sleep(0.2 * (attempt + 1)) - continue - logger.error("IBus engine socket not found") - return False - except Exception as e: - logger.error(f"Failed to inject text via IBus: {e}") - return False + logger.error(f"Failed to inject text via IBus: {e}") + return False + except FileNotFoundError: + if attempt < max_attempts - 1: + logger.warning( + "IBus engine socket disappeared on attempt " + f"{attempt + 1}/{max_attempts}; retrying..." + ) + if not is_engine_process_running() and not restart_engine_process(): + return False + time.sleep(0.2 * (attempt + 1)) + continue + logger.error("IBus engine socket not found") + return False + except Exception as e: + logger.error(f"Failed to inject text via IBus: {e}") + return False + finally: + if restore_engine: + logger.debug(f"Restoring IBus engine after injection: {restore_engine}") + switch_engine(restore_engine) return False diff --git a/src/vocalinux/text_injection/text_injector.py b/src/vocalinux/text_injection/text_injector.py index 42b83a7..bd6712f 100644 --- a/src/vocalinux/text_injection/text_injector.py +++ b/src/vocalinux/text_injection/text_injector.py @@ -263,7 +263,10 @@ class TextInjector: return try: - self._ibus_injector._setup_engine() + # Keep the Vocalinux IBus process warm without selecting it as the + # user's active keyboard engine. Activation is scoped to each text + # commit so normal typing keeps layout-specific compose/dead keys. + self._ibus_injector.prepare_engine() with self._state_lock: self._ibus_ready = True self._ibus_init_failed = False diff --git a/tests/test_ibus_engine.py b/tests/test_ibus_engine.py index 3ba6bcc..2f88318 100644 --- a/tests/test_ibus_engine.py +++ b/tests/test_ibus_engine.py @@ -440,6 +440,64 @@ class TestIBusTextInjector(unittest.TestCase): self.assertEqual(injector._previous_xkb_layout, ("fr", "azerty", "")) mock_restore_xkb.assert_called_once_with("fr", "azerty", "") + @patch("vocalinux.text_injection.ibus_engine.IBUS_AVAILABLE", True) + @patch("vocalinux.text_injection.ibus_engine.ensure_ibus_dir") + @patch("vocalinux.text_injection.ibus_engine.start_engine_process", return_value=True) + @patch("vocalinux.text_injection.ibus_engine.IBusTextInjector._wait_for_engine_ready") + @patch("vocalinux.text_injection.ibus_engine.switch_engine") + def test_prepare_engine_does_not_activate_vocalinux_engine( + self, + mock_switch, + mock_wait_ready, + mock_start_engine, + mock_ensure_dir, + ): + """Test warmup starts the process without replacing the user's IBus engine.""" + from vocalinux.text_injection.ibus_engine import IBusTextInjector + + injector = IBusTextInjector(auto_activate=False) + injector.prepare_engine() + + mock_start_engine.assert_called_once() + mock_wait_ready.assert_called_once_with(require_active=False) + mock_switch.assert_not_called() + + @patch("vocalinux.text_injection.ibus_engine.get_current_engine", return_value="xkb:fr::fra") + @patch("vocalinux.text_injection.ibus_engine.is_engine_active", return_value=False) + @patch("vocalinux.text_injection.ibus_engine.switch_engine", return_value=True) + @patch("socket.socket") + @patch("vocalinux.text_injection.ibus_engine.SOCKET_PATH") + @patch("vocalinux.text_injection.ibus_engine.IBUS_AVAILABLE", True) + @patch("vocalinux.text_injection.ibus_engine.ensure_ibus_dir") + def test_inject_text_restores_previous_engine_after_scoped_activation( + self, + mock_ensure_dir, + mock_socket_path, + mock_socket_cls, + mock_switch, + mock_is_active, + mock_get_current, + ): + """Test injection restores the user's original IBus engine after commit.""" + 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("Bonjour") + + self.assertTrue(result) + mock_sock.sendall.assert_called_once_with("Bonjour".encode("utf-8")) + self.assertEqual( + [call.args[0] for call in mock_switch.call_args_list], + [ENGINE_NAME, "xkb:fr::fra"], + ) + @patch("vocalinux.text_injection.ibus_engine.IBUS_AVAILABLE", True) @patch("vocalinux.text_injection.ibus_engine.ensure_ibus_dir") def test_inject_text_empty(self, mock_ensure_dir): @@ -496,12 +554,16 @@ class TestIBusTextInjector(unittest.TestCase): server_sock.close() self.assertTrue(result) + @patch("vocalinux.text_injection.ibus_engine.get_current_engine", return_value="xkb:fr::fra") + @patch("vocalinux.text_injection.ibus_engine.is_engine_active", return_value=False) @patch("vocalinux.text_injection.ibus_engine.switch_engine") @patch("vocalinux.text_injection.ibus_engine.IBUS_AVAILABLE", True) @patch("vocalinux.text_injection.ibus_engine.ensure_ibus_dir") - def test_inject_text_no_engine_retries_and_recovers(self, mock_ensure_dir, mock_switch): + def test_inject_text_no_engine_retries_and_recovers( + self, mock_ensure_dir, mock_switch, mock_is_active, mock_get_current + ): """Test inject_text retries once on NO_ENGINE and succeeds on second attempt.""" - from vocalinux.text_injection.ibus_engine import IBusTextInjector + from vocalinux.text_injection.ibus_engine import ENGINE_NAME, IBusTextInjector server_sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) server_sock.bind(str(self.socket_path)) @@ -532,7 +594,7 @@ class TestIBusTextInjector(unittest.TestCase): server_sock.close() self.assertTrue(result) self.assertEqual(call_count, 2) - mock_switch.assert_called_with("vocalinux") + self.assertIn(ENGINE_NAME, [call.args[0] for call in mock_switch.call_args_list]) @patch("vocalinux.text_injection.ibus_engine.switch_engine") @patch("vocalinux.text_injection.ibus_engine.IBUS_AVAILABLE", True) @@ -563,6 +625,235 @@ class TestIBusTextInjector(unittest.TestCase): server_sock.close() self.assertFalse(result) + @patch("vocalinux.text_injection.ibus_engine.get_current_engine", return_value="xkb:fr::fra") + @patch("vocalinux.text_injection.ibus_engine.is_engine_active", return_value=False) + @patch("vocalinux.text_injection.ibus_engine.switch_engine", return_value=False) + @patch("vocalinux.text_injection.ibus_engine.IBUS_AVAILABLE", True) + @patch("vocalinux.text_injection.ibus_engine.ensure_ibus_dir") + def test_inject_text_returns_false_when_scoped_activation_fails( + self, + mock_ensure_dir, + mock_switch, + mock_is_active, + mock_get_current, + ): + """Test inject_text returns False when switch to Vocalinux engine fails.""" + from vocalinux.text_injection.ibus_engine import IBusTextInjector + + injector = IBusTextInjector(auto_activate=False) + result = injector.inject_text("Hello") + + self.assertFalse(result) + mock_switch.assert_called_once_with("vocalinux") + + @patch("vocalinux.text_injection.ibus_engine.get_current_engine", return_value=None) + @patch("vocalinux.text_injection.ibus_engine.is_engine_active", return_value=False) + @patch("vocalinux.text_injection.ibus_engine.switch_engine") + @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") + def test_inject_text_warns_when_current_engine_unknown( + self, + mock_socket_path, + mock_ensure_dir, + mock_switch, + mock_is_active, + mock_get_current, + ): + """Test inject_text proceeds without switching when current engine is unknown.""" + 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" + + with patch("socket.socket", return_value=mock_sock): + injector = IBusTextInjector(auto_activate=False) + result = injector.inject_text("Hello") + + self.assertTrue(result) + mock_switch.assert_not_called() + + @patch("vocalinux.text_injection.ibus_engine.get_current_engine", return_value=None) + @patch("vocalinux.text_injection.ibus_engine.is_engine_active", return_value=False) + @patch("vocalinux.text_injection.ibus_engine.switch_engine") + @patch("vocalinux.text_injection.ibus_engine.IBUS_AVAILABLE", True) + @patch("vocalinux.text_injection.ibus_engine.ensure_ibus_dir") + def test_inject_text_no_engine_fails_when_cannot_reactivate( + self, mock_ensure_dir, mock_switch, mock_is_active, mock_get_current + ): + """Test NO_ENGINE retry fails when previous engine could not be captured.""" + from vocalinux.text_injection.ibus_engine import IBusTextInjector + + server_sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + server_sock.bind(str(self.socket_path)) + server_sock.listen(1) + + def handle_connection(): + conn, _ = server_sock.accept() + with conn: + conn.recv(65536) + conn.sendall(b"NO_ENGINE") + + server_thread = threading.Thread(target=handle_connection, daemon=True) + server_thread.start() + + with patch("vocalinux.text_injection.ibus_engine.SOCKET_PATH", self.socket_path): + with patch("vocalinux.text_injection.ibus_engine.time"): + injector = IBusTextInjector(auto_activate=False) + result = injector.inject_text("Hello") + + server_sock.close() + self.assertFalse(result) + + @patch("vocalinux.text_injection.ibus_engine.get_current_engine", return_value="xkb:fr::fra") + @patch("vocalinux.text_injection.ibus_engine.is_engine_active", return_value=False) + @patch("vocalinux.text_injection.ibus_engine.switch_engine", return_value=True) + @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.is_engine_process_running", return_value=True) + @patch("vocalinux.text_injection.ibus_engine.time") + def test_inject_text_connection_refused_exhausts_retries( + self, + mock_time, + mock_process_running, + mock_socket_path, + mock_ensure_dir, + mock_switch, + mock_is_active, + mock_get_current, + ): + """Test ConnectionRefusedError on final attempt returns False.""" + mock_socket_path.exists.return_value = True + + from vocalinux.text_injection.ibus_engine import IBusTextInjector + + with patch("socket.socket") as mock_socket_class: + socket_context = MagicMock() + sock = MagicMock() + sock.connect.side_effect = ConnectionRefusedError("refused") + socket_context.__enter__.return_value = sock + socket_context.__exit__.return_value = False + mock_socket_class.return_value = socket_context + + injector = IBusTextInjector(auto_activate=False) + result = injector.inject_text("Hello") + + self.assertFalse(result) + + @patch("vocalinux.text_injection.ibus_engine.get_current_engine", return_value="xkb:fr::fra") + @patch("vocalinux.text_injection.ibus_engine.is_engine_active", return_value=False) + @patch("vocalinux.text_injection.ibus_engine.switch_engine", return_value=True) + @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.start_engine_process", return_value=True) + @patch("vocalinux.text_injection.ibus_engine.is_engine_process_running", return_value=True) + @patch("vocalinux.text_injection.ibus_engine.time") + def test_inject_text_file_not_found_retries_then_succeeds( + self, + mock_time, + mock_process_running, + mock_start_engine, + mock_socket_path, + mock_ensure_dir, + mock_switch, + mock_is_active, + mock_get_current, + ): + """Test FileNotFoundError on first attempt retries and succeeds on second.""" + mock_socket_path.exists.return_value = True + + from vocalinux.text_injection.ibus_engine import IBusTextInjector + + with patch("socket.socket") as mock_socket_class: + socket_context = MagicMock() + mock_socket_class.return_value = socket_context + + first_sock = MagicMock() + second_sock = MagicMock() + first_sock.connect.side_effect = FileNotFoundError("gone") + second_sock.connect.return_value = None + second_sock.recv.return_value = b"OK" + socket_context.__enter__.side_effect = [first_sock, second_sock] + socket_context.__exit__.return_value = False + + injector = IBusTextInjector(auto_activate=False) + result = injector.inject_text("Hello") + + self.assertTrue(result) + + @patch("vocalinux.text_injection.ibus_engine.get_current_engine", return_value="xkb:fr::fra") + @patch("vocalinux.text_injection.ibus_engine.is_engine_active", return_value=False) + @patch("vocalinux.text_injection.ibus_engine.switch_engine", return_value=True) + @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") + def test_inject_text_generic_exception_returns_false( + self, + mock_socket_path, + mock_ensure_dir, + mock_switch, + mock_is_active, + mock_get_current, + ): + """Test generic exception in inject_text returns False.""" + mock_socket_path.exists.return_value = True + + from vocalinux.text_injection.ibus_engine import IBusTextInjector + + with patch("socket.socket") as mock_socket_class: + socket_context = MagicMock() + sock = MagicMock() + sock.connect.side_effect = RuntimeError("unexpected error") + socket_context.__enter__.return_value = sock + socket_context.__exit__.return_value = False + mock_socket_class.return_value = socket_context + + injector = IBusTextInjector(auto_activate=False) + result = injector.inject_text("Hello") + + self.assertFalse(result) + + @patch("vocalinux.text_injection.ibus_engine.get_current_engine", return_value="xkb:fr::fra") + @patch("vocalinux.text_injection.ibus_engine.is_engine_active", return_value=False) + @patch("vocalinux.text_injection.ibus_engine.switch_engine", return_value=True) + @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.is_engine_process_running", return_value=True) + @patch("vocalinux.text_injection.ibus_engine.time") + def test_inject_text_file_not_found_exhausts_retries( + self, + mock_time, + mock_process_running, + mock_socket_path, + mock_ensure_dir, + mock_switch, + mock_is_active, + mock_get_current, + ): + """Test FileNotFoundError on all attempts returns False.""" + mock_socket_path.exists.return_value = True + + from vocalinux.text_injection.ibus_engine import IBusTextInjector + + with patch("socket.socket") as mock_socket_class: + socket_context = MagicMock() + sock = MagicMock() + sock.connect.side_effect = FileNotFoundError("gone") + socket_context.__enter__.return_value = sock + socket_context.__exit__.return_value = False + mock_socket_class.return_value = socket_context + + injector = IBusTextInjector(auto_activate=False) + result = injector.inject_text("Hello") + + self.assertFalse(result) + @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") diff --git a/tests/test_ibus_engine_core.py b/tests/test_ibus_engine_core.py index 2682aa1..0a14c9b 100644 --- a/tests/test_ibus_engine_core.py +++ b/tests/test_ibus_engine_core.py @@ -1000,6 +1000,7 @@ class TestIBusEngineStartupReadiness(unittest.TestCase): @patch("vocalinux.text_injection.ibus_engine.ensure_ibus_dir") @patch("vocalinux.text_injection.ibus_engine.switch_engine") + @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("socket.socket") @patch("vocalinux.text_injection.ibus_engine.SOCKET_PATH") @@ -1008,6 +1009,7 @@ class TestIBusEngineStartupReadiness(unittest.TestCase): mock_socket_path, mock_socket_cls, mock_is_active, + mock_get_current, mock_switch_engine, mock_ensure_dir, ): @@ -1024,7 +1026,10 @@ class TestIBusEngineStartupReadiness(unittest.TestCase): result = injector.inject_text("hello") self.assertTrue(result) - mock_switch_engine.assert_called_once_with(ENGINE_NAME) + self.assertEqual( + [call.args[0] for call in mock_switch_engine.call_args_list], + [ENGINE_NAME, "xkb:us::eng"], + ) if __name__ == "__main__": diff --git a/tests/test_text_injector_ext.py b/tests/test_text_injector_ext.py index 8c80bf6..ae620b3 100644 --- a/tests/test_text_injector_ext.py +++ b/tests/test_text_injector_ext.py @@ -415,7 +415,7 @@ class TestBackgroundIBusInitialization(unittest.TestCase): obj = _make_injector(DesktopEnvironment.WAYLAND) obj._ibus_injector = MagicMock() - obj._ibus_injector._setup_engine.side_effect = RuntimeError("not ready") + obj._ibus_injector.prepare_engine.side_effect = RuntimeError("not ready") obj._initialize_ibus_in_background()