fix(ibus): start engine process before checking registration (fixes #360) (#361)

* fix(ibus): start engine process before checking registration

_setup_engine() gated on is_engine_registered() which queries
`ibus list-engine`. On systems where the IBus daemon doesn't scan
user-level component directories (~/.local/share/ibus/component/),
the engine never appeared in the list, blocking start_engine_process()
from ever being called — even though its register_component() D-Bus
call is the correct registration path for standalone-launched engines.

Reorder _setup_engine() to start the engine process first, relying on
register_component() for runtime registration. Remove the XML
install step from the setup path since install_ibus_component()
triggers `ibus restart` which kills the engine process.

Follow-up to PR #304 which added register_component() support.

* fix(ibus): remove dead code and add engine readiness check

Address review feedback:
- Remove install_ibus_component(), is_engine_registered(),
  is_component_up_to_date(), _get_expected_component_xml()
  and their tests — no longer called after setup order fix
- Add SOCKET_PATH poll after start_engine_process() to verify
  engine is fully initialized before activation
- Rebase onto current main
This commit is contained in:
Sbenazar
2026-04-05 17:31:35 -07:00
committed by GitHub
parent 04c8645e2a
commit c4f5b797ce
5 changed files with 35 additions and 465 deletions
-2
View File
@@ -6,9 +6,7 @@ from . import text_injector # noqa: F401
from .ibus_engine import ( # noqa: F401
IBusSetupError,
IBusTextInjector,
install_ibus_component,
is_engine_active,
is_engine_registered,
is_ibus_active_input_method,
is_ibus_available,
start_ibus_daemon,
+23 -162
View File
@@ -259,76 +259,6 @@ def _get_exec_command() -> str:
return f"{sys.executable} {engine_script} --ibus"
def _get_expected_component_xml() -> str:
"""Generate the expected component XML content for the current installation."""
e = _ENGINE_META
c = _COMPONENT_META
return f"""<?xml version="1.0" encoding="utf-8"?>
<component>
<name>{COMPONENT_NAME}</name>
<description>{ENGINE_DESCRIPTION}</description>
<exec>{_get_exec_command()}</exec>
<version>{c['version']}</version>
<author>{c['author']}</author>
<license>{c['license']}</license>
<homepage>{c['homepage']}</homepage>
<textdomain>{c['textdomain']}</textdomain>
<engines>
<engine>
<name>{ENGINE_NAME}</name>
<longname>{ENGINE_LONGNAME}</longname>
<language>{e['language']}</language>
<license>{e['license']}</license>
<author>{e['author']}</author>
<icon>{e['icon']}</icon>
<layout>{e['layout']}</layout>
<description>{ENGINE_DESCRIPTION}</description>
<rank>{ENGINE_RANK}</rank>
</engine>
</engines>
</component>
"""
def is_component_up_to_date() -> bool:
"""
Check if the installed IBus component XML matches the current installation.
Returns:
True if component exists and matches expected content, False otherwise
"""
component_file = Path.home() / ".local" / "share" / "ibus" / "component" / "vocalinux.xml"
if not component_file.exists():
return False
try:
installed_content = component_file.read_text()
expected_content = _get_expected_component_xml()
return installed_content.strip() == expected_content.strip()
except Exception as e:
logger.debug(f"Failed to check component XML: {e}")
return False
def is_engine_registered() -> bool:
"""Check if the Vocalinux IBus engine is registered."""
if not IBUS_AVAILABLE:
return False
try:
result = subprocess.run(
["ibus", "list-engine"],
capture_output=True,
text=True,
timeout=5,
)
return ENGINE_NAME in result.stdout
except (subprocess.SubprocessError, FileNotFoundError):
return False
def is_engine_active() -> bool:
"""Check if the Vocalinux IBus engine is currently active."""
try:
@@ -837,11 +767,12 @@ class IBusTextInjector:
This class connects to the Vocalinux IBus engine via Unix socket
and sends text to be injected. On initialization, it automatically:
1. Installs the IBus component if not registered
2. Saves the current engine
1. Starts the engine process (registers via D-Bus register_component)
2. Saves the current engine and XKB layout
3. Switches to the Vocalinux engine
4. Restores the XKB layout
On cleanup (stop), it restores the previous engine.
On cleanup (stop), it restores the previous engine and layout.
"""
def __init__(self, auto_activate: bool = True):
@@ -863,27 +794,28 @@ class IBusTextInjector:
def _setup_engine(self) -> None:
"""Install and activate the IBus engine."""
# Install or update component if needed
if not is_engine_registered():
logger.info("IBus engine not registered, installing...")
if not install_ibus_component(system_wide=False):
logger.info("User-level install failed, trying system-wide...")
install_ibus_component(system_wide=True)
elif not is_component_up_to_date():
logger.info("IBus component XML is outdated, updating...")
install_ibus_component(system_wide=False)
# Check again after installation
if not is_engine_registered():
raise IBusSetupError(
"Failed to register IBus engine. "
"Try running 'ibus list-engine' to verify installation."
)
# Start the engine process (Vocalinux starts it, not IBus)
# Start the engine process — it calls register_component() via
# D-Bus which makes the engine available even when the IBus daemon
# doesn't scan ~/.local/share/ibus/component/.
# Follow-up to PR #304: register_component() is the reliable path,
# so we no longer gate on is_engine_registered() / ibus list-engine.
if not start_engine_process():
raise IBusSetupError("Failed to start IBus engine process. Check logs for details.")
# Verify the engine is fully ready before proceeding.
# start_engine_process() only confirms the subprocess is alive —
# register_component() and socket setup may still be in progress.
for _attempt in range(15):
if SOCKET_PATH.exists():
logger.debug("Engine socket is ready")
break
time.sleep(0.2)
else:
logger.warning(
"Engine process started but socket not ready after retries; "
"proceeding with activation attempt"
)
# Capture current XKB layout before switching engines
# This is critical for preserving the user's keyboard layout
# when IBus engine switching might override it
@@ -995,77 +927,6 @@ class IBusTextInjector:
return False
def install_ibus_component(system_wide: bool = False) -> bool:
"""
Install the IBus component XML file.
Args:
system_wide: If True, install to /usr/share/ibus/component/ (requires root).
If False, install to user directory and set IBUS_COMPONENT_PATH.
Returns:
True if installation was successful, False otherwise
"""
component_xml = _get_expected_component_xml()
try:
if system_wide:
component_dir = Path("/usr/share/ibus/component")
component_file = component_dir / "vocalinux.xml"
# Write to temp file and move with sudo
import tempfile
with tempfile.NamedTemporaryFile(mode="w", suffix=".xml", delete=False) as f:
f.write(component_xml)
temp_path = f.name
result = subprocess.run(
["sudo", "cp", temp_path, str(component_file)],
capture_output=True,
text=True,
)
os.unlink(temp_path)
if result.returncode != 0:
logger.error(f"Failed to install component: {result.stderr}")
return False
subprocess.run(
["sudo", "chmod", "644", str(component_file)],
capture_output=True,
)
else:
# User-level installation
component_dir = Path.home() / ".local" / "share" / "ibus" / "component"
component_dir.mkdir(parents=True, exist_ok=True)
component_file = component_dir / "vocalinux.xml"
component_file.write_text(component_xml)
logger.info(f"Component installed to {component_file}")
# Refresh IBus - restart with component path set so it picks up user components
import time
user_component_dir = Path.home() / ".local" / "share" / "ibus" / "component"
env = os.environ.copy()
env["IBUS_COMPONENT_PATH"] = f"{user_component_dir}:/usr/share/ibus/component"
subprocess.run(["ibus", "write-cache"], capture_output=True, env=env)
subprocess.run(["ibus", "restart"], capture_output=True, env=env)
time.sleep(1) # Give IBus time to restart
logger.info(
"IBus component installed successfully.\n"
"Set 'Vocalinux' as your input method in system settings "
"to enable voice dictation."
)
return True
except Exception as e:
logger.error(f"Failed to install IBus component: {e}")
return False
def _get_engines_xml() -> str:
"""Return engine XML for IBus --xml discovery.
+10 -186
View File
@@ -54,31 +54,9 @@ class TestIBusTextInjectorSetupFailures(unittest.TestCase):
@patch("vocalinux.text_injection.ibus_engine.IBUS_AVAILABLE", True)
@patch("vocalinux.text_injection.ibus_engine.ensure_ibus_dir")
@patch("vocalinux.text_injection.ibus_engine.install_ibus_component")
@patch("vocalinux.text_injection.ibus_engine.is_engine_registered")
def test_raises_on_registration_failure(
self, mock_is_registered, mock_install, mock_ensure_dir
):
"""Test that IBusSetupError is raised when engine registration fails."""
mock_is_registered.return_value = False # Registration fails
mock_install.return_value = True
from vocalinux.text_injection.ibus_engine import IBusSetupError, IBusTextInjector
with self.assertRaises(IBusSetupError) as context:
IBusTextInjector(auto_activate=True)
self.assertIn("Failed to register 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.is_engine_registered")
@patch("vocalinux.text_injection.ibus_engine.start_engine_process")
def test_raises_on_engine_process_failure(
self, mock_start_engine, mock_is_registered, mock_ensure_dir
):
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_is_registered.return_value = True
mock_start_engine.return_value = False # Process start fails
from vocalinux.text_injection.ibus_engine import IBusSetupError, IBusTextInjector
@@ -90,7 +68,7 @@ class TestIBusTextInjectorSetupFailures(unittest.TestCase):
@patch("vocalinux.text_injection.ibus_engine.IBUS_AVAILABLE", True)
@patch("vocalinux.text_injection.ibus_engine.ensure_ibus_dir")
@patch("vocalinux.text_injection.ibus_engine.is_engine_registered")
@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")
@@ -101,12 +79,12 @@ class TestIBusTextInjectorSetupFailures(unittest.TestCase):
mock_get_current,
mock_is_active,
mock_start_engine,
mock_is_registered,
mock_socket_path,
mock_ensure_dir,
):
"""Test that IBusSetupError is raised when engine activation fails."""
mock_is_registered.return_value = True
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
@@ -138,102 +116,6 @@ class TestIBusEngineHelpers(unittest.TestCase):
self.assertEqual(is_ibus_available(), IBUS_AVAILABLE)
class TestIsEngineRegistered(unittest.TestCase):
"""Tests for is_engine_registered function."""
@patch("vocalinux.text_injection.ibus_engine.IBUS_AVAILABLE", True)
@patch("subprocess.run")
def test_engine_registered(self, mock_run):
"""Test detection when engine is registered."""
mock_run.return_value = MagicMock(stdout="vocalinux\nxkb:us::eng\n", returncode=0)
from vocalinux.text_injection.ibus_engine import is_engine_registered
result = is_engine_registered()
self.assertTrue(result)
mock_run.assert_called_once()
@patch("vocalinux.text_injection.ibus_engine.IBUS_AVAILABLE", True)
@patch("subprocess.run")
def test_engine_not_registered(self, mock_run):
"""Test detection when engine is not registered."""
mock_run.return_value = MagicMock(stdout="xkb:us::eng\n", returncode=0)
from vocalinux.text_injection.ibus_engine import is_engine_registered
result = is_engine_registered()
self.assertFalse(result)
@patch("vocalinux.text_injection.ibus_engine.IBUS_AVAILABLE", True)
@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_registered
result = is_engine_registered()
self.assertFalse(result)
@patch("vocalinux.text_injection.ibus_engine.IBUS_AVAILABLE", False)
def test_ibus_not_available(self):
"""Test returns False when IBus is not available."""
from vocalinux.text_injection.ibus_engine import is_engine_registered
result = is_engine_registered()
self.assertFalse(result)
class TestIsComponentUpToDate(unittest.TestCase):
"""Tests for is_component_up_to_date function."""
def test_component_missing(self):
"""Test returns False when component file doesn't exist."""
with patch("pathlib.Path.home") as mock_home:
mock_home.return_value = Path("/nonexistent")
from vocalinux.text_injection.ibus_engine import is_component_up_to_date
result = is_component_up_to_date()
self.assertFalse(result)
def test_component_matches(self):
"""Test returns True when component content matches."""
with tempfile.TemporaryDirectory() as tmpdir:
component_dir = Path(tmpdir) / ".local" / "share" / "ibus" / "component"
component_dir.mkdir(parents=True)
component_file = component_dir / "vocalinux.xml"
with patch("pathlib.Path.home", return_value=Path(tmpdir)):
from vocalinux.text_injection.ibus_engine import (
_get_expected_component_xml,
is_component_up_to_date,
)
# Write expected content
component_file.write_text(_get_expected_component_xml())
result = is_component_up_to_date()
self.assertTrue(result)
def test_component_differs(self):
"""Test returns False when component content differs."""
with tempfile.TemporaryDirectory() as tmpdir:
component_dir = Path(tmpdir) / ".local" / "share" / "ibus" / "component"
component_dir.mkdir(parents=True)
component_file = component_dir / "vocalinux.xml"
with patch("pathlib.Path.home", return_value=Path(tmpdir)):
from vocalinux.text_injection.ibus_engine import is_component_up_to_date
# Write stale content
component_file.write_text("<component>stale</component>")
result = is_component_up_to_date()
self.assertFalse(result)
class TestIsEngineActive(unittest.TestCase):
"""Tests for is_engine_active function."""
@@ -368,8 +250,7 @@ class TestIBusTextInjector(unittest.TestCase):
@patch("vocalinux.text_injection.ibus_engine.IBUS_AVAILABLE", True)
@patch("vocalinux.text_injection.ibus_engine.ensure_ibus_dir")
@patch("vocalinux.text_injection.ibus_engine.install_ibus_component")
@patch("vocalinux.text_injection.ibus_engine.is_engine_registered")
@patch("vocalinux.text_injection.ibus_engine.SOCKET_PATH")
@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")
@@ -380,12 +261,11 @@ class TestIBusTextInjector(unittest.TestCase):
mock_get_current,
mock_start_engine,
mock_is_active,
mock_is_registered,
mock_install,
mock_socket_path,
mock_ensure_dir,
):
"""Test initialization with auto_activate=True."""
mock_is_registered.return_value = 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"
@@ -487,27 +367,28 @@ class TestIBusTextInjector(unittest.TestCase):
@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.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.is_engine_registered", return_value=True)
@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_registered,
mock_active,
mock_get_engine,
mock_switch,
mock_start_proc,
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", "")
injector = IBusTextInjector(auto_activate=False)
@@ -621,63 +502,6 @@ class TestIBusTextInjector(unittest.TestCase):
self.assertFalse(result)
class TestInstallIBusComponent(unittest.TestCase):
"""Tests for install_ibus_component function."""
@patch("subprocess.run")
def test_user_install_success(self, mock_run):
"""Test successful user-level installation."""
mock_run.return_value = MagicMock(returncode=0)
with tempfile.TemporaryDirectory() as temp_dir:
component_dir = Path(temp_dir)
with patch(
"vocalinux.text_injection.ibus_engine.Path.home",
return_value=Path(temp_dir),
):
from vocalinux.text_injection.ibus_engine import install_ibus_component
# Force reload to use patched home
result = install_ibus_component(system_wide=False)
self.assertTrue(result)
@patch("subprocess.run")
def test_system_install_needs_sudo(self, mock_run):
"""Test system-wide installation calls sudo."""
mock_run.return_value = MagicMock(returncode=0)
from vocalinux.text_injection.ibus_engine import install_ibus_component
with patch("tempfile.NamedTemporaryFile"):
with patch("os.unlink"):
result = install_ibus_component(system_wide=True)
# Verify sudo was called
sudo_calls = [
call for call in mock_run.call_args_list if call[0][0] and "sudo" in call[0][0]
]
self.assertTrue(len(sudo_calls) > 0)
@patch("subprocess.run")
def test_system_install_failure(self, mock_run):
"""Test system-wide installation failure."""
mock_run.return_value = MagicMock(returncode=1, stderr="Permission denied")
from vocalinux.text_injection.ibus_engine import install_ibus_component
with patch("tempfile.NamedTemporaryFile") as mock_tmp:
mock_tmp.return_value.__enter__ = MagicMock()
mock_tmp.return_value.__exit__ = MagicMock()
mock_tmp.return_value.name = "/tmp/test.xml"
mock_tmp.return_value.write = MagicMock()
with patch("os.unlink"):
result = install_ibus_component(system_wide=True)
self.assertFalse(result)
class TestTextInjectorWithIBus(unittest.TestCase):
"""Tests for TextInjector integration with IBus."""
+2 -21
View File
@@ -138,24 +138,6 @@ class TestIBusEngineModuleFunctions(unittest.TestCase):
result = is_ibus_active_input_method()
self.assertFalse(result)
@patch("subprocess.run")
def test_is_engine_registered_success(self, mock_run):
"""Test engine registration detection."""
mock_run.return_value = MagicMock(returncode=0, stdout="vocalinux other-engine")
from vocalinux.text_injection.ibus_engine import is_engine_registered
result = is_engine_registered()
self.assertTrue(result)
@patch("subprocess.run")
def test_is_engine_registered_not_found(self, mock_run):
"""Test when engine is not registered."""
mock_run.return_value = MagicMock(returncode=0, stdout="other-engine")
from vocalinux.text_injection.ibus_engine import is_engine_registered
result = is_engine_registered()
self.assertFalse(result)
@patch("subprocess.run")
def test_is_engine_active_success(self, mock_run):
"""Test when engine is active."""
@@ -498,12 +480,11 @@ class TestIBusTextInjector(unittest.TestCase):
self.assertIsNone(injector._previous_engine)
@patch("vocalinux.text_injection.ibus_engine.ensure_ibus_dir")
@patch("vocalinux.text_injection.ibus_engine.is_engine_registered", return_value=True)
@patch("vocalinux.text_injection.ibus_engine.is_component_up_to_date", return_value=True)
@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_active", return_value=True)
def test_ibus_text_injector_init_with_auto_activate(
self, mock_active, mock_start, mock_update, mock_registered, mock_ensure_dir
self, mock_active, mock_start, mock_socket_path, mock_ensure_dir
):
"""Test IBusTextInjector initialization with auto-activation."""
from vocalinux.text_injection.ibus_engine import IBusTextInjector
-94
View File
@@ -180,100 +180,6 @@ class TestStartIBusDaemon(unittest.TestCase):
self.assertFalse(result)
class TestGetExpectedComponentXml(unittest.TestCase):
"""Tests for _get_expected_component_xml function."""
def test_get_expected_component_xml_returns_string(self):
"""Test that function returns valid XML string."""
from vocalinux.text_injection.ibus_engine import _get_expected_component_xml
result = _get_expected_component_xml()
self.assertIsInstance(result, str)
self.assertIn("<?xml", result)
self.assertIn("</component>", result)
def test_get_expected_component_xml_contains_vocalinux(self):
"""Test that XML contains Vocalinux references."""
from vocalinux.text_injection.ibus_engine import _get_expected_component_xml
result = _get_expected_component_xml()
self.assertIn("vocalinux", result.lower())
class TestIsComponentUpToDate(unittest.TestCase):
"""Tests for is_component_up_to_date function."""
@patch("pathlib.Path.exists")
def test_is_component_up_to_date_not_exists(self, mock_exists):
"""Test when component file doesn't exist."""
from vocalinux.text_injection.ibus_engine import is_component_up_to_date
mock_exists.return_value = False
result = is_component_up_to_date()
self.assertFalse(result)
@patch("vocalinux.text_injection.ibus_engine._get_expected_component_xml")
@patch("pathlib.Path.read_text")
@patch("pathlib.Path.exists")
def test_is_component_up_to_date_matches(self, mock_exists, mock_read, mock_get_xml):
"""Test when XML matches."""
from vocalinux.text_injection.ibus_engine import is_component_up_to_date
xml_content = "<?xml><component></component>"
mock_exists.return_value = True
mock_read.return_value = xml_content
mock_get_xml.return_value = xml_content
result = is_component_up_to_date()
self.assertTrue(result)
@patch("vocalinux.text_injection.ibus_engine._get_expected_component_xml")
@patch("pathlib.Path.read_text")
@patch("pathlib.Path.exists")
def test_is_component_up_to_date_mismatch(self, mock_exists, mock_read, mock_get_xml):
"""Test when XML differs."""
from vocalinux.text_injection.ibus_engine import is_component_up_to_date
mock_exists.return_value = True
mock_read.return_value = "<?xml><old></old>"
mock_get_xml.return_value = "<?xml><new></new>"
result = is_component_up_to_date()
self.assertFalse(result)
class TestIsEngineRegistered(unittest.TestCase):
"""Tests for is_engine_registered function."""
@patch("vocalinux.text_injection.ibus_engine.IBUS_AVAILABLE", False)
def test_is_engine_registered_not_available(self):
"""Test when IBus is not available."""
from vocalinux.text_injection.ibus_engine import is_engine_registered
result = is_engine_registered()
self.assertFalse(result)
@patch("subprocess.run")
@patch("vocalinux.text_injection.ibus_engine.IBUS_AVAILABLE", True)
def test_is_engine_registered_true(self, mock_run):
"""Test when engine is registered."""
from vocalinux.text_injection.ibus_engine import ENGINE_NAME, is_engine_registered
mock_run.return_value.stdout = f"other\n{ENGINE_NAME}\nmore"
result = is_engine_registered()
self.assertTrue(result)
@patch("subprocess.run")
@patch("vocalinux.text_injection.ibus_engine.IBUS_AVAILABLE", True)
def test_is_engine_registered_false(self, mock_run):
"""Test when engine is not registered."""
from vocalinux.text_injection.ibus_engine import is_engine_registered
mock_run.return_value.stdout = "other\nengines"
result = is_engine_registered()
self.assertFalse(result)
class TestIsEngineActive(unittest.TestCase):
"""Tests for is_engine_active function."""