""" Deep coverage tests for recognition_manager.py. Properly isolated - saves/restores sys.modules to avoid polluting other tests. """ import importlib import json import os import queue import sys import threading import time import unittest from typing import Optional from unittest.mock import MagicMock, call, patch # ---- Safely mock external deps, preserving stdlib ---- _MOCK_KEYS = [ "vosk", "whisper", "torch", "pyaudio", "pywhispercpp", "pywhispercpp.model", ] _ORIG_MODULES = {} for _k in _MOCK_KEYS: _ORIG_MODULES[_k] = sys.modules.get(_k) if _k not in sys.modules: sys.modules[_k] = MagicMock() if "gi" not in sys.modules: sys.modules["gi"] = MagicMock() if "gi.repository" not in sys.modules: sys.modules["gi.repository"] = MagicMock() from vocalinux.common_types import RecognitionState # noqa: E402 from vocalinux.speech_recognition.recognition_manager import ( # noqa: E402 SpeechRecognitionManager, _AudioSegment, _get_supported_channels, _get_supported_sample_rate, _ReleaseGate, _setup_alsa_error_handler, get_audio_input_devices, ) # Restore stdlib modules immediately after import for _k, _v in _ORIG_MODULES.items(): if _v is not None: sys.modules[_k] = _v elif _k in sys.modules and isinstance(sys.modules[_k], MagicMock): del sys.modules[_k] def _make_manager(engine="whisper_cpp", **kw): """Create a manager with deferred init to avoid actually loading models.""" with ( patch.object(SpeechRecognitionManager, "_init_vosk"), patch.object(SpeechRecognitionManager, "_init_whisper"), patch.object(SpeechRecognitionManager, "_init_whispercpp"), patch.object(SpeechRecognitionManager, "_init_parakeet"), patch.object(SpeechRecognitionManager, "_init_faster_whisper"), ): mgr = SpeechRecognitionManager( engine=engine, model_size="small", language="en-us", defer_download=True, **kw ) return mgr def _identity(text: str) -> str: return text class TestPushToTalkReleaseGate(unittest.TestCase): def test_gated_segment_is_held_until_release(self) -> None: """A gated segment is transcribed but delivers callbacks only on release.""" manager = _make_manager(engine="vosk") manager._voice_commands_enabled = True gate = _ReleaseGate() segment = _AudioSegment([b"audio"], None, release_gate=gate) delivered: list[tuple[str, str]] = [] def on_segment(text: str, _started_at: float) -> None: delivered.append(("segment", text)) def on_text(text: str) -> None: delivered.append(("text", text)) def on_action(action: str) -> None: delivered.append(("action", action)) manager.segment_callbacks.append(on_segment) manager.text_callbacks.append(on_text) manager.action_callbacks.append(on_action) with ( patch.object(manager, "_transcribe_with_vosk", return_value="raw"), patch.object(manager, "_apply_dictionary_corrections", side_effect=_identity), patch.object( manager.command_processor, "process_text", return_value=("hello", ["action"]), ), ): manager._process_audio_buffer(segment) self.assertEqual(delivered, []) self.assertEqual(len(gate.held), 1) self.assertEqual( (gate.held[0].text, gate.held[0].actions, gate.held[0].started_at), ("hello", ["action"], None), ) gate.set() manager._release_held_transcripts(gate) self.assertEqual( delivered, [("segment", "hello"), ("text", "hello"), ("action", "action")], ) self.assertEqual(gate.held, []) def test_old_session_gate_does_not_release_new_session_text(self) -> None: """A slow worker's gate must not release a newer session's held text.""" manager = _make_manager(engine="vosk") manager._voice_commands_enabled = False delivered: list[str] = [] manager.text_callbacks.append(delivered.append) old_gate = _ReleaseGate() new_gate = _ReleaseGate() def transcribe(audio_buffer: list[bytes]) -> str: return audio_buffer[0].decode() with ( patch.object(manager, "_transcribe_with_vosk", side_effect=transcribe), patch.object(manager, "_apply_dictionary_corrections", side_effect=_identity), ): manager._process_audio_buffer(_AudioSegment([b"old"], None, release_gate=old_gate)) manager._process_audio_buffer(_AudioSegment([b"new"], None, release_gate=new_gate)) self.assertEqual(delivered, []) # The old session's key goes up; the new session is still held. old_gate.set() manager._release_held_transcripts(old_gate) self.assertEqual(delivered, ["old"]) self.assertEqual(len(new_gate.held), 1) new_gate.set() manager._release_held_transcripts(new_gate) self.assertEqual(delivered, ["old", "new"]) def test_empty_transcription_while_held_does_not_flush(self) -> None: """A blank mid-hold result must not release text before the key is up.""" manager = _make_manager(engine="vosk") manager._voice_commands_enabled = False delivered: list[str] = [] manager.text_callbacks.append(delivered.append) gate = _ReleaseGate() with ( patch.object(manager, "_transcribe_with_vosk", return_value="hello"), patch.object(manager, "_apply_dictionary_corrections", side_effect=_identity), ): manager._process_audio_buffer(_AudioSegment([b"audio"], None, release_gate=gate)) self.assertEqual(delivered, []) self.assertEqual(len(gate.held), 1) with patch.object(manager, "_transcribe_with_vosk", return_value=""): manager._process_audio_buffer(_AudioSegment([b"blank"], None, release_gate=gate)) self.assertEqual(delivered, []) self.assertEqual(len(gate.held), 1) gate.set() manager._release_held_transcripts(gate) self.assertEqual(delivered, ["hello"]) def test_ungated_and_released_segments_deliver_immediately(self) -> None: """Segments without a pending gate never block callback delivery.""" manager = _make_manager(engine="vosk") manager._voice_commands_enabled = False delivered: list[str] = [] manager.text_callbacks.append(delivered.append) opened_gate = _ReleaseGate() opened_gate.set() with ( patch.object(manager, "_transcribe_with_vosk", return_value="hello"), patch.object(manager, "_apply_dictionary_corrections", side_effect=_identity), ): for gate in (None, opened_gate): manager._process_audio_buffer(_AudioSegment([b"audio"], None, release_gate=gate)) self.assertEqual(delivered, ["hello", "hello"]) def test_worker_transcribes_ahead_and_delivers_after_release(self) -> None: """Queued segments transcribe during a hold and deliver FIFO after release.""" manager = _make_manager(engine="vosk") manager._voice_commands_enabled = False manager.should_record = True manager._segment_queue = queue.Queue() gate = _ReleaseGate() manager._key_release_gate = gate transcribed = {name: threading.Event() for name in ("first", "second", "final")} processed_both = threading.Event() delivered: list[str] = [] manager.text_callbacks.append(delivered.append) first = _AudioSegment([b"first"], None, release_gate=gate) second = _AudioSegment([b"second"], None, release_gate=gate) final = _AudioSegment([b"final"], None, release_gate=gate) for segment in (first, second): manager._segment_queue.put((segment, "en-us")) def transcribe(audio_buffer: list[bytes]) -> str: text = audio_buffer[0].decode() transcribed[text].set() return text original_process = manager._process_audio_buffer processed_count = 0 def process_and_signal( audio_buffer: list[bytes], language: Optional[str] = None, ) -> None: nonlocal processed_count original_process(audio_buffer, language) processed_count += 1 if processed_count == 2: processed_both.set() with ( patch.object(manager, "_transcribe_with_vosk", side_effect=transcribe), patch.object(manager, "_apply_dictionary_corrections", side_effect=_identity), patch.object(manager, "_process_audio_buffer", side_effect=process_and_signal), ): worker = threading.Thread(target=manager._perform_recognition) worker.start() try: self.assertTrue(transcribed["first"].wait(timeout=2)) self.assertTrue(transcribed["second"].wait(timeout=2)) self.assertTrue(processed_both.wait(timeout=2)) self.assertEqual(len(gate.held), 2) self.assertEqual(delivered, []) self.assertTrue(worker.is_alive()) gate.set() manager._segment_queue.put((final, "en-us")) finally: gate.set() manager.should_record = False manager._segment_queue.put(None) worker.join(timeout=2) self.assertFalse(worker.is_alive()) self.assertEqual(delivered, ["first", "second", "final"]) def test_gate_release_without_final_segment_drains_on_timeout(self) -> None: """Held transcripts drain when the open gate is noticed on queue timeout.""" manager = _make_manager(engine="vosk") manager._voice_commands_enabled = False manager.should_record = True manager._segment_queue = queue.Queue() gate = _ReleaseGate() manager._key_release_gate = gate transcribed = {name: threading.Event() for name in ("first", "second")} processed_both = threading.Event() delivered_both = threading.Event() delivered: list[str] = [] def on_text(text: str) -> None: delivered.append(text) if len(delivered) == 2: delivered_both.set() manager.text_callbacks.append(on_text) for name in ("first", "second"): manager._segment_queue.put( (_AudioSegment([name.encode()], None, release_gate=gate), "en-us") ) def transcribe(audio_buffer: list[bytes]) -> str: text = audio_buffer[0].decode() transcribed[text].set() return text original_process = manager._process_audio_buffer processed_count = 0 def process_and_signal( audio_buffer: list[bytes], language: Optional[str] = None, ) -> None: nonlocal processed_count original_process(audio_buffer, language) processed_count += 1 if processed_count == 2: processed_both.set() with ( patch.object(manager, "_transcribe_with_vosk", side_effect=transcribe), patch.object(manager, "_apply_dictionary_corrections", side_effect=_identity), patch.object(manager, "_process_audio_buffer", side_effect=process_and_signal), ): worker = threading.Thread(target=manager._perform_recognition) worker.start() try: self.assertTrue(transcribed["first"].wait(timeout=2)) self.assertTrue(transcribed["second"].wait(timeout=2)) self.assertTrue(processed_both.wait(timeout=2)) self.assertEqual(delivered, []) gate.set() self.assertTrue(delivered_both.wait(timeout=2)) self.assertEqual(delivered, ["first", "second"]) finally: gate.set() manager.should_record = False manager._segment_queue.put(None) worker.join(timeout=2) self.assertFalse(worker.is_alive()) def test_worker_exit_force_releases_held_transcripts(self) -> None: """The worker's finally block delivers held text even if the gate stays closed.""" manager = _make_manager(engine="vosk") manager._voice_commands_enabled = False manager.should_record = False manager._segment_queue = queue.Queue() gate = _ReleaseGate() manager._key_release_gate = gate delivered: list[str] = [] manager.text_callbacks.append(delivered.append) for name in ("first", "second"): manager._segment_queue.put( (_AudioSegment([name.encode()], None, release_gate=gate), "en-us") ) manager._segment_queue.put(None) def transcribe(audio_buffer: list[bytes]) -> str: return audio_buffer[0].decode() with ( patch.object( manager, "_transcribe_with_vosk", side_effect=transcribe, ), patch.object(manager, "_apply_dictionary_corrections", side_effect=_identity), ): worker = threading.Thread(target=manager._perform_recognition) worker.start() try: worker.join(timeout=2) self.assertFalse(worker.is_alive()) self.assertFalse(gate.is_set()) finally: if worker.is_alive(): gate.set() worker.join(timeout=2) self.assertEqual(delivered, ["first", "second"]) self.assertEqual(gate.held, []) class TestALSAErrorHandler(unittest.TestCase): def test_setup_success(self): with patch("ctypes.CDLL") as mock_cdll: mock_cdll.return_value = MagicMock() _setup_alsa_error_handler() mock_cdll.assert_called_once_with("libasound.so.2") def test_setup_oserror(self): with patch("ctypes.CDLL", side_effect=OSError("not found")): _setup_alsa_error_handler() # Should not raise class TestGetAudioInputDevices(unittest.TestCase): def test_returns_devices(self): mock_pa_inst = MagicMock() mock_pa_inst.get_device_count.return_value = 3 mock_pa_inst.get_device_info_by_index.side_effect = [ {"name": "Mic 1", "maxInputChannels": 2, "index": 0}, {"name": "Speaker", "maxInputChannels": 0, "index": 1}, {"name": "Mic 2", "maxInputChannels": 1, "index": 2}, ] mock_pyaudio = MagicMock() mock_pyaudio.PyAudio.return_value = mock_pa_inst with patch.dict("sys.modules", {"pyaudio": mock_pyaudio}): devices = get_audio_input_devices() self.assertEqual(len(devices), 2) def test_handles_error(self): mock_pyaudio = MagicMock() mock_pyaudio.PyAudio.side_effect = OSError("no audio") with patch.dict("sys.modules", {"pyaudio": mock_pyaudio}): devices = get_audio_input_devices() self.assertEqual(devices, []) class TestSupportedChannels(unittest.TestCase): def _run_with_pyaudio(self, func): mock_pyaudio = MagicMock() mock_pyaudio.paInt16 = 8 with patch.dict("sys.modules", {"pyaudio": mock_pyaudio}): return func() def test_mono_supported(self): mock_pa = MagicMock() mock_pa.is_format_supported.return_value = True result = self._run_with_pyaudio(lambda: _get_supported_channels(mock_pa, None)) self.assertEqual(result, 1) def test_mono_fails_stereo_ok(self): mock_pa = MagicMock() call_count = [0] mock_stream = MagicMock() def open_side_effect(**kw): call_count[0] += 1 if kw.get("channels") == 1: raise IOError("Invalid number of channels -9998") return mock_stream mock_pa.open.side_effect = open_side_effect result = self._run_with_pyaudio(lambda: _get_supported_channels(mock_pa, None)) self.assertEqual(result, 2) def test_both_fail(self): mock_pa = MagicMock() mock_pa.is_format_supported.side_effect = ValueError("nope") result = self._run_with_pyaudio(lambda: _get_supported_channels(mock_pa, None)) self.assertEqual(result, 1) class TestSupportedSampleRate(unittest.TestCase): def _run_with_pyaudio(self, func): mock_pyaudio = MagicMock() mock_pyaudio.paInt16 = 8 with patch.dict("sys.modules", {"pyaudio": mock_pyaudio}): return func() def test_default_rate_used(self): mock_pa = MagicMock() mock_pa.get_default_input_device_info.return_value = {"defaultSampleRate": 48000} mock_stream = MagicMock() mock_pa.open.return_value = mock_stream result = self._run_with_pyaudio(lambda: _get_supported_sample_rate(mock_pa, None, 1)) self.assertEqual(result, 48000) def test_all_open_fail_returns_16000(self): mock_pa = MagicMock() mock_pa.get_default_input_device_info.return_value = {"defaultSampleRate": 0} mock_pa.open.side_effect = IOError("no rate works") result = self._run_with_pyaudio(lambda: _get_supported_sample_rate(mock_pa, None, 1)) self.assertEqual(result, 16000) class TestManagerConstruction(unittest.TestCase): def test_vosk_engine(self): mgr = _make_manager(engine="vosk") self.assertEqual(mgr.engine, "vosk") def test_whisper_engine(self): mgr = _make_manager(engine="whisper") self.assertEqual(mgr.engine, "whisper") def test_whispercpp_engine(self): mgr = _make_manager(engine="whisper_cpp") self.assertEqual(mgr.engine, "whisper_cpp") def test_default_state(self): mgr = _make_manager() self.assertEqual(mgr.state, RecognitionState.IDLE) self.assertIsNone(mgr.model) class TestCallbacks(unittest.TestCase): def test_register_text_callback(self): mgr = _make_manager() cb = MagicMock() mgr.register_text_callback(cb) self.assertIn(cb, mgr.text_callbacks) def test_register_state_callback(self): mgr = _make_manager() cb = MagicMock() mgr.register_state_callback(cb) self.assertIn(cb, mgr.state_callbacks) def test_register_action_callback(self): mgr = _make_manager() cb = MagicMock() mgr.register_action_callback(cb) self.assertIn(cb, mgr.action_callbacks) def test_register_audio_level_callback(self): mgr = _make_manager() cb = MagicMock() mgr.register_audio_level_callback(cb) self.assertIn(cb, mgr._audio_level_callbacks) def test_update_state(self): mgr = _make_manager() cb = MagicMock() mgr.register_state_callback(cb) mgr._update_state(RecognitionState.LISTENING) self.assertEqual(mgr.state, RecognitionState.LISTENING) cb.assert_called_with(RecognitionState.LISTENING) class TestCancelDownload(unittest.TestCase): def test_cancel_download(self): mgr = _make_manager() mgr._download_cancelled = False mgr.cancel_download() self.assertTrue(mgr._download_cancelled) class TestStartStopRecognition(unittest.TestCase): def test_stop_recognition(self) -> None: mgr = _make_manager() mgr.state = RecognitionState.LISTENING mgr.should_record = True gate = _ReleaseGate() mgr._key_release_gate = gate mgr.audio_thread = MagicMock() mgr.audio_thread.is_alive.return_value = True mgr.audio_thread.join = MagicMock() mgr.stop_recognition() self.assertFalse(mgr.should_record) self.assertTrue(gate.is_set()) def test_stop_enqueues_final_buffer_with_gate_only_for_push_to_talk(self) -> None: for mode in ("push_to_talk", "toggle"): with self.subTest(mode=mode): mgr = _make_manager() mgr.state = RecognitionState.LISTENING mgr.should_record = True mgr._recognition_mode = mode mgr._key_release_gate = _ReleaseGate() mgr.audio_buffer = [b"speech"] mgr._recording_segment_has_speech = True mgr._enqueue_audio_segment = MagicMock() with patch("vocalinux.speech_recognition.recognition_manager.play_stop_sound"): mgr.stop_recognition() expected_gate = mgr._key_release_gate if mode == "push_to_talk" else None mgr._enqueue_audio_segment.assert_called_once_with( [b"speech"], release_gate=expected_gate, ) def test_stop_sound_plays_after_audio_thread_join(self): mgr = _make_manager() mgr.state = RecognitionState.LISTENING mgr.should_record = True mgr.audio_thread = MagicMock() mgr.audio_thread.is_alive.return_value = True events = [] mgr.audio_thread.join.side_effect = lambda timeout=None: events.append("join") with patch( "vocalinux.speech_recognition.recognition_manager.play_stop_sound", side_effect=lambda: events.append("sound"), ): mgr.stop_recognition() # sound must play after the audio thread has joined so the cue cannot be captured self.assertEqual(events[:2], ["join", "sound"]) def test_stop_sound_guard_chunk_calculation(self): mgr = _make_manager() self.assertEqual(mgr._get_stop_sound_guard_chunks(), 3) mgr.stop_sound_guard_ms = 0 self.assertEqual(mgr._get_stop_sound_guard_chunks(), 0) class TestReconfigure(unittest.TestCase): def test_reconfigure_language(self): mgr = _make_manager() mgr.state = RecognitionState.IDLE with patch.object(mgr, "_init_whispercpp"): mgr.reconfigure(language="fr") self.assertEqual(mgr.language, "fr") def test_reconfigure_model_size(self): mgr = _make_manager() mgr.state = RecognitionState.IDLE with patch.object(mgr, "_init_whispercpp"): mgr.reconfigure(model_size="medium", force_download=False) self.assertEqual(mgr.model_size, "medium") def test_reconfigure_engine(self): mgr = _make_manager() mgr.state = RecognitionState.IDLE with patch.object(mgr, "_init_vosk"): mgr.reconfigure(engine="vosk", force_download=False) self.assertEqual(mgr.engine, "vosk") def test_reconfigure_language_candidates(self) -> None: mgr = _make_manager() mgr.state = RecognitionState.IDLE with patch.object(mgr, "_init_whispercpp"): mgr.reconfigure(whispercpp_language_candidates="en,es", force_download=False) self.assertEqual(mgr.whispercpp_language_candidates, ["en", "es"]) class TestProcessFinalBuffer(unittest.TestCase): def test_process_final_buffer_whispercpp(self): mgr = _make_manager(engine="whisper_cpp") mgr.audio_buffer = [b"\x00\x00" * 512] * 5 mgr._capture_sample_rate = 16000 cb = MagicMock() mgr.register_text_callback(cb) mgr.command_processor = MagicMock() mgr.command_processor.process_text.return_value = "hello" with patch.object(mgr, "_transcribe_with_whispercpp", return_value="hello world"): mgr._process_final_buffer() def test_process_final_buffer_whisper(self): mgr = _make_manager(engine="whisper") mgr.audio_buffer = [b"\x00\x00" * 512] * 5 mgr._capture_sample_rate = 16000 mgr.command_processor = MagicMock() mgr.command_processor.process_text.return_value = "hello" with patch.object(mgr, "_transcribe_with_whisper", return_value="hello world"): mgr._process_final_buffer() def test_process_final_buffer_empty(self): mgr = _make_manager(engine="whisper_cpp") mgr.audio_buffer = [] mgr._process_final_buffer() # Should handle empty gracefully class TestTranscribeWhispercpp(unittest.TestCase): def test_normalize_language_candidates(self) -> None: self.assertEqual( SpeechRecognitionManager._normalize_language_candidates(" en-US, es ; fr "), ["en", "es", "fr"], ) self.assertEqual( SpeechRecognitionManager._normalize_language_candidates(["EN", "es", "en"]), ["en", "es"], ) self.assertEqual(SpeechRecognitionManager._normalize_language_candidates(""), []) self.assertEqual(SpeechRecognitionManager._normalize_language_candidates(None), []) def test_transcribe_success(self): mgr = _make_manager() mgr.language = "en-us" mock_segment = MagicMock() mock_segment.text = "hello world" mgr.model = MagicMock() mgr.model.transcribe.return_value = [mock_segment] mock_np = MagicMock() mock_np.frombuffer.return_value = MagicMock(__len__=lambda s: 16000) mock_np.frombuffer.return_value.astype.return_value = mock_np.frombuffer.return_value mock_np.int16 = "int16" mock_np.float32 = "float32" with patch.dict("sys.modules", {"numpy": mock_np}): result = mgr._transcribe_with_whispercpp([b"\x00\x00" * 512]) # Should return transcribed text def test_transcribe_model_none(self): mgr = _make_manager() mgr.model = None mock_np = MagicMock() mock_np.frombuffer.return_value = MagicMock(__len__=lambda s: 16000) mock_np.frombuffer.return_value.astype.return_value = mock_np.frombuffer.return_value mock_np.int16 = "int16" mock_np.float32 = "float32" with patch.dict("sys.modules", {"numpy": mock_np}): result = mgr._transcribe_with_whispercpp([b"\x00\x00" * 512]) self.assertEqual(result, "") def test_transcribe_empty_buffer(self): mgr = _make_manager() mock_np = MagicMock() with patch.dict("sys.modules", {"numpy": mock_np}): result = mgr._transcribe_with_whispercpp([]) self.assertEqual(result, "") def test_transcribe_exception(self): mgr = _make_manager() mgr.model = MagicMock() mgr.model.transcribe.side_effect = RuntimeError("crash") mock_np = MagicMock() mock_np.frombuffer.return_value = MagicMock(__len__=lambda s: 16000) mock_np.frombuffer.return_value.astype.return_value = mock_np.frombuffer.return_value mock_np.int16 = "int16" mock_np.float32 = "float32" with patch.dict("sys.modules", {"numpy": mock_np}): result = mgr._transcribe_with_whispercpp([b"\x00\x00" * 512]) self.assertEqual(result, "") def test_transcribe_auto_language(self): mgr = _make_manager() mgr.language = "auto" mock_segment = MagicMock() mock_segment.text = "bonjour" mgr.model = MagicMock() mgr.model.transcribe.return_value = [mock_segment] mock_np = MagicMock() mock_np.frombuffer.return_value = MagicMock(__len__=lambda s: 16000) mock_np.frombuffer.return_value.astype.return_value = mock_np.frombuffer.return_value mock_np.int16 = "int16" mock_np.float32 = "float32" with patch.dict("sys.modules", {"numpy": mock_np}): result = mgr._transcribe_with_whispercpp([b"\x00\x00" * 512]) self.assertEqual(result, "bonjour") mgr.model.transcribe.assert_called_once() self.assertIsNone(mgr.model.transcribe.call_args.kwargs["language"]) def test_transcribe_specific_language(self) -> None: mgr = _make_manager() mgr.language = "es" mock_segment = MagicMock() mock_segment.text = "hola" mgr.model = MagicMock() mgr.model.transcribe.return_value = [mock_segment] mock_np = MagicMock() mock_np.frombuffer.return_value = MagicMock(__len__=lambda s: 16000) mock_np.frombuffer.return_value.astype.return_value = mock_np.frombuffer.return_value mock_np.int16 = "int16" mock_np.float32 = "float32" with patch.dict("sys.modules", {"numpy": mock_np}): result = mgr._transcribe_with_whispercpp([b"\x00\x00" * 512]) self.assertEqual(result, "hola") mgr.model.transcribe.assert_called_once() self.assertEqual(mgr.model.transcribe.call_args.kwargs["language"], "es") def test_transcribe_candidate_languages_restricts_to_english(self) -> None: mgr = _make_manager(whispercpp_language_candidates="en,es") mgr.language = "auto" mock_segment = MagicMock() mock_segment.text = "hello" mgr.model = MagicMock() mgr.model.auto_detect_language.return_value = (("en", 0.8), {"en": 0.7, "es": 0.3}) mgr.model.transcribe.return_value = [mock_segment] mock_np = MagicMock() mock_np.frombuffer.return_value = MagicMock(__len__=lambda s: 16000) mock_np.frombuffer.return_value.astype.return_value = mock_np.frombuffer.return_value mock_np.int16 = "int16" mock_np.float32 = "float32" with patch.dict("sys.modules", {"numpy": mock_np}): result = mgr._transcribe_with_whispercpp([b"\x00\x00" * 512]) self.assertEqual(result, "hello") mgr.model.auto_detect_language.assert_called_once() mgr.model.transcribe.assert_called_once() self.assertEqual(mgr.model.transcribe.call_args.kwargs["language"], "en") def test_transcribe_candidate_languages_restricts_to_spanish(self) -> None: mgr = _make_manager(whispercpp_language_candidates=["en", "es", "fr"]) mgr.language = "auto" mock_segment = MagicMock() mock_segment.text = "hola" mgr.model = MagicMock() mgr.model.auto_detect_language.return_value = ( ("de", 0.6), {"en": 0.2, "es": 0.5, "fr": 0.1}, ) mgr.model.transcribe.return_value = [mock_segment] mock_np = MagicMock() mock_np.frombuffer.return_value = MagicMock(__len__=lambda s: 16000) mock_np.frombuffer.return_value.astype.return_value = mock_np.frombuffer.return_value mock_np.int16 = "int16" mock_np.float32 = "float32" with patch.dict("sys.modules", {"numpy": mock_np}): result = mgr._transcribe_with_whispercpp([b"\x00\x00" * 512]) self.assertEqual(result, "hola") mgr.model.auto_detect_language.assert_called_once() mgr.model.transcribe.assert_called_once() self.assertEqual(mgr.model.transcribe.call_args.kwargs["language"], "es") class TestTranscribeWhisper(unittest.TestCase): def test_transcribe_success(self): mgr = _make_manager(engine="whisper") mgr.language = "en-us" mgr.model = MagicMock() mgr.model.transcribe.return_value = {"text": "hello world"} mgr.model.device = MagicMock() mock_np = MagicMock() mock_np.frombuffer.return_value = MagicMock(__len__=lambda s: 16000) mock_np.frombuffer.return_value.astype.return_value = mock_np.frombuffer.return_value mock_np.int16 = "int16" mock_np.float32 = "float32" mock_torch = MagicMock() with patch.dict("sys.modules", {"numpy": mock_np, "torch": mock_torch}): result = mgr._transcribe_with_whisper([b"\x00\x00" * 512]) def test_transcribe_empty(self): mgr = _make_manager(engine="whisper") mock_np = MagicMock() with patch.dict("sys.modules", {"numpy": mock_np}): result = mgr._transcribe_with_whisper([]) self.assertEqual(result, "") def test_transcribe_model_none(self): mgr = _make_manager(engine="whisper") mgr.model = None mock_np = MagicMock() mock_np.frombuffer.return_value = MagicMock(__len__=lambda s: 16000) mock_np.frombuffer.return_value.astype.return_value = mock_np.frombuffer.return_value mock_np.int16 = "int16" mock_np.float32 = "float32" with patch.dict("sys.modules", {"numpy": mock_np, "torch": MagicMock()}): result = mgr._transcribe_with_whisper([b"\x00\x00" * 512]) self.assertEqual(result, "") class TestTranscribeParakeet(unittest.TestCase): def test_transcribe_create_stream_exception(self): mgr = _make_manager() mgr.model = MagicMock() mgr.model.create_stream.side_effect = RuntimeError("decode boom") mock_np = MagicMock() mock_np.frombuffer.return_value = MagicMock() mock_np.frombuffer.return_value.astype.return_value = MagicMock() with patch.dict("sys.modules", {"numpy": mock_np}): result = mgr._transcribe_with_parakeet([b"\x00\x00" * 512]) self.assertEqual(result, "") mgr.model.create_stream.assert_called_once() class TestInitVosk(unittest.TestCase): def test_init_vosk_model_exists(self): mgr = _make_manager(engine="vosk") with patch("os.path.exists", return_value=True): with patch("os.path.isdir", return_value=True): mock_vosk = MagicMock() mock_model = MagicMock() mock_recognizer = MagicMock() mock_vosk.Model.return_value = mock_model mock_vosk.KaldiRecognizer.return_value = mock_recognizer with patch.dict("sys.modules", {"vosk": mock_vosk}): mgr._init_vosk() self.assertTrue(mgr._model_initialized) self.assertEqual(mgr.model, mock_model) self.assertEqual(mgr.recognizer, mock_recognizer) def test_init_vosk_no_model(self): mgr = _make_manager(engine="vosk") mock_vosk = MagicMock() with patch("os.path.exists", return_value=False): with patch.dict("sys.modules", {"vosk": mock_vosk}): mgr._init_vosk() # When model doesn't exist and defer_download is True, should not initialize self.assertFalse(mgr._model_initialized) class TestInitWhisper(unittest.TestCase): def test_init_whisper(self): mgr = _make_manager(engine="whisper") mock_model = MagicMock() mock_whisper = MagicMock() mock_whisper.load_model.return_value = mock_model mock_torch = MagicMock() mock_torch.cuda.is_available.return_value = False with patch.dict("sys.modules", {"whisper": mock_whisper, "torch": mock_torch}): with patch("os.path.exists", return_value=True): mgr._init_whisper() self.assertTrue(mgr._model_initialized) self.assertEqual(mgr.model, mock_model) class TestInitWhispercpp(unittest.TestCase): def test_init_whispercpp_vulkan_fallback(self): mgr = _make_manager(engine="whisper_cpp") with patch("os.path.exists", return_value=True): mock_pywhispercpp = MagicMock() call_count = [0] def model_side_effect(*a, **kw): call_count[0] += 1 if call_count[0] == 1: raise RuntimeError("16-bit storage not supported") return MagicMock() mock_pywhispercpp.Model.side_effect = model_side_effect with patch.dict( "sys.modules", { "pywhispercpp": MagicMock(), "pywhispercpp.model": mock_pywhispercpp, }, ): with patch.object( SpeechRecognitionManager, "_whispercpp_model_is_verified", return_value=True, ): try: mgr._init_whispercpp() except Exception: pass def test_cpu_fallback_filters_unsupported_whispercpp_params(self) -> None: for override, original_threads, expected_threads in ( (None, 2, 6), (3, 2, 3), ): with self.subTest(override=override): mgr = _make_manager(engine="whisper_cpp", whispercpp_n_threads=override) model = MagicMock() mock_pywhispercpp = MagicMock() mock_pywhispercpp.Model.return_value = model model_kwargs = {"n_threads": original_threads, "no_context": True} with patch.dict( "sys.modules", { "pywhispercpp": MagicMock(), "pywhispercpp.model": mock_pywhispercpp, }, ): with patch.object( mgr, "_get_supported_whispercpp_params", return_value={"n_threads"}, ): with patch( "vocalinux.speech_recognition.recognition_manager.multiprocessing.cpu_count", return_value=8, ): with patch( "vocalinux.speech_recognition.recognition_manager._show_notification" ): with patch.dict("os.environ", {}, clear=False): loaded_backend = mgr._handle_gpu_fallback( RuntimeError("16-bit storage not supported"), "/tmp/model.bin", model_kwargs, "cpu", ) self.assertEqual(loaded_backend, "cpu") self.assertIs(mgr.model, model) self.assertEqual(mock_pywhispercpp.Model.call_count, 1) self.assertEqual( mock_pywhispercpp.Model.call_args.kwargs, {"n_threads": expected_threads}, ) self.assertEqual(model_kwargs["n_threads"], original_threads) class TestDownloadVoskModel(unittest.TestCase): pass class TestBufferManagement(unittest.TestCase): def test_set_buffer_limit(self): mgr = _make_manager() mgr.set_buffer_limit(100) self.assertEqual(mgr._max_buffer_size, 100) def test_get_buffer_stats(self): mgr = _make_manager() stats = mgr.get_buffer_stats() self.assertIsInstance(stats, dict) class TestWhispercppModelKwargs(unittest.TestCase): def test_default_model_kwargs(self) -> None: mgr = _make_manager(engine="whisper_cpp") kwargs = mgr._build_whispercpp_model_kwargs(n_threads=4) self.assertTrue(kwargs.get("no_timestamps")) self.assertTrue(kwargs.get("no_context")) self.assertFalse(kwargs["print_progress"]) self.assertNotIn("initial_prompt", kwargs) self.assertEqual(kwargs.get("temperature"), 0.0) self.assertEqual(kwargs.get("n_threads"), 4) def test_model_kwargs_all_switches_off(self): mgr = _make_manager( engine="whisper_cpp", whispercpp_no_timestamps=False, whispercpp_no_context=False, ) kwargs = mgr._build_whispercpp_model_kwargs(n_threads=4) self.assertNotIn("no_timestamps", kwargs) self.assertNotIn("no_context", kwargs) def test_model_kwargs_with_initial_prompt(self): mgr = _make_manager( engine="whisper_cpp", whispercpp_initial_prompt="test prompt", ) kwargs = mgr._build_whispercpp_model_kwargs(n_threads=4) self.assertEqual(kwargs.get("initial_prompt"), "test prompt") def test_model_kwargs_custom_numerics(self): mgr = _make_manager( engine="whisper_cpp", whispercpp_temperature=0.7, whispercpp_temperature_inc=0.2, whispercpp_entropy_thold=3.0, whispercpp_logprob_thold=-2.5, whispercpp_no_speech_thold=0.8, ) kwargs = mgr._build_whispercpp_model_kwargs(n_threads=8) self.assertEqual(kwargs.get("temperature"), 0.7) self.assertEqual(kwargs.get("temperature_inc"), 0.2) self.assertEqual(kwargs.get("entropy_thold"), 3.0) self.assertEqual(kwargs.get("logprob_thold"), -2.5) self.assertEqual(kwargs.get("no_speech_thold"), 0.8) self.assertEqual(kwargs.get("n_threads"), 8) def test_model_kwargs_filter_unsupported_native_params(self) -> None: mgr = _make_manager(engine="whisper_cpp") kwargs = mgr._build_whispercpp_model_kwargs(n_threads=4) filtered = mgr._filter_whispercpp_model_kwargs( kwargs, supported_params={ "n_threads", "suppress_blank", "no_speech_thold", "entropy_thold", "logprob_thold", "temperature", "temperature_inc", "no_context", "initial_prompt", }, ) self.assertNotIn("no_timestamps", filtered) self.assertNotIn("print_progress", filtered) self.assertEqual(filtered["n_threads"], 4) self.assertTrue(filtered["no_context"]) def test_reconfigure_whispercpp_params(self): mgr = _make_manager(engine="whisper_cpp") original_temp_inc = mgr.whispercpp_temperature_inc with patch.object(SpeechRecognitionManager, "_init_whispercpp"): mgr.reconfigure( whispercpp_temperature=0.5, whispercpp_no_timestamps=False, whispercpp_initial_prompt="hello", force_download=False, ) self.assertEqual(mgr.whispercpp_temperature, 0.5) self.assertFalse(mgr.whispercpp_no_timestamps) self.assertEqual(mgr.whispercpp_initial_prompt, "hello") self.assertEqual(mgr.whispercpp_temperature_inc, original_temp_inc) class TestVoiceCommandsProperty(unittest.TestCase): def test_voice_commands_vosk_default(self): mgr = _make_manager(engine="vosk") # For vosk engine, voice commands should be enabled by default self.assertTrue(mgr._resolve_voice_commands_enabled()) def test_voice_commands_whisper_default(self): mgr = _make_manager(engine="whisper") self.assertFalse(mgr._resolve_voice_commands_enabled()) def test_voice_commands_explicit_on(self): mgr = _make_manager(engine="whisper", voice_commands_enabled=True) self.assertTrue(mgr._resolve_voice_commands_enabled()) def test_voice_commands_explicit_off(self): mgr = _make_manager(engine="vosk", voice_commands_enabled=False) self.assertFalse(mgr._resolve_voice_commands_enabled()) class TestPendingTranscription(unittest.TestCase): def test_live_non_current_worker_is_pending(self) -> None: manager = _make_manager() started = threading.Event() release = threading.Event() def worker() -> None: started.set() release.wait(timeout=2.0) thread = threading.Thread(target=worker, daemon=True) thread.start() try: self.assertTrue(started.wait(timeout=1.0)) manager.recognition_thread = thread self.assertTrue(manager.has_pending_transcription()) finally: release.set() thread.join(timeout=1.0) def test_superseded_worker_is_still_pending(self) -> None: manager = _make_manager() started = threading.Event() release = threading.Event() def worker_a() -> None: started.set() release.wait(timeout=2.0) def worker_b() -> None: return first = threading.Thread(target=worker_a, daemon=True) manager._track_recognition_worker(first) first.start() self.assertTrue(started.wait(timeout=1.0)) second = threading.Thread(target=worker_b, daemon=True) second.start() second.join(timeout=1.0) manager.recognition_thread = second try: self.assertTrue(manager.has_pending_transcription()) finally: release.set() first.join(timeout=1.0) self.assertFalse(manager.has_pending_transcription()) def test_track_recognition_worker_prunes_finished_workers(self) -> None: manager = _make_manager() def finished_worker() -> None: return finished = threading.Thread(target=finished_worker, daemon=True) finished.start() finished.join(timeout=1.0) manager._track_recognition_worker(finished) started = threading.Event() release = threading.Event() def active_worker() -> None: started.set() release.wait(timeout=2.0) active = threading.Thread(target=active_worker, daemon=True) manager._track_recognition_worker(active) active.start() try: self.assertTrue(started.wait(timeout=1.0)) self.assertEqual(manager._recognition_workers, [active]) finally: release.set() active.join(timeout=1.0) def test_start_recognition_tracks_worker(self) -> None: manager = _make_manager() manager._model_initialized = True manager.model = MagicMock() started = threading.Event() release = threading.Event() def record_audio() -> None: return def perform_recognition() -> None: started.set() release.wait(timeout=2.0) manager._record_audio = record_audio manager._perform_recognition = perform_recognition try: with ( patch("vocalinux.speech_recognition.recognition_manager.play_start_sound"), patch("vocalinux.speech_recognition.recognition_manager.play_stop_sound"), patch.object(manager, "_arm_playback_duck"), ): self.assertTrue(manager.start_recognition()) worker = manager.recognition_thread self.assertIsNotNone(worker) self.assertIn(worker, manager._recognition_workers) self.assertTrue(started.wait(timeout=1.0)) release.set() manager.stop_recognition() finally: release.set() if manager.recognition_thread is not None: manager.recognition_thread.join(timeout=1.0) def test_queued_audio_without_live_worker_is_not_pending(self) -> None: manager = _make_manager() manager._segment_queue.put(([], "en")) self.assertFalse(manager.has_pending_transcription()) def test_idle_manager_without_queued_audio_is_not_pending(self) -> None: manager = _make_manager() self.assertFalse(manager.has_pending_transcription()) def test_stop_recognition_signal_is_not_pending(self) -> None: manager = _make_manager() manager.state = RecognitionState.LISTENING manager.should_record = True started = threading.Event() def worker() -> None: started.set() manager._perform_recognition() thread = threading.Thread(target=worker, daemon=True) manager.recognition_thread = thread thread.start() self.assertTrue(started.wait(timeout=1.0)) with patch("vocalinux.speech_recognition.recognition_manager.play_stop_sound"): manager.stop_recognition() self.assertFalse(manager.has_pending_transcription()) def test_orphaned_final_segment_is_not_pending(self) -> None: manager = _make_manager() manager.state = RecognitionState.LISTENING manager.should_record = True started = threading.Event() audio_release = threading.Event() def worker() -> None: started.set() manager._perform_recognition() def audio_worker() -> None: audio_release.wait(timeout=1.0) recognition_thread = threading.Thread(target=worker, daemon=True) manager.recognition_thread = recognition_thread recognition_thread.start() self.assertTrue(started.wait(timeout=1.0)) audio_thread = threading.Thread(target=audio_worker, daemon=True) manager.audio_thread = audio_thread audio_thread.start() manager.audio_buffer = [b"\x00\x00"] * 8 manager._recording_segment_has_speech = True try: with patch("vocalinux.speech_recognition.recognition_manager.play_stop_sound"): manager.stop_recognition() finally: audio_release.set() audio_thread.join(timeout=1.0) recognition_thread.join(timeout=1.0) with manager._segment_queue.mutex: self.assertTrue(any(item is not None for item in manager._segment_queue.queue)) self.assertFalse(manager.has_pending_transcription()) def test_worker_does_not_count_itself_as_pending(self) -> None: manager = _make_manager() manager.recognition_thread = threading.current_thread() self.assertFalse(manager.has_pending_transcription()) if __name__ == "__main__": unittest.main()