From 3e4089a45ce648472b1d6067d106c0a6a87ed4ec Mon Sep 17 00:00:00 2001 From: jatinkrmalik Date: Sat, 2 May 2026 17:33:29 -0700 Subject: [PATCH] fix(ibus): handle engine instance destruction on layout switch (fixes #388, closes #389) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Override do_destroy() to clear _active_instance when IBus destroys the engine on keyboard layout switch (Super+Space). Add retry logic in inject_text() — on NO_ENGINE, re-activates via switch_engine() and retries once to recover cleanly. Co-authored-by: Sbenazar <127324689+Sbenazar@users.noreply.github.com> --- src/vocalinux/text_injection/ibus_engine.py | 79 +++++++++------ tests/test_ibus_engine.py | 104 ++++++++++++++++++-- 2 files changed, 144 insertions(+), 39 deletions(-) diff --git a/src/vocalinux/text_injection/ibus_engine.py b/src/vocalinux/text_injection/ibus_engine.py index d218436..4f20ba2 100644 --- a/src/vocalinux/text_injection/ibus_engine.py +++ b/src/vocalinux/text_injection/ibus_engine.py @@ -565,6 +565,14 @@ class VocalinuxEngine(IBus.Engine if IBUS_AVAILABLE else object): # want to be able to inject text. The engine process keeps running and # the instance remains valid for text injection. + def do_destroy(self) -> None: + """Called when the engine instance is destroyed by IBus (e.g. layout switch).""" + logger.debug("VocalinuxEngine instance destroyed") + if VocalinuxEngine._active_instance is self: + VocalinuxEngine._active_instance = None + if IBUS_AVAILABLE: + super().do_destroy() + def do_focus_in(self) -> None: """Called when the engine gains focus.""" logger.debug("VocalinuxEngine focus in") @@ -894,37 +902,50 @@ class IBusTextInjector: logger.debug("Vocalinux engine not active, re-activating...") switch_engine(ENGINE_NAME) - try: - if not SOCKET_PATH.exists(): - 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 - else: - logger.error(f"Text injection failed: {response}") + # Try injection, retry once if engine instance was destroyed + # (e.g. user switched keyboard layout and IBus called do_destroy) + for attempt in range(2): + try: + if not SOCKET_PATH.exists(): + logger.error( + "IBus engine socket not found. " + "Make sure Vocalinux IBus engine is running." + ) return False - except socket.timeout: - logger.error("Timeout connecting to IBus engine") - return False - except FileNotFoundError: - 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 + # 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 == 0: + # 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) + continue + else: + logger.error(f"Text injection failed: {response}") + return False + + except socket.timeout: + logger.error("Timeout connecting to IBus engine") + return False + except FileNotFoundError: + 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 + + return False def _get_engines_xml() -> str: diff --git a/tests/test_ibus_engine.py b/tests/test_ibus_engine.py index a89a87a..f6b4bdc 100644 --- a/tests/test_ibus_engine.py +++ b/tests/test_ibus_engine.py @@ -495,29 +495,69 @@ class TestIBusTextInjector(unittest.TestCase): server_sock.close() self.assertTrue(result) + @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_engine_error(self, mock_ensure_dir): - """Test text injection when engine returns error.""" + def test_inject_text_no_engine_retries_and_recovers(self, mock_ensure_dir, mock_switch): + """Test inject_text retries once on NO_ENGINE and succeeds on second attempt.""" from vocalinux.text_injection.ibus_engine import IBusTextInjector - # Create a mock server socket that returns error server_sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) server_sock.bind(str(self.socket_path)) - server_sock.listen(1) + server_sock.listen(2) + + call_count = 0 def handle_connection(): - conn, _ = server_sock.accept() - with conn: - conn.recv(65536) - conn.sendall(b"NO_ENGINE") + nonlocal call_count + for _ in range(2): + conn, _ = server_sock.accept() + with conn: + conn.recv(65536) + call_count += 1 + if call_count == 1: + conn.sendall(b"NO_ENGINE") + else: + conn.sendall(b"OK") server_thread = threading.Thread(target=handle_connection, daemon=True) server_thread.start() with patch("vocalinux.text_injection.ibus_engine.SOCKET_PATH", self.socket_path): - injector = IBusTextInjector(auto_activate=False) - result = injector.inject_text("Hello") + with patch("vocalinux.text_injection.ibus_engine.time"): + injector = IBusTextInjector(auto_activate=False) + result = injector.inject_text("Hello") + + server_sock.close() + self.assertTrue(result) + self.assertEqual(call_count, 2) + mock_switch.assert_called_with("vocalinux") + + @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_after_retry(self, mock_ensure_dir, mock_switch): + """Test inject_text returns False when NO_ENGINE persists after retry.""" + 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(2) + + def handle_connection(): + for _ in range(2): + 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) @@ -1105,5 +1145,49 @@ class TestWaylandXkbLayoutSkipping(unittest.TestCase): self.assertFalse(_is_wayland_session()) +class TestVocalinuxEngineDestroy(unittest.TestCase): + """Tests for VocalinuxEngine.do_destroy (layout switch resilience).""" + + @patch("vocalinux.text_injection.ibus_engine.IBUS_AVAILABLE", False) + def test_do_destroy_clears_active_instance(self): + """Test do_destroy clears _active_instance when called on the active engine.""" + from vocalinux.text_injection.ibus_engine import VocalinuxEngine + + engine = VocalinuxEngine.__new__(VocalinuxEngine) + VocalinuxEngine._active_instance = engine + + engine.do_destroy() + + self.assertIsNone(VocalinuxEngine._active_instance) + + @patch("vocalinux.text_injection.ibus_engine.IBUS_AVAILABLE", False) + def test_do_destroy_ignores_different_instance(self): + """Test do_destroy doesn't clear _active_instance if called on a different engine.""" + from vocalinux.text_injection.ibus_engine import VocalinuxEngine + + active_engine = VocalinuxEngine.__new__(VocalinuxEngine) + old_engine = VocalinuxEngine.__new__(VocalinuxEngine) + VocalinuxEngine._active_instance = active_engine + + old_engine.do_destroy() + + self.assertIs(VocalinuxEngine._active_instance, active_engine) + + def test_do_destroy_calls_super_when_ibus_available(self): + """Test do_destroy calls super().do_destroy() when IBus is available.""" + from vocalinux.text_injection.ibus_engine import VocalinuxEngine + + engine = VocalinuxEngine.__new__(VocalinuxEngine) + VocalinuxEngine._active_instance = None + + mock_ibus.Engine.do_destroy = MagicMock() + try: + with patch("vocalinux.text_injection.ibus_engine.IBUS_AVAILABLE", True): + engine.do_destroy() + mock_ibus.Engine.do_destroy.assert_called_once() + finally: + del mock_ibus.Engine.do_destroy + + if __name__ == "__main__": unittest.main()