"""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