diff --git a/install.sh b/install.sh
index cbdcf52..659a1ab 100755
--- a/install.sh
+++ b/install.sh
@@ -865,13 +865,13 @@ install_system_dependencies() {
fi
# Define package names for different distributions
- local APT_PACKAGES_UBUNTU="python3-pip python3-gi python3-gi-cairo gir1.2-gtk-3.0 gir1.2-appindicator3-0.1 libgirepository1.0-dev python3-dev build-essential portaudio19-dev python3-venv pkg-config wget curl unzip vulkan-tools libvulkan-dev $VULKAN_SHADER_PKG"
- local APT_PACKAGES_DEBIAN_BASE="python3-pip python3-gi python3-gi-cairo gir1.2-gtk-3.0 libcairo2-dev python3-dev build-essential portaudio19-dev python3-venv pkg-config wget curl unzip vulkan-tools libvulkan-dev $VULKAN_SHADER_PKG"
+ local APT_PACKAGES_UBUNTU="python3-pip python3-gi python3-gi-cairo gir1.2-gtk-3.0 gir1.2-appindicator3-0.1 gir1.2-ibus-1.0 libgirepository1.0-dev python3-dev build-essential portaudio19-dev python3-venv pkg-config wget curl unzip vulkan-tools libvulkan-dev $VULKAN_SHADER_PKG"
+ local APT_PACKAGES_DEBIAN_BASE="python3-pip python3-gi python3-gi-cairo gir1.2-gtk-3.0 gir1.2-ibus-1.0 libcairo2-dev python3-dev build-essential portaudio19-dev python3-venv pkg-config wget curl unzip vulkan-tools libvulkan-dev $VULKAN_SHADER_PKG"
local APT_PACKAGES_DEBIAN_11_12="$APT_PACKAGES_DEBIAN_BASE libgirepository1.0-dev gir1.2-ayatanaappindicator3-0.1"
local APT_PACKAGES_DEBIAN_13_PLUS="$APT_PACKAGES_DEBIAN_BASE libgirepository-2.0-dev gir1.2-ayatanaappindicator3-0.1"
- local DNF_PACKAGES="python3-pip python3-gobject gtk3 libappindicator-gtk3 gobject-introspection-devel python3-devel portaudio-devel python3-virtualenv pkg-config wget curl unzip vulkan-tools vulkan-loader-devel glslang"
- local PACMAN_PACKAGES="python-pip python-gobject gtk3 libappindicator gobject-introspection python-cairo portaudio python-virtualenv pkg-config wget curl unzip base-devel vulkan-tools vulkan-headers glslang"
- local ZYPPER_PACKAGES="python3-pip python3-gobject python3-gobject-cairo gtk3 libappindicator-gtk3 gobject-introspection-devel python3-devel portaudio-devel python3-virtualenv pkg-config wget curl unzip vulkan-tools vulkan-devel glslang"
+ local DNF_PACKAGES="python3-pip python3-gobject gtk3 libappindicator-gtk3 ibus-devel gobject-introspection-devel python3-devel portaudio-devel python3-virtualenv pkg-config wget curl unzip vulkan-tools vulkan-loader-devel glslang"
+ local PACMAN_PACKAGES="python-pip python-gobject gtk3 libappindicator-gtk3 ibus gobject-introspection python-cairo portaudio python-virtualenv pkg-config wget curl unzip base-devel vulkan-tools vulkan-headers glslang"
+ local ZYPPER_PACKAGES="python3-pip python3-gobject python3-gobject-cairo gtk3 libappindicator-gtk3 ibus-devel gobject-introspection-devel python3-devel portaudio-devel python3-virtualenv pkg-config wget curl unzip vulkan-tools vulkan-devel glslang"
# Gentoo uses Portage and different package naming convention
local EMERGE_PACKAGES="dev-python/pygobject:3 x11-libs/gtk+:3 dev-libs/libayatana-appindicator media-libs/portaudio dev-lang/python:3.8 pkgconf dev-util/glslang"
# Alpine Linux uses apk and has musl libc
diff --git a/src/vocalinux/text_injection/__init__.py b/src/vocalinux/text_injection/__init__.py
index 87fe298..87c8766 100644
--- a/src/vocalinux/text_injection/__init__.py
+++ b/src/vocalinux/text_injection/__init__.py
@@ -3,3 +3,10 @@ Text injection module for Vocalinux
"""
from . import text_injector # noqa: F401
+from .ibus_engine import ( # noqa: F401
+ IBusTextInjector,
+ install_ibus_component,
+ is_engine_active,
+ is_engine_registered,
+ is_ibus_available,
+)
diff --git a/src/vocalinux/text_injection/ibus_engine.py b/src/vocalinux/text_injection/ibus_engine.py
new file mode 100644
index 0000000..43d50d4
--- /dev/null
+++ b/src/vocalinux/text_injection/ibus_engine.py
@@ -0,0 +1,758 @@
+"""
+IBus engine for Vocalinux text injection.
+
+This module provides an IBus input method engine that acts as a transparent
+proxy - it passes all keystrokes through normally while allowing Vocalinux
+to inject text directly via commit_text().
+
+The engine should be set as the user's default input method. It will:
+1. Pass all keyboard input through unchanged
+2. Listen for text injection requests via a Unix socket or file
+3. Commit injected text directly without switching engines
+
+This is the preferred method for Wayland environments as it works universally
+without requiring compositor-specific protocols.
+"""
+
+import logging
+import os
+import signal
+import socket
+import struct
+import subprocess
+import sys
+import threading
+from pathlib import Path
+from typing import Optional
+
+logger = logging.getLogger(__name__)
+
+# Check if IBus is available
+try:
+ import gi
+
+ gi.require_version("IBus", "1.0")
+ from gi.repository import GLib, GObject, IBus
+
+ IBUS_AVAILABLE = True
+except (ImportError, ValueError) as e:
+ logger.debug(f"IBus not available: {e}")
+ IBUS_AVAILABLE = False
+ IBus = None
+ GLib = None
+ GObject = None
+
+
+# File paths for communication
+VOCALINUX_IBUS_DIR = Path.home() / ".local" / "share" / "vocalinux-ibus"
+SOCKET_PATH = VOCALINUX_IBUS_DIR / "inject.sock"
+PID_FILE = VOCALINUX_IBUS_DIR / "engine.pid"
+
+# Engine identification
+ENGINE_NAME = "vocalinux"
+ENGINE_LONGNAME = "Vocalinux"
+ENGINE_DESCRIPTION = "Vocalinux voice dictation (use as default input method)"
+COMPONENT_NAME = "org.freedesktop.IBus.Vocalinux"
+
+
+def ensure_ibus_dir() -> None:
+ """Ensure the IBus data directory exists with secure permissions."""
+ VOCALINUX_IBUS_DIR.mkdir(parents=True, exist_ok=True)
+ # Secure the directory - only owner can access (prevents socket hijacking)
+ VOCALINUX_IBUS_DIR.chmod(0o700)
+
+
+def verify_peer_credentials(conn: socket.socket) -> bool:
+ """
+ Verify that the connecting process belongs to the same user.
+
+ Uses SO_PEERCRED to get the UID of the peer process and compares
+ it to our own UID. This prevents other users from injecting text.
+
+ Args:
+ conn: The connected socket
+
+ Returns:
+ True if peer is same user, False otherwise
+ """
+ try:
+ # SO_PEERCRED returns a struct with pid, uid, gid
+ # struct ucred { pid_t pid; uid_t uid; gid_t gid; }
+ cred = conn.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED, struct.calcsize("iII"))
+ pid, uid, gid = struct.unpack("iII", cred)
+
+ my_uid = os.getuid()
+ if uid != my_uid:
+ logger.warning(f"Rejected connection from UID {uid} (expected {my_uid})")
+ return False
+
+ logger.debug(f"Accepted connection from PID {pid}, UID {uid}")
+ return True
+ except (OSError, struct.error) as e:
+ logger.error(f"Failed to verify peer credentials: {e}")
+ return False
+
+
+def is_ibus_available() -> bool:
+ """Check if IBus is available on the system."""
+ return IBUS_AVAILABLE
+
+
+def _get_expected_component_xml() -> str:
+ """Generate the expected component XML content for the current installation."""
+ engine_script = Path(__file__).resolve()
+ python_exec = sys.executable
+
+ return f"""
+
+ {COMPONENT_NAME}
+ {ENGINE_DESCRIPTION}
+ {python_exec} {engine_script}
+ 1.0
+ Vocalinux
+ GPL-3.0
+ https://github.com/jatinkrmalik/vocalinux
+ vocalinux
+
+
+ {ENGINE_NAME}
+ other
+ GPL-3.0
+ Vocalinux
+ audio-input-microphone
+ default
+ {ENGINE_LONGNAME}
+ {ENGINE_DESCRIPTION}
+ 50
+
+
+
+"""
+
+
+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:
+ import subprocess
+
+ result = subprocess.run(
+ ["ibus", "engine"],
+ capture_output=True,
+ text=True,
+ timeout=5,
+ )
+ return result.returncode == 0 and ENGINE_NAME in result.stdout.strip()
+ except (subprocess.SubprocessError, FileNotFoundError):
+ return False
+
+
+def get_current_engine() -> Optional[str]:
+ """Get the currently active IBus engine name."""
+ try:
+ import subprocess
+
+ result = subprocess.run(
+ ["ibus", "engine"],
+ capture_output=True,
+ text=True,
+ timeout=5,
+ )
+ if result.returncode == 0:
+ return result.stdout.strip()
+ except (subprocess.SubprocessError, FileNotFoundError):
+ pass
+ return None
+
+
+def switch_engine(engine_name: str) -> bool:
+ """Switch to the specified IBus engine."""
+ import time
+
+ try:
+ import subprocess
+
+ subprocess.run(
+ ["ibus", "engine", engine_name],
+ capture_output=True,
+ text=True,
+ timeout=5,
+ )
+ # ibus engine command may return non-zero even on success
+ # So we verify by checking the current engine
+ time.sleep(0.2)
+ current = get_current_engine()
+ if current == engine_name:
+ return True
+ else:
+ logger.warning(f"Engine switch failed: expected {engine_name}, got {current}")
+ return False
+ except (subprocess.SubprocessError, FileNotFoundError) as e:
+ logger.error(f"Failed to switch IBus engine: {e}")
+ return False
+
+
+def is_engine_process_running() -> bool:
+ """Check if the Vocalinux IBus engine process is running."""
+ try:
+ if not PID_FILE.exists():
+ return False
+
+ pid = int(PID_FILE.read_text().strip())
+ # Check if process exists and is our engine
+ os.kill(pid, 0) # Raises OSError if process doesn't exist
+ # Verify it's actually our process by checking cmdline
+ cmdline_path = Path(f"/proc/{pid}/cmdline")
+ if cmdline_path.exists():
+ cmdline = cmdline_path.read_text()
+ return "ibus_engine.py" in cmdline and "vocalinux" in cmdline
+ return True # Process exists but can't verify cmdline
+ except (OSError, ValueError, FileNotFoundError):
+ # Process doesn't exist or PID file is invalid
+ if PID_FILE.exists():
+ PID_FILE.unlink() # Clean up stale PID file
+ return False
+
+
+def start_engine_process() -> bool:
+ """
+ Start the IBus engine process in the background.
+
+ This should be called by Vocalinux on startup to ensure the engine
+ is running before attempting to switch to it.
+
+ Returns:
+ True if the engine was started or is already running, False otherwise
+ """
+ import time
+
+ if is_engine_process_running():
+ logger.debug("IBus engine process already running")
+ return True
+
+ engine_script = Path(__file__).resolve()
+ logger.info(f"Starting IBus engine process: {engine_script}")
+
+ try:
+ # Start the engine process using the same Python interpreter
+ # and inherit the current environment (for venv compatibility)
+ env = os.environ.copy()
+ # Ensure PYTHONPATH includes current site-packages if in a venv
+ if hasattr(sys, "prefix") and sys.prefix != sys.base_prefix:
+ # We're in a virtual environment - ensure it's preserved
+ env["VIRTUAL_ENV"] = sys.prefix
+ # Keep PATH so the venv's bin is first
+
+ process = subprocess.Popen(
+ [sys.executable, str(engine_script)],
+ stdout=subprocess.DEVNULL,
+ stderr=subprocess.DEVNULL,
+ start_new_session=True,
+ env=env,
+ )
+
+ # Write PID file for tracking
+ ensure_ibus_dir()
+ PID_FILE.write_text(str(process.pid))
+
+ # Wait a bit for the engine to start
+ for _ in range(10):
+ time.sleep(0.2)
+ if is_engine_process_running():
+ logger.info("IBus engine process started successfully")
+ return True
+
+ logger.error("IBus engine process failed to start")
+ return False
+
+ except Exception as e:
+ logger.error(f"Failed to start IBus engine process: {e}")
+ return False
+
+
+def stop_engine_process() -> None:
+ """Stop the IBus engine process if running."""
+ try:
+ if not PID_FILE.exists():
+ logger.debug("No PID file found, engine not running")
+ return
+
+ pid = int(PID_FILE.read_text().strip())
+ # Verify it's our process before killing
+ cmdline_path = Path(f"/proc/{pid}/cmdline")
+ if cmdline_path.exists():
+ cmdline = cmdline_path.read_text()
+ if "ibus_engine.py" not in cmdline or "vocalinux" not in cmdline:
+ logger.warning(f"PID {pid} is not our engine process, skipping kill")
+ PID_FILE.unlink()
+ return
+
+ os.kill(pid, signal.SIGTERM)
+ logger.info(f"IBus engine process (PID {pid}) stopped")
+ PID_FILE.unlink()
+ except (OSError, ValueError, FileNotFoundError) as e:
+ logger.debug(f"Failed to stop IBus engine process: {e}")
+ if PID_FILE.exists():
+ PID_FILE.unlink()
+
+
+class VocalinuxEngine(IBus.Engine if IBUS_AVAILABLE else object):
+ """
+ IBus proxy engine for Vocalinux text injection.
+
+ This engine acts as a transparent proxy:
+ - All keyboard input passes through unchanged
+ - Text injection requests are received via Unix socket
+ - Injected text is committed directly to the focused application
+
+ Users should set this as their default input method for seamless
+ voice dictation support.
+ """
+
+ if IBUS_AVAILABLE:
+ __gtype_name__ = "VocalinuxEngine"
+
+ # Class-level reference to active engine instance
+ _active_instance: Optional["VocalinuxEngine"] = None
+ _socket_server: Optional[threading.Thread] = None
+ _server_socket: Optional[socket.socket] = None
+ _server_running: bool = False
+
+ def __init__(self):
+ """Initialize the Vocalinux IBus engine."""
+ if IBUS_AVAILABLE:
+ super().__init__()
+ logger.debug("VocalinuxEngine instance created")
+
+ def do_enable(self) -> None:
+ """Called when the engine is enabled/selected."""
+ logger.info("VocalinuxEngine enabled - ready for text injection")
+ VocalinuxEngine._active_instance = self
+
+ # Start socket server if not already running
+ if VocalinuxEngine._socket_server is None:
+ self._start_socket_server()
+
+ def do_disable(self) -> None:
+ """Called when the engine is disabled/deselected."""
+ logger.debug("VocalinuxEngine disabled (keeping instance for injection)")
+ # NOTE: We intentionally do NOT clear _active_instance here.
+ # IBus calls do_disable when focus changes between windows, but we still
+ # want to be able to inject text. The engine process keeps running and
+ # the instance remains valid for text injection.
+
+ def do_focus_in(self) -> None:
+ """Called when the engine gains focus."""
+ logger.debug("VocalinuxEngine focus in")
+ VocalinuxEngine._active_instance = self
+
+ def do_focus_out(self) -> None:
+ """Called when the engine loses focus."""
+ logger.debug("VocalinuxEngine focus out")
+
+ def do_process_key_event(self, keyval: int, keycode: int, state: int) -> bool:
+ """
+ Process key events.
+
+ We pass through ALL keys unchanged since this is a proxy engine.
+ The actual keyboard layout handling is done by the system.
+ """
+ return False # Don't consume any keys - pass through
+
+ def inject_text(self, text: str) -> bool:
+ """
+ Inject text into the currently focused application.
+
+ Args:
+ text: The text to inject
+
+ Returns:
+ True if injection was successful, False otherwise
+ """
+ if not text:
+ return True
+
+ try:
+ logger.debug(f"Injecting text: {text[:50]}...")
+ self.commit_text(IBus.Text.new_from_string(text))
+ logger.info(f"Text injected via IBus: '{text[:20]}...' ({len(text)} chars)")
+ return True
+ except Exception as e:
+ logger.error(f"Failed to inject text: {e}")
+ return False
+
+ @classmethod
+ def _start_socket_server(cls) -> None:
+ """Start the Unix socket server for receiving injection requests."""
+ ensure_ibus_dir()
+
+ # Remove existing socket
+ if SOCKET_PATH.exists():
+ SOCKET_PATH.unlink()
+
+ cls._server_running = True
+
+ def server_thread():
+ try:
+ cls._server_socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+ cls._server_socket.bind(str(SOCKET_PATH))
+ cls._server_socket.listen(1)
+ SOCKET_PATH.chmod(0o600) # Only user can access
+
+ logger.info(f"Socket server listening on {SOCKET_PATH}")
+
+ while cls._server_running:
+ try:
+ conn, _ = cls._server_socket.accept()
+ with conn:
+ # Verify peer is same user (defense in depth)
+ if not verify_peer_credentials(conn):
+ conn.sendall(b"UNAUTHORIZED")
+ continue
+
+ data = conn.recv(65536) # Max text size
+ if data:
+ text = data.decode("utf-8")
+ # Schedule injection on main thread
+ if cls._active_instance:
+
+ def do_inject(t):
+ cls._active_instance.inject_text(t)
+ return False # Run only once
+
+ GLib.idle_add(do_inject, text)
+ conn.sendall(b"OK")
+ else:
+ logger.warning("No active engine instance")
+ conn.sendall(b"NO_ENGINE")
+ except OSError as e:
+ # Check if we're shutting down
+ if not cls._server_running:
+ logger.debug("Socket server shutting down")
+ break
+ logger.error(f"Socket connection error: {e}")
+ except Exception as e:
+ logger.error(f"Socket connection error: {e}")
+
+ except Exception as e:
+ if cls._server_running:
+ logger.error(f"Socket server error: {e}")
+
+ cls._socket_server = threading.Thread(target=server_thread, daemon=True)
+ cls._socket_server.start()
+
+ @classmethod
+ def stop_socket_server(cls) -> None:
+ """Stop the socket server."""
+ cls._server_running = False
+ if cls._server_socket:
+ try:
+ cls._server_socket.close()
+ except Exception:
+ pass
+ cls._server_socket = None
+
+ if SOCKET_PATH.exists():
+ SOCKET_PATH.unlink()
+
+
+class VocalinuxEngineApplication:
+ """
+ Application wrapper for running the Vocalinux IBus engine.
+
+ This is used when the engine is launched by IBus as a separate process.
+ """
+
+ def __init__(self):
+ """Initialize the engine application."""
+ if not IBUS_AVAILABLE:
+ raise RuntimeError("IBus is not available")
+
+ self.mainloop = GLib.MainLoop()
+ self.bus = IBus.Bus()
+ self.bus.connect("disconnected", self._on_disconnected)
+
+ self.factory = IBus.Factory.new(self.bus.get_connection())
+ self.factory.add_engine(
+ ENGINE_NAME,
+ GObject.type_from_name("VocalinuxEngine"),
+ )
+
+ self.bus.request_name(COMPONENT_NAME, 0)
+ logger.info("Vocalinux IBus engine started")
+
+ # Start the socket server immediately so it's ready for connections
+ # even before the engine is activated via `ibus engine vocalinux`
+ VocalinuxEngine._start_socket_server()
+
+ def _on_disconnected(self, bus: "IBus.Bus") -> None:
+ """Handle IBus disconnection."""
+ logger.info("IBus disconnected, exiting")
+ VocalinuxEngine.stop_socket_server()
+ self.mainloop.quit()
+
+ def run(self) -> None:
+ """Run the engine main loop."""
+ self.mainloop.run()
+
+
+class IBusTextInjector:
+ """
+ Text injector that uses IBus for text injection.
+
+ 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
+ 3. Switches to the Vocalinux engine
+
+ On cleanup (stop), it restores the previous engine.
+ """
+
+ def __init__(self, auto_activate: bool = True):
+ """
+ Initialize the IBus text injector.
+
+ Args:
+ auto_activate: If True, automatically install and activate the engine
+ """
+ if not IBUS_AVAILABLE:
+ raise RuntimeError("IBus is not available")
+
+ ensure_ibus_dir()
+ self._previous_engine: Optional[str] = None
+
+ if auto_activate:
+ self._setup_engine()
+
+ 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():
+ logger.error("Failed to register IBus engine")
+ return
+
+ # Start the engine process (Vocalinux starts it, not IBus)
+ if not start_engine_process():
+ logger.error("Failed to start IBus engine process")
+ return
+
+ # Save current engine and switch to Vocalinux
+ if not is_engine_active():
+ self._previous_engine = get_current_engine()
+ if self._previous_engine:
+ logger.info(f"Saving current engine: {self._previous_engine}")
+
+ logger.info("Activating Vocalinux IBus engine...")
+ if switch_engine(ENGINE_NAME):
+ logger.info("Vocalinux IBus engine activated")
+ else:
+ logger.error("Failed to activate Vocalinux IBus engine")
+
+ def stop(self) -> None:
+ """
+ Stop the IBus text injector and restore previous engine.
+
+ Call this when Vocalinux is shutting down.
+ """
+ if self._previous_engine:
+ logger.info(f"Restoring previous engine: {self._previous_engine}")
+ switch_engine(self._previous_engine)
+ self._previous_engine = None
+
+ # Stop the engine process
+ stop_engine_process()
+
+ def inject_text(self, text: str) -> bool:
+ """
+ Inject text using the IBus engine.
+
+ Args:
+ text: The text to inject
+
+ Returns:
+ True if injection was successful, False otherwise
+ """
+ if not text or not text.strip():
+ logger.debug("Empty text provided, skipping injection")
+ return True
+
+ logger.info(f"Starting IBus text injection: '{text[:20]}...' (length: {len(text)})")
+
+ # Ensure vocalinux engine is active (user may have switched keyboard)
+ if not is_engine_active():
+ logger.debug("Vocalinux engine not active, re-activating...")
+ switch_engine(ENGINE_NAME)
+
+ try:
+ if not SOCKET_PATH.exists():
+ logger.error(
+ "IBus engine socket not found. " "Make sure Vocalinux IBus engine is running."
+ )
+ return False
+
+ # Connect to engine socket and send text
+ with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as sock:
+ sock.settimeout(5.0)
+ sock.connect(str(SOCKET_PATH))
+ sock.sendall(text.encode("utf-8"))
+
+ # Wait for response
+ response = sock.recv(64).decode("utf-8")
+ if response == "OK":
+ logger.debug("Text injection successful")
+ return True
+ else:
+ logger.error(f"Text injection failed: {response}")
+ return False
+
+ except socket.timeout:
+ logger.error("Timeout connecting to IBus engine")
+ return False
+ except FileNotFoundError:
+ logger.error("IBus engine socket not found")
+ return False
+ except Exception as e:
+ logger.error(f"Failed to inject text via IBus: {e}")
+ return False
+
+
+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 main():
+ """Entry point when run as IBus engine process."""
+ logging.basicConfig(
+ level=logging.DEBUG,
+ format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+ )
+
+ if not IBUS_AVAILABLE:
+ logger.error("IBus is not available")
+ return 1
+
+ IBus.init()
+ app = VocalinuxEngineApplication()
+ app.run()
+ return 0
+
+
+if __name__ == "__main__":
+ import sys
+
+ sys.exit(main())
diff --git a/src/vocalinux/text_injection/text_injector.py b/src/vocalinux/text_injection/text_injector.py
index 7244162..abf21b7 100644
--- a/src/vocalinux/text_injection/text_injector.py
+++ b/src/vocalinux/text_injection/text_injector.py
@@ -13,6 +13,11 @@ import time
from enum import Enum
from typing import Optional # noqa: F401
+from .ibus_engine import (
+ IBusTextInjector,
+ is_ibus_available,
+)
+
logger = logging.getLogger(__name__)
@@ -22,6 +27,7 @@ class DesktopEnvironment(Enum):
X11 = "x11"
WAYLAND = "wayland"
WAYLAND_XDOTOOL = "wayland-xdotool" # Wayland with XWayland fallback
+ WAYLAND_IBUS = "wayland-ibus" # Wayland with IBus engine (preferred)
UNKNOWN = "unknown"
@@ -40,6 +46,7 @@ class TextInjector:
Args:
wayland_mode: Force Wayland compatibility mode
"""
+ self._ibus_injector: Optional[IBusTextInjector] = None
self.environment = self._detect_environment()
# Force Wayland mode if requested
@@ -88,6 +95,17 @@ class TextInjector:
except Exception as e:
logger.error(f"XWayland fallback test failed: {e}")
+ def stop(self) -> None:
+ """
+ Clean up resources and restore previous state.
+
+ Call this when shutting down Vocalinux.
+ """
+ if self._ibus_injector:
+ logger.info("Stopping IBus text injector")
+ self._ibus_injector.stop()
+ self._ibus_injector = None
+
def _detect_environment(self) -> DesktopEnvironment:
"""
Detect the current desktop environment (X11 or Wayland).
@@ -118,8 +136,19 @@ class TextInjector:
logger.error("xdotool not found. Please install it with: sudo apt install xdotool")
raise RuntimeError("Missing required dependency: xdotool")
else:
- # Check for wtype or ydotool for Wayland
- # Prefer ydotool as it works universally (including GNOME Wayland)
+ # For Wayland, prefer IBus as it works universally with any keyboard layout
+ # IBus is usually pre-installed on GNOME systems
+ if is_ibus_available():
+ try:
+ self._ibus_injector = IBusTextInjector(auto_activate=True)
+ self.environment = DesktopEnvironment.WAYLAND_IBUS
+ self.wayland_tool = "ibus"
+ logger.info("Using IBus for Wayland text injection (best compatibility)")
+ return
+ except Exception as e:
+ logger.warning(f"IBus initialization failed: {e}, trying alternatives")
+
+ # Fallback: Check for wtype or ydotool for Wayland
wtype_available = shutil.which("wtype") is not None
ydotool_available = shutil.which("ydotool") is not None
xdotool_available = shutil.which("xdotool") is not None
@@ -131,9 +160,7 @@ class TextInjector:
["ydotool", "type", ""], check=True, stderr=subprocess.PIPE, timeout=2
)
self.wayland_tool = "ydotool"
- logger.info(
- f"Using {self.wayland_tool} for Wayland text injection (universal compatibility)"
- )
+ logger.info(f"Using {self.wayland_tool} for Wayland text injection")
except (
subprocess.CalledProcessError,
subprocess.TimeoutExpired,
@@ -158,6 +185,7 @@ class TextInjector:
else:
logger.error(
"No text injection tools found. Please install one of:\n"
+ "- IBus (recommended, usually pre-installed)\n"
"- wtype: sudo apt install wtype\n"
"- ydotool: sudo apt install ydotool\n"
"- xdotool: sudo apt install xdotool"
@@ -219,7 +247,14 @@ class TextInjector:
logger.debug(f"Text to inject: '{text}'")
try:
- if (
+ if self.environment == DesktopEnvironment.WAYLAND_IBUS:
+ # Use IBus for text injection
+ if self._ibus_injector is not None:
+ return self._ibus_injector.inject_text(text)
+ else:
+ logger.error("IBus injector not initialized")
+ return False
+ elif (
self.environment == DesktopEnvironment.X11
or self.environment == DesktopEnvironment.WAYLAND_XDOTOOL
):
diff --git a/src/vocalinux/ui/tray_indicator.py b/src/vocalinux/ui/tray_indicator.py
index 354de1e..4b28f88 100644
--- a/src/vocalinux/ui/tray_indicator.py
+++ b/src/vocalinux/ui/tray_indicator.py
@@ -339,6 +339,10 @@ class TrayIndicator:
# Stop the keyboard shortcut manager
self.shortcut_manager.stop()
+ # Stop the text injector (restores previous IBus engine)
+ if hasattr(self, "text_injector") and self.text_injector is not None:
+ self.text_injector.stop()
+
Gtk.main_quit()
def run(self):
diff --git a/tests/test_ibus_engine.py b/tests/test_ibus_engine.py
new file mode 100644
index 0000000..ee4868d
--- /dev/null
+++ b/tests/test_ibus_engine.py
@@ -0,0 +1,642 @@
+"""
+Tests for IBus engine functionality.
+"""
+
+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 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 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("stale")
+
+ result = is_component_up_to_date()
+ self.assertFalse(result)
+
+
+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 (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 (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.is_engine_registered")
+ @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_init_auto_activate(
+ self,
+ mock_switch,
+ mock_get_current,
+ mock_is_active,
+ mock_is_registered,
+ mock_ensure_dir,
+ ):
+ """Test initialization with auto_activate=True."""
+ mock_is_registered.return_value = True
+ mock_is_active.return_value = False
+ mock_get_current.return_value = "xkb:us::eng"
+ mock_switch.return_value = True
+
+ from vocalinux.text_injection.ibus_engine import IBusTextInjector
+
+ injector = IBusTextInjector(auto_activate=True)
+
+ # ensure_ibus_dir is called twice: once in __init__ and once in start_engine_process
+ self.assertGreaterEqual(mock_ensure_dir.call_count, 1)
+ mock_switch.assert_called_once_with("vocalinux")
+ 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.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.IBUS_AVAILABLE", True)
+ @patch("vocalinux.text_injection.ibus_engine.ensure_ibus_dir")
+ def test_inject_text_engine_error(self, mock_ensure_dir):
+ """Test text injection when engine returns error."""
+ from vocalinux.text_injection.ibus_engine import IBusTextInjector
+
+ # Create a mock server socket that returns error
+ server_sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+ server_sock.bind(str(self.socket_path))
+ server_sock.listen(1)
+
+ 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):
+ injector = IBusTextInjector(auto_activate=False)
+ result = injector.inject_text("Hello")
+
+ server_sock.close()
+ 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)
+
+
+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."""
+
+ 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.IBusTextInjector")
+ def test_wayland_prefers_ibus(self, mock_injector_class, mock_ibus_available):
+ """Test that Wayland environment prefers IBus when available."""
+ 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 (
+ 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.IBusTextInjector",
+ mock_injector_class,
+ ):
+ injector = TextInjector()
+
+ # Should be using IBus
+ self.assertEqual(injector.environment, DesktopEnvironment.WAYLAND_IBUS)
+
+ @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()
+
+ mock_injector_instance.stop.assert_called_once()
+
+
+class TestDesktopEnvironmentEnumWithIBus(unittest.TestCase):
+ """Tests for DesktopEnvironment enum including IBus variant."""
+
+ 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")
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/test_text_injector.py b/tests/test_text_injector.py
index d7162e9..f71a48c 100644
--- a/tests/test_text_injector.py
+++ b/tests/test_text_injector.py
@@ -34,6 +34,12 @@ class TestTextInjector(unittest.TestCase):
self.patch_sleep = patch("time.sleep")
self.mock_sleep = self.patch_sleep.start()
+ # Disable IBus for these tests (testing fallback paths)
+ self.patch_ibus_available = patch(
+ "vocalinux.text_injection.text_injector.is_ibus_available", return_value=False
+ )
+ self.patch_ibus_available.start()
+
# Setup environment variable patching
self.env_patcher = patch.dict("os.environ", {"XDG_SESSION_TYPE": "x11", "DISPLAY": ":0"})
self.env_patcher.start()
@@ -56,6 +62,7 @@ class TestTextInjector(unittest.TestCase):
self.patch_which.stop()
self.patch_subprocess.stop()
self.patch_sleep.stop()
+ self.patch_ibus_available.stop()
self.env_patcher.stop()
def test_detect_x11_environment(self):
@@ -487,6 +494,7 @@ class TestDesktopEnvironmentEnum(unittest.TestCase):
self.assertEqual(DesktopEnvironment.X11.value, "x11")
self.assertEqual(DesktopEnvironment.WAYLAND.value, "wayland")
self.assertEqual(DesktopEnvironment.WAYLAND_XDOTOOL.value, "wayland-xdotool")
+ self.assertEqual(DesktopEnvironment.WAYLAND_IBUS.value, "wayland-ibus")
self.assertEqual(DesktopEnvironment.UNKNOWN.value, "unknown")
@@ -505,6 +513,12 @@ class TestTextInjectorEdgeCases(unittest.TestCase):
self.patch_sleep = patch("time.sleep")
self.mock_sleep = self.patch_sleep.start()
+ # Disable IBus for these tests (testing fallback paths)
+ self.patch_ibus_available = patch(
+ "vocalinux.text_injection.text_injector.is_ibus_available", return_value=False
+ )
+ self.patch_ibus_available.start()
+
mock_process = MagicMock()
mock_process.returncode = 0
mock_process.stdout = "1234"
@@ -516,6 +530,7 @@ class TestTextInjectorEdgeCases(unittest.TestCase):
self.patch_which.stop()
self.patch_subprocess.stop()
self.patch_sleep.stop()
+ self.patch_ibus_available.stop()
def test_wtype_test_exception(self):
"""Test wtype test handling exceptions gracefully."""