From db075907615b134f473d2ea46fc0a7d651c71df5 Mon Sep 17 00:00:00 2001 From: Web-Benefits <78207756+webenefits@users.noreply.github.com> Date: Wed, 30 Sep 2026 06:19:29 +0200 Subject: [PATCH] feat(activation): D-Bus activation for compositor global shortcuts (#568) Rebased onto latest main (conflict-resolution squash for PR #568). Co-authored-by: Alexander Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- docs/USER_GUIDE.md | 49 +++++ src/vocalinux/dbus_service.py | 208 ++++++++++++++++++ src/vocalinux/main.py | 66 +++++- src/vocalinux/ui/config_manager.py | 4 + src/vocalinux/ui/settings_dialog.py | 65 +++++- src/vocalinux/ui/tray_indicator.py | 96 ++++++++- tests/test_dbus_activation.py | 316 ++++++++++++++++++++++++++++ tests/test_settings_shortcuts.py | 61 +++++- tests/test_suspend_handler.py | 19 ++ tests/test_tray_indicator_ext.py | 202 ++++++++++++++++++ 10 files changed, 1077 insertions(+), 9 deletions(-) create mode 100644 src/vocalinux/dbus_service.py create mode 100644 tests/test_dbus_activation.py diff --git a/docs/USER_GUIDE.md b/docs/USER_GUIDE.md index 266cb49..3337ca1 100644 --- a/docs/USER_GUIDE.md +++ b/docs/USER_GUIDE.md @@ -104,6 +104,55 @@ the hotkey; Parakeet does not use the language picker at all. Active-layout detection uses GNOME's input-source settings; on other desktops it falls back to the configured primary layout. +### Activation via a KDE Plasma global shortcut + +Instead of the built-in hotkey listener, you can let your desktop's global +shortcut system trigger Vocalinux. On Wayland the built-in listener reads +`/dev/input` (requiring membership in the `input` group and effectively acting +as a system-wide key reader). Delegating activation to the compositor avoids +this entirely: no `/dev/input` access and no `input` group needed just to +start/stop dictation. Text injection is unaffected. + +Enable it in Vocalinux: + +1. Open **Settings → Shortcuts** (the Keyboard Shortcuts group on the Dictation page) +2. Turn on **External Activation (Desktop Shortcut)** + +The change applies immediately; you do not need to restart. The built-in key +listener stops, and a running instance exposes a D-Bus service on the session +bus (`com.vocalinux.Vocalinux`). The CLI can forward commands to it: + +```bash +vocalinux --toggle # start if idle, stop if active +vocalinux --start # start voice typing +vocalinux --stop # stop voice typing +``` + +Then bind a compositor shortcut to `vocalinux --toggle`. On KDE Plasma: + +1. Open **System Settings -> Keyboard -> Shortcuts -> Add New -> Command or Script** + (older Plasma: **System Settings -> Shortcuts -> Custom Shortcuts -> Edit -> + New -> Global Shortcut -> Command/URL**). + +2. Bind a key combination of your choice to the command `vocalinux --toggle`. + +Now your chosen key combination toggles dictation, handled by the compositor +rather than by Vocalinux reading the keyboard directly. This works the same way +on other compositors that support binding a key to a command (e.g. GNOME custom +shortcuts, Sway/Hyprland `bindsym`/`bind`). + +As an advanced alternative, you can set the same option in +`~/.config/vocalinux/config.json` under `shortcuts`: + +```json +"shortcuts": { + "disable_internal_hotkey": true +} +``` + +Quit Vocalinux before editing that file so the running app cannot overwrite +the edit from its cached config. Start Vocalinux again after saving. + ### Model size (whisper.cpp / Whisper) | Size | Approx. size | Tradeoff | diff --git a/src/vocalinux/dbus_service.py b/src/vocalinux/dbus_service.py new file mode 100644 index 0000000..e68b706 --- /dev/null +++ b/src/vocalinux/dbus_service.py @@ -0,0 +1,208 @@ +""" +Session-bus D-Bus service for Vocalinux. + +Exposes ``com.vocalinux.Vocalinux`` with ``Toggle``/``Start``/``Stop`` methods +so an external trigger (e.g. a KDE Plasma global shortcut running +``vocalinux --toggle``) can control a running instance without the internal +evdev/pynput hotkey listener reading ``/dev/input``. + +Mirrors the GDBus usage already present in ``suspend_handler.py``. +""" + +import logging +from typing import Callable, Optional + +from gi.repository import Gio, GLib + +logger = logging.getLogger(__name__) + +BUS_NAME = "com.vocalinux.Vocalinux" +OBJECT_PATH = "/com/vocalinux/Vocalinux" +INTERFACE_NAME = "com.vocalinux.Vocalinux" + +# Maps CLI trigger names to the D-Bus method names exposed below. +COMMAND_METHODS = { + "toggle": "Toggle", + "start": "Start", + "stop": "Stop", +} + +_INTROSPECTION_XML = """ + + + + + + + +""" + + +class VocalinuxDBusService: + """Owns the Vocalinux session-bus name and dispatches control methods. + + Method calls are marshalled onto the GLib/GTK main thread via + :func:`GLib.idle_add` before the callbacks run, so recognition start/stop + happens on the same thread as the rest of the tray indicator. + + Usage:: + + service = VocalinuxDBusService( + on_toggle=indicator._toggle_recognition, + on_start=indicator._start_recognition, + on_stop=indicator._stop_recognition, + ) + # ... later, on shutdown: + service.shutdown() + """ + + def __init__( + self, + on_toggle: Callable[[], None], + on_start: Callable[[], None], + on_stop: Callable[[], None], + on_registration_failed: Optional[Callable[[], None]] = None, + ) -> None: + self._callbacks = { + "Toggle": on_toggle, + "Start": on_start, + "Stop": on_stop, + } + self._on_registration_failed = on_registration_failed + self._owner_id = 0 + self._registration_id = 0 + self._connection: Optional[Gio.DBusConnection] = None + self._node_info = Gio.DBusNodeInfo.new_for_xml(_INTROSPECTION_XML) + self._register() + + def _register(self) -> None: + """Request ownership of the bus name on the session bus.""" + try: + self._owner_id = Gio.bus_own_name( + Gio.BusType.SESSION, + BUS_NAME, + Gio.BusNameOwnerFlags.NONE, + self._on_bus_acquired, + self._on_name_acquired, + self._on_name_lost, + ) + logger.info("Requested D-Bus name %s on session bus", BUS_NAME) + except GLib.Error: + logger.warning("Failed to own D-Bus name %s", BUS_NAME, exc_info=True) + self._notify_registration_failed() + + def _on_bus_acquired(self, connection: Gio.DBusConnection, name: str) -> None: + """Register the control object once the bus connection is available.""" + self._connection = connection + try: + self._registration_id = connection.register_object( + OBJECT_PATH, + self._node_info.interfaces[0], + self._handle_method_call, + None, + None, + ) + except GLib.Error: + logger.warning("Failed to register D-Bus object", exc_info=True) + self._notify_registration_failed() + + def _on_name_acquired(self, connection: Gio.DBusConnection, name: str) -> None: + logger.info("Acquired D-Bus name %s", name) + + def _on_name_lost(self, connection: Gio.DBusConnection, name: str) -> None: + logger.warning("Could not acquire or lost D-Bus name %s", name) + self._notify_registration_failed() + + def _notify_registration_failed(self) -> None: + """Tell the caller the service is not reachable on the bus. + + Without this, a config that also disables the internal evdev/pynput + listener would leave the user with no way to start or stop dictation. + """ + if self._on_registration_failed is not None: + self._on_registration_failed() + + def _handle_method_call( + self, + connection: Gio.DBusConnection, + sender: str, + object_path: str, + interface_name: str, + method_name: str, + parameters: GLib.Variant, + invocation: Gio.DBusMethodInvocation, + ) -> None: + """Dispatch an incoming method call to the matching callback.""" + callback = self._callbacks.get(method_name) + if callback is None: + invocation.return_error_literal( + Gio.dbus_error_quark(), + Gio.DBusError.UNKNOWN_METHOD, + f"Unknown method: {method_name}", + ) + return + + # Marshal onto the GTK main thread; recognition start/stop is expected + # to run there, matching the rest of the tray indicator. + GLib.idle_add(self._invoke, callback) + invocation.return_value(None) + + @staticmethod + def _invoke(callback: Callable[[], None]) -> bool: + try: + callback() + except Exception: + logger.error("Error in D-Bus method handler", exc_info=True) + return GLib.SOURCE_REMOVE # type: ignore[no-any-return] + + def shutdown(self) -> None: + """Unregister the object and release the bus name.""" + if self._registration_id and self._connection is not None: + try: + self._connection.unregister_object(self._registration_id) + except GLib.Error: + logger.debug("Failed to unregister D-Bus object", exc_info=True) + self._registration_id = 0 + if self._owner_id: + try: + Gio.bus_unown_name(self._owner_id) + except GLib.Error: + logger.debug("Failed to unown D-Bus name", exc_info=True) + self._owner_id = 0 + logger.info("D-Bus service shut down") + + +def send_command(command: str) -> bool: + """Forward a control command to a running instance over the session bus. + + Args: + command: One of ``"toggle"``, ``"start"``, ``"stop"``. + + Returns: + True if the call reached a running instance, False otherwise. + """ + method = COMMAND_METHODS.get(command) + if method is None: + raise ValueError(f"Unknown command: {command}") + + try: + proxy = Gio.DBusProxy.new_for_bus_sync( + Gio.BusType.SESSION, + Gio.DBusProxyFlags.DO_NOT_AUTO_START_AT_CONSTRUCTION, + None, + BUS_NAME, + OBJECT_PATH, + INTERFACE_NAME, + None, + ) + proxy.call_sync(method, None, Gio.DBusCallFlags.NONE, -1, None) + return True + except GLib.Error as exc: + logger.error("Failed to send '%s' command via D-Bus: %s", command, exc) + return False + except Exception as exc: + # e.g. no session bus available (DBUS_SESSION_BUS_ADDRESS unset in an + # SSH/systemd context). Guarantee a deterministic non-zero CLI exit + # instead of an unhandled traceback. + logger.error("Failed to send '%s' command (no session bus?): %s", command, exc) + return False diff --git a/src/vocalinux/main.py b/src/vocalinux/main.py index 226deff..89aa22a 100644 --- a/src/vocalinux/main.py +++ b/src/vocalinux/main.py @@ -7,6 +7,7 @@ import argparse import atexit import logging import sys +from typing import Optional from .utils.vosk_model_info import SUPPORTED_LANGUAGES from .version import __version__ @@ -91,9 +92,58 @@ def parse_arguments(): action="store_true", help="Start minimized to system tray", ) + # External activation triggers: forward a control command to a running + # instance over D-Bus (e.g. from a KDE Plasma global shortcut) and exit. + parser.add_argument( + "--toggle", + action="store_true", + help="Toggle voice typing on a running instance (via D-Bus) and exit", + ) + parser.add_argument( + "--start", + action="store_true", + help="Start voice typing on a running instance (via D-Bus) and exit", + ) + parser.add_argument( + "--stop", + action="store_true", + help="Stop voice typing on a running instance (via D-Bus) and exit", + ) return parser.parse_args() +# CLI flags that trigger a running instance instead of starting a new one. +_TRIGGER_FLAGS = ("toggle", "start", "stop") + + +def _selected_trigger(args: argparse.Namespace) -> Optional[str]: + """Return the external-activation command requested via CLI, if any.""" + for name in _TRIGGER_FLAGS: + # Explicit `is True` guards against MagicMock args in tests, whose + # attributes are truthy by default. + if getattr(args, name, False) is True: + return name + return None + + +def _dispatch_trigger(command: str) -> int: + """Forward a control command to a running instance over D-Bus. + + Returns a process exit code (0 on success, 1 if no instance is reachable). + """ + from .dbus_service import send_command + + if send_command(command): + logger.info("Sent '%s' command to running Vocalinux instance", command) + return 0 + + logger.error( + "Could not reach a running Vocalinux instance to '%s'. Is Vocalinux running?", + command, + ) + return 1 + + def check_dependencies(): """Check for required dependencies and provide helpful error messages.""" missing_system_deps = [] @@ -252,6 +302,17 @@ def main(): # another instance already holds the single-instance lock args = parse_arguments() + # Configure debug logging if requested + if args.debug: + logging.getLogger().setLevel(logging.DEBUG) + logger.debug("Debug logging enabled") + + # External activation: forward the command to a running instance over D-Bus + # and exit without acquiring the lock or starting a second full instance. + trigger = _selected_trigger(args) + if trigger is not None: + sys.exit(_dispatch_trigger(trigger)) + # Check for single instance BEFORE any initialization from . import single_instance @@ -279,11 +340,6 @@ def main(): # Register cleanup to release lock on exit atexit.register(single_instance.release_lock) - # Configure debug logging if requested - if args.debug: - logging.getLogger().setLevel(logging.DEBUG) - logger.debug("Debug logging enabled") - # Check dependencies first (before importing GTK-dependent modules) if not check_dependencies(): logger.error("Cannot start Vocalinux due to missing dependencies") diff --git a/src/vocalinux/ui/config_manager.py b/src/vocalinux/ui/config_manager.py index 5c37821..6d41c9a 100644 --- a/src/vocalinux/ui/config_manager.py +++ b/src/vocalinux/ui/config_manager.py @@ -148,6 +148,10 @@ DEFAULT_CONFIG = { "shortcuts": { "toggle_recognition": "right_alt+right_alt", "mode": "push_to_talk", # "toggle" or "push_to_talk" + # When True, the internal evdev/pynput listener is not started; activation + # comes in over D-Bus instead (e.g. a KDE Plasma global shortcut running + # `vocalinux --toggle`). Avoids /dev/input access and the input group. + "disable_internal_hotkey": False, # Pure-modifier gestures: "ctrl+ctrl", "alt+alt", "shift+shift" (and # left_/right_ variants) — double-tap (toggle) or hold (push_to_talk). # Modifier+key combos are also supported, e.g. "alt+r", "ctrl+alt+r", diff --git a/src/vocalinux/ui/settings_dialog.py b/src/vocalinux/ui/settings_dialog.py index edfa276..99519b1 100644 --- a/src/vocalinux/ui/settings_dialog.py +++ b/src/vocalinux/ui/settings_dialog.py @@ -2424,6 +2424,7 @@ class SettingsDialog(Gtk.Dialog): pending_update: Optional[ReleaseInfo] = None, update_status_callback: callable = None, overlay_enabled_callback: Optional[Callable[[bool], None]] = None, + hotkey_listener_update_callback: Optional[Callable[[], None]] = None, ): super().__init__(title="Vocalinux Settings", transient_for=parent, flags=0) # Force window decorations (title-bar close) on all WMs. An in-window @@ -2437,6 +2438,7 @@ class SettingsDialog(Gtk.Dialog): self.shortcut_update_callback = shortcut_update_callback self.update_status_callback = update_status_callback self.overlay_enabled_callback = overlay_enabled_callback + self.hotkey_listener_update_callback = hotkey_listener_update_callback self._test_active = False self._test_result = "" self._initializing = True # Flag to prevent auto-apply during initialization @@ -4110,6 +4112,22 @@ class SettingsDialog(Gtk.Dialog): description="Configure the shortcut to control voice recognition", ) + # External activation: drive start/stop from a desktop/compositor global + # shortcut (bound to `vocalinux --toggle`) instead of the built-in key + # listener. Avoids reading /dev/input (no keylogging, no `input` group). + self.disable_internal_hotkey_switch = Gtk.Switch() + self.disable_internal_hotkey_switch.set_tooltip_text( + "Turn off the built-in key listener and trigger voice typing from a " + "desktop global shortcut bound to 'vocalinux --toggle'. Avoids reading " + "/dev/input (no keylogging, no 'input' group needed)." + ) + external_row = PreferenceRow( + title="External Activation (Desktop Shortcut)", + subtitle="Use a compositor global shortcut instead of the built-in key listener", + widget=self.disable_internal_hotkey_switch, + ) + group.add_row(external_row) + # Mode selection (Toggle vs Push-to-Talk) self.shortcut_mode_combo = Gtk.ComboBoxText() _style_combo(self.shortcut_mode_combo) @@ -4129,12 +4147,12 @@ class SettingsDialog(Gtk.Dialog): if not self.shortcut_mode_combo.set_active_id(current_mode): self.shortcut_mode_combo.set_active_id(DEFAULT_SHORTCUT_MODE) - mode_row = PreferenceRow( + self.mode_row = PreferenceRow( title="Shortcut Mode", subtitle="How the shortcut behaves", widget=self.shortcut_mode_combo, ) - group.add_row(mode_row) + group.add_row(self.mode_row) # Shortcut selection combo self.shortcut_combo = Gtk.ComboBoxText() @@ -4231,10 +4249,47 @@ class SettingsDialog(Gtk.Dialog): # Connect signals self.shortcut_combo.connect("changed", self._on_shortcut_changed) self.shortcut_mode_combo.connect("changed", self._on_shortcut_mode_changed) + self.disable_internal_hotkey_switch.connect( + "state-set", self._on_disable_internal_hotkey_toggled + ) # Update UI based on initial mode self._update_shortcut_ui_for_mode(current_mode) + def _update_internal_hotkey_sensitivity(self, disabled: bool) -> None: + """Grey out the built-in shortcut controls when external activation is on.""" + for row in (self.mode_row, self.shortcut_row, self.custom_shortcut_row): + row.set_sensitive(not disabled) + + if disabled: + self.shortcut_info_label.set_text( + "External activation is on: the built-in key listener is off. " + "Bind a desktop global shortcut to 'vocalinux --toggle' to start/stop." + ) + else: + # Restore the mode-appropriate hint. + self._update_shortcut_ui_for_mode( + self.config_manager.get_str("shortcuts", "mode", "toggle") + ) + + def _on_disable_internal_hotkey_toggled(self, widget: Gtk.Switch, state: bool) -> bool: + """Handle toggle of the external-activation switch.""" + if self._initializing or self._applying_settings: + return False + + disabled = bool(state) + logger.info(f"External activation (internal hotkey disabled) toggled: {disabled}") + self.config_manager.set("shortcuts", "disable_internal_hotkey", disabled) + self.config_manager.save_settings() + + self._update_internal_hotkey_sensitivity(disabled) + + # Live-apply: start or stop the built-in listener without a restart. + if self.hotkey_listener_update_callback: + self.hotkey_listener_update_callback() + + return False + def _is_preset_shortcut(self, shortcut: str) -> bool: """Return True if shortcut is one of the built-in double-tap presets.""" return shortcut in SUPPORTED_SHORTCUTS @@ -5943,6 +5998,12 @@ class SettingsDialog(Gtk.Dialog): if not self.model_keepalive_timeout_combo.set_active_id(str(timeout_seconds)): self.model_keepalive_timeout_combo.set_active_id("300") + disable_internal_hotkey = self.config_manager.get_bool( + "shortcuts", "disable_internal_hotkey", False + ) + self.disable_internal_hotkey_switch.set_active(disable_internal_hotkey) + self._update_internal_hotkey_sensitivity(disable_internal_hotkey) + available_engines = get_available_engines() available_count = 0 diff --git a/src/vocalinux/ui/tray_indicator.py b/src/vocalinux/ui/tray_indicator.py index d2040c6..7b3edc2 100644 --- a/src/vocalinux/ui/tray_indicator.py +++ b/src/vocalinux/ui/tray_indicator.py @@ -37,6 +37,7 @@ from gi.repository import GdkPixbuf, Gio, GLib, GObject, Gtk # Import local modules - Use protocols to avoid circular imports from ..auto_pause_monitor import DEFAULT_POLL_INTERVAL_SECONDS, AutoPauseMonitor from ..common_types import RecognitionState, SpeechRecognitionManagerProtocol, TextInjectorProtocol +from ..dbus_service import VocalinuxDBusService from ..gateway_embed import GatewayStatus, get_gateway_embed_manager from ..model_keepalive import DEFAULT_IDLE_TIMEOUT_SECONDS, ModelKeepAlive from ..suspend_handler import SuspendHandler @@ -131,6 +132,11 @@ class TrayIndicator: # Shared with main() and the settings dialog: separate instances would # overwrite each other's saves with stale in-memory copies. self.config_manager = get_shared_config_manager() + # Set once by _on_dbus_registration_failed and never cleared: the + # service does not retry, so every later reconfigure (mode change, + # settings toggle, resume) must keep honoring the fallback rather + # than reapplying a disable_internal_hotkey setting D-Bus cannot serve. + self._external_activation_unavailable = False self._syncing_autostart_menu = False # Get configured shortcut and mode from config @@ -227,8 +233,63 @@ class TrayIndicator: # Set up keyboard shortcuts with mode support self._setup_keyboard_shortcuts() + # Register the session-bus service so external triggers (e.g. a KDE + # Plasma global shortcut running `vocalinux --toggle`) can control this + # running instance. Handlers marshal onto the GTK main thread. + self._dbus_service = VocalinuxDBusService( + on_toggle=self._toggle_recognition, + on_start=self._external_start, + on_stop=self._external_stop, + on_registration_failed=self._on_dbus_registration_failed, + ) + + def _external_activation_active(self) -> bool: + """Whether the internal listener should stay off right now. + + True only when the saved setting asks for it *and* the D-Bus service + has not already failed to register this run. The service does not + retry, so once it has failed, this stays False for the rest of the + process regardless of the saved setting — every later reconfigure + (mode change, settings toggle, resume) must keep using the fallback + instead of re-disabling the only working activation path. + """ + if self._external_activation_unavailable: + return False + return self.config_manager.get_bool("shortcuts", "disable_internal_hotkey", False) + + def _on_dbus_registration_failed(self) -> None: + """Fall back to the internal listener if external activation cannot work. + + Runs when the D-Bus service could not claim its bus name or register + its object (no session bus, name already owned, etc.). If the internal + evdev/pynput listener is also disabled, the user would otherwise be + left with no way to start or stop dictation at all. + """ + if not self.config_manager.get_bool("shortcuts", "disable_internal_hotkey", False): + return + logger.error( + "D-Bus activation unavailable and the internal listener is disabled; " + "falling back to the internal listener" + ) + notifications.notify( + "External activation unavailable", + "The D-Bus service could not start, so Vocalinux is using the " + "internal keyboard shortcut instead. Check Settings -> Shortcuts.", + "dialog-warning", + ) + self._external_activation_unavailable = True + self._setup_keyboard_shortcuts() + def _setup_keyboard_shortcuts(self): """Set up keyboard shortcuts based on configured mode.""" + # Reconfiguring (e.g. live-toggling external activation) tears down the + # release callback below. A push-to-talk session held at that moment + # would then never see its release, leaving recognition and the + # microphone running with no way back except another control surface. + if self.speech_engine.state != RecognitionState.IDLE: + logger.info("Stopping active recognition before reconfiguring shortcuts") + self._stop_recognition() + # Stop existing shortcut manager if running if self.shortcut_manager.active: logger.info("Stopping existing shortcut manager before reconfiguration") @@ -239,6 +300,13 @@ class TrayIndicator: self.shortcut_manager.register_press_callback(None) self.shortcut_manager.register_release_callback(None) + # External-activation mode: skip the internal evdev/pynput listener + # entirely so no /dev/input access is required. Activation then comes + # in over D-Bus (see VocalinuxDBusService). + if self._external_activation_active(): + logger.info("Internal hotkey listener disabled (external activation via D-Bus)") + return + # Get configured mode from config mode = self.config_manager.get_str("shortcuts", "mode", DEFAULT_SHORTCUT_MODE) logger.info(f"Setting up keyboard shortcuts with mode: {mode}") @@ -471,6 +539,21 @@ class TrayIndicator: if self.speech_engine.state == RecognitionState.IDLE: self.speech_engine.start_recognition(mode="push_to_talk") + def _external_start(self) -> None: + """Start recognition for an external (D-Bus) trigger. + + Uses normal start semantics — not push-to-talk — so a single + `vocalinux --start` transcribes immediately/with silence detection + rather than deferring until a Stop. + """ + if self.speech_engine.state == RecognitionState.IDLE: + self.speech_engine.start_recognition() + + def _external_stop(self) -> None: + """Stop recognition for an external (D-Bus) trigger.""" + if self.speech_engine.state != RecognitionState.IDLE: + self.speech_engine.stop_recognition() + def _stop_recognition(self): """Stop voice recognition (for push-to-talk mode).""" if self.speech_engine.state != RecognitionState.IDLE: @@ -843,6 +926,7 @@ class TrayIndicator: available, release, notify=False ), overlay_enabled_callback=self.set_overlay_enabled, + hotkey_listener_update_callback=self._setup_keyboard_shortcuts, ) dialog.connect("response", self._on_settings_dialog_response) dialog.connect("destroy", self._on_settings_dialog_destroyed) @@ -1098,7 +1182,14 @@ class TrayIndicator: logger.info("Skipping resume reinit: auto-pause still active") else: GLib.timeout_add_seconds(2, self._reinit_speech_after_resume) - GLib.timeout_add_seconds(2, self._start_input_device_monitor) + + # External-activation mode never started the /dev/input listener in + # the first place; watching it here on every resume would open the + # very file descriptor that mode promises to avoid. + if self._external_activation_active(): + logger.info("Skipping input device monitor: external activation via D-Bus") + else: + GLib.timeout_add_seconds(2, self._start_input_device_monitor) def _reinit_speech_after_resume(self): try: @@ -1224,6 +1315,9 @@ class TrayIndicator: if getattr(self, "_update_monitor", None) is not None: self._update_monitor.shutdown() + if getattr(self, "_dbus_service", None) is not None: + self._dbus_service.shutdown() + self._cleanup_input_monitor() # Stop the keyboard shortcut manager diff --git a/tests/test_dbus_activation.py b/tests/test_dbus_activation.py new file mode 100644 index 0000000..122d01e --- /dev/null +++ b/tests/test_dbus_activation.py @@ -0,0 +1,316 @@ +""" +Tests for D-Bus based external activation. + +Covers: +- CLI dispatch: --toggle/--start/--stop forward the right D-Bus command. +- Config gate: external mode skips the internal hotkey listener. +- D-Bus handlers route to the correct recognition callbacks. +""" + +import sys +from typing import TYPE_CHECKING +from unittest.mock import MagicMock, patch + +import pytest + +if TYPE_CHECKING: + from vocalinux.dbus_service import VocalinuxDBusService + +# gi is mocked globally by conftest. All vocalinux imports are done lazily +# inside the individual tests. Importing them at module level would bind their +# `gi.repository` references at collection time and clash with test_main.py, +# which replaces sys.modules["gi"] / sys.modules["gi.repository"] wholesale -- +# leading to MagicMock identity mismatches (e.g. GLib.SOURCE_REMOVE) in +# unrelated tests that run in the same session. + +# -- CLI dispatch ---------------------------------------------------------- + + +@pytest.mark.parametrize( + "flag,command", + [("--toggle", "toggle"), ("--start", "start"), ("--stop", "stop")], +) +def test_cli_flag_sends_matching_command(flag, command): + """Each trigger flag forwards the matching command and exits with 0.""" + from vocalinux.main import main + + with patch("vocalinux.dbus_service.send_command", return_value=True) as mock_send: + with patch.object(sys, "argv", ["vocalinux", flag]): + with pytest.raises(SystemExit) as exc_info: + main() + assert exc_info.value.code == 0 + mock_send.assert_called_once_with(command) + + +def test_cli_flag_exits_nonzero_when_no_instance(): + """When no instance answers, the trigger exits with a non-zero code.""" + from vocalinux.main import main + + with patch("vocalinux.dbus_service.send_command", return_value=False): + with patch.object(sys, "argv", ["vocalinux", "--toggle"]): + with pytest.raises(SystemExit) as exc_info: + main() + assert exc_info.value.code == 1 + + +def test_no_trigger_flag_does_not_dispatch(): + """Without a trigger flag, main() proceeds past the dispatch shortcut.""" + from vocalinux.main import main + + # Mock single_instance via sys.modules (not patch("...acquire_lock")) so we + # don't import the real module and set it as a vocalinux package attribute, + # which would defeat patch.dict-based single-instance tests elsewhere. + mock_single_instance = MagicMock() + mock_single_instance.acquire_lock.return_value = True + with patch("vocalinux.dbus_service.send_command") as mock_send: + # check_dependencies returns False so main() exits early after dispatch. + with patch.dict(sys.modules, {"vocalinux.single_instance": mock_single_instance}): + with patch("vocalinux.main.check_dependencies", return_value=False): + with patch.object(sys, "argv", ["vocalinux"]): + with pytest.raises(SystemExit): + main() + mock_send.assert_not_called() + + +# Config-gate tests for TrayIndicator._setup_keyboard_shortcuts live in +# test_tray_indicator_ext.py (TestExternalActivationGate). They require the real +# tray_indicator module, which must not be imported this early in the session -- +# doing so freezes tray_indicator.GLib against a stale gi.repository mock and +# breaks unrelated GLib-identity assertions elsewhere. + +# -- D-Bus handlers -------------------------------------------------------- + + +@pytest.mark.parametrize( + "method,expected", + [("Toggle", "on_toggle"), ("Start", "on_start"), ("Stop", "on_stop")], +) +def test_method_call_routes_to_callback(method, expected): + """A D-Bus method call invokes the matching callback on the main thread.""" + from vocalinux import dbus_service + + callbacks = { + "on_toggle": MagicMock(), + "on_start": MagicMock(), + "on_stop": MagicMock(), + } + service = dbus_service.VocalinuxDBusService( + on_toggle=callbacks["on_toggle"], + on_start=callbacks["on_start"], + on_stop=callbacks["on_stop"], + ) + + invocation = MagicMock() + # Make idle_add run the marshalled call synchronously for the assertion. + with patch.object(dbus_service.GLib, "idle_add", side_effect=lambda fn, *a: fn(*a)): + service._handle_method_call( + MagicMock(), + "sender", + dbus_service.OBJECT_PATH, + dbus_service.INTERFACE_NAME, + method, + None, + invocation, + ) + + callbacks[expected].assert_called_once_with() + for name, cb in callbacks.items(): + if name != expected: + cb.assert_not_called() + invocation.return_value.assert_called_once() + + +def test_unknown_method_returns_error(): + """An unknown method name returns a D-Bus error and no callback runs.""" + from vocalinux import dbus_service + + on_toggle = MagicMock() + service = dbus_service.VocalinuxDBusService( + on_toggle=on_toggle, on_start=MagicMock(), on_stop=MagicMock() + ) + invocation = MagicMock() + service._handle_method_call( + MagicMock(), + "sender", + dbus_service.OBJECT_PATH, + dbus_service.INTERFACE_NAME, + "Bogus", + None, + invocation, + ) + invocation.return_error_literal.assert_called_once() + on_toggle.assert_not_called() + + +def test_shutdown_releases_name_and_object(): + """shutdown() unregisters the object and unowns the bus name, idempotently.""" + from vocalinux import dbus_service + + service = dbus_service.VocalinuxDBusService( + on_toggle=MagicMock(), on_start=MagicMock(), on_stop=MagicMock() + ) + # Simulate a successful registration having happened. + connection = MagicMock() + service._connection = connection + service._registration_id = 42 + service._owner_id = 7 + + with patch.object(dbus_service.Gio, "bus_unown_name") as mock_unown: + service.shutdown() + connection.unregister_object.assert_called_once_with(42) + mock_unown.assert_called_once_with(7) + + # A second shutdown must be a no-op (ids cleared). + connection.unregister_object.reset_mock() + mock_unown.reset_mock() + service.shutdown() + connection.unregister_object.assert_not_called() + mock_unown.assert_not_called() + + +def test_send_command_uses_proxy_call(): + """send_command forwards the mapped method name through a D-Bus proxy.""" + from vocalinux import dbus_service + + mock_proxy = MagicMock() + with patch.object(dbus_service.Gio.DBusProxy, "new_for_bus_sync", return_value=mock_proxy): + result = dbus_service.send_command("toggle") + assert result is True + args = mock_proxy.call_sync.call_args.args + assert args[0] == "Toggle" + + +def test_send_command_rejects_unknown(): + """An unknown command name raises ValueError.""" + from vocalinux import dbus_service + + with pytest.raises(ValueError): + dbus_service.send_command("explode") + + +def test_send_command_returns_false_without_session_bus(): + """A generic failure (e.g. no session bus) yields False, not a traceback.""" + from vocalinux import dbus_service + + # GLib.Error is a MagicMock under the test harness, so patch it to a real + # exception class; the raised error is a different type and must fall + # through to the generic handler. + with patch.object(dbus_service.GLib, "Error", RuntimeError): + with patch.object( + dbus_service.Gio.DBusProxy, + "new_for_bus_sync", + side_effect=ValueError("no session bus"), + ): + assert dbus_service.send_command("toggle") is False + + +def test_send_command_returns_false_on_glib_error(): + """A GLib.Error from the proxy call is caught and yields False.""" + from vocalinux import dbus_service + + # Patch GLib.Error to a real class so both the raise and the `except` clause + # reference the same type and the GLib.Error branch is exercised. + with patch.object(dbus_service.GLib, "Error", RuntimeError): + with patch.object( + dbus_service.Gio.DBusProxy, + "new_for_bus_sync", + side_effect=RuntimeError("bus error"), + ): + assert dbus_service.send_command("toggle") is False + + +# -- Service lifecycle & error paths -------------------------------------- + + +def _make_service() -> "VocalinuxDBusService": + from vocalinux import dbus_service + + return dbus_service.VocalinuxDBusService( + on_toggle=MagicMock(), on_start=MagicMock(), on_stop=MagicMock() + ) + + +def test_register_handles_bus_own_name_failure(): + """A failure while requesting the bus name is logged, not raised.""" + from vocalinux import dbus_service + + # Patch GLib.Error to a real class so both the raise and the `except` clause + # reference the same type and the GLib.Error branch is exercised. + with patch.object(dbus_service.GLib, "Error", RuntimeError): + with patch.object(dbus_service.Gio, "bus_own_name", side_effect=RuntimeError("no bus")): + # __init__ calls _register(); it must not propagate the error. + service = _make_service() + assert service._owner_id == 0 + + +def test_on_bus_acquired_registers_object(): + """_on_bus_acquired stores the connection and registers the control object.""" + from vocalinux import dbus_service + + service = _make_service() + connection = MagicMock() + connection.register_object.return_value = 99 + + service._on_bus_acquired(connection, dbus_service.BUS_NAME) + + assert service._connection is connection + assert service._registration_id == 99 + connection.register_object.assert_called_once() + + +def test_on_bus_acquired_handles_register_failure(): + """A failed object registration is caught and leaves the id unset.""" + from vocalinux import dbus_service + + service = _make_service() + connection = MagicMock() + connection.register_object.side_effect = RuntimeError("register failed") + + # Patch GLib.Error to a real class so both the raise and the `except` clause + # reference the same type and the GLib.Error branch is exercised. + with patch.object(dbus_service.GLib, "Error", RuntimeError): + service._on_bus_acquired(connection, "com.vocalinux.Vocalinux") + + assert service._registration_id == 0 + + +def test_name_acquired_and_lost_are_safe_to_call(): + """The name-acquired/lost callbacks only log and never raise.""" + service = _make_service() + connection = MagicMock() + service._on_name_acquired(connection, "com.vocalinux.Vocalinux") + service._on_name_lost(connection, "com.vocalinux.Vocalinux") + + +def test_invoke_swallows_callback_errors(): + """_invoke logs a raising callback and still returns SOURCE_REMOVE.""" + from vocalinux import dbus_service + + failing = MagicMock(side_effect=RuntimeError("boom")) + result = dbus_service.VocalinuxDBusService._invoke(failing) + failing.assert_called_once_with() + assert result == dbus_service.GLib.SOURCE_REMOVE + + +def test_shutdown_swallows_unregister_and_unown_errors(): + """shutdown() ignores failures from unregister_object and bus_unown_name.""" + from vocalinux import dbus_service + + service = _make_service() + connection = MagicMock() + connection.unregister_object.side_effect = RuntimeError("unregister failed") + service._connection = connection + service._registration_id = 42 + service._owner_id = 7 + + # Patch GLib.Error to a real class so both the raise and the `except` clause + # reference the same type and the GLib.Error branch is exercised. + with patch.object(dbus_service.GLib, "Error", RuntimeError): + with patch.object( + dbus_service.Gio, "bus_unown_name", side_effect=RuntimeError("unown failed") + ): + # Must not raise despite both cleanup calls failing. + service.shutdown() + + assert service._registration_id == 0 + assert service._owner_id == 0 diff --git a/tests/test_settings_shortcuts.py b/tests/test_settings_shortcuts.py index 72a2215..ef76e7d 100644 --- a/tests/test_settings_shortcuts.py +++ b/tests/test_settings_shortcuts.py @@ -8,7 +8,7 @@ import os import unittest -def _get_source_code(): +def _get_source_code() -> str: """Read the settings dialog source file.""" source_path = os.path.join( os.path.dirname(__file__), @@ -253,5 +253,64 @@ class TestConfigManagerShortcuts(unittest.TestCase): self.assertEqual(config.get("shortcuts", "mode"), "push_to_talk") +class TestExternalActivationToggle(unittest.TestCase): + """Source-inspection tests for the external-activation switch. + + The dialog subclasses a mocked Gtk.Dialog, so its methods can't be called + directly; we assert the wiring is present in source (same approach as + TestSettingsDialogShortcutsSection above). + """ + + def setUp(self): + self.source_code = _get_source_code() + + def test_switch_widget_created(self): + """The external-activation switch is created in the shortcuts section.""" + self.assertIn("self.disable_internal_hotkey_switch = Gtk.Switch()", self.source_code) + + def test_switch_signal_connected(self): + """The switch is wired to its state-set handler.""" + self.assertIn("self.disable_internal_hotkey_switch.connect(", self.source_code) + self.assertIn("self._on_disable_internal_hotkey_toggled", self.source_code) + + def test_handler_persists_config_key(self): + """The handler saves the disable_internal_hotkey flag.""" + self.assertIn( + 'self.config_manager.set("shortcuts", "disable_internal_hotkey", disabled)', + self.source_code, + ) + self.assertIn("self.config_manager.save_settings()", self.source_code) + + def test_handler_live_applies_via_callback(self): + """Toggling live-applies by invoking the listener update callback.""" + self.assertIn("self.hotkey_listener_update_callback()", self.source_code) + + def test_handler_guards_during_initialization(self): + """The handler is inert while the dialog is initializing/applying.""" + self.assertIn( + "def _on_disable_internal_hotkey_toggled(self, widget: Gtk.Switch, state: bool)" + " -> bool:", + self.source_code, + ) + self.assertIn("if self._initializing or self._applying_settings:", self.source_code) + + def test_sensitivity_helper_greys_internal_rows(self): + """External activation greys out the built-in shortcut rows.""" + self.assertIn( + "def _update_internal_hotkey_sensitivity(self, disabled: bool) -> None:", + self.source_code, + ) + self.assertIn( + "for row in (self.mode_row, self.shortcut_row, self.custom_shortcut_row):", + self.source_code, + ) + self.assertIn("row.set_sensitive(not disabled)", self.source_code) + + def test_initial_state_loaded(self): + """The switch's initial state is loaded from config during populate.""" + self.assertIn("self.disable_internal_hotkey_switch.set_active(", self.source_code) + self.assertIn('"shortcuts", "disable_internal_hotkey", False', self.source_code) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_suspend_handler.py b/tests/test_suspend_handler.py index 428d528..c106c63 100644 --- a/tests/test_suspend_handler.py +++ b/tests/test_suspend_handler.py @@ -302,6 +302,9 @@ class TestTrayIndicatorResumeFlow(unittest.TestCase): indicator = TrayIndicator.__new__(TrayIndicator) indicator.speech_engine = MagicMock() + indicator.config_manager = MagicMock() + indicator.config_manager.get_bool.return_value = False + indicator._external_activation_unavailable = False indicator._setup_keyboard_shortcuts = MagicMock() indicator._input_monitor = None indicator._settle_timer_id = None @@ -343,6 +346,19 @@ class TestTrayIndicatorResumeFlow(unittest.TestCase): assert calls[1][0][0] == 2 assert calls[1][0][1] == indicator._start_input_device_monitor + @patch("vocalinux.ui.tray_indicator.GLib") + def test_on_system_resume_skips_monitor_for_external_activation(self, mock_glib): + """External activation promises no /dev/input access; a resume must + not open the device monitor that (re)starts the internal listener.""" + indicator = self._make_tray_indicator() + indicator.config_manager.get_bool.return_value = True + + indicator._on_system_resume() + + calls = mock_glib.timeout_add_seconds.call_args_list + assert len(calls) == 1 + assert calls[0][0][1] == indicator._reinit_speech_after_resume + @patch("vocalinux.ui.tray_indicator.Gio") @patch("vocalinux.ui.tray_indicator.GLib") def test_start_input_device_monitor_sets_up_gio_monitor(self, mock_glib, mock_gio): @@ -497,6 +513,7 @@ class TestTrayAutoPauseWiring(unittest.TestCase): indicator.speech_engine.reinitialize_after_resume = MagicMock() indicator.speech_engine.is_auto_paused = False indicator.config_manager = MagicMock() + indicator._external_activation_unavailable = False indicator._setup_keyboard_shortcuts = MagicMock() indicator._input_monitor = None indicator._settle_timer_id = None @@ -565,6 +582,7 @@ class TestTrayAutoPauseWiring(unittest.TestCase): def test_system_resume_skips_reinit_when_auto_paused(self, mock_glib): indicator = self._make_tray_indicator() indicator.speech_engine.is_auto_paused = True + indicator.config_manager.get_bool.return_value = False indicator._start_input_device_monitor = MagicMock() indicator._on_system_resume() @@ -578,6 +596,7 @@ class TestTrayAutoPauseWiring(unittest.TestCase): def test_system_resume_reinits_when_not_auto_paused(self, mock_glib): indicator = self._make_tray_indicator() indicator.speech_engine.is_auto_paused = False + indicator.config_manager.get_bool.return_value = False indicator._reinit_speech_after_resume = MagicMock() indicator._start_input_device_monitor = MagicMock() diff --git a/tests/test_tray_indicator_ext.py b/tests/test_tray_indicator_ext.py index cacde70..41c6f02 100644 --- a/tests/test_tray_indicator_ext.py +++ b/tests/test_tray_indicator_ext.py @@ -513,5 +513,207 @@ class TestTrayIndicatorSignalHandling(unittest.TestCase): assert isinstance(result, bool) +class TestExternalActivationGate(unittest.TestCase): + """Tests for the D-Bus external-activation config gate. + + These live here (and not in test_dbus_activation.py) so that the real + tray_indicator module is imported late in the session -- after the tests + that swap sys.modules["gi.repository"] at runtime. Importing it earlier + would freeze tray_indicator.GLib against a stale mock and break unrelated + GLib-identity assertions (e.g. in test_suspend_handler.py). + """ + + @staticmethod + def _fake_indicator(active: bool) -> MagicMock: + """Minimal stand-in for calling _setup_keyboard_shortcuts in isolation. + + ``active`` stubs _external_activation_active() directly rather than + the config it normally derives from -- that derivation (including the + once-failed-stays-off persistence) has its own tests below. + """ + fake = MagicMock() + fake.shortcut_manager.active = False + fake.config_manager.get_str.return_value = "toggle" + fake._external_activation_active.return_value = active + return fake + + def test_external_mode_skips_listener_start(self): + """With the option enabled, the evdev/pynput listener is not started.""" + from vocalinux.ui.tray_indicator import TrayIndicator + + fake = self._fake_indicator(active=True) + TrayIndicator._setup_keyboard_shortcuts(fake) + fake.shortcut_manager.start.assert_not_called() + fake.shortcut_manager.register_toggle_callback.assert_called_once_with(None) + + def test_default_mode_starts_listener(self): + """With the option disabled (default), the listener starts as before.""" + from vocalinux.ui.tray_indicator import TrayIndicator + + fake = self._fake_indicator(active=False) + TrayIndicator._setup_keyboard_shortcuts(fake) + fake.shortcut_manager.start.assert_called_once() + + def test_reconfigure_stops_active_recognition_first(self): + """An active push-to-talk session is stopped before the listener that + would have delivered its release callback is torn down, so it cannot + be left running with no way back to idle.""" + from vocalinux.common_types import RecognitionState + from vocalinux.ui.tray_indicator import TrayIndicator + + fake = self._fake_indicator(active=True) + fake.speech_engine.state = RecognitionState.LISTENING + TrayIndicator._setup_keyboard_shortcuts(fake) + fake._stop_recognition.assert_called_once_with() + + def test_reconfigure_leaves_idle_recognition_alone(self): + """No spurious stop is issued when nothing is active.""" + from vocalinux.common_types import RecognitionState + from vocalinux.ui.tray_indicator import TrayIndicator + + fake = self._fake_indicator(active=False) + fake.speech_engine.state = RecognitionState.IDLE + TrayIndicator._setup_keyboard_shortcuts(fake) + fake._stop_recognition.assert_not_called() + + +class TestExternalActivationActiveGate(unittest.TestCase): + """Tests for _external_activation_active(): the saved setting, unless the + D-Bus service has already failed to register this run.""" + + @staticmethod + def _fake_indicator(disable_internal_hotkey: bool, unavailable: bool) -> MagicMock: + fake = MagicMock() + fake.config_manager.get_bool.return_value = disable_internal_hotkey + fake._external_activation_unavailable = unavailable + return fake + + def test_true_when_configured_and_available(self): + from vocalinux.ui.tray_indicator import TrayIndicator + + fake = self._fake_indicator(disable_internal_hotkey=True, unavailable=False) + assert TrayIndicator._external_activation_active(fake) is True + + def test_false_when_not_configured(self): + from vocalinux.ui.tray_indicator import TrayIndicator + + fake = self._fake_indicator(disable_internal_hotkey=False, unavailable=False) + assert TrayIndicator._external_activation_active(fake) is False + + def test_false_once_unavailable_even_if_still_configured(self): + """Once the service has failed, the saved setting is overridden for + the rest of the process -- it does not retry.""" + from vocalinux.ui.tray_indicator import TrayIndicator + + fake = self._fake_indicator(disable_internal_hotkey=True, unavailable=True) + assert TrayIndicator._external_activation_active(fake) is False + + +class TestDBusRegistrationFailedFallback(unittest.TestCase): + """Tests for falling back to the internal listener when the D-Bus service + could not register, so the user is never left with no activation path.""" + + @staticmethod + def _fake_indicator(disable_internal_hotkey: bool) -> MagicMock: + fake = MagicMock() + fake.config_manager.get_bool.return_value = disable_internal_hotkey + return fake + + @patch("vocalinux.ui.tray_indicator.notifications") + def test_falls_back_when_external_activation_configured(self, mock_notifications): + from vocalinux.ui.tray_indicator import TrayIndicator + + fake = self._fake_indicator(disable_internal_hotkey=True) + TrayIndicator._on_dbus_registration_failed(fake) + assert fake._external_activation_unavailable is True + fake._setup_keyboard_shortcuts.assert_called_once_with() + mock_notifications.notify.assert_called_once() + + @patch("vocalinux.ui.tray_indicator.notifications") + def test_fallback_persists_across_a_later_reconfigure(self, mock_notifications): + """The exact regression this fixes: a reconfigure after the fallback + (mode change, settings toggle, resume) must not re-disable the only + working activation path just because the saved setting still asks + for external activation.""" + from vocalinux.ui.tray_indicator import TrayIndicator + + with patch.object(TrayIndicator, "__init__", lambda self: None): + indicator = TrayIndicator.__new__(TrayIndicator) + indicator._external_activation_unavailable = False + indicator.config_manager = MagicMock() + indicator.config_manager.get_bool.return_value = True + indicator.config_manager.get_str.return_value = "toggle" + indicator.speech_engine = MagicMock() + indicator.shortcut_manager = MagicMock() + indicator.shortcut_manager.active = False + + indicator._on_dbus_registration_failed() + indicator.shortcut_manager.start.assert_called_once() + + # A later reconfigure -- e.g. the mode combo changing in Settings -- + # calls this with no knowledge of the earlier failure. + indicator.shortcut_manager.reset_mock() + indicator._setup_keyboard_shortcuts() + indicator.shortcut_manager.start.assert_called_once() + + def test_noop_when_internal_listener_already_active(self): + """Nothing to fall back to/from if the internal listener was already + the active mode.""" + from vocalinux.ui.tray_indicator import TrayIndicator + + fake = self._fake_indicator(disable_internal_hotkey=False) + TrayIndicator._on_dbus_registration_failed(fake) + fake._setup_keyboard_shortcuts.assert_not_called() + + +class TestExternalTriggerHandlers(unittest.TestCase): + """Tests for the D-Bus external Start/Stop handlers on TrayIndicator. + + These use normal (non push-to-talk) start semantics and guard on the + current recognition state, so repeated `vocalinux --start`/`--stop` calls + are idempotent. + """ + + def test_external_start_starts_when_idle(self): + """_external_start starts recognition only when currently idle.""" + from vocalinux.common_types import RecognitionState + from vocalinux.ui.tray_indicator import TrayIndicator + + fake = MagicMock() + fake.speech_engine.state = RecognitionState.IDLE + TrayIndicator._external_start(fake) + fake.speech_engine.start_recognition.assert_called_once_with() + + def test_external_start_noop_when_already_running(self): + """_external_start does nothing if recognition is already active.""" + from vocalinux.common_types import RecognitionState + from vocalinux.ui.tray_indicator import TrayIndicator + + fake = MagicMock() + fake.speech_engine.state = RecognitionState.LISTENING + TrayIndicator._external_start(fake) + fake.speech_engine.start_recognition.assert_not_called() + + def test_external_stop_stops_when_active(self): + """_external_stop stops recognition when it is not idle.""" + from vocalinux.common_types import RecognitionState + from vocalinux.ui.tray_indicator import TrayIndicator + + fake = MagicMock() + fake.speech_engine.state = RecognitionState.LISTENING + TrayIndicator._external_stop(fake) + fake.speech_engine.stop_recognition.assert_called_once_with() + + def test_external_stop_noop_when_idle(self): + """_external_stop does nothing if recognition is already idle.""" + from vocalinux.common_types import RecognitionState + from vocalinux.ui.tray_indicator import TrayIndicator + + fake = MagicMock() + fake.speech_engine.state = RecognitionState.IDLE + TrayIndicator._external_stop(fake) + fake.speech_engine.stop_recognition.assert_not_called() + + if __name__ == "__main__": unittest.main()