feat(activation): D-Bus activation for compositor global shortcuts (#568)

Rebased onto latest main (conflict-resolution squash for PR #568).

Co-authored-by: Alexander <ab@digital-benefits.de>
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
Web-Benefits
2026-09-29 21:19:29 -07:00
committed by GitHub
co-authored by Alexander Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
parent c0217a8aec
commit db07590761
10 changed files with 1077 additions and 9 deletions
+49
View File
@@ -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 |
+208
View File
@@ -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 = """
<node>
<interface name="com.vocalinux.Vocalinux">
<method name="Toggle"/>
<method name="Start"/>
<method name="Stop"/>
</interface>
</node>
"""
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
+61 -5
View File
@@ -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")
+4
View File
@@ -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",
+63 -2
View File
@@ -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
+95 -1
View File
@@ -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
+316
View File
@@ -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
+60 -1
View File
@@ -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()
+19
View File
@@ -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()
+202
View File
@@ -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()