Jatin K MalikandGitHub
e1384aad2e
* 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
1730 lines
72 KiB
Python
1730 lines
72 KiB
Python
"""
|
|
Tests for IBus engine functionality.
|
|
"""
|
|
|
|
import os
|
|
import socket
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
# Mock GI imports before importing the module
|
|
mock_gi = MagicMock()
|
|
mock_ibus = MagicMock()
|
|
mock_glib = MagicMock()
|
|
mock_gobject = MagicMock()
|
|
|
|
# Set up IBus mock
|
|
mock_ibus.Engine = MagicMock
|
|
mock_ibus.Bus = MagicMock
|
|
mock_ibus.Factory = MagicMock
|
|
mock_ibus.Text = MagicMock()
|
|
mock_ibus.Text.new_from_string = MagicMock(return_value=MagicMock())
|
|
|
|
sys.modules["gi"] = mock_gi
|
|
sys.modules["gi.repository"] = MagicMock()
|
|
sys.modules["gi.repository"].IBus = mock_ibus
|
|
sys.modules["gi.repository"].GLib = mock_glib
|
|
sys.modules["gi.repository"].GObject = mock_gobject
|
|
|
|
|
|
class TestIBusSetupError(unittest.TestCase):
|
|
"""Tests for IBusSetupError exception."""
|
|
|
|
def test_is_runtime_error_subclass(self):
|
|
"""Test that IBusSetupError inherits from RuntimeError."""
|
|
from vocalinux.text_injection.ibus_engine import IBusSetupError
|
|
|
|
self.assertTrue(issubclass(IBusSetupError, RuntimeError))
|
|
|
|
def test_accepts_message(self):
|
|
"""Test that IBusSetupError accepts a message."""
|
|
from vocalinux.text_injection.ibus_engine import IBusSetupError
|
|
|
|
error = IBusSetupError("Test message")
|
|
self.assertEqual(str(error), "Test message")
|
|
|
|
|
|
class TestIBusTextInjectorSetupFailures(unittest.TestCase):
|
|
"""Tests for IBusTextInjector raising IBusSetupError on setup failures."""
|
|
|
|
@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")
|
|
def test_raises_on_engine_process_failure(self, mock_start_engine, mock_ensure_dir):
|
|
"""Test that IBusSetupError is raised when engine process fails to start."""
|
|
mock_start_engine.return_value = False # Process start fails
|
|
|
|
from vocalinux.text_injection.ibus_engine import IBusSetupError, IBusTextInjector
|
|
|
|
with self.assertRaises(IBusSetupError) as context:
|
|
IBusTextInjector(auto_activate=True)
|
|
|
|
self.assertIn("Failed to start IBus engine process", str(context.exception))
|
|
|
|
@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")
|
|
@patch("vocalinux.text_injection.ibus_engine.is_engine_active")
|
|
@patch("vocalinux.text_injection.ibus_engine.get_current_engine")
|
|
@patch("vocalinux.text_injection.ibus_engine.switch_engine")
|
|
def test_raises_on_activation_failure(
|
|
self,
|
|
mock_switch,
|
|
mock_get_current,
|
|
mock_is_active,
|
|
mock_start_engine,
|
|
mock_socket_path,
|
|
mock_ensure_dir,
|
|
):
|
|
"""Test that IBusSetupError is raised when engine activation fails."""
|
|
mock_start_engine.return_value = True
|
|
mock_socket_path.exists.return_value = True
|
|
mock_is_active.return_value = False
|
|
mock_get_current.return_value = "xkb:us::eng"
|
|
mock_switch.return_value = False # Activation fails
|
|
|
|
from vocalinux.text_injection.ibus_engine import IBusSetupError, IBusTextInjector
|
|
|
|
with self.assertRaises(IBusSetupError) as context:
|
|
IBusTextInjector(auto_activate=True)
|
|
|
|
self.assertIn("Failed to activate Vocalinux IBus engine", str(context.exception))
|
|
|
|
@patch("vocalinux.text_injection.ibus_engine.IBUS_AVAILABLE", True)
|
|
@patch("vocalinux.text_injection.ibus_engine.ensure_ibus_dir")
|
|
@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")
|
|
@patch("vocalinux.text_injection.ibus_engine.is_engine_active")
|
|
@patch("vocalinux.text_injection.ibus_engine.get_current_engine")
|
|
@patch("vocalinux.text_injection.ibus_engine.switch_engine")
|
|
def test_warns_and_proceeds_when_socket_not_ready(
|
|
self,
|
|
mock_switch,
|
|
mock_get_engine,
|
|
mock_is_active,
|
|
mock_restore_xkb,
|
|
mock_get_xkb,
|
|
mock_start_engine,
|
|
mock_wait_ready,
|
|
mock_ensure_dir,
|
|
):
|
|
"""Readiness probe is invoked as part of setup before returning."""
|
|
mock_start_engine.return_value = True
|
|
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
|
|
|
|
injector = IBusTextInjector(auto_activate=True)
|
|
|
|
mock_wait_ready.assert_called_once_with()
|
|
self.assertIsNotNone(injector)
|
|
|
|
|
|
class TestIBusEngineHelpers(unittest.TestCase):
|
|
"""Tests for IBus engine helper functions."""
|
|
|
|
def test_ensure_ibus_dir_creates_directory(self):
|
|
"""Test that ensure_ibus_dir creates the directory."""
|
|
with patch("vocalinux.text_injection.ibus_engine.VOCALINUX_IBUS_DIR") as mock_dir:
|
|
mock_dir.mkdir = MagicMock()
|
|
from vocalinux.text_injection.ibus_engine import ensure_ibus_dir
|
|
|
|
ensure_ibus_dir()
|
|
mock_dir.mkdir.assert_called_once_with(parents=True, exist_ok=True)
|
|
|
|
def test_is_ibus_available_returns_constant(self):
|
|
"""Test is_ibus_available returns the module constant."""
|
|
from vocalinux.text_injection.ibus_engine import IBUS_AVAILABLE, is_ibus_available
|
|
|
|
self.assertEqual(is_ibus_available(), IBUS_AVAILABLE)
|
|
|
|
|
|
class TestIsEngineActive(unittest.TestCase):
|
|
"""Tests for is_engine_active function."""
|
|
|
|
@patch("subprocess.run")
|
|
def test_engine_active(self, mock_run):
|
|
"""Test detection when engine is active."""
|
|
mock_run.return_value = MagicMock(stdout="vocalinux", returncode=0)
|
|
|
|
from vocalinux.text_injection.ibus_engine import is_engine_active
|
|
|
|
result = is_engine_active()
|
|
self.assertTrue(result)
|
|
|
|
@patch("subprocess.run")
|
|
def test_engine_not_active(self, mock_run):
|
|
"""Test detection when engine is not active."""
|
|
mock_run.return_value = MagicMock(stdout="xkb:us::eng", returncode=0)
|
|
|
|
from vocalinux.text_injection.ibus_engine import is_engine_active
|
|
|
|
result = is_engine_active()
|
|
self.assertFalse(result)
|
|
|
|
@patch("subprocess.run")
|
|
def test_subprocess_error(self, mock_run):
|
|
"""Test handling of subprocess errors."""
|
|
import subprocess
|
|
|
|
mock_run.side_effect = subprocess.SubprocessError("Command failed")
|
|
|
|
from vocalinux.text_injection.ibus_engine import is_engine_active
|
|
|
|
result = is_engine_active()
|
|
self.assertFalse(result)
|
|
|
|
|
|
class TestGetCurrentEngine(unittest.TestCase):
|
|
"""Tests for get_current_engine function."""
|
|
|
|
@patch("subprocess.run")
|
|
def test_get_current_engine_success(self, mock_run):
|
|
"""Test getting current engine successfully."""
|
|
mock_run.return_value = MagicMock(stdout="xkb:fr::fra\n", returncode=0)
|
|
|
|
from vocalinux.text_injection.ibus_engine import get_current_engine
|
|
|
|
result = get_current_engine()
|
|
self.assertEqual(result, "xkb:fr::fra")
|
|
|
|
@patch("subprocess.run")
|
|
def test_get_current_engine_error(self, mock_run):
|
|
"""Test handling of errors when getting current engine."""
|
|
mock_run.return_value = MagicMock(returncode=1)
|
|
|
|
from vocalinux.text_injection.ibus_engine import get_current_engine
|
|
|
|
result = get_current_engine()
|
|
self.assertIsNone(result)
|
|
|
|
@patch("subprocess.run")
|
|
def test_subprocess_exception(self, mock_run):
|
|
"""Test handling of subprocess exceptions."""
|
|
import subprocess
|
|
|
|
mock_run.side_effect = subprocess.SubprocessError("Failed")
|
|
|
|
from vocalinux.text_injection.ibus_engine import get_current_engine
|
|
|
|
result = get_current_engine()
|
|
self.assertIsNone(result)
|
|
|
|
|
|
class TestSwitchEngine(unittest.TestCase):
|
|
"""Tests for switch_engine function."""
|
|
|
|
@patch("subprocess.run")
|
|
def test_switch_engine_success(self, mock_run):
|
|
"""Test switching engine successfully."""
|
|
# First call: ibus engine <name> (switch)
|
|
# Second call: ibus engine (get current) - for verification
|
|
switch_result = MagicMock(returncode=0)
|
|
verify_result = MagicMock(returncode=0, stdout="vocalinux\n")
|
|
mock_run.side_effect = [switch_result, verify_result]
|
|
|
|
from vocalinux.text_injection.ibus_engine import switch_engine
|
|
|
|
result = switch_engine("vocalinux")
|
|
self.assertTrue(result)
|
|
self.assertEqual(mock_run.call_count, 2)
|
|
|
|
@patch("subprocess.run")
|
|
def test_switch_engine_failure(self, mock_run):
|
|
"""Test switching engine failure."""
|
|
# First call: ibus engine <name> (switch)
|
|
# Second call: ibus engine (get current) - returns different engine
|
|
switch_result = MagicMock(returncode=1)
|
|
verify_result = MagicMock(returncode=0, stdout="xkb:us::eng\n")
|
|
mock_run.side_effect = [switch_result, verify_result]
|
|
|
|
from vocalinux.text_injection.ibus_engine import switch_engine
|
|
|
|
result = switch_engine("nonexistent")
|
|
self.assertFalse(result)
|
|
|
|
@patch("subprocess.run")
|
|
def test_switch_engine_exception(self, mock_run):
|
|
"""Test handling of subprocess exceptions."""
|
|
import subprocess
|
|
|
|
mock_run.side_effect = subprocess.SubprocessError("Failed")
|
|
|
|
from vocalinux.text_injection.ibus_engine import switch_engine
|
|
|
|
result = switch_engine("vocalinux")
|
|
self.assertFalse(result)
|
|
|
|
|
|
class TestIBusTextInjector(unittest.TestCase):
|
|
"""Tests for IBusTextInjector class."""
|
|
|
|
def setUp(self):
|
|
"""Set up test fixtures."""
|
|
# Create temp directory for socket
|
|
self.temp_dir = tempfile.mkdtemp()
|
|
self.socket_path = Path(self.temp_dir) / "inject.sock"
|
|
|
|
def tearDown(self):
|
|
"""Clean up test fixtures."""
|
|
import shutil
|
|
|
|
shutil.rmtree(self.temp_dir, ignore_errors=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.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")
|
|
@patch("vocalinux.text_injection.ibus_engine.switch_engine")
|
|
def test_init_auto_activate(
|
|
self,
|
|
mock_switch,
|
|
mock_get_current,
|
|
mock_start_engine,
|
|
mock_is_active,
|
|
mock_wait_ready,
|
|
mock_socket_path,
|
|
mock_ensure_dir,
|
|
):
|
|
"""Test initialization with auto_activate=True."""
|
|
mock_socket_path.exists.return_value = True
|
|
mock_is_active.return_value = False
|
|
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
|
|
|
|
injector = IBusTextInjector(auto_activate=True)
|
|
|
|
# ensure_ibus_dir is called at least once
|
|
self.assertGreaterEqual(mock_ensure_dir.call_count, 1)
|
|
# Engine should be started
|
|
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)
|
|
@patch("vocalinux.text_injection.ibus_engine.ensure_ibus_dir")
|
|
def test_init_no_auto_activate(self, mock_ensure_dir):
|
|
"""Test initialization with auto_activate=False."""
|
|
from vocalinux.text_injection.ibus_engine import IBusTextInjector
|
|
|
|
injector = IBusTextInjector(auto_activate=False)
|
|
|
|
mock_ensure_dir.assert_called_once()
|
|
self.assertIsNone(injector._previous_engine)
|
|
|
|
@patch("vocalinux.text_injection.ibus_engine.IBUS_AVAILABLE", False)
|
|
def test_init_ibus_not_available(self):
|
|
"""Test initialization raises when IBus is not available."""
|
|
from vocalinux.text_injection.ibus_engine import IBusTextInjector
|
|
|
|
with self.assertRaises(RuntimeError):
|
|
IBusTextInjector()
|
|
|
|
@patch("vocalinux.text_injection.ibus_engine.IBUS_AVAILABLE", True)
|
|
@patch("vocalinux.text_injection.ibus_engine.ensure_ibus_dir")
|
|
@patch("vocalinux.text_injection.ibus_engine.switch_engine")
|
|
def test_stop_restores_engine(self, mock_switch, mock_ensure_dir):
|
|
"""Test stop() restores previous engine."""
|
|
from vocalinux.text_injection.ibus_engine import IBusTextInjector
|
|
|
|
injector = IBusTextInjector(auto_activate=False)
|
|
injector._previous_engine = "xkb:fr::fra"
|
|
|
|
injector.stop()
|
|
|
|
mock_switch.assert_called_once_with("xkb:fr::fra")
|
|
self.assertIsNone(injector._previous_engine)
|
|
|
|
@patch("vocalinux.text_injection.ibus_engine.IBUS_AVAILABLE", True)
|
|
@patch("vocalinux.text_injection.ibus_engine.ensure_ibus_dir")
|
|
def test_stop_no_previous_engine(self, mock_ensure_dir):
|
|
"""Test stop() when no previous engine was saved."""
|
|
from vocalinux.text_injection.ibus_engine import IBusTextInjector
|
|
|
|
injector = IBusTextInjector(auto_activate=False)
|
|
|
|
# Should not raise
|
|
injector.stop()
|
|
|
|
@patch("vocalinux.text_injection.ibus_engine.restore_xkb_layout")
|
|
@patch("vocalinux.text_injection.ibus_engine.stop_engine_process")
|
|
@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_stop_restores_xkb_layout(
|
|
self, mock_ensure_dir, mock_switch, mock_stop_proc, mock_restore_xkb
|
|
):
|
|
"""Test stop() restores the captured XKB layout (#292)."""
|
|
from vocalinux.text_injection.ibus_engine import IBusTextInjector
|
|
|
|
injector = IBusTextInjector(auto_activate=False)
|
|
injector._previous_xkb_layout = ("es", "catalan", "compose:menu")
|
|
|
|
injector.stop()
|
|
|
|
mock_restore_xkb.assert_called_once_with("es", "catalan", "compose:menu")
|
|
self.assertIsNone(injector._previous_xkb_layout)
|
|
|
|
@patch("vocalinux.text_injection.ibus_engine.restore_xkb_layout")
|
|
@patch("vocalinux.text_injection.ibus_engine.stop_engine_process")
|
|
@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_stop_skips_xkb_restore_when_no_layout(
|
|
self, mock_ensure_dir, mock_switch, mock_stop_proc, mock_restore_xkb
|
|
):
|
|
"""Test stop() skips XKB restore when no layout was captured."""
|
|
from vocalinux.text_injection.ibus_engine import IBusTextInjector
|
|
|
|
injector = IBusTextInjector(auto_activate=False)
|
|
injector._previous_xkb_layout = None
|
|
|
|
injector.stop()
|
|
|
|
mock_restore_xkb.assert_not_called()
|
|
|
|
@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")
|
|
@patch("vocalinux.text_injection.ibus_engine.is_engine_active", return_value=False)
|
|
@patch("vocalinux.text_injection.ibus_engine.IBUS_AVAILABLE", True)
|
|
@patch("vocalinux.text_injection.ibus_engine.ensure_ibus_dir")
|
|
def test_setup_engine_captures_and_restores_xkb(
|
|
self,
|
|
mock_ensure_dir,
|
|
mock_active,
|
|
mock_get_engine,
|
|
mock_switch,
|
|
mock_start_proc,
|
|
mock_wait_ready,
|
|
mock_socket_path,
|
|
mock_get_xkb,
|
|
mock_restore_xkb,
|
|
):
|
|
"""Test _setup_engine captures XKB layout and restores it after activation (#292)."""
|
|
from vocalinux.text_injection.ibus_engine import IBusTextInjector
|
|
|
|
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()
|
|
|
|
mock_get_xkb.assert_called_once()
|
|
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):
|
|
"""Test inject_text with empty text returns True."""
|
|
from vocalinux.text_injection.ibus_engine import IBusTextInjector
|
|
|
|
injector = IBusTextInjector(auto_activate=False)
|
|
|
|
result = injector.inject_text("")
|
|
self.assertTrue(result)
|
|
|
|
result = injector.inject_text(" ")
|
|
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")
|
|
def test_inject_text_socket_not_found(self, mock_socket_path, mock_ensure_dir):
|
|
"""Test inject_text when socket doesn't exist."""
|
|
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)
|
|
|
|
@patch("vocalinux.text_injection.ibus_engine.IBUS_AVAILABLE", True)
|
|
@patch("vocalinux.text_injection.ibus_engine.ensure_ibus_dir")
|
|
def test_inject_text_success(self, mock_ensure_dir):
|
|
"""Test successful text injection via socket."""
|
|
from vocalinux.text_injection.ibus_engine import IBusTextInjector
|
|
|
|
# Create a mock server socket
|
|
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:
|
|
data = conn.recv(65536)
|
|
self.assertEqual(data.decode("utf-8"), "Hello World")
|
|
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 World")
|
|
|
|
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, 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 ENGINE_NAME, IBusTextInjector
|
|
|
|
server_sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
server_sock.bind(str(self.socket_path))
|
|
server_sock.listen(2)
|
|
|
|
call_count = 0
|
|
|
|
def handle_connection():
|
|
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):
|
|
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)
|
|
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)
|
|
@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)
|
|
|
|
@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")
|
|
def test_inject_text_timeout(self, mock_socket_path, mock_ensure_dir):
|
|
"""Test text injection timeout handling."""
|
|
mock_socket_path.exists.return_value = True
|
|
|
|
from vocalinux.text_injection.ibus_engine import IBusTextInjector
|
|
|
|
with patch("socket.socket") as mock_socket_class:
|
|
mock_socket_instance = MagicMock()
|
|
mock_socket_class.return_value.__enter__ = MagicMock(return_value=mock_socket_instance)
|
|
mock_socket_class.return_value.__exit__ = MagicMock(return_value=False)
|
|
mock_socket_instance.connect.side_effect = socket.timeout("Connection timeout")
|
|
|
|
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")
|
|
@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."""
|
|
|
|
def setUp(self):
|
|
"""Set up test fixtures."""
|
|
self.patch_which = patch("shutil.which")
|
|
self.mock_which = self.patch_which.start()
|
|
|
|
self.patch_subprocess = patch("subprocess.run")
|
|
self.mock_subprocess = self.patch_subprocess.start()
|
|
|
|
self.patch_sleep = patch("time.sleep")
|
|
self.mock_sleep = self.patch_sleep.start()
|
|
|
|
# Default to having xdotool available
|
|
self.mock_which.return_value = "/usr/bin/xdotool"
|
|
|
|
mock_process = MagicMock()
|
|
mock_process.returncode = 0
|
|
mock_process.stdout = "1234"
|
|
mock_process.stderr = ""
|
|
self.mock_subprocess.return_value = mock_process
|
|
|
|
def tearDown(self):
|
|
"""Clean up after tests."""
|
|
self.patch_which.stop()
|
|
self.patch_subprocess.stop()
|
|
self.patch_sleep.stop()
|
|
|
|
@patch("vocalinux.text_injection.ibus_engine.IBUS_AVAILABLE", True)
|
|
@patch("vocalinux.text_injection.ibus_engine.is_ibus_available")
|
|
@patch("vocalinux.text_injection.ibus_engine.is_ibus_active_input_method")
|
|
@patch("vocalinux.text_injection.ibus_engine.IBusTextInjector")
|
|
def test_wayland_prefers_ibus(
|
|
self, mock_injector_class, mock_is_active_im, mock_ibus_available
|
|
):
|
|
"""Test that Wayland environment prefers IBus when available."""
|
|
mock_ibus_available.return_value = True
|
|
mock_is_active_im.return_value = True
|
|
mock_injector_instance = MagicMock()
|
|
mock_injector_class.return_value = mock_injector_instance
|
|
|
|
with patch.dict("os.environ", {"XDG_SESSION_TYPE": "wayland"}):
|
|
from vocalinux.text_injection.text_injector import DesktopEnvironment, TextInjector
|
|
|
|
# Patch the import inside text_injector
|
|
with patch(
|
|
"vocalinux.text_injection.text_injector.is_ibus_available",
|
|
return_value=True,
|
|
):
|
|
with patch(
|
|
"vocalinux.text_injection.text_injector.is_ibus_active_input_method",
|
|
return_value=True,
|
|
):
|
|
with patch(
|
|
"vocalinux.text_injection.text_injector.IBusTextInjector",
|
|
mock_injector_class,
|
|
):
|
|
injector = TextInjector()
|
|
|
|
# Should be using IBus
|
|
self.assertEqual(injector.environment, DesktopEnvironment.WAYLAND_IBUS)
|
|
|
|
@patch("vocalinux.text_injection.ibus_engine.IBUS_AVAILABLE", True)
|
|
@patch("vocalinux.text_injection.ibus_engine.is_ibus_available")
|
|
@patch("vocalinux.text_injection.ibus_engine.is_ibus_active_input_method")
|
|
@patch("vocalinux.text_injection.ibus_engine.IBusTextInjector")
|
|
def test_x11_prefers_ibus(self, mock_injector_class, mock_is_active_im, mock_ibus_available):
|
|
"""Test that X11 environment prefers IBus when available."""
|
|
mock_ibus_available.return_value = True
|
|
mock_is_active_im.return_value = True
|
|
mock_injector_instance = MagicMock()
|
|
mock_injector_class.return_value = mock_injector_instance
|
|
|
|
with patch.dict("os.environ", {"XDG_SESSION_TYPE": "x11"}):
|
|
from vocalinux.text_injection.text_injector import DesktopEnvironment, TextInjector
|
|
|
|
with patch(
|
|
"vocalinux.text_injection.text_injector.is_ibus_available",
|
|
return_value=True,
|
|
):
|
|
with patch(
|
|
"vocalinux.text_injection.text_injector.is_ibus_active_input_method",
|
|
return_value=True,
|
|
):
|
|
with patch(
|
|
"vocalinux.text_injection.text_injector.IBusTextInjector",
|
|
mock_injector_class,
|
|
):
|
|
injector = TextInjector()
|
|
|
|
self.assertEqual(injector.environment, DesktopEnvironment.X11_IBUS)
|
|
|
|
@patch("vocalinux.text_injection.text_injector.is_ibus_available")
|
|
def test_x11_fallback_when_ibus_unavailable(self, mock_ibus_available):
|
|
"""Test X11 falls back to xdotool when IBus unavailable."""
|
|
mock_ibus_available.return_value = False
|
|
|
|
with patch.dict("os.environ", {"XDG_SESSION_TYPE": "x11"}):
|
|
self.mock_which.side_effect = lambda cmd: (
|
|
"/usr/bin/xdotool" if cmd == "xdotool" else None
|
|
)
|
|
|
|
from vocalinux.text_injection.text_injector import DesktopEnvironment, TextInjector
|
|
|
|
injector = TextInjector()
|
|
|
|
self.assertEqual(injector.environment, DesktopEnvironment.X11)
|
|
|
|
@patch("vocalinux.text_injection.text_injector.is_ibus_available")
|
|
def test_wayland_fallback_when_ibus_unavailable(self, mock_ibus_available):
|
|
"""Test Wayland falls back to other tools when IBus unavailable."""
|
|
mock_ibus_available.return_value = False
|
|
|
|
with patch.dict("os.environ", {"XDG_SESSION_TYPE": "wayland"}):
|
|
# Make ydotool available
|
|
self.mock_which.side_effect = lambda cmd: (
|
|
"/usr/bin/ydotool" if cmd == "ydotool" else None
|
|
)
|
|
|
|
from vocalinux.text_injection.text_injector import DesktopEnvironment, TextInjector
|
|
|
|
injector = TextInjector()
|
|
|
|
# Should fall back to WAYLAND with ydotool
|
|
self.assertEqual(injector.environment, DesktopEnvironment.WAYLAND)
|
|
self.assertEqual(injector.wayland_tool, "ydotool")
|
|
|
|
@patch("vocalinux.text_injection.text_injector.is_ibus_available")
|
|
@patch("vocalinux.text_injection.text_injector.IBusTextInjector")
|
|
def test_ibus_inject_text(self, mock_injector_class, mock_ibus_available):
|
|
"""Test text injection via IBus."""
|
|
mock_ibus_available.return_value = True
|
|
mock_injector_instance = MagicMock()
|
|
mock_injector_instance.inject_text.return_value = True
|
|
mock_injector_class.return_value = mock_injector_instance
|
|
|
|
with patch.dict("os.environ", {"XDG_SESSION_TYPE": "wayland"}):
|
|
from vocalinux.text_injection.text_injector import TextInjector
|
|
|
|
injector = TextInjector()
|
|
result = injector.inject_text("Hello via IBus")
|
|
|
|
self.assertTrue(result)
|
|
mock_injector_instance.inject_text.assert_called_once_with("Hello via IBus")
|
|
|
|
@patch("vocalinux.text_injection.text_injector.is_ibus_available")
|
|
@patch("vocalinux.text_injection.text_injector.IBusTextInjector")
|
|
def test_stop_calls_ibus_stop(self, mock_injector_class, mock_ibus_available):
|
|
"""Test that stop() calls IBus injector stop."""
|
|
mock_ibus_available.return_value = True
|
|
mock_injector_instance = MagicMock()
|
|
mock_injector_class.return_value = mock_injector_instance
|
|
|
|
with patch.dict("os.environ", {"XDG_SESSION_TYPE": "wayland"}):
|
|
from vocalinux.text_injection.text_injector import TextInjector
|
|
|
|
injector = TextInjector()
|
|
injector.stop()
|
|
|
|
@patch("vocalinux.text_injection.text_injector.is_ibus_available")
|
|
@patch("vocalinux.text_injection.text_injector.is_ibus_active_input_method")
|
|
@patch("vocalinux.text_injection.text_injector.IBusTextInjector")
|
|
def test_ibus_inject_text(self, mock_injector_class, mock_is_active_im, mock_ibus_available):
|
|
"""Test text injection via IBus."""
|
|
mock_ibus_available.return_value = True
|
|
mock_is_active_im.return_value = True
|
|
mock_injector_instance = MagicMock()
|
|
mock_injector_instance.inject_text.return_value = True
|
|
mock_injector_class.return_value = mock_injector_instance
|
|
|
|
with patch.dict("os.environ", {"XDG_SESSION_TYPE": "wayland"}):
|
|
from vocalinux.text_injection.text_injector import TextInjector
|
|
|
|
injector = TextInjector()
|
|
result = injector.inject_text("Hello via IBus")
|
|
|
|
self.assertTrue(result)
|
|
mock_injector_instance.inject_text.assert_called_once_with("Hello via IBus")
|
|
|
|
@patch("vocalinux.text_injection.text_injector.is_ibus_available")
|
|
@patch("vocalinux.text_injection.text_injector.is_ibus_active_input_method")
|
|
@patch("vocalinux.text_injection.text_injector.IBusTextInjector")
|
|
def test_stop_calls_ibus_stop(
|
|
self, mock_injector_class, mock_is_active_im, mock_ibus_available
|
|
):
|
|
"""Test that stop() calls IBus injector stop."""
|
|
mock_ibus_available.return_value = True
|
|
mock_is_active_im.return_value = True
|
|
mock_injector_instance = MagicMock()
|
|
mock_injector_class.return_value = mock_injector_instance
|
|
|
|
with patch.dict("os.environ", {"XDG_SESSION_TYPE": "wayland"}):
|
|
from vocalinux.text_injection.text_injector import TextInjector
|
|
|
|
injector = TextInjector()
|
|
injector.stop()
|
|
|
|
mock_injector_instance.stop.assert_called_once()
|
|
|
|
|
|
class TestDesktopEnvironmentEnumWithIBus(unittest.TestCase):
|
|
"""Tests for DesktopEnvironment enum including IBus variants."""
|
|
|
|
def test_enum_includes_wayland_ibus(self):
|
|
"""Test that WAYLAND_IBUS enum value exists."""
|
|
from vocalinux.text_injection.text_injector import DesktopEnvironment
|
|
|
|
self.assertEqual(DesktopEnvironment.WAYLAND_IBUS.value, "wayland-ibus")
|
|
|
|
def test_enum_includes_x11_ibus(self):
|
|
"""Test that X11_IBUS enum value exists."""
|
|
from vocalinux.text_injection.text_injector import DesktopEnvironment
|
|
|
|
self.assertEqual(DesktopEnvironment.X11_IBUS.value, "x11-ibus")
|
|
|
|
|
|
class TestIsIbusActiveInputMethod(unittest.TestCase):
|
|
"""Tests for is_ibus_active_input_method function."""
|
|
|
|
def test_detects_ibus_via_gtk_im_module(self):
|
|
"""Test detection when GTK_IM_MODULE is set to ibus."""
|
|
with patch.dict(os.environ, {"GTK_IM_MODULE": "ibus"}, clear=True):
|
|
from vocalinux.text_injection.ibus_engine import is_ibus_active_input_method
|
|
|
|
result = is_ibus_active_input_method()
|
|
self.assertTrue(result)
|
|
|
|
def test_detects_ibus_python_via_gtk_im_module(self):
|
|
"""Test detection when GTK_IM_MODULE is set to ibus-python."""
|
|
with patch.dict(os.environ, {"GTK_IM_MODULE": "ibus-python"}, clear=True):
|
|
from vocalinux.text_injection.ibus_engine import is_ibus_active_input_method
|
|
|
|
result = is_ibus_active_input_method()
|
|
self.assertTrue(result)
|
|
|
|
def test_detects_ibus_via_qt_im_module(self):
|
|
"""Test detection when QT_IM_MODULE is set to ibus."""
|
|
with patch.dict(os.environ, {"QT_IM_MODULE": "ibus"}, clear=True):
|
|
from vocalinux.text_injection.ibus_engine import is_ibus_active_input_method
|
|
|
|
result = is_ibus_active_input_method()
|
|
self.assertTrue(result)
|
|
|
|
def test_detects_ibus_via_xmodifiers(self):
|
|
"""Test detection when XMODIFIERS contains @im=ibus."""
|
|
with patch.dict(os.environ, {"XMODIFIERS": "@im=ibus"}, clear=True):
|
|
from vocalinux.text_injection.ibus_engine import is_ibus_active_input_method
|
|
|
|
result = is_ibus_active_input_method()
|
|
self.assertTrue(result)
|
|
|
|
def test_not_active_when_gtk_is_xim(self):
|
|
"""Test returns False when GTK_IM_MODULE is xim."""
|
|
with patch.dict(os.environ, {"GTK_IM_MODULE": "xim"}, clear=True):
|
|
from vocalinux.text_injection.ibus_engine import is_ibus_active_input_method
|
|
|
|
result = is_ibus_active_input_method()
|
|
self.assertFalse(result)
|
|
|
|
def test_not_active_when_qt_is_fcitx(self):
|
|
"""Test returns False when QT_IM_MODULE is fcitx."""
|
|
with patch.dict(os.environ, {"QT_IM_MODULE": "fcitx"}, clear=True):
|
|
from vocalinux.text_injection.ibus_engine import is_ibus_active_input_method
|
|
|
|
result = is_ibus_active_input_method()
|
|
self.assertFalse(result)
|
|
|
|
def test_not_active_when_xmodifiers_is_fcitx(self):
|
|
"""Test returns False when XMODIFIERS is @im=fcitx."""
|
|
with patch.dict(os.environ, {"XMODIFIERS": "@im=fcitx"}, clear=True):
|
|
from vocalinux.text_injection.ibus_engine import is_ibus_active_input_method
|
|
|
|
result = is_ibus_active_input_method()
|
|
self.assertFalse(result)
|
|
|
|
@patch("vocalinux.text_injection.ibus_engine.is_ibus_daemon_running", return_value=False)
|
|
def test_not_active_when_no_env_vars_and_no_daemon(self, mock_daemon):
|
|
"""Test returns False when no env vars set and daemon is not running."""
|
|
with patch.dict(os.environ, {}, clear=True):
|
|
from vocalinux.text_injection.ibus_engine import is_ibus_active_input_method
|
|
|
|
result = is_ibus_active_input_method()
|
|
self.assertFalse(result)
|
|
|
|
@patch("vocalinux.text_injection.ibus_engine.get_current_engine", return_value="xkb:us::eng")
|
|
@patch("vocalinux.text_injection.ibus_engine.is_ibus_daemon_running", return_value=True)
|
|
def test_active_via_daemon_when_no_env_vars(self, mock_daemon, mock_engine):
|
|
"""Test returns True when no env vars but daemon is running with active engine."""
|
|
with patch.dict(os.environ, {}, clear=True):
|
|
from vocalinux.text_injection.ibus_engine import is_ibus_active_input_method
|
|
|
|
result = is_ibus_active_input_method()
|
|
self.assertTrue(result)
|
|
|
|
@patch("vocalinux.text_injection.ibus_engine.get_current_engine", return_value=None)
|
|
@patch("vocalinux.text_injection.ibus_engine.is_ibus_daemon_running", return_value=True)
|
|
def test_not_active_when_daemon_running_but_no_engine(self, mock_daemon, mock_engine):
|
|
"""Test returns False when daemon is running but no engine is active."""
|
|
with patch.dict(os.environ, {}, clear=True):
|
|
from vocalinux.text_injection.ibus_engine import is_ibus_active_input_method
|
|
|
|
result = is_ibus_active_input_method()
|
|
self.assertFalse(result)
|
|
|
|
def test_not_active_when_other_im_configured(self):
|
|
"""Test returns False when another IM is explicitly configured, even if daemon runs."""
|
|
with patch.dict(os.environ, {"GTK_IM_MODULE": "fcitx"}, clear=True):
|
|
from vocalinux.text_injection.ibus_engine import is_ibus_active_input_method
|
|
|
|
result = is_ibus_active_input_method()
|
|
self.assertFalse(result)
|
|
|
|
@patch("vocalinux.text_injection.ibus_engine.is_ibus_daemon_running", return_value=False)
|
|
def test_empty_string_env_vars_return_false(self, mock_daemon):
|
|
"""Test returns False when env vars are empty strings and daemon not running."""
|
|
with patch.dict(
|
|
os.environ,
|
|
{"GTK_IM_MODULE": "", "QT_IM_MODULE": "", "XMODIFIERS": ""},
|
|
clear=True,
|
|
):
|
|
from vocalinux.text_injection.ibus_engine import is_ibus_active_input_method
|
|
|
|
result = is_ibus_active_input_method()
|
|
self.assertFalse(result)
|
|
|
|
def test_gtk_im_module_case_insensitive(self):
|
|
"""Test detection is case-insensitive for GTK_IM_MODULE."""
|
|
with patch.dict(os.environ, {"GTK_IM_MODULE": "IBUS"}, clear=True):
|
|
from vocalinux.text_injection.ibus_engine import is_ibus_active_input_method
|
|
|
|
result = is_ibus_active_input_method()
|
|
self.assertTrue(result)
|
|
|
|
def test_qt_im_module_case_insensitive(self):
|
|
"""Test detection is case-insensitive for QT_IM_MODULE."""
|
|
with patch.dict(os.environ, {"QT_IM_MODULE": "IBUS"}, clear=True):
|
|
from vocalinux.text_injection.ibus_engine import is_ibus_active_input_method
|
|
|
|
result = is_ibus_active_input_method()
|
|
self.assertTrue(result)
|
|
|
|
def test_xmodifiers_case_insensitive(self):
|
|
"""Test detection is case-insensitive for XMODIFIERS."""
|
|
with patch.dict(os.environ, {"XMODIFIERS": "@IM=IBUS"}, clear=True):
|
|
from vocalinux.text_injection.ibus_engine import is_ibus_active_input_method
|
|
|
|
result = is_ibus_active_input_method()
|
|
self.assertTrue(result)
|
|
|
|
def test_priority_order_gtk_over_qt(self):
|
|
"""Test GTK_IM_MODULE takes priority over QT_IM_MODULE."""
|
|
with patch.dict(
|
|
os.environ,
|
|
{"GTK_IM_MODULE": "ibus", "QT_IM_MODULE": "fcitx"},
|
|
clear=True,
|
|
):
|
|
from vocalinux.text_injection.ibus_engine import is_ibus_active_input_method
|
|
|
|
result = is_ibus_active_input_method()
|
|
self.assertTrue(result)
|
|
|
|
def test_priority_order_qt_over_xmodifiers(self):
|
|
"""Test QT_IM_MODULE takes priority over XMODIFIERS."""
|
|
with patch.dict(
|
|
os.environ,
|
|
{"QT_IM_MODULE": "ibus", "XMODIFIERS": "@im=fcitx"},
|
|
clear=True,
|
|
):
|
|
from vocalinux.text_injection.ibus_engine import is_ibus_active_input_method
|
|
|
|
result = is_ibus_active_input_method()
|
|
self.assertTrue(result)
|
|
|
|
|
|
class TestTextInjectorWithIbusActiveInputMethod(unittest.TestCase):
|
|
"""Tests for TextInjector with is_ibus_active_input_method check."""
|
|
|
|
def setUp(self):
|
|
"""Set up test fixtures."""
|
|
self.patch_which = patch("shutil.which")
|
|
self.mock_which = self.patch_which.start()
|
|
|
|
self.patch_subprocess = patch("subprocess.run")
|
|
self.mock_subprocess = self.patch_subprocess.start()
|
|
|
|
# Default to having ydotool available
|
|
self.mock_which.side_effect = lambda cmd: ("/usr/bin/ydotool" if cmd == "ydotool" else None)
|
|
|
|
mock_process = MagicMock()
|
|
mock_process.returncode = 0
|
|
mock_process.stdout = ""
|
|
mock_process.stderr = ""
|
|
self.mock_subprocess.return_value = mock_process
|
|
|
|
def tearDown(self):
|
|
"""Clean up after tests."""
|
|
self.patch_which.stop()
|
|
self.patch_subprocess.stop()
|
|
|
|
@patch("vocalinux.text_injection.text_injector.is_ibus_available")
|
|
@patch("vocalinux.text_injection.text_injector.is_ibus_active_input_method")
|
|
@patch("vocalinux.text_injection.text_injector.is_ibus_daemon_running")
|
|
def test_skips_ibus_when_not_active_input_method(
|
|
self, mock_daemon_running, mock_is_active_im, mock_is_available
|
|
):
|
|
"""Test that IBus is skipped when not the active input method."""
|
|
mock_is_available.return_value = True
|
|
mock_is_active_im.return_value = False # IBus installed but not active
|
|
mock_daemon_running.return_value = True
|
|
|
|
with patch.dict("os.environ", {"XDG_SESSION_TYPE": "wayland"}):
|
|
from vocalinux.text_injection.text_injector import DesktopEnvironment, TextInjector
|
|
|
|
injector = TextInjector()
|
|
|
|
# Should fall back to WAYLAND with ydotool, not WAYLAND_IBUS
|
|
self.assertEqual(injector.environment, DesktopEnvironment.WAYLAND)
|
|
self.assertEqual(injector.wayland_tool, "ydotool")
|
|
|
|
@patch("vocalinux.text_injection.text_injector.is_ibus_available")
|
|
@patch("vocalinux.text_injection.text_injector.is_ibus_active_input_method")
|
|
@patch("vocalinux.text_injection.text_injector.is_ibus_daemon_running")
|
|
@patch("vocalinux.text_injection.text_injector.IBusTextInjector")
|
|
def test_uses_ibus_when_active_input_method(
|
|
self, mock_injector_class, mock_daemon_running, mock_is_active_im, mock_is_available
|
|
):
|
|
"""Test that IBus is used when it is the active input method."""
|
|
mock_is_available.return_value = True
|
|
mock_is_active_im.return_value = True # IBus is the active input method
|
|
mock_daemon_running.return_value = True
|
|
|
|
mock_injector_instance = MagicMock()
|
|
mock_injector_class.return_value = mock_injector_instance
|
|
|
|
with patch.dict("os.environ", {"XDG_SESSION_TYPE": "wayland"}):
|
|
from vocalinux.text_injection.text_injector import DesktopEnvironment, TextInjector
|
|
|
|
with patch(
|
|
"vocalinux.text_injection.text_injector.IBusTextInjector",
|
|
mock_injector_class,
|
|
):
|
|
injector = TextInjector()
|
|
|
|
# Should use IBus
|
|
self.assertEqual(injector.environment, DesktopEnvironment.WAYLAND_IBUS)
|
|
# Should use IBus
|
|
self.assertEqual(injector.environment, DesktopEnvironment.WAYLAND_IBUS)
|
|
|
|
@patch("vocalinux.text_injection.text_injector.shutil.which")
|
|
@patch("vocalinux.text_injection.text_injector.is_ibus_available")
|
|
@patch("vocalinux.text_injection.text_injector.is_ibus_active_input_method")
|
|
@patch("vocalinux.text_injection.text_injector.is_ibus_daemon_running")
|
|
def test_x11_falls_back_when_ibus_not_active_input_method(
|
|
self, mock_daemon_running, mock_is_active_im, mock_is_available, mock_which
|
|
):
|
|
"""Test that X11 falls back to xdotool when IBus is not the active input method."""
|
|
mock_is_available.return_value = True
|
|
mock_is_active_im.return_value = False # IBus installed but not active
|
|
mock_daemon_running.return_value = True
|
|
mock_which.return_value = "/usr/bin/xdotool" # xdotool is available
|
|
|
|
with patch.dict("os.environ", {"XDG_SESSION_TYPE": "x11"}):
|
|
from vocalinux.text_injection.text_injector import DesktopEnvironment, TextInjector
|
|
|
|
injector = TextInjector()
|
|
|
|
# Should fall back to X11 (xdotool), not X11_IBUS
|
|
self.assertEqual(injector.environment, DesktopEnvironment.X11)
|
|
|
|
@patch("vocalinux.text_injection.text_injector.is_ibus_available")
|
|
@patch("vocalinux.text_injection.text_injector.is_ibus_active_input_method")
|
|
@patch("vocalinux.text_injection.text_injector.is_ibus_daemon_running")
|
|
@patch("vocalinux.text_injection.text_injector.IBusTextInjector")
|
|
def test_ibus_setup_exception_falls_back(
|
|
self, mock_injector_class, mock_daemon_running, mock_is_active_im, mock_is_available
|
|
):
|
|
"""Test that exceptions during IBus setup fall back to alternative methods."""
|
|
mock_is_available.return_value = True
|
|
mock_is_active_im.return_value = True # IBus is the active input method
|
|
mock_daemon_running.return_value = True
|
|
mock_injector_class.side_effect = Exception("IBus setup failed")
|
|
|
|
with patch.dict("os.environ", {"XDG_SESSION_TYPE": "wayland"}):
|
|
from vocalinux.text_injection.text_injector import DesktopEnvironment, TextInjector
|
|
|
|
injector = TextInjector()
|
|
|
|
# Should fall back to WAYLAND with ydotool after IBus setup fails
|
|
self.assertEqual(injector.environment, DesktopEnvironment.WAYLAND)
|
|
self.assertEqual(injector.wayland_tool, "ydotool")
|
|
|
|
|
|
class TestWaylandXkbLayoutSkipping(unittest.TestCase):
|
|
"""Tests for skipping setxkbmap operations on Wayland sessions."""
|
|
|
|
def test_get_current_xkb_layout_returns_empty_on_wayland(self):
|
|
"""Test that get_current_xkb_layout returns empty tuple on Wayland."""
|
|
with patch.dict(os.environ, {"XDG_SESSION_TYPE": "wayland"}, clear=False):
|
|
from vocalinux.text_injection.ibus_engine import get_current_xkb_layout
|
|
|
|
layout, variant, option = get_current_xkb_layout()
|
|
self.assertEqual(layout, "")
|
|
self.assertEqual(variant, "")
|
|
self.assertEqual(option, "")
|
|
|
|
@patch("subprocess.run")
|
|
def test_get_current_xkb_layout_queries_setxkbmap_on_x11(self, mock_run):
|
|
"""Test that get_current_xkb_layout uses setxkbmap on X11."""
|
|
mock_run.return_value = MagicMock(
|
|
returncode=0,
|
|
stdout="layout: de\nvariant: nodeadkeys\noptions: ctrl:nocaps\n",
|
|
)
|
|
with patch.dict(os.environ, {"XDG_SESSION_TYPE": "x11"}, clear=False):
|
|
from vocalinux.text_injection.ibus_engine import get_current_xkb_layout
|
|
|
|
layout, variant, option = get_current_xkb_layout()
|
|
self.assertEqual(layout, "de")
|
|
self.assertEqual(variant, "nodeadkeys")
|
|
self.assertEqual(option, "ctrl:nocaps")
|
|
mock_run.assert_called_once()
|
|
|
|
@patch("subprocess.run")
|
|
def test_get_current_xkb_layout_skips_subprocess_on_wayland(self, mock_run):
|
|
"""Test that setxkbmap is never called on Wayland."""
|
|
with patch.dict(os.environ, {"XDG_SESSION_TYPE": "wayland"}, clear=False):
|
|
from vocalinux.text_injection.ibus_engine import get_current_xkb_layout
|
|
|
|
get_current_xkb_layout()
|
|
mock_run.assert_not_called()
|
|
|
|
def test_restore_xkb_layout_skips_empty_layout(self):
|
|
"""Test that restore_xkb_layout returns False for empty layout string."""
|
|
from vocalinux.text_injection.ibus_engine import restore_xkb_layout
|
|
|
|
result = restore_xkb_layout("", "", "")
|
|
self.assertFalse(result)
|
|
|
|
def test_is_wayland_session_detects_wayland(self):
|
|
"""Test _is_wayland_session returns True for Wayland."""
|
|
with patch.dict(os.environ, {"XDG_SESSION_TYPE": "wayland"}, clear=False):
|
|
from vocalinux.text_injection.ibus_engine import _is_wayland_session
|
|
|
|
self.assertTrue(_is_wayland_session())
|
|
|
|
def test_is_wayland_session_detects_x11(self):
|
|
"""Test _is_wayland_session returns False for X11."""
|
|
with patch.dict(os.environ, {"XDG_SESSION_TYPE": "x11"}, clear=False):
|
|
from vocalinux.text_injection.ibus_engine import _is_wayland_session
|
|
|
|
self.assertFalse(_is_wayland_session())
|
|
|
|
def test_is_wayland_session_returns_false_when_unset(self):
|
|
"""Test _is_wayland_session returns False when env var not set."""
|
|
env = os.environ.copy()
|
|
env.pop("XDG_SESSION_TYPE", None)
|
|
with patch.dict(os.environ, env, clear=True):
|
|
from vocalinux.text_injection.ibus_engine import _is_wayland_session
|
|
|
|
self.assertFalse(_is_wayland_session())
|
|
|
|
|
|
class TestVocalinuxEngineDestroy(unittest.TestCase):
|
|
"""Tests for VocalinuxEngine.do_destroy (layout switch resilience)."""
|
|
|
|
def test_do_destroy_clears_active_instance(self):
|
|
"""Test destroy handler clears active instance when called on active engine."""
|
|
from vocalinux.text_injection.ibus_engine import _handle_engine_destroy
|
|
|
|
engine = object()
|
|
next_active = _handle_engine_destroy(
|
|
active_instance=engine,
|
|
current_instance=engine,
|
|
ibus_available=False,
|
|
)
|
|
|
|
self.assertIsNone(next_active)
|
|
|
|
def test_do_destroy_ignores_different_instance(self):
|
|
"""Test destroy handler keeps active instance for different engine."""
|
|
from vocalinux.text_injection.ibus_engine import _handle_engine_destroy
|
|
|
|
active_engine = object()
|
|
old_engine = object()
|
|
|
|
next_active = _handle_engine_destroy(
|
|
active_instance=active_engine,
|
|
current_instance=old_engine,
|
|
ibus_available=False,
|
|
)
|
|
|
|
self.assertIs(next_active, active_engine)
|
|
|
|
def test_do_destroy_calls_super_when_ibus_available(self):
|
|
"""Test destroy handler calls parent destroy callback when IBus available."""
|
|
from vocalinux.text_injection.ibus_engine import _handle_engine_destroy
|
|
|
|
active_engine = object()
|
|
mock_super_destroy = MagicMock()
|
|
|
|
_handle_engine_destroy(
|
|
active_instance=active_engine,
|
|
current_instance=object(),
|
|
ibus_available=True,
|
|
super_destroy=mock_super_destroy,
|
|
)
|
|
|
|
mock_super_destroy.assert_called_once()
|
|
|
|
def test_do_destroy_skips_super_when_ibus_unavailable(self):
|
|
"""Test destroy handler skips parent destroy callback when IBus unavailable."""
|
|
from vocalinux.text_injection.ibus_engine import _handle_engine_destroy
|
|
|
|
mock_super_destroy = MagicMock()
|
|
|
|
_handle_engine_destroy(
|
|
active_instance=object(),
|
|
current_instance=object(),
|
|
ibus_available=False,
|
|
super_destroy=mock_super_destroy,
|
|
)
|
|
|
|
mock_super_destroy.assert_not_called()
|
|
|
|
def test_do_destroy_handles_missing_super_callback(self):
|
|
"""Test destroy handler works when no parent destroy callback is provided."""
|
|
from vocalinux.text_injection.ibus_engine import _handle_engine_destroy
|
|
|
|
active_engine = object()
|
|
next_active = _handle_engine_destroy(
|
|
active_instance=active_engine,
|
|
current_instance=active_engine,
|
|
ibus_available=True,
|
|
super_destroy=None,
|
|
)
|
|
|
|
self.assertIsNone(next_active)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|