xavierk/voculinux
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity
Files
1ae453e7d9b10adea17308993ab9b276c6298fb7
voculinux/tests/test_suspend_handler.py
T
MichaelGitHubJatin K Malik
 e66bd218d9 fix(audio): prevent dictation hangs on stalled microphones (#974) 
* fix(audio): bound capture reads and support cancellation

* fix(recognition): bound and serialize capture shutdown

* fix(recognition): cancel retries and report stalled microphones

* fix(audio): prefer pulse when the ALSA default is unavailable

* fix(recognition): guard late callbacks and finalize capture tests

* docs(audio): explain input selection and stall recovery

* fix(speech): enforce auto-pause across buffered reload timeouts

* fix(speech): recover auto-resume after reload timeout

* test(speech): fix auto-pause follow-up test regressions

* fix(speech): retain auto-resume after reconfigure timeout

* fix(tray): preserve auto-pause recovery across suspend

* test(capture-shutdown): give the bare tray the suspend path's shortcut managers

_on_system_suspend now releases the main and per-language hotkey
listeners before sleep (#1025); the fixture builds TrayIndicator without
__init__, so it needs the two attributes present.

* fix(tray): release hotkey listeners on suspend even while auto-paused

Merging #1025's suspend cleanup with this branch's auto-pause handoff
left an early return that skipped shortcut_manager.stop() whenever a
configured game was active, keeping EVIOCGRABs held through sleep. The
paused branch now only owns the model-lifecycle work; listener release
and stopping active recognition run for every suspend.

* fix(tray): hold the apply lock only around the download's verify and save

Wrapping the whole recommended-model download in _apply_settings_lock
blocked every GTK-thread settings apply (e.g. Test Dictation) for the
length of the transfer. The engine already serializes reconfigures, so
the lock only needs to cover verifying the download still owns the
receipt and saving it; an apply that lands mid-download now proceeds
instead of freezing the app.

---------

Co-authored-by: Jatin K Malik <jatinkrmalik@gmail.com>
2026-10-26 21:25:50 -07:00

864 lines
33 KiB
Python
Raw Blame History

"""
Tests for suspend/resume handling.
Covers SuspendHandler and the reinitialize_after_resume path in
SpeechRecognitionManager.
"""
import sys
import threading
import unittest
from unittest.mock import MagicMock, patch
mock_pywhispercpp = MagicMock()
mock_pywhispercpp.model = MagicMock()
mock_pywhispercpp.model.Model = MagicMock()
sys.modules["vosk"] = MagicMock()
sys.modules["whisper"] = MagicMock()
sys.modules["pywhispercpp"] = mock_pywhispercpp
sys.modules["pywhispercpp.model"] = mock_pywhispercpp.model
sys.modules["requests"] = MagicMock()
sys.modules["pyaudio"] = MagicMock()
sys.modules["wave"] = MagicMock()
sys.modules["tqdm"] = MagicMock()
sys.modules["numpy"] = MagicMock()
sys.modules["torch"] = MagicMock()
sys.modules["psutil"] = MagicMock()
from conftest import mock_audio_feedback # noqa: E402
from vocalinux.common_types import RecognitionState # noqa: E402
from vocalinux.suspend_handler import SuspendHandler # noqa: E402
class TestSuspendHandlerInit(unittest.TestCase):
"""Tests for SuspendHandler initialization."""
@patch("vocalinux.suspend_handler.Gio")
def test_init_connects_to_logind(self, mock_gio):
mock_proxy = MagicMock()
mock_gio.DBusProxy.new_for_bus_sync.return_value = mock_proxy
handler = SuspendHandler()
mock_gio.DBusProxy.new_for_bus_sync.assert_called_once_with(
bus_type=mock_gio.BusType.SYSTEM,
flags=mock_gio.DBusProxyFlags.NONE,
info=None,
name="org.freedesktop.login1",
object_path="/org/freedesktop/login1",
interface_name="org.freedesktop.login1.Manager",
cancellable=None,
)
mock_proxy.connect.assert_called_once_with("g-signal", handler._on_signal)
assert handler.active is True
@patch("vocalinux.suspend_handler.Gio")
def test_init_handles_dbus_unavailable(self, mock_gio):
mock_gio.DBusProxy.new_for_bus_sync.side_effect = Exception("no logind")
handler = SuspendHandler()
assert handler.active is False
@patch("vocalinux.suspend_handler.Gio")
def test_shutdown_disconnects(self, mock_gio):
mock_proxy = MagicMock()
mock_gio.DBusProxy.new_for_bus_sync.return_value = mock_proxy
handler = SuspendHandler()
handler.shutdown()
mock_proxy.disconnect_by_func.assert_called_once_with(handler._on_signal)
assert handler.active is False
@patch("vocalinux.suspend_handler.Gio")
def test_shutdown_when_not_connected(self, mock_gio):
from gi.repository import GLib
mock_gio.DBusProxy.new_for_bus_sync.side_effect = GLib.Error("no logind")
handler = SuspendHandler()
handler.shutdown()
class TestSuspendHandlerCallbacks(unittest.TestCase):
"""Tests for suspend/resume callback invocation."""
@patch("vocalinux.suspend_handler.Gio")
def test_suspend_callback_invoked(self, mock_gio):
mock_proxy = MagicMock()
mock_gio.DBusProxy.new_for_bus_sync.return_value = mock_proxy
suspend_cb = MagicMock()
resume_cb = MagicMock()
handler = SuspendHandler(on_suspend=suspend_cb, on_resume=resume_cb)
# Simulate PrepareForSleep(true)
mock_params = MagicMock()
mock_params.get_type_string.return_value = "(b)"
mock_params.unpack.return_value = (True,)
handler._on_signal(mock_proxy, "org.freedesktop.login1", "PrepareForSleep", mock_params)
suspend_cb.assert_called_once()
resume_cb.assert_not_called()
@patch("vocalinux.suspend_handler.Gio")
def test_resume_callback_invoked(self, mock_gio):
mock_proxy = MagicMock()
mock_gio.DBusProxy.new_for_bus_sync.return_value = mock_proxy
suspend_cb = MagicMock()
resume_cb = MagicMock()
handler = SuspendHandler(on_suspend=suspend_cb, on_resume=resume_cb)
mock_params = MagicMock()
mock_params.get_type_string.return_value = "(b)"
mock_params.unpack.return_value = (True,)
handler._on_signal(mock_proxy, "org.freedesktop.login1", "PrepareForSleep", mock_params)
@patch("vocalinux.suspend_handler.Gio")
def test_resume_callback_invoked(self, mock_gio):
mock_proxy = MagicMock()
mock_gio.DBusProxy.new_for_bus_sync.return_value = mock_proxy
suspend_cb = MagicMock()
resume_cb = MagicMock()
handler = SuspendHandler(on_suspend=suspend_cb, on_resume=resume_cb)
mock_params = MagicMock()
mock_params.get_type_string.return_value = "(b)"
mock_params.unpack.return_value = (False,)
handler._on_signal(mock_proxy, "org.freedesktop.login1", "PrepareForSleep", mock_params)
resume_cb.assert_called_once()
suspend_cb.assert_not_called()
@patch("vocalinux.suspend_handler.Gio")
def test_ignores_other_signals(self, mock_gio):
mock_proxy = MagicMock()
mock_gio.DBusProxy.new_for_bus_sync.return_value = mock_proxy
suspend_cb = MagicMock()
resume_cb = MagicMock()
handler = SuspendHandler(on_suspend=suspend_cb, on_resume=resume_cb)
mock_params = MagicMock()
handler._on_signal(mock_proxy, "org.freedesktop.login1", "SomeOtherSignal", mock_params)
suspend_cb.assert_not_called()
resume_cb.assert_not_called()
@patch("vocalinux.suspend_handler.Gio")
def test_ignores_unexpected_param_type(self, mock_gio):
mock_proxy = MagicMock()
mock_gio.DBusProxy.new_for_bus_sync.return_value = mock_proxy
suspend_cb = MagicMock()
resume_cb = MagicMock()
handler = SuspendHandler(on_suspend=suspend_cb, on_resume=resume_cb)
mock_params = MagicMock()
mock_params.get_type_string.return_value = "(s)"
handler._on_signal(mock_proxy, "org.freedesktop.login1", "PrepareForSleep", mock_params)
suspend_cb.assert_not_called()
resume_cb.assert_not_called()
@patch("vocalinux.suspend_handler.Gio")
def test_callback_exception_is_caught(self, mock_gio):
mock_proxy = MagicMock()
mock_gio.DBusProxy.new_for_bus_sync.return_value = mock_proxy
suspend_cb = MagicMock(side_effect=RuntimeError("boom"))
handler = SuspendHandler(on_suspend=suspend_cb)
mock_params = MagicMock()
mock_params.get_type_string.return_value = "(b)"
mock_params.unpack.return_value = (True,)
handler._on_signal(mock_proxy, "org.freedesktop.login1", "PrepareForSleep", mock_params)
@patch("vocalinux.suspend_handler.Gio")
def test_no_callback_does_not_crash(self, mock_gio):
mock_proxy = MagicMock()
mock_gio.DBusProxy.new_for_bus_sync.return_value = mock_proxy
handler = SuspendHandler()
mock_params = MagicMock()
mock_params.get_type_string.return_value = "(b)"
mock_params.unpack.return_value = (True,)
handler._on_signal(mock_proxy, "org.freedesktop.login1", "PrepareForSleep", mock_params)
class TestReinitializeAfterResume(unittest.TestCase):
"""Tests for SpeechRecognitionManager.reinitialize_after_resume."""
def _make_manager(self):
from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager
with patch.object(SpeechRecognitionManager, "__init__", lambda self: None):
mgr = SpeechRecognitionManager()
mgr.engine = "vosk"
mgr.state = RecognitionState.IDLE
mgr.model = MagicMock()
mgr.recognizer = MagicMock()
mgr._model_initialized = True
mgr._model_lock = MagicMock()
mgr._model_lock.__enter__ = MagicMock(return_value=None)
mgr._model_lock.__exit__ = MagicMock(return_value=False)
mgr._http_session = None
mgr._faster_whisper_engine = None
mgr._reconnection_attempts = 3
mgr._update_state = MagicMock()
mgr._init_vosk = MagicMock()
mgr._init_whisper = MagicMock()
mgr._init_whispercpp = MagicMock()
mgr._init_parakeet = MagicMock()
mgr._init_faster_whisper = MagicMock()
mgr.stop_recognition = MagicMock()
return mgr
def test_reinitializes_vosk_engine(self):
mgr = self._make_manager()
mgr.reinitialize_after_resume()
mgr._init_vosk.assert_called_once()
assert mgr._reconnection_attempts == 0
def test_reinitializes_whisper_engine(self):
mgr = self._make_manager()
mgr.engine = "whisper"
mgr.reinitialize_after_resume()
mgr._init_whisper.assert_called_once()
def test_reinitializes_whispercpp_engine(self):
mgr = self._make_manager()
mgr.engine = "whisper_cpp"
mgr.reinitialize_after_resume()
mgr._init_whispercpp.assert_called_once()
def test_reinitializes_parakeet_engine(self):
mgr = self._make_manager()
mgr.engine = "parakeet"
mgr.reinitialize_after_resume()
mgr._init_parakeet.assert_called_once()
def test_reinitializes_faster_whisper_engine(self):
mgr = self._make_manager()
mgr.engine = "faster_whisper"
mgr.reinitialize_after_resume()
mgr._init_faster_whisper.assert_called_once()
def test_stops_active_recognition_first(self):
mgr = self._make_manager()
mgr.state = RecognitionState.LISTENING
mgr.reinitialize_after_resume()
mgr.stop_recognition.assert_called_once()
def test_clears_model_state(self):
mgr = self._make_manager()
mgr.reinitialize_after_resume()
assert mgr.model is None
assert mgr.recognizer is None
assert mgr._model_initialized is False
def test_sets_error_state_on_failure(self):
mgr = self._make_manager()
mgr._init_vosk.side_effect = RuntimeError("model load failed")
mgr.reinitialize_after_resume()
mgr._update_state.assert_called_once_with(
RecognitionState.ERROR, model_request=mgr._model_request
)
class TestTrayIndicatorResumeFlow(unittest.TestCase):
"""Tests for TrayIndicator resume handling."""
def _make_tray_indicator(self):
from vocalinux.ui.tray_indicator import TrayIndicator
with patch.object(TrayIndicator, "__init__", lambda self: None):
indicator = TrayIndicator.__new__(TrayIndicator)
indicator.speech_engine = MagicMock()
indicator.speech_engine.prepare_auto_resume.return_value = object()
indicator.speech_engine.is_auto_paused = False
indicator._auto_pause_monitor = None
indicator.config_manager = MagicMock()
indicator.config_manager.get_bool.return_value = False
indicator._external_activation_unavailable = False
indicator._setup_keyboard_shortcuts = MagicMock()
indicator.shortcut_manager = MagicMock()
indicator._language_shortcut_managers = []
indicator._input_monitor = None
indicator._settle_timer_id = None
indicator._monitor_timeout_id = None
return indicator
def test_reinit_speech_calls_engine_reinit(self):
indicator = self._make_tray_indicator()
indicator._reinit_speech_after_resume()
indicator.speech_engine.reinitialize_after_resume.assert_called_once_with(
auto_resume=True, expected_request=None
)
def test_reinit_speech_returns_source_remove(self):
indicator = self._make_tray_indicator()
result = indicator._reinit_speech_after_resume()
assert result is False
def test_reinit_speech_failure_is_caught(self):
indicator = self._make_tray_indicator()
indicator.speech_engine.reinitialize_after_resume.side_effect = RuntimeError("boom")
indicator._reinit_speech_after_resume()
def test_reinit_speech_uses_captured_request(self) -> None:
indicator = self._make_tray_indicator()
request = indicator.speech_engine.prepare_auto_resume.return_value
indicator._reinit_speech_after_resume(request)
indicator.speech_engine.reinitialize_after_resume.assert_called_once_with(
auto_resume=True, expected_request=request
)
def test_reinit_speech_skips_new_game_pause(self) -> None:
indicator = self._make_tray_indicator()
indicator._auto_pause_monitor = MagicMock(paused=True)
indicator._model_keepalive = MagicMock()
indicator._reinit_speech_after_resume(object())
indicator.speech_engine.reinitialize_after_resume.assert_not_called()
indicator._model_keepalive.bump.assert_not_called()
def test_reinit_speech_skips_new_suspend(self) -> None:
indicator = self._make_tray_indicator()
indicator._model_keepalive = MagicMock()
indicator._on_system_suspend()
indicator._reinit_speech_after_resume(object())
indicator.speech_engine.reinitialize_after_resume.assert_not_called()
indicator._model_keepalive.bump.assert_not_called()
indicator.speech_engine.cancel_pending_auto_pause.assert_called_once_with()
def test_suspend_stops_active_recognition(self) -> None:
indicator = self._make_tray_indicator()
indicator.speech_engine.state = RecognitionState.LISTENING
indicator._model_keepalive = MagicMock()
indicator._on_system_suspend()
assert indicator._system_suspended
indicator.speech_engine.stop_recognition.assert_called_once_with()
indicator._model_keepalive.cancel.assert_called_once_with()
@patch("vocalinux.ui.tray_indicator.GLib")
def test_on_system_resume_schedules_speech_and_monitor(self, mock_glib):
indicator = self._make_tray_indicator()
indicator._on_system_resume()
calls = mock_glib.timeout_add_seconds.call_args_list
assert len(calls) == 2
assert calls[0][0][0] == 2
assert calls[0][0][1] == indicator._reinit_speech_after_resume
assert calls[0][0][2] is indicator.speech_engine.prepare_auto_resume.return_value
indicator.speech_engine.prepare_auto_resume.assert_called_once_with()
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
assert calls[0][0][2] is indicator.speech_engine.prepare_auto_resume.return_value
@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):
indicator = self._make_tray_indicator()
mock_file = MagicMock()
mock_gio.File.new_for_path.return_value = mock_file
mock_monitor = MagicMock()
mock_file.monitor_directory.return_value = mock_monitor
result = indicator._start_input_device_monitor()
mock_gio.File.new_for_path.assert_called_once_with("/dev/input")
mock_file.monitor_directory.assert_called_once_with(mock_gio.FileMonitorFlags.NONE, None)
mock_monitor.connect.assert_called_once_with("changed", indicator._on_input_device_changed)
assert indicator._input_monitor is mock_monitor
assert mock_glib.timeout_add_seconds.call_count == 2
assert result == mock_glib.SOURCE_REMOVE
@patch("vocalinux.ui.tray_indicator.Gio")
@patch("vocalinux.ui.tray_indicator.GLib")
def test_start_input_device_monitor_falls_back_on_failure(self, mock_glib, mock_gio):
indicator = self._make_tray_indicator()
mock_gio.File.new_for_path.side_effect = Exception("no inotify")
result = indicator._start_input_device_monitor()
from vocalinux.ui.tray_indicator import _FALLBACK_KEYBOARD_RESTART_SECONDS
mock_glib.timeout_add_seconds.assert_called_once_with(
_FALLBACK_KEYBOARD_RESTART_SECONDS, indicator._reinit_keyboard_fallback
)
assert result == mock_glib.SOURCE_REMOVE
@patch("vocalinux.ui.tray_indicator.GLib")
def test_on_input_device_changed_resets_settle_timer(self, mock_glib):
indicator = self._make_tray_indicator()
indicator._settle_timer_id = 42
indicator._on_input_device_changed(MagicMock(), MagicMock(), None, MagicMock())
mock_glib.source_remove.assert_called_once_with(42)
from vocalinux.ui.tray_indicator import _INPUT_SETTLE_SECONDS
mock_glib.timeout_add_seconds.assert_called_once_with(
_INPUT_SETTLE_SECONDS, indicator._on_devices_settled
)
@patch("vocalinux.ui.tray_indicator.GLib")
def test_on_input_device_changed_creates_timer_when_none(self, mock_glib):
indicator = self._make_tray_indicator()
indicator._on_input_device_changed(MagicMock(), MagicMock(), None, MagicMock())
mock_glib.source_remove.assert_not_called()
mock_glib.timeout_add_seconds.assert_called_once()
@patch("vocalinux.ui.tray_indicator.GLib")
def test_on_devices_settled_cleans_up_and_restarts_keyboard(self, mock_glib):
indicator = self._make_tray_indicator()
mock_monitor = MagicMock()
indicator._input_monitor = mock_monitor
indicator._settle_timer_id = 10
indicator._monitor_timeout_id = 20
result = indicator._on_devices_settled()
assert indicator._input_monitor is None
assert indicator._settle_timer_id is None
assert indicator._monitor_timeout_id is None
mock_monitor.cancel.assert_called_once()
indicator._setup_keyboard_shortcuts.assert_called_once()
assert result == mock_glib.SOURCE_REMOVE
@patch("vocalinux.ui.tray_indicator.GLib")
def test_on_input_monitor_timeout_cleans_up_and_restarts_keyboard(self, mock_glib):
indicator = self._make_tray_indicator()
mock_monitor = MagicMock()
indicator._input_monitor = mock_monitor
indicator._settle_timer_id = 10
indicator._monitor_timeout_id = 20
result = indicator._on_input_monitor_timeout()
assert indicator._input_monitor is None
indicator._setup_keyboard_shortcuts.assert_called_once()
assert result == mock_glib.SOURCE_REMOVE
def test_reinit_keyboard_fallback_calls_setup(self):
indicator = self._make_tray_indicator()
result = indicator._reinit_keyboard_fallback()
indicator._setup_keyboard_shortcuts.assert_called_once()
from gi.repository import GLib
assert result == GLib.SOURCE_REMOVE
def test_reinit_keyboard_fallback_failure_is_caught(self):
indicator = self._make_tray_indicator()
indicator._setup_keyboard_shortcuts.side_effect = RuntimeError("no devices")
try:
indicator._reinit_keyboard_fallback()
except RuntimeError:
pass
@patch("vocalinux.ui.tray_indicator.GLib")
def test_cleanup_input_monitor_cancels_everything(self, mock_glib):
indicator = self._make_tray_indicator()
mock_monitor = MagicMock()
indicator._input_monitor = mock_monitor
indicator._settle_timer_id = 10
indicator._monitor_timeout_id = 20
indicator._cleanup_input_monitor()
mock_glib.source_remove.assert_any_call(10)
mock_glib.source_remove.assert_any_call(20)
mock_monitor.cancel.assert_called_once()
assert indicator._input_monitor is None
assert indicator._settle_timer_id is None
assert indicator._monitor_timeout_id is None
@patch("vocalinux.ui.tray_indicator.GLib")
def test_cleanup_input_monitor_noop_when_nothing_active(self, mock_glib):
indicator = self._make_tray_indicator()
indicator._cleanup_input_monitor()
mock_glib.source_remove.assert_not_called()
def test_tray_hotkey_cancels_late_worker_after_stop_join_timeout(self) -> None:
import vocalinux.speech_recognition.recognition_manager as recognition_module
from vocalinux.speech_recognition.recognition_manager import (
SpeechRecognitionManager,
)
duck = MagicMock()
duck.enabled.return_value = False
with patch.object(SpeechRecognitionManager, "_init_whispercpp"):
manager = SpeechRecognitionManager(
engine="whisper_cpp",
model_size="small",
language="en-us",
playback_duck=duck,
)
manager.model = MagicMock()
manager._model_initialized = True
manager._voice_commands_enabled = False
transcriber_started = threading.Event()
release_transcriber = threading.Event()
delivered: list[str] = []
def record_audio() -> None:
return None
def transcribe(_audio: list[bytes], _language: str | None = None) -> str:
transcriber_started.set()
release_transcriber.wait(timeout=3.0)
return "cancelled text"
manager._record_audio = record_audio
manager._transcribe_with_whispercpp = transcribe
manager.register_text_callback(delivered.append)
with patch.object(recognition_module, "play_start_sound"):
with patch.object(manager, "_arm_playback_duck"):
assert manager.start_recognition()
manager._enqueue_audio_segment([b"slow audio"])
assert transcriber_started.wait(timeout=1.0)
worker_exited = manager._worker_exited
assert worker_exited is not None
worker_thread = manager.recognition_thread
assert worker_thread is not None
try:
with patch.object(SpeechRecognitionManager, "_STOP_WORKER_JOIN_TIMEOUT_S", 0.05):
with patch.object(
SpeechRecognitionManager, "_STOP_WORKER_FINAL_JOIN_TIMEOUT_S", 0.01
):
with patch.object(recognition_module, "play_stop_sound"):
manager.stop_recognition()
assert manager.state is RecognitionState.PROCESSING
assert manager.is_transcribing()
tray = self._make_tray_indicator()
tray.speech_engine = manager
tray._ptt_lock = threading.Lock()
tray._ptt_owner = None
tray._toggle_recognition()
assert manager.state is RecognitionState.IDLE
finally:
release_transcriber.set()
worker_thread.join(timeout=2.0)
assert not worker_thread.is_alive()
assert worker_exited.is_set()
assert manager.state is RecognitionState.IDLE
assert delivered == []
class TestTrayAutoPauseWiring(unittest.TestCase):
"""TrayIndicator auto-pause config + pause/resume handlers.
Kept here (not in test_auto_pause_monitor) so tray_indicator is only
imported from this module's existing tray tests, avoiding early import
under a collection-time gi MagicMock that would pin a stale GLib on the
module and break SOURCE_REMOVE comparisons later in the suite.
"""
def _make_tray_indicator(self):
from vocalinux.ui.tray_indicator import TrayIndicator
with patch.object(TrayIndicator, "__init__", lambda self: None):
indicator = TrayIndicator.__new__(TrayIndicator)
indicator.speech_engine = MagicMock()
indicator.speech_engine.state = RecognitionState.IDLE
indicator.speech_engine.unload_model = MagicMock()
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
indicator._monitor_timeout_id = None
indicator._model_keepalive = MagicMock()
indicator._auto_pause_monitor = MagicMock()
indicator._auto_pause_monitor.paused = False
indicator.shortcut_manager = MagicMock()
indicator._language_shortcut_managers = []
return indicator
def test_get_auto_pause_config(self):
indicator = self._make_tray_indicator()
indicator.config_manager.get_bool.return_value = True
indicator.config_manager.get.return_value = ["steam"]
indicator.config_manager.get_float.return_value = 7.0
enabled, apps, interval = indicator._get_auto_pause_config()
assert enabled is True
assert apps == ["steam"]
assert interval == 7.0
indicator.config_manager.get_bool.assert_called_with("auto_pause", "enabled", False)
indicator.config_manager.get.assert_called_with("auto_pause", "apps", [])
def test_on_auto_pause_calls_unload(self):
indicator = self._make_tray_indicator()
indicator._on_auto_pause()
indicator.speech_engine.unload_model.assert_called_once_with(reason="auto_pause")
indicator._model_keepalive.cancel.assert_called_once()
def test_on_auto_resume_calls_reinit(self):
indicator = self._make_tray_indicator()
indicator._on_auto_resume()
indicator.speech_engine.reinitialize_after_resume.assert_called_once_with(auto_resume=True)
indicator._model_keepalive.bump.assert_called_once()
def test_auto_resume_during_suspend_waits_for_wake(self) -> None:
indicator = self._make_tray_indicator()
indicator._on_system_suspend()
indicator._on_auto_resume()
indicator.speech_engine.reinitialize_after_resume.assert_not_called()
indicator._model_keepalive.bump.assert_not_called()
def test_suspend_reissues_unload_when_game_stays_active(self) -> None:
indicator = self._make_tray_indicator()
indicator._auto_pause_monitor.paused = True
indicator._on_system_suspend()
assert indicator._system_suspended
indicator.speech_engine.unload_model.assert_called_once_with(reason="auto_pause")
indicator.speech_engine.cancel_pending_auto_pause.assert_not_called()
indicator._model_keepalive.cancel.assert_called_once_with()
def test_get_model_keepalive_config(self):
indicator = self._make_tray_indicator()
indicator.config_manager.get_bool.return_value = True
indicator.config_manager.get_float.return_value = 600.0
enabled, timeout = indicator._get_model_keepalive_config()
assert enabled is True
assert timeout == 600.0
indicator.config_manager.get_bool.assert_called_with("model_keepalive", "enabled", False)
def test_on_keepalive_idle_unload(self):
indicator = self._make_tray_indicator()
indicator._on_keepalive_idle_unload()
indicator.speech_engine.unload_model.assert_called_once_with(reason="idle_keepalive")
def test_is_safe_for_keepalive_unload(self):
indicator = self._make_tray_indicator()
indicator.speech_engine.state = RecognitionState.IDLE
indicator.speech_engine.is_auto_paused = False
assert indicator._is_safe_for_keepalive_unload() is True
indicator.speech_engine.state = RecognitionState.LISTENING
assert indicator._is_safe_for_keepalive_unload() is False
indicator.speech_engine.state = RecognitionState.IDLE
indicator.speech_engine.is_auto_paused = True
assert indicator._is_safe_for_keepalive_unload() is False
@patch("vocalinux.ui.tray_indicator.GLib")
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._auto_pause_monitor.paused = True
indicator.config_manager.get_bool.return_value = False
indicator._start_input_device_monitor = MagicMock()
indicator._on_system_resume()
assert mock_glib.timeout_add_seconds.call_count == 1
assert (
mock_glib.timeout_add_seconds.call_args[0][1] == indicator._start_input_device_monitor
)
indicator.speech_engine.prepare_auto_resume.assert_not_called()
@patch("vocalinux.ui.tray_indicator.GLib")
def test_system_resume_rechecks_monitor_even_without_manager_guard(
self, mock_glib: MagicMock
) -> None:
indicator = self._make_tray_indicator()
indicator._auto_pause_monitor.paused = True
indicator.config_manager.get_bool.return_value = False
indicator._on_system_resume()
mock_glib.timeout_add_seconds.assert_called_once_with(
2, indicator._start_input_device_monitor
)
@patch("vocalinux.ui.tray_indicator.GLib")
def test_system_resume_schedules_guarded_recovery_when_game_is_gone(
self, mock_glib: MagicMock
) -> None:
indicator = self._make_tray_indicator()
indicator.speech_engine.is_auto_paused = True
indicator.config_manager.get_bool.return_value = False
indicator._system_suspended = True
indicator._on_system_resume()
assert not indicator._system_suspended
assert mock_glib.timeout_add_seconds.call_args_list[0].args == (
2,
indicator._reinit_speech_after_resume,
indicator.speech_engine.prepare_auto_resume.return_value,
)
@patch("vocalinux.ui.tray_indicator.GLib")
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()
indicator._on_system_resume()
assert mock_glib.timeout_add_seconds.call_count == 2
def test_get_auto_pause_config_coerces_non_list_apps(self):
indicator = self._make_tray_indicator()
indicator.config_manager.get_bool.return_value = True
indicator.config_manager.get.return_value = "steam" # not a list
indicator.config_manager.get_float.return_value = 5.0
enabled, apps, interval = indicator._get_auto_pause_config()
assert enabled is True
assert apps == []
assert interval == 5.0
def test_get_auto_pause_config_treats_none_apps_as_empty(self):
indicator = self._make_tray_indicator()
indicator.config_manager.get_bool.return_value = False
indicator.config_manager.get.return_value = None
indicator.config_manager.get_float.return_value = 5.0
enabled, apps, _interval = indicator._get_auto_pause_config()
assert enabled is False
assert apps == []
def test_on_auto_pause_fallback_stops_when_no_unload(self):
indicator = self._make_tray_indicator()
# Non-callable attribute so getattr finds it but callable() is False
indicator.speech_engine.unload_model = None
indicator.speech_engine.state = RecognitionState.LISTENING
indicator.speech_engine.stop_recognition = MagicMock()
indicator._on_auto_pause()
indicator.speech_engine.stop_recognition.assert_called_once()
def test_on_auto_pause_fallback_idle_is_noop(self):
indicator = self._make_tray_indicator()
indicator.speech_engine.unload_model = None
indicator.speech_engine.state = RecognitionState.IDLE
indicator.speech_engine.stop_recognition = MagicMock()
indicator._on_auto_pause()
indicator.speech_engine.stop_recognition.assert_not_called()
def test_on_auto_resume_swallows_reinit_errors(self):
indicator = self._make_tray_indicator()
indicator.speech_engine.reinitialize_after_resume.side_effect = RuntimeError("boom")
indicator._on_auto_resume() # must not raise
indicator.speech_engine.reinitialize_after_resume.assert_called_once_with(auto_resume=True)
def test_on_auto_resume_without_reinit_method(self):
indicator = self._make_tray_indicator()
indicator.speech_engine.reinitialize_after_resume = None
indicator._on_auto_resume() # no-op path
indicator._model_keepalive.bump.assert_called_once()
def test_quit_shuts_down_keepalive(self):
indicator = self._make_tray_indicator()
monitor = MagicMock()
indicator._auto_pause_monitor = monitor
indicator._suspend_handler = MagicMock()
indicator._cleanup_input_monitor = MagicMock()
indicator.shortcut_manager = MagicMock()
indicator.text_injector = None
with patch("vocalinux.ui.tray_indicator.Gtk") as mock_gtk:
indicator._quit()
monitor.shutdown.assert_called_once()
indicator._model_keepalive.shutdown.assert_called_once()
indicator._suspend_handler.shutdown.assert_called_once()
mock_gtk.main_quit.assert_called_once()
if __name__ == "__main__":
unittest.main()
Reference in New Issue View Git Blame Copy Permalink