Jatin K MalikandGitHub
9646011e77
* fix(settings): speech model follow-ups from #834 Keep the Unused downloads expander in sync with search, recommend Faster Whisper .en weights for English, parse nvidia-smi MiB VRAM, stop simple mode from clobbering Faster Whisper/Parakeet, and ceil RAM so ~7.4 GiB counts as 8. * fix(settings): keep FW .en picker ids and test unused search Represent Faster Whisper English catalog variants in the Settings size combo so tray-saved small.en is not rejected and overwritten. Add behavioral unused-downloads search coverage for match/nonmatch/clear.
527 lines
20 KiB
Python
527 lines
20 KiB
Python
"""Tests for offering the recommended model when dictation finds none."""
|
|
|
|
import importlib
|
|
import sys
|
|
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
|
|
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.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
|
|
# 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()
|
|
|
|
|
|
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
|