e66bd218d9
* 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>
795 lines
31 KiB
Python
795 lines
31 KiB
Python
"""Tests for offering the recommended model when dictation finds none."""
|
|
|
|
import importlib
|
|
import json
|
|
import sys
|
|
import threading
|
|
from unittest.mock import MagicMock, Mock, patch
|
|
|
|
import pytest
|
|
|
|
import vocalinux.ui
|
|
|
|
|
|
def _load(module_name, extra_bases=()):
|
|
"""Import a GTK module with real base classes so its methods stay callable.
|
|
|
|
conftest swaps gi for a MagicMock, which would leave every ``class X(Gtk.Y)``
|
|
as a mock and make its methods unreachable.
|
|
|
|
Reimporting rebinds the module both in ``sys.modules`` and as an attribute
|
|
on its package. Leaving those two pointing at different objects breaks any
|
|
later test that patches the module by name, so the caller restores both;
|
|
doing this at collection time is what desynced test_model_deletion in #686.
|
|
"""
|
|
repository = sys.modules["gi.repository"]
|
|
names = ("Box", "ListBoxRow", "Dialog") + tuple(extra_bases)
|
|
bases = {name: type(name, (), {}) for name in names}
|
|
attribute = module_name.rsplit(".", 1)[1]
|
|
saved_module = sys.modules.pop(module_name, None)
|
|
saved_attribute = getattr(vocalinux.ui, attribute, None)
|
|
try:
|
|
with patch.object(repository, "Gtk", MagicMock(**bases)):
|
|
yield importlib.import_module(module_name)
|
|
finally:
|
|
sys.modules.pop(module_name, None)
|
|
if saved_module is not None:
|
|
sys.modules[module_name] = saved_module
|
|
if saved_attribute is not None:
|
|
setattr(vocalinux.ui, attribute, saved_attribute)
|
|
elif hasattr(vocalinux.ui, attribute):
|
|
delattr(vocalinux.ui, attribute)
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def settings_dialog():
|
|
yield from _load("vocalinux.ui.settings_dialog")
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def tray_indicator(settings_dialog):
|
|
yield from _load("vocalinux.ui.tray_indicator")
|
|
|
|
|
|
@pytest.fixture
|
|
def TrayIndicator(tray_indicator):
|
|
return tray_indicator.TrayIndicator
|
|
|
|
|
|
@pytest.fixture
|
|
def recommendation(settings_dialog):
|
|
return settings_dialog.ModelRecommendation(
|
|
model_id="small.en", reason="Vulkan GPU", display_name="Small EN", size_label="466 MB"
|
|
)
|
|
|
|
|
|
def _tray_stub():
|
|
tray = Mock()
|
|
tray._model_download_active = False
|
|
tray.config_manager.get_str.return_value = "en-us"
|
|
tray.speech_engine.model_ready = True
|
|
tray.speech_engine.engine = "whisper_cpp"
|
|
tray.speech_engine.model_size = "small.en"
|
|
tray.speech_engine.is_auto_paused = False
|
|
return tray
|
|
|
|
|
|
class TestRecommendedModelForEngine:
|
|
"""The offer must name the model the picker stars, not a fixed one."""
|
|
|
|
def test_whisper_cpp_follows_the_hardware_recommendation(self, settings_dialog):
|
|
with patch.object(
|
|
settings_dialog,
|
|
"get_recommended_whispercpp_model",
|
|
return_value=("small", "Vulkan GPU"),
|
|
):
|
|
result = settings_dialog.recommended_model_for_engine("whisper_cpp", "de")
|
|
|
|
assert result.model_id == "small"
|
|
assert "Vulkan GPU" in result.reason
|
|
assert result.size_label
|
|
|
|
def test_english_selection_gets_the_english_only_variant(self, settings_dialog):
|
|
with patch.object(
|
|
settings_dialog,
|
|
"get_recommended_whispercpp_model",
|
|
return_value=("small", "Vulkan GPU"),
|
|
):
|
|
result = settings_dialog.recommended_model_for_engine("whisper_cpp", "en-us")
|
|
|
|
assert result.model_id == "small.en"
|
|
|
|
def test_vosk_uses_its_own_recommendation(self, settings_dialog):
|
|
with patch.object(
|
|
settings_dialog, "_get_recommended_vosk_model", return_value=("medium", "8GB RAM")
|
|
):
|
|
result = settings_dialog.recommended_model_for_engine("vosk", "en-us")
|
|
|
|
assert result.model_id == "medium"
|
|
|
|
def test_remote_api_has_nothing_to_download(self, settings_dialog):
|
|
assert settings_dialog.recommended_model_for_engine("remote_api", "auto") is None
|
|
|
|
def test_faster_whisper_english_gets_the_english_only_variant(self, settings_dialog):
|
|
with patch.object(
|
|
settings_dialog,
|
|
"get_recommended_faster_whisper_model",
|
|
return_value=("small", "CUDA GPU"),
|
|
):
|
|
result = settings_dialog.recommended_model_for_engine("faster_whisper", "en-us")
|
|
|
|
assert result.model_id == "small.en"
|
|
assert "CUDA GPU" in result.reason
|
|
assert result.model_id in settings_dialog.ENGINE_MODELS["faster_whisper"]
|
|
|
|
def test_faster_whisper_non_english_keeps_the_multilingual_model(self, settings_dialog):
|
|
with patch.object(
|
|
settings_dialog,
|
|
"get_recommended_faster_whisper_model",
|
|
return_value=("tiny", "4GB RAM"),
|
|
):
|
|
result = settings_dialog.recommended_model_for_engine("faster_whisper", "de")
|
|
|
|
assert result.model_id == "tiny"
|
|
|
|
def test_faster_whisper_skips_english_variant_when_catalog_has_none(self, settings_dialog):
|
|
with patch.object(
|
|
settings_dialog,
|
|
"get_recommended_faster_whisper_model",
|
|
return_value=("large-v3", "lots of RAM"),
|
|
):
|
|
result = settings_dialog.recommended_model_for_engine("faster_whisper", "en-us")
|
|
|
|
assert result.model_id == "large-v3"
|
|
|
|
|
|
class TestOffer:
|
|
"""What the notification carries."""
|
|
|
|
def _show(self, tray_indicator, tray, engine, rec, supports_actions):
|
|
"""Show the offer notification, return the notifications mock."""
|
|
with patch.object(tray_indicator, "notifications") as mock_notifications:
|
|
mock_notifications.supports_actions.return_value = supports_actions
|
|
tray_indicator.TrayIndicator._show_model_offer(tray, engine, rec)
|
|
return mock_notifications
|
|
|
|
def test_offer_is_handed_to_the_main_loop(self, tray_indicator, TrayIndicator, recommendation):
|
|
"""Dictation can start on the shortcut thread, which must not touch the UI."""
|
|
tray = _tray_stub()
|
|
|
|
with patch.object(tray_indicator, "GLib") as glib:
|
|
with patch.object(
|
|
tray_indicator, "recommended_model_for_engine", return_value=recommendation
|
|
):
|
|
handled = TrayIndicator._offer_recommended_model(tray, "whisper_cpp")
|
|
|
|
assert handled is True
|
|
glib.idle_add.assert_called_once_with(tray._show_model_offer, "whisper_cpp", recommendation)
|
|
|
|
def test_offer_carries_a_download_button(self, tray_indicator, recommendation):
|
|
tray = _tray_stub()
|
|
|
|
mock_notifications = self._show(
|
|
tray_indicator, tray, "whisper_cpp", recommendation, supports_actions=True
|
|
)
|
|
|
|
action = mock_notifications.notify.call_args.kwargs["action"]
|
|
assert action[0] == "Download Small EN (466 MB)"
|
|
assert "from this notification" in mock_notifications.notify.call_args.args[1]
|
|
|
|
# The button downloads exactly what was offered.
|
|
action[1]()
|
|
tray._download_recommended_model.assert_called_once_with("whisper_cpp", recommendation)
|
|
|
|
def test_offer_without_server_action_support_is_still_shown(
|
|
self, tray_indicator, settings_dialog
|
|
):
|
|
"""No button means the body must not tell the user to click one."""
|
|
tray = _tray_stub()
|
|
rec = settings_dialog.ModelRecommendation(
|
|
model_id="small", reason="8GB RAM", display_name="Small", size_label="466 MB"
|
|
)
|
|
|
|
mock_notifications = self._show(tray_indicator, tray, "vosk", rec, supports_actions=False)
|
|
|
|
body = mock_notifications.notify.call_args.args[1]
|
|
assert mock_notifications.notify.call_args.kwargs["action"] is None
|
|
assert "Settings" in body
|
|
assert "from this notification" not in body
|
|
|
|
|
|
class _RestartGatedEngine:
|
|
"""A speech engine that gates re-init the way reconfigure() really does.
|
|
|
|
Only engine, model and language changes set restart_needed, and
|
|
force_download is read during that re-init — so a caller asking for the
|
|
model it is already configured for downloads nothing unless it also asks
|
|
for the re-init.
|
|
"""
|
|
|
|
def __init__(self, engine, model_size, language):
|
|
self.engine = engine
|
|
self.model_size = model_size
|
|
self.language = language
|
|
self.model_ready = False
|
|
self.is_auto_paused = False
|
|
self.reinits = 0
|
|
self.state = None
|
|
|
|
def set_download_progress_callback(self, callback):
|
|
pass
|
|
|
|
def try_begin_download(self):
|
|
return True
|
|
|
|
def end_download(self):
|
|
pass
|
|
|
|
def reconfigure(
|
|
self,
|
|
engine=None,
|
|
model_size=None,
|
|
language=None,
|
|
force_download=True,
|
|
force_reinit=False,
|
|
**kwargs,
|
|
):
|
|
restart_needed = force_reinit
|
|
if engine is not None and engine != self.engine:
|
|
self.engine = engine
|
|
restart_needed = True
|
|
if model_size is not None and model_size != self.model_size:
|
|
self.model_size = model_size
|
|
restart_needed = True
|
|
if language is not None and language != self.language:
|
|
self.language = language
|
|
restart_needed = True
|
|
|
|
if restart_needed and force_download:
|
|
self.reinits += 1
|
|
self.model_ready = True
|
|
|
|
|
|
class TestDownload:
|
|
"""What the button does once clicked."""
|
|
|
|
def _run(self, tray_indicator, tray, rec, progress=None):
|
|
with patch.object(tray_indicator, "notifications"):
|
|
with patch.object(tray_indicator, "GLib") as glib:
|
|
glib.idle_add.side_effect = lambda func, *args: func(*args)
|
|
tray_indicator.TrayIndicator._run_recommended_model_download(
|
|
tray, "whisper_cpp", rec, progress or object()
|
|
)
|
|
|
|
def test_successful_download_is_saved_for_that_engine(
|
|
self, tray_indicator, tray, recommendation
|
|
):
|
|
self._run(tray_indicator, tray, recommendation)
|
|
|
|
tray.speech_engine.reconfigure.assert_called_once_with(
|
|
engine="whisper_cpp",
|
|
model_size="small.en",
|
|
force_download=True,
|
|
force_reinit=True,
|
|
)
|
|
tray.config_manager.set_model_size_for_engine.assert_called_once_with(
|
|
"whisper_cpp", "small.en"
|
|
)
|
|
tray.config_manager.save_config.assert_called_once()
|
|
assert tray._model_download_active is False
|
|
|
|
def test_the_model_already_configured_is_still_fetched(
|
|
self, tray_indicator, tray, recommendation
|
|
):
|
|
"""The case that breaks the feature: recommendation == configured model.
|
|
|
|
First-run defaults are whisper_cpp + tiny, and tiny is exactly what a
|
|
CPU-only machine is recommended, so nothing about the request differs
|
|
from what the engine already holds. Without asking for the re-init the
|
|
click downloads nothing and still reports the model as ready.
|
|
"""
|
|
tray.speech_engine = _RestartGatedEngine("whisper_cpp", recommendation.model_id, "en-us")
|
|
|
|
self._run(tray_indicator, tray, recommendation)
|
|
|
|
assert tray.speech_engine.reinits == 1
|
|
assert tray.speech_engine.model_ready is True
|
|
tray.config_manager.set_model_size_for_engine.assert_called_once_with(
|
|
"whisper_cpp", recommendation.model_id
|
|
)
|
|
tray.config_manager.save_config.assert_called_once()
|
|
|
|
def test_a_model_that_did_not_load_is_not_reported_as_ready(
|
|
self, tray_indicator, tray, recommendation
|
|
):
|
|
"""reconfigure() returning without raising is not proof of a model."""
|
|
tray.speech_engine.model_ready = False
|
|
|
|
with patch.object(tray_indicator, "notifications") as mock_notifications:
|
|
with patch.object(tray_indicator, "GLib") as glib:
|
|
glib.idle_add.side_effect = lambda func, *args: func(*args)
|
|
tray_indicator.TrayIndicator._run_recommended_model_download(
|
|
tray, "whisper_cpp", recommendation, object()
|
|
)
|
|
|
|
tray.config_manager.set_model_size_for_engine.assert_not_called()
|
|
tray.config_manager.save_config.assert_not_called()
|
|
titles = [call.args[0] for call in mock_notifications.notify.call_args_list]
|
|
assert "Model download failed" in titles
|
|
assert "Speech model ready" not in titles
|
|
|
|
def test_failed_download_is_not_saved(self, tray_indicator, tray, recommendation):
|
|
tray.speech_engine.reconfigure.side_effect = RuntimeError("network down")
|
|
|
|
self._run(tray_indicator, tray, recommendation)
|
|
|
|
tray.config_manager.set_model_size_for_engine.assert_not_called()
|
|
# The engine must not keep reporting progress to a finished download.
|
|
tray.speech_engine.set_download_progress_callback.assert_called_with(None)
|
|
assert tray._model_download_active is False
|
|
|
|
def test_the_first_notification_is_created_on_the_main_loop(
|
|
self, tray_indicator, tray, recommendation
|
|
):
|
|
"""libnotify is not thread-safe, so the worker only gets a handle."""
|
|
with patch.object(tray_indicator, "notifications") as mock_notifications:
|
|
with patch.object(tray_indicator.threading, "Thread") as thread:
|
|
tray_indicator.TrayIndicator._download_recommended_model(
|
|
tray, "whisper_cpp", recommendation
|
|
)
|
|
|
|
mock_notifications.notify.assert_called_once()
|
|
assert mock_notifications.notify.call_args.args[0] == "Downloading speech model"
|
|
handle = mock_notifications.notify.return_value
|
|
assert thread.call_args.kwargs["args"] == ("whisper_cpp", recommendation, handle)
|
|
|
|
def test_main_loop_callbacks_do_not_ask_to_run_again(
|
|
self, tray_indicator, tray, recommendation
|
|
):
|
|
"""GLib.idle_add repeats a callback that returns truthy, so ours must not.
|
|
|
|
The notification helpers return handles and success flags; handing them
|
|
to idle_add directly would re-show the notification on every idle cycle.
|
|
"""
|
|
results = []
|
|
|
|
def fake_reconfigure(**kwargs):
|
|
# Progress arrives while the download inside reconfigure() runs.
|
|
on_progress = tray.speech_engine.set_download_progress_callback.call_args[0][0]
|
|
on_progress(0.5, 10.0, "50%")
|
|
|
|
tray.speech_engine.reconfigure.side_effect = fake_reconfigure
|
|
|
|
with patch.object(tray_indicator, "notifications") as mock_notifications:
|
|
mock_notifications.update.return_value = True
|
|
with patch.object(tray_indicator, "GLib") as glib:
|
|
glib.idle_add.side_effect = lambda func, *args: results.append(func(*args))
|
|
tray_indicator.TrayIndicator._run_recommended_model_download(
|
|
tray, "whisper_cpp", recommendation, object()
|
|
)
|
|
|
|
assert results, "expected callbacks scheduled on the main loop"
|
|
assert all(not result for result in results)
|
|
|
|
def test_a_second_click_does_not_start_a_second_download(self, tray_indicator):
|
|
tray = _tray_stub()
|
|
tray._model_download_active = True
|
|
|
|
with patch.object(tray_indicator.threading, "Thread") as thread:
|
|
tray_indicator.TrayIndicator._download_recommended_model(tray, "vosk", Mock())
|
|
|
|
thread.assert_not_called()
|
|
|
|
|
|
@pytest.fixture
|
|
def tray():
|
|
return _tray_stub()
|
|
|
|
|
|
@pytest.mark.parametrize("handler_result", [True, False])
|
|
def test_engine_defers_to_the_registered_handler(handler_result):
|
|
"""The engine only shows its own notification when nobody handled it."""
|
|
from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager
|
|
|
|
manager = Mock()
|
|
manager._model_missing_handler = Mock(return_value=handler_result)
|
|
manager.engine = "whisper_cpp"
|
|
|
|
assert SpeechRecognitionManager._notify_model_missing(manager) is handler_result
|
|
manager._model_missing_handler.assert_called_once_with("whisper_cpp")
|
|
|
|
|
|
def test_engine_without_a_handler_keeps_its_own_notification():
|
|
from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager
|
|
|
|
manager = Mock()
|
|
manager._model_missing_handler = None
|
|
|
|
assert SpeechRecognitionManager._notify_model_missing(manager) is False
|
|
|
|
|
|
def _reconfigure_manager():
|
|
from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager
|
|
|
|
manager_class = SpeechRecognitionManager
|
|
manager = MagicMock()
|
|
manager.engine = "whisper_cpp"
|
|
manager.model_size = "tiny"
|
|
manager.language = "en-us"
|
|
manager._defer_download = True
|
|
# Request gating needs an allocated token and concrete resume state.
|
|
manager._model_request_lock = threading.Lock()
|
|
manager._model_request = object()
|
|
manager._auto_resume_pending = False
|
|
manager._reconfigure_request = None
|
|
manager._download_claim = threading.Lock()
|
|
manager._download_cancelled = False
|
|
manager.recognition_thread = None
|
|
manager._recognition_workers = []
|
|
for name in (
|
|
"_settings_reconfigure_context",
|
|
"_reconfigure_needs_restart",
|
|
"_finish_recognition_before_reconfigure",
|
|
"_apply_reconfigure",
|
|
"_check_reconfigure_cancelled",
|
|
):
|
|
setattr(manager, name, getattr(manager_class, name).__get__(manager, manager_class))
|
|
manager._new_model_request = lambda **kwargs: manager_class._new_model_request(
|
|
manager, **kwargs
|
|
)
|
|
manager._hot_reconfigure_settings = manager_class._hot_reconfigure_settings
|
|
# reconfigure() dispatches through _init_selected_engine(); bind the real
|
|
# method so force_reinit still reaches _init_whispercpp on this mock.
|
|
manager._init_selected_engine = lambda: manager_class._init_selected_engine(manager)
|
|
manager._snapshot_reconfigure_state = lambda: manager_class._snapshot_reconfigure_state(manager)
|
|
manager._restore_reconfigure_state = lambda previous: manager_class._restore_reconfigure_state(
|
|
manager, previous
|
|
)
|
|
return manager_class, manager
|
|
|
|
|
|
def test_force_reinit_reinitializes_an_engine_that_did_not_change():
|
|
"""The production contract behind the offer: force_download needs a re-init.
|
|
|
|
Downloads only happen while the engine re-initializes, and that is gated on
|
|
engine, model or language actually changing.
|
|
"""
|
|
manager_class, manager = _reconfigure_manager()
|
|
seen = {}
|
|
manager._init_whispercpp.side_effect = lambda: seen.update(defer=manager._defer_download)
|
|
|
|
manager_class.reconfigure(
|
|
manager,
|
|
engine="whisper_cpp",
|
|
model_size="tiny",
|
|
language="en-us",
|
|
force_download=True,
|
|
force_reinit=True,
|
|
)
|
|
|
|
manager._init_whispercpp.assert_called_once()
|
|
assert seen["defer"] is False
|
|
|
|
|
|
def test_an_unchanged_reconfigure_still_does_nothing_by_default():
|
|
"""Callers that only tweak VAD must not pay for a re-init."""
|
|
manager_class, manager = _reconfigure_manager()
|
|
|
|
manager_class.reconfigure(
|
|
manager,
|
|
engine="whisper_cpp",
|
|
model_size="tiny",
|
|
language="en-us",
|
|
force_download=True,
|
|
)
|
|
|
|
manager._init_whispercpp.assert_not_called()
|
|
|
|
|
|
@pytest.fixture
|
|
def download_manager(monkeypatch):
|
|
from vocalinux.speech_recognition import recognition_manager as rm
|
|
|
|
monkeypatch.setattr(rm, "load_silero_vad", lambda: None)
|
|
monkeypatch.setattr(rm.SpeechRecognitionManager, "_init_whispercpp", lambda self: None)
|
|
manager = rm.SpeechRecognitionManager(engine="whisper_cpp", model_size="tiny", language="en-us")
|
|
manager.model = ("whisper_cpp", "tiny")
|
|
manager._model_initialized = True
|
|
|
|
def initialize():
|
|
manager.model = (manager.engine, manager.model_size)
|
|
manager._model_initialized = True
|
|
|
|
monkeypatch.setattr(manager, "_init_selected_engine", initialize)
|
|
monkeypatch.setattr(manager, "_init_whispercpp", initialize)
|
|
return manager
|
|
|
|
|
|
@pytest.fixture
|
|
def download_config(monkeypatch, tmp_path):
|
|
from vocalinux.ui import config_manager
|
|
|
|
monkeypatch.setattr(config_manager, "CONFIG_DIR", str(tmp_path))
|
|
monkeypatch.setattr(config_manager, "CONFIG_FILE", str(tmp_path / "config.json"))
|
|
config = config_manager.ConfigManager()
|
|
config.update_speech_recognition_settings(
|
|
{"engine": "whisper_cpp", "model_size": "tiny", "language": "en-us"}
|
|
)
|
|
assert config.save_config()
|
|
return config
|
|
|
|
|
|
def test_tray_download_during_auto_pause_saves_the_model_for_resume(
|
|
monkeypatch, tray_indicator, recommendation, download_manager, download_config, tmp_path
|
|
):
|
|
"""A completed transfer stays accepted when a game unloads its target model."""
|
|
manager = download_manager
|
|
downloading, finish_download = threading.Event(), threading.Event()
|
|
downloaded = []
|
|
|
|
def download():
|
|
downloading.set()
|
|
assert finish_download.wait(3)
|
|
downloaded.append(manager.model_size)
|
|
|
|
monkeypatch.setattr(manager, "_download_selected_model_if_missing", download)
|
|
notifications = Mock()
|
|
monkeypatch.setattr(tray_indicator, "notifications", notifications)
|
|
monkeypatch.setattr(tray_indicator, "_idle_once", lambda func, *args: func(*args))
|
|
tray = Mock(speech_engine=manager, config_manager=download_config)
|
|
tray._model_download_active = True
|
|
tray._model_keepalive = None
|
|
tray._system_suspended = False
|
|
assert manager.try_begin_download()
|
|
worker = threading.Thread(
|
|
target=tray_indicator.TrayIndicator._run_recommended_model_download,
|
|
args=(tray, "whisper_cpp", recommendation, object()),
|
|
)
|
|
worker.start()
|
|
try:
|
|
assert downloading.wait(1)
|
|
tray_indicator.TrayIndicator._on_auto_pause(tray)
|
|
assert manager.is_auto_paused
|
|
assert manager.model is None
|
|
finally:
|
|
finish_download.set()
|
|
worker.join(2)
|
|
assert not worker.is_alive()
|
|
assert downloaded == [recommendation.model_id]
|
|
assert not manager.model_ready
|
|
assert not manager.download_in_progress
|
|
assert tray._model_download_active is False
|
|
title, body, _icon = notifications.notify.call_args.args
|
|
assert title == "Speech model downloaded"
|
|
assert "resumes" in body
|
|
assert "Press" not in body
|
|
with (tmp_path / "config.json").open() as handle:
|
|
saved = json.load(handle)["speech_recognition"]
|
|
assert saved["model_size"] == recommendation.model_id
|
|
|
|
tray_indicator.TrayIndicator._on_auto_resume(tray)
|
|
assert not manager.is_auto_paused
|
|
assert manager.model == (saved["engine"], saved["model_size"])
|
|
assert manager.model_ready
|
|
|
|
|
|
@pytest.mark.parametrize("failure", ["download", "initialize"])
|
|
def test_paused_download_or_initializer_exception_still_fails_without_saving(
|
|
monkeypatch,
|
|
tray_indicator,
|
|
recommendation,
|
|
download_manager,
|
|
download_config,
|
|
tmp_path,
|
|
failure,
|
|
):
|
|
"""Being paused must never turn a real transfer/load failure into success."""
|
|
manager = download_manager
|
|
manager.unload_model(reason="auto_pause")
|
|
|
|
def initialize():
|
|
if manager.model_size == recommendation.model_id:
|
|
raise RuntimeError("model load failed")
|
|
manager.model = (manager.engine, manager.model_size)
|
|
manager._model_initialized = True
|
|
|
|
if failure == "download":
|
|
download = Mock(side_effect=OSError("network failed"))
|
|
else:
|
|
download = Mock()
|
|
monkeypatch.setattr(manager, "_init_selected_engine", initialize)
|
|
monkeypatch.setattr(manager, "_download_selected_model_if_missing", download)
|
|
notifications = Mock()
|
|
monkeypatch.setattr(tray_indicator, "notifications", notifications)
|
|
monkeypatch.setattr(tray_indicator, "_idle_once", lambda func, *args: func(*args))
|
|
tray = Mock(speech_engine=manager, config_manager=download_config)
|
|
tray._model_download_active = True
|
|
assert manager.try_begin_download()
|
|
tray_indicator.TrayIndicator._run_recommended_model_download(
|
|
tray, "whisper_cpp", recommendation, object()
|
|
)
|
|
assert notifications.notify.call_args.args[0] == "Model download failed"
|
|
assert manager.model_size == "tiny"
|
|
with (tmp_path / "config.json").open() as handle:
|
|
assert json.load(handle)["speech_recognition"]["model_size"] == "tiny"
|
|
assert not manager.download_in_progress
|
|
assert tray._model_download_active is False
|
|
assert manager._download_progress_callback is None
|
|
|
|
|
|
def test_paused_resident_model_is_reported_as_waiting_for_resume(
|
|
tray_indicator, tray, recommendation
|
|
):
|
|
"""A pending unload can leave resident weights while dictation is blocked."""
|
|
tray.speech_engine.is_auto_paused = True
|
|
with (
|
|
patch.object(tray_indicator, "notifications") as notifications,
|
|
patch.object(tray_indicator, "_idle_once", side_effect=lambda func, *args: func(*args)),
|
|
):
|
|
tray_indicator.TrayIndicator._run_recommended_model_download(
|
|
tray, "whisper_cpp", recommendation, object()
|
|
)
|
|
assert notifications.notify.call_args.args[0] == "Speech model downloaded"
|
|
assert "resumes" in notifications.notify.call_args.args[1]
|
|
tray.config_manager.save_config.assert_called_once_with()
|
|
|
|
|
|
@pytest.mark.parametrize("paused", [False, True])
|
|
@pytest.mark.parametrize("changed", [{"engine": "vosk"}, {"model_size": "medium"}])
|
|
def test_superseded_download_target_is_not_saved_even_with_a_ready_model(
|
|
tray_indicator, tray, recommendation, changed, paused
|
|
):
|
|
"""Readiness alone must not save a downloaded choice over a newer selection."""
|
|
tray.speech_engine.is_auto_paused = paused
|
|
|
|
def supersede(**kwargs):
|
|
for name, value in changed.items():
|
|
setattr(tray.speech_engine, name, value)
|
|
|
|
tray.speech_engine.reconfigure.side_effect = supersede
|
|
with (
|
|
patch.object(tray_indicator, "notifications") as notifications,
|
|
patch.object(tray_indicator, "_idle_once", side_effect=lambda func, *args: func(*args)),
|
|
):
|
|
tray_indicator.TrayIndicator._run_recommended_model_download(
|
|
tray, "whisper_cpp", recommendation, object()
|
|
)
|
|
tray.config_manager.set_model_size_for_engine.assert_not_called()
|
|
tray.config_manager.save_config.assert_not_called()
|
|
assert notifications.notify.call_args.args[0] == "Model download failed"
|
|
|
|
|
|
def test_download_save_finishes_before_a_new_settings_model_apply(
|
|
monkeypatch, settings_dialog, tray_indicator, recommendation, download_manager, download_config
|
|
):
|
|
"""Settings cannot overtake the tray between its model apply and config save."""
|
|
manager = download_manager
|
|
save_entered, finish_save, settings_started, settings_done = (
|
|
threading.Event(),
|
|
threading.Event(),
|
|
threading.Event(),
|
|
threading.Event(),
|
|
)
|
|
monkeypatch.setattr(manager, "_download_selected_model_if_missing", lambda: None)
|
|
save = download_config.save_config
|
|
|
|
def tray_save():
|
|
if threading.current_thread() is worker:
|
|
save_entered.set()
|
|
assert finish_save.wait(3)
|
|
return save()
|
|
|
|
monkeypatch.setattr(download_config, "save_config", tray_save)
|
|
monkeypatch.setattr(tray_indicator, "notifications", Mock())
|
|
monkeypatch.setattr(tray_indicator, "_idle_once", lambda func, *args: func(*args))
|
|
tray = Mock(speech_engine=manager, config_manager=download_config)
|
|
dialog = Mock(speech_engine=manager, config_manager=download_config)
|
|
dialog._save_selected_settings = settings_dialog.SettingsDialog._save_selected_settings.__get__(
|
|
dialog
|
|
)
|
|
errors = []
|
|
|
|
def apply_settings():
|
|
settings_started.set()
|
|
try:
|
|
assert settings_dialog.SettingsDialog._apply_settings_internal(
|
|
dialog,
|
|
{"engine": "whisper_cpp", "model_size": "medium"},
|
|
raise_errors=True,
|
|
)
|
|
except Exception as error:
|
|
errors.append(error)
|
|
finally:
|
|
settings_done.set()
|
|
|
|
worker = threading.Thread(
|
|
target=tray_indicator.TrayIndicator._run_recommended_model_download,
|
|
args=(tray, "whisper_cpp", recommendation, object()),
|
|
)
|
|
settings_worker = threading.Thread(target=apply_settings)
|
|
assert manager.try_begin_download()
|
|
worker.start()
|
|
try:
|
|
assert save_entered.wait(1)
|
|
settings_worker.start()
|
|
assert settings_started.wait(1)
|
|
assert not settings_done.wait(0.1)
|
|
assert manager.model_size == recommendation.model_id
|
|
finally:
|
|
finish_save.set()
|
|
worker.join(2)
|
|
if settings_worker.ident is not None:
|
|
settings_worker.join(2)
|
|
assert not worker.is_alive()
|
|
assert not settings_worker.is_alive()
|
|
assert errors == []
|
|
assert manager.model_size == "medium"
|
|
assert download_config.get("speech_recognition", "model_size") == "medium"
|
|
|
|
|
|
class TestOneDownloadAtATime:
|
|
"""The tray and Settings share one engine, so only one of them downloads."""
|
|
|
|
def _run(self, tray_indicator, tray, rec):
|
|
with patch.object(tray_indicator, "notifications"):
|
|
with patch.object(tray_indicator, "GLib") as glib:
|
|
glib.idle_add.side_effect = lambda func, *args: func(*args)
|
|
tray_indicator.TrayIndicator._run_recommended_model_download(
|
|
tray, "whisper_cpp", rec, object()
|
|
)
|
|
|
|
def test_tray_refuses_when_another_download_holds_the_engine(self, tray_indicator):
|
|
tray = _tray_stub()
|
|
tray.speech_engine.try_begin_download.return_value = False
|
|
|
|
with patch.object(tray_indicator, "notifications") as mock_notifications:
|
|
with patch.object(tray_indicator.threading, "Thread") as thread:
|
|
tray_indicator.TrayIndicator._download_recommended_model(tray, "vosk", Mock())
|
|
|
|
thread.assert_not_called()
|
|
assert tray._model_download_active is False
|
|
assert mock_notifications.notify.call_args.args[0] == "Download already in progress"
|
|
|
|
def test_tray_releases_the_engine_when_the_download_ends(
|
|
self, tray_indicator, tray, recommendation
|
|
):
|
|
self._run(tray_indicator, tray, recommendation)
|
|
|
|
tray.speech_engine.end_download.assert_called_once_with()
|
|
|
|
def test_tray_releases_the_engine_when_the_download_fails(
|
|
self, tray_indicator, tray, recommendation
|
|
):
|
|
tray.speech_engine.reconfigure.side_effect = RuntimeError("boom")
|
|
|
|
self._run(tray_indicator, tray, recommendation)
|
|
|
|
tray.speech_engine.end_download.assert_called_once_with()
|
|
|
|
def test_download_claim_is_exclusive_until_released(self):
|
|
import threading
|
|
|
|
from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager
|
|
|
|
manager = MagicMock()
|
|
manager._download_claim = threading.Lock()
|
|
|
|
assert SpeechRecognitionManager.try_begin_download(manager) is True
|
|
assert SpeechRecognitionManager.try_begin_download(manager) is False
|
|
assert SpeechRecognitionManager.download_in_progress.fget(manager) is True
|
|
|
|
SpeechRecognitionManager.end_download(manager)
|
|
|
|
assert SpeechRecognitionManager.download_in_progress.fget(manager) is False
|
|
assert SpeechRecognitionManager.try_begin_download(manager) is True
|
|
|
|
def test_releasing_an_unclaimed_engine_is_harmless(self):
|
|
import threading
|
|
|
|
from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager
|
|
|
|
manager = MagicMock()
|
|
manager._download_claim = threading.Lock()
|
|
|
|
SpeechRecognitionManager.end_download(manager) # must not raise
|