xavierk/voculinux
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

fix(ibus): handle engine instance destruction on layout switch (fixes #388, closes #389)

Browse Source 
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>
This commit is contained in:
jatinkrmalik
2026-05-02 17:33:42 -07:00
co-authored by Sbenazar
parent d04947374f
commit 3e4089a45c
2 changed files with 144 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/vocalinux/text_injection/ibus_engine.py
+50 -29
View File
@@ -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:
tests/test_ibus_engine.py
+94 -10
View File
@@ -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()