From a1725f7cfc4c838c64eda09be387e063e47679cc Mon Sep 17 00:00:00 2001 From: Antoine Monnet <39551484+AntoineMontane@users.noreply.github.com> Date: Wed, 18 Feb 2026 07:15:06 +0100 Subject: [PATCH] feat(text-injection): add IBus engine for Wayland text injection (#211) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(text-injection): add IBus engine for Wayland text injection Add a custom IBus input method engine for reliable text injection on Wayland, solving keyboard layout issues with xdotool/wtype. Key features: - IBus proxy engine that passes through normal keyboard input - Text injection via Unix socket IPC - User-level component installation (no sudo required) - Auto-reactivation when user switches keyboard layouts - Graceful fallback when IBus is unavailable Changes: - New ibus_engine.py with VocalinuxEngine and IBusTextInjector - TextInjector now prefers IBus on Wayland for best compatibility - Add IBus GObject bindings to system dependencies in install.sh - Add comprehensive tests for IBus engine 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 * security: secure IBus socket with permissions and peer verification - Set directory to 0o700 (owner-only access) - Set socket to 0o600 (owner-only read/write) - Add SO_PEERCRED verification to reject connections from other users This provides defense-in-depth: even if directory permissions are bypassed, the socket server verifies the connecting process UID. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 * fix: address PR review feedback for IBus text injection - Fix infinite loop in socket server by adding _server_running flag - Use sys.executable instead of hardcoded python3 for venv compatibility - Also inherit environment (env=os.environ.copy()) for venv deps - Fix component XML to use sys.executable in tag - Replace unsafe pgrep with PID file tracking and /proc verification - Integrate TextInjector.stop() in TrayIndicator._quit() for cleanup - Remove unused is_engine_registered import - Fix redundant hasattr check to use proper None comparison - Auto-detect and update stale IBus component XML on startup 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 --------- Co-authored-by: AntoineMontane Co-authored-by: Claude Opus 4.5 --- install.sh | 10 +- src/vocalinux/text_injection/__init__.py | 7 + src/vocalinux/text_injection/ibus_engine.py | 758 ++++++++++++++++++ src/vocalinux/text_injection/text_injector.py | 47 +- src/vocalinux/ui/tray_indicator.py | 4 + tests/test_ibus_engine.py | 642 +++++++++++++++ tests/test_text_injector.py | 15 + 7 files changed, 1472 insertions(+), 11 deletions(-) create mode 100644 src/vocalinux/text_injection/ibus_engine.py create mode 100644 tests/test_ibus_engine.py 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."""