soubarnak
74ea855220
Alongside Toggle/Start/Stop, the renamed interface now publishes dictation state (idle/listening/processing) as a StateChanged signal and a readable State property, so status bars and scripts can follow dictation without scraping the tray. Signals fire only on change, so double start/stop emits no duplicates; command-line triggers behave exactly as before. Refs #21
459 lines
16 KiB
Python
459 lines
16 KiB
Python
"""
|
|
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 voculinux 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", ["voculinux", 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", ["voculinux", "--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 voculinux 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", ["voculinux"]):
|
|
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.voculinux.Voculinux")
|
|
|
|
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.voculinux.Voculinux")
|
|
service._on_name_lost(connection, "com.voculinux.Voculinux")
|
|
|
|
|
|
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
|
|
|
|
|
|
# -- Dictation state signal & property (ticket #21) -------------------------
|
|
|
|
|
|
def _connected_service():
|
|
"""A service with a mocked bus connection for emission assertions."""
|
|
from vocalinux import dbus_service
|
|
|
|
service = _make_service()
|
|
service._connection = MagicMock()
|
|
return dbus_service, service
|
|
|
|
|
|
def test_state_starts_idle():
|
|
"""Reading the state property right after start returns idle."""
|
|
_dbus_service, service = _connected_service()
|
|
assert service.get_state() == "idle"
|
|
|
|
|
|
def test_full_dictation_cycle_emits_each_transition():
|
|
"""idle -> listening -> processing -> idle emits one signal per change."""
|
|
dbus_service, service = _connected_service()
|
|
|
|
assert service.set_state("listening") is True
|
|
assert service.set_state("processing") is True
|
|
assert service.set_state("idle") is True
|
|
|
|
emitted = [call.args[3] for call in service._connection.emit_signal.call_args_list]
|
|
assert emitted == ["StateChanged", "StateChanged", "StateChanged"]
|
|
assert service.get_state() == "idle"
|
|
|
|
|
|
def test_repeat_state_does_not_emit_duplicate():
|
|
"""Starting twice or stopping while idle emits nothing further."""
|
|
_dbus_service, service = _connected_service()
|
|
|
|
assert service.set_state("listening") is True
|
|
assert service.set_state("listening") is False
|
|
assert service.set_state("idle") is True
|
|
assert service.set_state("idle") is False
|
|
|
|
assert service._connection.emit_signal.call_count == 2
|
|
assert service.get_state() == "idle"
|
|
|
|
|
|
def test_unknown_state_is_ignored():
|
|
"""An unknown state never changes the property nor emits."""
|
|
_dbus_service, service = _connected_service()
|
|
|
|
assert service.set_state("bogus") is False
|
|
|
|
assert service.get_state() == "idle"
|
|
service._connection.emit_signal.assert_not_called()
|
|
|
|
|
|
def test_state_from_recognition_maps_all_states():
|
|
"""Engine states map to bus states; error reads back as idle."""
|
|
from vocalinux import dbus_service
|
|
from vocalinux.common_types import RecognitionState
|
|
|
|
assert dbus_service.state_from_recognition(RecognitionState.IDLE) == "idle"
|
|
assert dbus_service.state_from_recognition(RecognitionState.LISTENING) == "listening"
|
|
assert dbus_service.state_from_recognition(RecognitionState.PROCESSING) == "processing"
|
|
assert dbus_service.state_from_recognition(RecognitionState.ERROR) == "idle"
|
|
|
|
|
|
def test_properties_get_returns_current_state():
|
|
"""A Properties.Get call for State returns the live value."""
|
|
dbus_service, service = _connected_service()
|
|
service.set_state("listening")
|
|
|
|
invocation = MagicMock()
|
|
parameters = MagicMock()
|
|
parameters.unpack.return_value = (dbus_service.INTERFACE_NAME, "State")
|
|
service._handle_method_call(
|
|
MagicMock(),
|
|
"sender",
|
|
dbus_service.OBJECT_PATH,
|
|
dbus_service.PROPERTIES_INTERFACE,
|
|
"Get",
|
|
parameters,
|
|
invocation,
|
|
)
|
|
|
|
invocation.return_value.assert_called_once()
|
|
assert service.get_state() == "listening"
|
|
|
|
|
|
def test_properties_get_unknown_returns_error():
|
|
"""A Properties.Get call for another interface errors without emitting."""
|
|
dbus_service, service = _connected_service()
|
|
|
|
invocation = MagicMock()
|
|
parameters = MagicMock()
|
|
parameters.unpack.return_value = ("org.example.Other", "State")
|
|
service._handle_method_call(
|
|
MagicMock(),
|
|
"sender",
|
|
dbus_service.OBJECT_PATH,
|
|
dbus_service.PROPERTIES_INTERFACE,
|
|
"Get",
|
|
parameters,
|
|
invocation,
|
|
)
|
|
|
|
invocation.return_error_literal.assert_called_once()
|
|
service._connection.emit_signal.assert_not_called()
|
|
|
|
|
|
def test_control_methods_still_dispatch_after_state_changes():
|
|
"""Toggle/Start/Stop keep working exactly as before (ticket #21)."""
|
|
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"],
|
|
)
|
|
service._connection = MagicMock()
|
|
service.set_state("processing")
|
|
|
|
invocation = MagicMock()
|
|
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,
|
|
"Stop",
|
|
None,
|
|
invocation,
|
|
)
|
|
|
|
callbacks["on_stop"].assert_called_once_with()
|
|
invocation.return_value.assert_called_once()
|
|
assert service.get_state() == "processing"
|