fix(text-injection): recover from IBus runtime failures (#411)
* fix(ibus): retry and recover from transient socket failures Harden IBus injection against connection-refused, timeout, and engine process races so dictation is more resilient in real-world desktop sessions. Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) * fix(text-injection): fallback to non-IBus backend on runtime IBus failures Allow dictation to continue when IBus runtime injection fails by automatically switching to xdotool/wtype/ydotool fallbacks instead of returning an immediate failure. Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) * fix(ibus): preserve live socket injection during recovery Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) * test(coverage): add runtime fallback and ibus retry coverage Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)
This commit is contained in:
Jatin K Malik
@@ -924,11 +924,33 @@ class IBusTextInjector:
|
||||
logger.debug("Vocalinux engine not active, re-activating...")
|
||||
switch_engine(ENGINE_NAME)
|
||||
|
||||
# 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 injection with bounded retries for transient socket/engine races.
|
||||
# This can happen if IBus re-created the engine instance or if the
|
||||
# engine process/socket is still coming up when dictation ends.
|
||||
max_attempts = 3
|
||||
|
||||
def restart_engine_process() -> bool:
|
||||
logger.warning("IBus engine process is not running, restarting...")
|
||||
if not start_engine_process():
|
||||
logger.error("Failed to restart IBus engine process")
|
||||
return False
|
||||
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():
|
||||
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."
|
||||
@@ -946,7 +968,7 @@ class IBusTextInjector:
|
||||
if response == "OK":
|
||||
logger.debug("Text injection successful")
|
||||
return True
|
||||
elif response == "NO_ENGINE" and attempt == 0:
|
||||
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...")
|
||||
@@ -958,9 +980,37 @@ class IBusTextInjector:
|
||||
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:
|
||||
@@ -1022,6 +1072,4 @@ def main():
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import sys
|
||||
|
||||
sys.exit(main())
|
||||
|
||||
@@ -160,7 +160,8 @@ class TextInjector:
|
||||
else:
|
||||
self.environment = DesktopEnvironment.WAYLAND_IBUS
|
||||
logger.info(
|
||||
f"Using IBus for {self.environment.value} text injection (best compatibility)"
|
||||
"Using IBus for "
|
||||
f"{self.environment.value} text injection (best compatibility)"
|
||||
)
|
||||
return
|
||||
except Exception as e:
|
||||
@@ -195,7 +196,9 @@ class TextInjector:
|
||||
if wtype_available:
|
||||
self.wayland_tool = "wtype"
|
||||
logger.info(
|
||||
f"Using {self.wayland_tool} for Wayland text injection (ydotoold not running)"
|
||||
"Using "
|
||||
f"{self.wayland_tool} for Wayland text injection "
|
||||
"(ydotoold not running)"
|
||||
)
|
||||
else:
|
||||
logger.warning("ydotool found but ydotoold daemon not running")
|
||||
@@ -224,6 +227,60 @@ class TextInjector:
|
||||
)
|
||||
raise RuntimeError("Missing required dependencies for text injection")
|
||||
|
||||
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:
|
||||
if shutil.which("xdotool"):
|
||||
self.environment = DesktopEnvironment.X11
|
||||
logger.warning("IBus injection failed, switching to X11 xdotool fallback")
|
||||
return True
|
||||
|
||||
logger.error("IBus fallback failed: xdotool is not available on X11")
|
||||
return False
|
||||
|
||||
if self.environment == DesktopEnvironment.WAYLAND_IBUS:
|
||||
ydotool_available = shutil.which("ydotool") is not None
|
||||
wtype_available = shutil.which("wtype") is not None
|
||||
xdotool_available = shutil.which("xdotool") is not None
|
||||
|
||||
if ydotool_available:
|
||||
try:
|
||||
subprocess.run(
|
||||
["ydotool", "type", ""],
|
||||
check=True,
|
||||
stderr=subprocess.PIPE,
|
||||
timeout=2,
|
||||
)
|
||||
self.wayland_tool = "ydotool"
|
||||
self.environment = DesktopEnvironment.WAYLAND
|
||||
logger.warning("IBus injection failed, switching to Wayland ydotool fallback")
|
||||
return True
|
||||
except (
|
||||
subprocess.CalledProcessError,
|
||||
subprocess.TimeoutExpired,
|
||||
FileNotFoundError,
|
||||
):
|
||||
logger.debug("ydotool fallback unavailable (daemon not running)")
|
||||
|
||||
if wtype_available:
|
||||
self.wayland_tool = "wtype"
|
||||
self.environment = DesktopEnvironment.WAYLAND
|
||||
logger.warning("IBus injection failed, switching to Wayland wtype fallback")
|
||||
return True
|
||||
|
||||
if xdotool_available:
|
||||
self.environment = DesktopEnvironment.WAYLAND_XDOTOOL
|
||||
logger.warning("IBus injection failed, switching to XWayland xdotool fallback")
|
||||
return True
|
||||
|
||||
logger.error(
|
||||
"IBus fallback failed: no Wayland text injection tools available "
|
||||
"(ydotool, wtype, xdotool)"
|
||||
)
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
def _test_xdotool_fallback(self):
|
||||
"""Test if xdotool is working correctly with XWayland."""
|
||||
try:
|
||||
@@ -452,11 +509,23 @@ class TextInjector:
|
||||
args=(text,),
|
||||
daemon=True,
|
||||
).start()
|
||||
return result
|
||||
return True
|
||||
|
||||
logger.warning(
|
||||
"IBus runtime injection failed. Falling back to non-IBus backend."
|
||||
)
|
||||
if not self._switch_to_non_ibus_backend():
|
||||
raise RuntimeError(
|
||||
"IBus injection failed and no non-IBus fallback is available"
|
||||
)
|
||||
else:
|
||||
logger.error("IBus injector not initialized")
|
||||
return False
|
||||
elif (
|
||||
logger.error("IBus injector not initialized, trying non-IBus fallback")
|
||||
if not self._switch_to_non_ibus_backend():
|
||||
raise RuntimeError(
|
||||
"IBus injector not initialized and no non-IBus fallback is available"
|
||||
)
|
||||
|
||||
if (
|
||||
self.environment == DesktopEnvironment.X11
|
||||
or self.environment == DesktopEnvironment.WAYLAND_XDOTOOL
|
||||
):
|
||||
|
||||
@@ -582,6 +582,217 @@ class TestIBusTextInjector(unittest.TestCase):
|
||||
|
||||
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")
|
||||
@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_connection_refused_then_success(
|
||||
self,
|
||||
mock_time,
|
||||
mock_process_running,
|
||||
mock_start_engine,
|
||||
mock_socket_path,
|
||||
mock_ensure_dir,
|
||||
):
|
||||
"""Test retry behavior when first socket connect is refused."""
|
||||
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 = ConnectionRefusedError("refused")
|
||||
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.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",
|
||||
side_effect=[False, True, True],
|
||||
)
|
||||
@patch("vocalinux.text_injection.ibus_engine.time")
|
||||
def test_inject_text_restarts_engine_when_not_running(
|
||||
self,
|
||||
mock_time,
|
||||
mock_process_running,
|
||||
mock_start_engine,
|
||||
mock_socket_path,
|
||||
mock_ensure_dir,
|
||||
):
|
||||
"""Test injector restarts the engine process when its socket is not ready."""
|
||||
mock_socket_path.exists.side_effect = [False, True, 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
|
||||
sock = MagicMock()
|
||||
sock.recv.return_value = b"OK"
|
||||
socket_context.__enter__.return_value = sock
|
||||
socket_context.__exit__.return_value = False
|
||||
|
||||
injector = IBusTextInjector(auto_activate=False)
|
||||
result = injector.inject_text("Hello")
|
||||
|
||||
self.assertTrue(result)
|
||||
mock_start_engine.assert_called_once()
|
||||
|
||||
@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=False)
|
||||
@patch("vocalinux.text_injection.ibus_engine.is_engine_process_running", return_value=False)
|
||||
def test_inject_text_returns_false_when_engine_restart_fails(
|
||||
self,
|
||||
mock_process_running,
|
||||
mock_start_engine,
|
||||
mock_socket_path,
|
||||
mock_ensure_dir,
|
||||
):
|
||||
"""Missing socket plus failed engine restart should abort cleanly."""
|
||||
mock_socket_path.exists.return_value = False
|
||||
|
||||
from vocalinux.text_injection.ibus_engine import IBusTextInjector
|
||||
|
||||
injector = IBusTextInjector(auto_activate=False)
|
||||
result = injector.inject_text("Hello")
|
||||
|
||||
self.assertFalse(result)
|
||||
mock_start_engine.assert_called_once()
|
||||
|
||||
@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_socket_missing_until_final_attempt(
|
||||
self,
|
||||
mock_time,
|
||||
mock_process_running,
|
||||
mock_socket_path,
|
||||
mock_ensure_dir,
|
||||
):
|
||||
"""Persistent missing socket should reach the final error branch."""
|
||||
mock_socket_path.exists.return_value = False
|
||||
|
||||
from vocalinux.text_injection.ibus_engine import IBusTextInjector
|
||||
|
||||
injector = IBusTextInjector(auto_activate=False)
|
||||
result = injector.inject_text("Hello")
|
||||
|
||||
self.assertFalse(result)
|
||||
self.assertEqual(mock_socket_path.exists.call_count, 3)
|
||||
|
||||
@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_times_out_until_final_attempt(
|
||||
self,
|
||||
mock_time,
|
||||
mock_process_running,
|
||||
mock_socket_path,
|
||||
mock_ensure_dir,
|
||||
):
|
||||
"""Repeated socket timeouts should exercise retry and terminal timeout branches."""
|
||||
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 = socket.timeout("slow")
|
||||
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)
|
||||
self.assertEqual(sock.connect.call_count, 3)
|
||||
|
||||
@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=False)
|
||||
@patch("vocalinux.text_injection.ibus_engine.is_engine_process_running", return_value=False)
|
||||
def test_inject_text_connection_refused_aborts_when_restart_fails(
|
||||
self,
|
||||
mock_process_running,
|
||||
mock_start_engine,
|
||||
mock_socket_path,
|
||||
mock_ensure_dir,
|
||||
):
|
||||
"""Connection refusal with a dead process should fail if restart fails."""
|
||||
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)
|
||||
mock_start_engine.assert_called_once()
|
||||
|
||||
@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=False)
|
||||
@patch("vocalinux.text_injection.ibus_engine.is_engine_process_running", return_value=False)
|
||||
def test_inject_text_socket_disappeared_aborts_when_restart_fails(
|
||||
self,
|
||||
mock_process_running,
|
||||
mock_start_engine,
|
||||
mock_socket_path,
|
||||
mock_ensure_dir,
|
||||
):
|
||||
"""Socket disappearance with a dead process should fail if restart fails."""
|
||||
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)
|
||||
mock_start_engine.assert_called_once()
|
||||
|
||||
|
||||
class TestTextInjectorWithIBus(unittest.TestCase):
|
||||
"""Tests for TextInjector integration with IBus."""
|
||||
|
||||
@@ -1415,3 +1415,204 @@ class TestIBusSetupErrorFallback(unittest.TestCase):
|
||||
# Should be able to inject text via xdotool
|
||||
result = injector.inject_text("Hello world")
|
||||
self.assertTrue(result)
|
||||
|
||||
|
||||
class TestIBusRuntimeFallback(unittest.TestCase):
|
||||
"""Tests for runtime fallback when IBus injection fails."""
|
||||
|
||||
def setUp(self):
|
||||
self.patch_which = patch("shutil.which")
|
||||
self.mock_which = self.patch_which.start()
|
||||
self.mock_which.return_value = "/usr/bin/xdotool"
|
||||
|
||||
self.patch_subprocess = patch("subprocess.run")
|
||||
self.mock_subprocess = self.patch_subprocess.start()
|
||||
mock_process = MagicMock()
|
||||
mock_process.returncode = 0
|
||||
mock_process.stdout = ""
|
||||
mock_process.stderr = ""
|
||||
self.mock_subprocess.return_value = mock_process
|
||||
|
||||
def tearDown(self):
|
||||
self.patch_which.stop()
|
||||
self.patch_subprocess.stop()
|
||||
|
||||
@patch("vocalinux.text_injection.text_injector.is_ibus_daemon_running", 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_available", return_value=True)
|
||||
@patch("vocalinux.text_injection.text_injector.IBusTextInjector")
|
||||
def test_runtime_fallback_from_x11_ibus_to_xdotool(
|
||||
self,
|
||||
mock_ibus_class,
|
||||
mock_ibus_available,
|
||||
mock_is_active,
|
||||
mock_daemon,
|
||||
):
|
||||
"""If IBus runtime injection fails on X11, fallback to xdotool should work."""
|
||||
mock_ibus_instance = MagicMock()
|
||||
mock_ibus_instance.inject_text.return_value = False
|
||||
mock_ibus_class.return_value = mock_ibus_instance
|
||||
|
||||
with patch.dict("os.environ", {"XDG_SESSION_TYPE": "x11", "DISPLAY": ":0"}):
|
||||
injector = TextInjector()
|
||||
self.assertEqual(injector.environment, DesktopEnvironment.X11_IBUS)
|
||||
|
||||
result = injector.inject_text("Hello from fallback")
|
||||
|
||||
self.assertTrue(result)
|
||||
self.assertEqual(injector.environment, DesktopEnvironment.X11)
|
||||
|
||||
@patch("vocalinux.text_injection.text_injector.is_ibus_daemon_running", 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_available", return_value=True)
|
||||
@patch("vocalinux.text_injection.text_injector.IBusTextInjector")
|
||||
def test_runtime_fallback_from_wayland_ibus_to_wtype(
|
||||
self,
|
||||
mock_ibus_class,
|
||||
mock_ibus_available,
|
||||
mock_is_active,
|
||||
mock_daemon,
|
||||
):
|
||||
"""If IBus runtime injection fails on Wayland, fallback to wtype should work."""
|
||||
mock_ibus_instance = MagicMock()
|
||||
mock_ibus_instance.inject_text.return_value = False
|
||||
mock_ibus_class.return_value = mock_ibus_instance
|
||||
|
||||
self.mock_which.side_effect = lambda cmd: {
|
||||
"wtype": "/usr/bin/wtype",
|
||||
"xdotool": "/usr/bin/xdotool",
|
||||
}.get(cmd)
|
||||
|
||||
with patch.dict("os.environ", {"XDG_SESSION_TYPE": "wayland"}):
|
||||
injector = TextInjector()
|
||||
self.assertEqual(injector.environment, DesktopEnvironment.WAYLAND_IBUS)
|
||||
|
||||
result = injector.inject_text("Hello via wayland fallback")
|
||||
|
||||
self.assertTrue(result)
|
||||
self.assertEqual(injector.environment, DesktopEnvironment.WAYLAND)
|
||||
self.assertEqual(injector.wayland_tool, "wtype")
|
||||
|
||||
def test_switch_from_x11_ibus_fails_without_xdotool(self):
|
||||
"""X11 IBus fallback should fail cleanly when xdotool is unavailable."""
|
||||
self.mock_which.return_value = None
|
||||
injector = TextInjector.__new__(TextInjector)
|
||||
injector.environment = DesktopEnvironment.X11_IBUS
|
||||
|
||||
result = injector._switch_to_non_ibus_backend()
|
||||
|
||||
self.assertFalse(result)
|
||||
self.assertEqual(injector.environment, DesktopEnvironment.X11_IBUS)
|
||||
|
||||
def test_switch_from_wayland_ibus_prefers_running_ydotool(self):
|
||||
"""Wayland IBus fallback should prefer ydotool when its daemon responds."""
|
||||
self.mock_which.side_effect = lambda cmd: {"ydotool": "/usr/bin/ydotool"}.get(cmd)
|
||||
injector = TextInjector.__new__(TextInjector)
|
||||
injector.environment = DesktopEnvironment.WAYLAND_IBUS
|
||||
injector.wayland_tool = None
|
||||
|
||||
result = injector._switch_to_non_ibus_backend()
|
||||
|
||||
self.assertTrue(result)
|
||||
self.assertEqual(injector.environment, DesktopEnvironment.WAYLAND)
|
||||
self.assertEqual(injector.wayland_tool, "ydotool")
|
||||
self.mock_subprocess.assert_any_call(
|
||||
["ydotool", "type", ""], check=True, stderr=subprocess.PIPE, timeout=2
|
||||
)
|
||||
|
||||
def test_switch_from_wayland_ibus_falls_back_when_ydotool_daemon_down(self):
|
||||
"""If ydotool exists but daemon fails, wtype should be selected next."""
|
||||
self.mock_which.side_effect = lambda cmd: {
|
||||
"ydotool": "/usr/bin/ydotool",
|
||||
"wtype": "/usr/bin/wtype",
|
||||
}.get(cmd)
|
||||
self.mock_subprocess.side_effect = subprocess.CalledProcessError(1, ["ydotool"])
|
||||
injector = TextInjector.__new__(TextInjector)
|
||||
injector.environment = DesktopEnvironment.WAYLAND_IBUS
|
||||
injector.wayland_tool = None
|
||||
|
||||
result = injector._switch_to_non_ibus_backend()
|
||||
|
||||
self.assertTrue(result)
|
||||
self.assertEqual(injector.environment, DesktopEnvironment.WAYLAND)
|
||||
self.assertEqual(injector.wayland_tool, "wtype")
|
||||
|
||||
def test_switch_from_wayland_ibus_uses_xwayland_as_last_tool(self):
|
||||
"""Wayland IBus fallback should use xdotool/XWayland when native tools are absent."""
|
||||
self.mock_which.side_effect = lambda cmd: {"xdotool": "/usr/bin/xdotool"}.get(cmd)
|
||||
injector = TextInjector.__new__(TextInjector)
|
||||
injector.environment = DesktopEnvironment.WAYLAND_IBUS
|
||||
injector.wayland_tool = None
|
||||
|
||||
result = injector._switch_to_non_ibus_backend()
|
||||
|
||||
self.assertTrue(result)
|
||||
self.assertEqual(injector.environment, DesktopEnvironment.WAYLAND_XDOTOOL)
|
||||
|
||||
def test_switch_from_wayland_ibus_fails_without_tools(self):
|
||||
"""Wayland IBus fallback should fail when no non-IBus tools are available."""
|
||||
self.mock_which.return_value = None
|
||||
injector = TextInjector.__new__(TextInjector)
|
||||
injector.environment = DesktopEnvironment.WAYLAND_IBUS
|
||||
injector.wayland_tool = None
|
||||
|
||||
result = injector._switch_to_non_ibus_backend()
|
||||
|
||||
self.assertFalse(result)
|
||||
self.assertEqual(injector.environment, DesktopEnvironment.WAYLAND_IBUS)
|
||||
|
||||
def test_switch_to_non_ibus_backend_noop_outside_ibus_mode(self):
|
||||
"""Non-IBus modes are already on fallback backends."""
|
||||
injector = TextInjector.__new__(TextInjector)
|
||||
injector.environment = DesktopEnvironment.X11
|
||||
|
||||
self.assertTrue(injector._switch_to_non_ibus_backend())
|
||||
self.assertEqual(injector.environment, DesktopEnvironment.X11)
|
||||
|
||||
@patch("vocalinux.text_injection.text_injector.is_ibus_daemon_running", 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_available", return_value=True)
|
||||
@patch("vocalinux.text_injection.text_injector.IBusTextInjector")
|
||||
def test_successful_ibus_injection_still_copies_to_clipboard_when_enabled(
|
||||
self,
|
||||
mock_ibus_class,
|
||||
mock_ibus_available,
|
||||
mock_is_active,
|
||||
mock_daemon,
|
||||
):
|
||||
"""Successful IBus injection should keep the existing optional clipboard copy path."""
|
||||
mock_ibus_instance = MagicMock()
|
||||
mock_ibus_instance.inject_text.return_value = True
|
||||
mock_ibus_class.return_value = mock_ibus_instance
|
||||
|
||||
with patch.dict("os.environ", {"XDG_SESSION_TYPE": "x11", "DISPLAY": ":0"}):
|
||||
injector = TextInjector()
|
||||
with patch.object(injector, "_should_copy_to_clipboard", return_value=True):
|
||||
with patch("threading.Thread") as mock_thread:
|
||||
result = injector.inject_text("copy me")
|
||||
|
||||
self.assertTrue(result)
|
||||
mock_thread.assert_called_once()
|
||||
mock_thread.return_value.start.assert_called_once()
|
||||
|
||||
@patch("vocalinux.text_injection.text_injector.is_ibus_daemon_running", 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_available", return_value=True)
|
||||
@patch("vocalinux.text_injection.text_injector.IBusTextInjector")
|
||||
def test_uninitialized_ibus_injector_uses_non_ibus_fallback(
|
||||
self,
|
||||
mock_ibus_class,
|
||||
mock_ibus_available,
|
||||
mock_is_active,
|
||||
mock_daemon,
|
||||
):
|
||||
"""If the IBus injector disappears at runtime, fallback backend should still run."""
|
||||
mock_ibus_class.return_value = MagicMock()
|
||||
|
||||
with patch.dict("os.environ", {"XDG_SESSION_TYPE": "x11", "DISPLAY": ":0"}):
|
||||
injector = TextInjector()
|
||||
injector._ibus_injector = None
|
||||
result = injector.inject_text("fallback without ibus instance")
|
||||
|
||||
self.assertTrue(result)
|
||||
self.assertEqual(injector.environment, DesktopEnvironment.X11)
|
||||
|
||||
@@ -7,6 +7,13 @@ from unittest.mock import MagicMock, patch
|
||||
|
||||
class TestXkbLayoutFunctions(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.env_patcher = patch.dict("os.environ", {"XDG_SESSION_TYPE": "x11"})
|
||||
self.env_patcher.start()
|
||||
|
||||
def tearDown(self):
|
||||
self.env_patcher.stop()
|
||||
|
||||
@patch("vocalinux.text_injection.ibus_engine.subprocess.run")
|
||||
def test_get_current_xkb_layout_success(self, mock_run):
|
||||
mock_run.return_value = MagicMock(
|
||||
|
||||
Reference in New Issue
Block a user