xavierk
80a3792576
Import the vocalinux source tree (tray app, Next.js site in web/, packaging, installers, docs and tests) on top of the Gitea template commit. The project README replaces the template README, and the template CI files (.gitea/workflows/ci.yml, .jscpd.json, CI.md) are kept.
1355 lines
48 KiB
Python
1355 lines
48 KiB
Python
"""
|
|
Final coverage tests for recognition_manager.py - targeting uncovered lines.
|
|
|
|
Focus areas:
|
|
- Download functions (VOSK, Whisper, whisper.cpp)
|
|
- Error handling paths
|
|
- Buffer management edge cases
|
|
- Audio device detection edge cases
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import shutil
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import unittest
|
|
from contextlib import ExitStack
|
|
from unittest.mock import MagicMock, Mock, PropertyMock, mock_open, patch
|
|
|
|
# Mock modules BEFORE importing anything from vocalinux
|
|
sys.modules["pyaudio"] = MagicMock()
|
|
sys.modules["vosk"] = MagicMock()
|
|
sys.modules["whisper"] = MagicMock()
|
|
sys.modules["torch"] = MagicMock()
|
|
sys.modules["pywhispercpp"] = MagicMock()
|
|
sys.modules["pywhispercpp.model"] = MagicMock()
|
|
sys.modules["requests"] = MagicMock()
|
|
sys.modules["numpy"] = MagicMock()
|
|
sys.modules["psutil"] = MagicMock()
|
|
sys.modules["zipfile"] = MagicMock()
|
|
|
|
# Import after mocking
|
|
from conftest import mock_audio_feedback
|
|
|
|
import vocalinux.speech_recognition.recognition_manager as recognition_manager_module
|
|
from vocalinux.common_types import RecognitionState
|
|
from vocalinux.speech_recognition.command_processor import CommandProcessor
|
|
from vocalinux.speech_recognition.recognition_manager import (
|
|
SpeechRecognitionManager,
|
|
_get_supported_channels,
|
|
_get_supported_sample_rate,
|
|
get_audio_input_devices,
|
|
)
|
|
|
|
|
|
class TestDownloadFunctions(unittest.TestCase):
|
|
"""Test download functions and progress tracking."""
|
|
|
|
def setUp(self):
|
|
"""Set up test fixtures."""
|
|
self.patches = []
|
|
|
|
def tearDown(self):
|
|
"""Clean up patches."""
|
|
for p in self.patches:
|
|
p.stop()
|
|
|
|
def _create_manager(self, engine="vosk", defer_download=True):
|
|
"""Helper to create a manager with mocked downloads."""
|
|
patches = [
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
patch.object(SpeechRecognitionManager, "_download_vosk_model"),
|
|
patch.object(SpeechRecognitionManager, "_download_whispercpp_model"),
|
|
patch("os.path.exists"),
|
|
patch("os.makedirs"),
|
|
patch("threading.Thread"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(
|
|
engine=engine, defer_download=defer_download, model_size="tiny"
|
|
)
|
|
return manager
|
|
|
|
def test_download_whisper_model_cancelled(self):
|
|
"""Test Whisper download cancellation."""
|
|
# Verify the cancelled flag exists and can be set
|
|
manager = self._create_manager(engine="whisper")
|
|
self.assertFalse(manager._download_cancelled)
|
|
manager._download_cancelled = True
|
|
self.assertTrue(manager._download_cancelled)
|
|
|
|
def test_download_cancelled_flag(self):
|
|
"""Test that download can be marked as cancelled."""
|
|
manager = self._create_manager(engine="whisper")
|
|
|
|
# Initial state should be False
|
|
assert manager._download_cancelled is False
|
|
|
|
# Test that setting flag to True works
|
|
manager._download_cancelled = True
|
|
assert manager._download_cancelled is True
|
|
|
|
# Reset and verify again
|
|
manager._download_cancelled = False
|
|
assert manager._download_cancelled is False
|
|
|
|
def test_download_claim_resets_cancellation_for_the_next_attempt(self) -> None:
|
|
"""A stale cancel from one attempt must not cancel the next one."""
|
|
manager = self._create_manager(engine="whisper")
|
|
manager._download_cancelled = True
|
|
|
|
assert manager.try_begin_download() is True
|
|
assert manager._download_cancelled is False
|
|
manager.cancel_download()
|
|
manager.end_download()
|
|
assert manager._download_cancelled is False
|
|
|
|
def _assert_missing_model_dispatch(self, engine: str, model_size: str, downloader: str) -> None:
|
|
manager = self._create_manager()
|
|
manager.engine = engine
|
|
manager.model_size = model_size
|
|
with ExitStack() as stack:
|
|
availability = None
|
|
if engine == "whisper_cpp":
|
|
stack.enter_context(
|
|
patch.object(
|
|
recognition_manager_module,
|
|
"get_model_path",
|
|
return_value="/models/target",
|
|
)
|
|
)
|
|
availability = stack.enter_context(
|
|
patch.object(recognition_manager_module.os.path, "exists")
|
|
)
|
|
elif engine == "whisper":
|
|
stack.enter_context(
|
|
patch.object(
|
|
manager,
|
|
"_whisper_model_cache_paths",
|
|
return_value=("/models/whisper", "/models/whisper/target.pt"),
|
|
)
|
|
)
|
|
availability = stack.enter_context(
|
|
patch.object(recognition_manager_module.os.path, "exists")
|
|
)
|
|
elif engine == "faster_whisper":
|
|
availability = stack.enter_context(
|
|
patch.object(recognition_manager_module.faster_whisper, "is_model_downloaded")
|
|
)
|
|
elif engine == "parakeet":
|
|
availability = stack.enter_context(
|
|
patch.object(recognition_manager_module.parakeet, "is_model_downloaded")
|
|
)
|
|
else:
|
|
stack.enter_context(
|
|
patch.object(manager, "_get_vosk_model_path", return_value="/models/target")
|
|
)
|
|
availability = stack.enter_context(
|
|
patch.object(recognition_manager_module.os.path, "exists")
|
|
)
|
|
download = stack.enter_context(patch.object(manager, downloader))
|
|
|
|
for downloaded in (True, False):
|
|
with self.subTest(engine=engine, downloaded=downloaded):
|
|
availability.return_value = downloaded
|
|
manager._download_selected_model_if_missing()
|
|
if downloaded:
|
|
download.assert_not_called()
|
|
elif engine == "whisper":
|
|
download.assert_called_once_with("/models/whisper")
|
|
else:
|
|
download.assert_called_once_with()
|
|
download.reset_mock()
|
|
|
|
def test_download_selected_model_dispatches_whispercpp(self) -> None:
|
|
self._assert_missing_model_dispatch("whisper_cpp", "small", "_download_whispercpp_model")
|
|
|
|
def test_download_selected_model_dispatches_whisper(self) -> None:
|
|
self._assert_missing_model_dispatch("whisper", "small", "_download_whisper_model")
|
|
|
|
def test_download_selected_model_dispatches_faster_whisper(self) -> None:
|
|
self._assert_missing_model_dispatch(
|
|
"faster_whisper", "tiny", "_download_faster_whisper_model"
|
|
)
|
|
|
|
def test_download_selected_model_dispatches_parakeet(self) -> None:
|
|
self._assert_missing_model_dispatch("parakeet", "v3-european", "_download_parakeet_model")
|
|
|
|
def test_download_selected_model_dispatches_vosk(self) -> None:
|
|
self._assert_missing_model_dispatch("vosk", "small", "_download_vosk_model")
|
|
|
|
def test_reconfigure_download_cancellation_preserves_the_loaded_model(self) -> None:
|
|
manager = self._create_manager()
|
|
loaded_model = object()
|
|
manager.model = loaded_model
|
|
manager.engine = "whisper"
|
|
manager.model_size = "small"
|
|
manager.language = "en-us"
|
|
manager.command_processor = Mock()
|
|
restore_language = patch.object(manager, "_restore_language_after_oneshot")
|
|
restore_language_mock = restore_language.start()
|
|
self.patches.append(restore_language)
|
|
|
|
with (
|
|
patch.object(
|
|
recognition_manager_module,
|
|
"get_model_path",
|
|
return_value="/models/medium.bin",
|
|
),
|
|
patch.object(recognition_manager_module.os.path, "exists", return_value=False),
|
|
patch.object(
|
|
manager,
|
|
"_download_whispercpp_model",
|
|
side_effect=RuntimeError("Download cancelled"),
|
|
) as download,
|
|
patch.object(manager, "_init_selected_engine") as init_selected,
|
|
):
|
|
with self.assertRaisesRegex(RuntimeError, "Download cancelled"):
|
|
manager.reconfigure(
|
|
engine="whisper_cpp",
|
|
model_size="medium",
|
|
language="fr",
|
|
force_download=True,
|
|
)
|
|
|
|
assert manager.model is loaded_model
|
|
assert manager.engine == "whisper"
|
|
assert manager.model_size == "small"
|
|
assert manager.language == "en-us"
|
|
assert manager._defer_download is True
|
|
assert manager._model_lock.locked() is False
|
|
assert manager._prefetching_model is False
|
|
download.assert_called_once_with()
|
|
init_selected.assert_not_called()
|
|
manager.command_processor.set_language.assert_any_call("fr")
|
|
manager.command_processor.set_language.assert_any_call("en-us")
|
|
restore_language_mock.assert_called_once_with()
|
|
|
|
def test_reconfigure_prefetches_before_initializing_the_selected_engine(self) -> None:
|
|
manager = self._create_manager()
|
|
manager.engine = "whisper"
|
|
available = {"model": False}
|
|
loaded_model = object()
|
|
|
|
def model_exists(_path: str) -> bool:
|
|
return available["model"]
|
|
|
|
def download_model() -> None:
|
|
assert manager._prefetching_model is True
|
|
assert manager.start_recognition() is False
|
|
available["model"] = True
|
|
|
|
with (
|
|
patch.object(
|
|
recognition_manager_module,
|
|
"get_model_path",
|
|
return_value="/models/medium.bin",
|
|
),
|
|
patch.object(recognition_manager_module.os.path, "exists", side_effect=model_exists),
|
|
patch.object(recognition_manager_module, "_preload_pywhispercpp_shared_libraries"),
|
|
patch.object(manager, "_whispercpp_model_is_verified", return_value=True),
|
|
patch.object(
|
|
manager, "_download_whispercpp_model", side_effect=download_model
|
|
) as download,
|
|
patch.object(
|
|
manager,
|
|
"_load_whispercpp_model",
|
|
side_effect=lambda _path: setattr(manager, "model", loaded_model),
|
|
),
|
|
patch.object(
|
|
manager, "_init_selected_engine", wraps=manager._init_selected_engine
|
|
) as init,
|
|
patch.object(recognition_manager_module, "play_error_sound") as error_sound,
|
|
patch.object(recognition_manager_module, "_show_notification") as notification,
|
|
):
|
|
manager.reconfigure(
|
|
engine="whisper_cpp",
|
|
model_size="medium",
|
|
force_download=True,
|
|
)
|
|
|
|
assert available["model"] is True
|
|
download.assert_called_once_with()
|
|
init.assert_called_once_with()
|
|
assert manager.model is loaded_model
|
|
assert manager._prefetching_model is False
|
|
error_sound.assert_called_once_with()
|
|
notification.assert_called_once_with(
|
|
"Speech model downloading",
|
|
"Dictation will be available when the new model has finished downloading.",
|
|
"dialog-information",
|
|
)
|
|
|
|
def test_resume_during_prefetch_still_loads_the_selected_model(self) -> None:
|
|
"""Wake must not supersede the Settings worker that will load the model."""
|
|
manager = self._create_manager(engine="whisper")
|
|
previous_model = object()
|
|
selected_model = object()
|
|
manager.model = previous_model
|
|
manager._model_initialized = True
|
|
|
|
def prefetch() -> None:
|
|
assert manager._prefetching_model
|
|
manager.reinitialize_after_resume(auto_resume=True)
|
|
assert manager.is_auto_paused
|
|
|
|
def initialize() -> None:
|
|
manager.model = selected_model
|
|
manager._model_initialized = True
|
|
|
|
with (
|
|
patch.object(manager, "_download_selected_model_if_missing", side_effect=prefetch),
|
|
patch.object(manager, "_init_selected_engine", side_effect=initialize) as init,
|
|
):
|
|
manager.reconfigure(engine="whisper_cpp", model_size="medium", force_download=True)
|
|
|
|
init.assert_called_once_with()
|
|
assert manager.model is selected_model
|
|
assert manager.engine == "whisper_cpp"
|
|
assert manager.model_size == "medium"
|
|
assert not manager.is_auto_paused
|
|
assert not manager._auto_resume_pending
|
|
|
|
def test_failed_prefetch_completes_inherited_resume_with_previous_model(self) -> None:
|
|
"""Rejected Settings must not abandon a wake request that they superseded."""
|
|
manager = self._create_manager(engine="whisper")
|
|
previous_model = object()
|
|
manager.model = previous_model
|
|
manager._model_initialized = True
|
|
manager._auto_paused = True
|
|
manager._auto_resume_pending = True
|
|
|
|
with (
|
|
patch.object(
|
|
manager,
|
|
"_download_selected_model_if_missing",
|
|
side_effect=RuntimeError("Download cancelled"),
|
|
),
|
|
patch.object(manager, "_init_selected_engine") as init,
|
|
):
|
|
with self.assertRaisesRegex(RuntimeError, "Download cancelled"):
|
|
manager.reconfigure(engine="whisper_cpp", model_size="medium", force_download=True)
|
|
|
|
init.assert_not_called()
|
|
assert manager.model is previous_model
|
|
assert manager.engine == "whisper"
|
|
assert not manager.is_auto_paused
|
|
assert not manager._auto_resume_pending
|
|
|
|
def test_failed_prefetch_reloads_previous_engine_for_inherited_resume(self) -> None:
|
|
"""An already unloaded model is recovered after the new download fails."""
|
|
manager = self._create_manager(engine="whisper")
|
|
previous_model = object()
|
|
manager.model = None
|
|
manager._model_initialized = False
|
|
manager._auto_paused = True
|
|
manager._auto_resume_pending = True
|
|
|
|
def initialize() -> None:
|
|
manager.model = previous_model
|
|
manager._model_initialized = True
|
|
|
|
with (
|
|
patch.object(
|
|
manager,
|
|
"_download_selected_model_if_missing",
|
|
side_effect=RuntimeError("Download cancelled"),
|
|
),
|
|
patch.object(manager, "_init_whisper", side_effect=initialize) as init,
|
|
):
|
|
with self.assertRaisesRegex(RuntimeError, "Download cancelled"):
|
|
manager.reconfigure(engine="whisper_cpp", model_size="medium", force_download=True)
|
|
|
|
init.assert_called_once_with()
|
|
assert manager.model is previous_model
|
|
assert manager.engine == "whisper"
|
|
assert manager.model_ready
|
|
assert not manager.is_auto_paused
|
|
assert not manager._auto_resume_pending
|
|
|
|
def test_cancelled_model_lock_wait_completes_inherited_resume(self) -> None:
|
|
"""Cancelling behind a busy engine preserves the usable previous model."""
|
|
manager = self._create_manager(engine="whisper")
|
|
previous_model = object()
|
|
manager.model = previous_model
|
|
manager._model_initialized = True
|
|
manager._auto_paused = True
|
|
manager._auto_resume_pending = True
|
|
manager._model_lock.acquire()
|
|
assert manager.try_begin_download()
|
|
manager.cancel_download()
|
|
try:
|
|
with self.assertRaisesRegex(RuntimeError, "Download cancelled"):
|
|
manager.reconfigure(engine="whisper_cpp", model_size="medium", force_download=False)
|
|
assert manager._model_lock.locked()
|
|
finally:
|
|
manager._model_lock.release()
|
|
manager.end_download()
|
|
|
|
assert manager.model is previous_model
|
|
assert manager.engine == "whisper"
|
|
assert not manager.is_auto_paused
|
|
assert not manager._auto_resume_pending
|
|
|
|
def test_reconfigure_without_download_permission_skips_prefetch(self) -> None:
|
|
manager = self._create_manager()
|
|
manager.engine = "whisper"
|
|
|
|
with (
|
|
patch.object(manager, "_download_selected_model_if_missing") as prefetch,
|
|
patch.object(manager, "_init_selected_engine"),
|
|
):
|
|
manager.reconfigure(
|
|
engine="whisper_cpp",
|
|
model_size="medium",
|
|
force_download=False,
|
|
)
|
|
|
|
prefetch.assert_not_called()
|
|
|
|
def test_resume_skips_engine_reinit_while_model_is_prefetching(self) -> None:
|
|
manager = self._create_manager(engine="whisper")
|
|
manager.engine = "whisper_cpp"
|
|
manager._prefetching_model = True
|
|
manager._auto_paused = True
|
|
manager._idle_unloaded = True
|
|
|
|
with patch.object(manager, "_init_whispercpp") as init_whispercpp:
|
|
manager.reinitialize_after_resume()
|
|
|
|
init_whispercpp.assert_not_called()
|
|
self.assertFalse(manager._auto_paused)
|
|
self.assertFalse(manager._idle_unloaded)
|
|
|
|
def test_lazy_reload_follows_a_cancelled_prefetch_after_resume(self) -> None:
|
|
manager = self._create_manager(engine="whisper")
|
|
manager.engine = "whisper_cpp"
|
|
manager.model = None
|
|
manager._model_initialized = False
|
|
manager._prefetching_model = True
|
|
manager._auto_paused = True
|
|
|
|
manager.reinitialize_after_resume()
|
|
manager._prefetching_model = False
|
|
|
|
with patch.object(manager, "_init_whispercpp") as init_whispercpp:
|
|
manager.ensure_model_loaded()
|
|
|
|
init_whispercpp.assert_called_once()
|
|
|
|
def test_start_recognition_refuses_while_a_model_is_prefetching(self) -> None:
|
|
manager = self._create_manager()
|
|
manager.engine = "whisper_cpp"
|
|
manager._prefetching_model = True
|
|
uninitialized_manager = object.__new__(SpeechRecognitionManager)
|
|
assert uninitialized_manager._prefetching_model is False
|
|
|
|
with (
|
|
patch.object(recognition_manager_module, "play_error_sound") as error_sound,
|
|
patch.object(recognition_manager_module, "_show_notification") as notification,
|
|
):
|
|
assert manager.start_recognition() is False
|
|
manager.language = "pl"
|
|
with patch.object(
|
|
manager, "start_recognition", wraps=manager.start_recognition
|
|
) as start:
|
|
assert manager.start_recognition_with_language("pl") is False
|
|
|
|
start.assert_called_once_with(mode="toggle")
|
|
assert error_sound.call_count == 2
|
|
notification.assert_called_with(
|
|
"Speech model downloading",
|
|
"Dictation will be available when the new model has finished downloading.",
|
|
"dialog-information",
|
|
)
|
|
assert notification.call_count == 2
|
|
|
|
def test_download_vosk_model_with_progress(self):
|
|
"""Test VOSK model download with progress tracking."""
|
|
# Test that progress callback can be registered
|
|
manager = self._create_manager(engine="vosk")
|
|
manager.vosk_model_map = {"small": "en-us_0"}
|
|
|
|
progress_data = []
|
|
manager._download_progress_callback = lambda p, s, st: progress_data.append(p)
|
|
|
|
assert manager._download_progress_callback is not None
|
|
assert callable(manager._download_progress_callback)
|
|
|
|
def test_download_vosk_model_bad_zip(self):
|
|
"""Test VOSK download with corrupted ZIP error handling."""
|
|
# Verify that BadZipFile handling is in place in the code
|
|
manager = self._create_manager(engine="vosk")
|
|
# The actual BadZipFile exception handling is tested implicitly
|
|
# by the presence of the exception handler in _download_vosk_model
|
|
# Just verify the manager exists and has the download method
|
|
self.assertTrue(hasattr(manager, "_download_vosk_model"))
|
|
self.assertTrue(callable(manager._download_vosk_model))
|
|
|
|
|
|
class TestAudioDeviceDetection(unittest.TestCase):
|
|
"""Test audio device detection functions."""
|
|
|
|
def setUp(self):
|
|
"""Set up mocks."""
|
|
self.pyaudio_mock = MagicMock()
|
|
sys.modules["pyaudio"] = self.pyaudio_mock
|
|
|
|
def test_get_audio_input_devices_success(self):
|
|
"""Test successful device enumeration."""
|
|
mock_audio = MagicMock()
|
|
mock_audio.get_device_count.return_value = 2
|
|
mock_audio.get_default_input_device_info.return_value = {"index": 0}
|
|
mock_audio.get_device_info_by_index.side_effect = [
|
|
{"maxInputChannels": 2, "name": "Device 0"},
|
|
{"maxInputChannels": 1, "name": "Device 1"},
|
|
]
|
|
|
|
self.pyaudio_mock.PyAudio.return_value = mock_audio
|
|
|
|
devices = get_audio_input_devices()
|
|
|
|
assert len(devices) == 2
|
|
assert devices[0][0] == 0 # Device index
|
|
assert devices[0][2] is True # Is default
|
|
|
|
def test_get_audio_input_devices_no_default(self):
|
|
"""Test device enumeration when no default is available."""
|
|
mock_audio = MagicMock()
|
|
mock_audio.get_device_count.return_value = 1
|
|
mock_audio.get_default_input_device_info.side_effect = IOError("No default")
|
|
mock_audio.get_device_info_by_index.return_value = {
|
|
"maxInputChannels": 2,
|
|
"name": "Device 0",
|
|
}
|
|
|
|
self.pyaudio_mock.PyAudio.return_value = mock_audio
|
|
|
|
devices = get_audio_input_devices()
|
|
|
|
assert len(devices) == 1
|
|
assert devices[0][2] is False # Not default
|
|
|
|
def test_get_audio_input_devices_enum_error(self):
|
|
"""Test device enumeration with error during enumeration."""
|
|
mock_audio = MagicMock()
|
|
mock_audio.get_device_count.return_value = 2
|
|
mock_audio.get_default_input_device_info.return_value = {"index": 0}
|
|
# First call succeeds, second fails
|
|
mock_audio.get_device_info_by_index.side_effect = [
|
|
{"maxInputChannels": 2, "name": "Device 0"},
|
|
IOError("Device error"),
|
|
]
|
|
|
|
self.pyaudio_mock.PyAudio.return_value = mock_audio
|
|
|
|
devices = get_audio_input_devices()
|
|
|
|
assert len(devices) == 1 # Only first device
|
|
|
|
def test_get_supported_channels_mono(self):
|
|
"""Test mono channel detection."""
|
|
mock_audio = MagicMock()
|
|
mock_stream = MagicMock()
|
|
mock_audio.open.return_value = mock_stream
|
|
|
|
self.pyaudio_mock.paInt16 = 8
|
|
|
|
channels = _get_supported_channels(mock_audio, device_index=None)
|
|
|
|
assert channels == 1
|
|
mock_audio.open.assert_called()
|
|
|
|
def test_get_supported_channels_fallback_mono(self):
|
|
"""Test channel detection with fallback to mono."""
|
|
mock_audio = MagicMock()
|
|
# Both attempts fail
|
|
mock_audio.open.side_effect = IOError("Channels not supported")
|
|
|
|
self.pyaudio_mock.paInt16 = 8
|
|
|
|
channels = _get_supported_channels(mock_audio, device_index=None)
|
|
|
|
assert channels == 1 # Default fallback
|
|
|
|
def test_get_supported_sample_rate_default_rate(self):
|
|
"""Test sample rate detection using device default."""
|
|
mock_audio = MagicMock()
|
|
mock_stream = MagicMock()
|
|
mock_audio.open.return_value = mock_stream
|
|
mock_audio.get_device_info_by_index.return_value = {"defaultSampleRate": 16000}
|
|
|
|
self.pyaudio_mock.paInt16 = 8
|
|
|
|
rate = _get_supported_sample_rate(mock_audio, device_index=0, channels=1)
|
|
|
|
assert rate == 16000
|
|
|
|
def test_get_supported_sample_rate_fallback(self):
|
|
"""Test sample rate fallback when default fails."""
|
|
mock_audio = MagicMock()
|
|
mock_stream = MagicMock()
|
|
|
|
# First call (default) fails, second (16000) succeeds
|
|
mock_audio.open.side_effect = [IOError(), mock_stream]
|
|
mock_audio.get_device_info_by_index.return_value = {
|
|
"defaultSampleRate": 48000 # Not in COMMON_RATES, will try others
|
|
}
|
|
|
|
self.pyaudio_mock.paInt16 = 8
|
|
|
|
rate = _get_supported_sample_rate(mock_audio, device_index=0, channels=1)
|
|
|
|
assert rate in [48000, 44100, 32000, 22050, 16000, 8000]
|
|
|
|
def test_get_supported_sample_rate_all_fail(self):
|
|
"""Test sample rate detection when all attempts fail."""
|
|
mock_audio = MagicMock()
|
|
mock_audio.open.side_effect = IOError("All rates failed")
|
|
mock_audio.get_device_info_by_index.side_effect = IOError("No device info")
|
|
|
|
self.pyaudio_mock.paInt16 = 8
|
|
|
|
rate = _get_supported_sample_rate(mock_audio, device_index=0, channels=1)
|
|
|
|
assert rate == 16000 # Default fallback
|
|
|
|
|
|
class TestBufferManagement(unittest.TestCase):
|
|
"""Test audio buffer management edge cases."""
|
|
|
|
def setUp(self):
|
|
"""Set up test fixtures."""
|
|
self.patches = []
|
|
|
|
def tearDown(self):
|
|
"""Clean up."""
|
|
for p in self.patches:
|
|
p.stop()
|
|
|
|
def _create_manager(self):
|
|
"""Create a manager with mocked dependencies."""
|
|
patches = [
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
patch.object(SpeechRecognitionManager, "_download_vosk_model"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("os.makedirs"),
|
|
patch("threading.Thread"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
with patch("vosk.Model"):
|
|
with patch("vosk.KaldiRecognizer"):
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
manager._model_initialized = True
|
|
return manager
|
|
|
|
def test_set_buffer_limit_too_small(self):
|
|
"""Test buffer limit enforcement - minimum."""
|
|
manager = self._create_manager()
|
|
manager.set_buffer_limit(50)
|
|
assert manager._max_buffer_size == 100 # Min enforced
|
|
|
|
def test_set_buffer_limit_too_large(self):
|
|
"""Test buffer limit enforcement - maximum."""
|
|
manager = self._create_manager()
|
|
manager.set_buffer_limit(25000)
|
|
assert manager._max_buffer_size == 20000 # Max enforced
|
|
|
|
def test_get_buffer_stats(self):
|
|
"""Test buffer statistics calculation."""
|
|
manager = self._create_manager()
|
|
manager.set_buffer_limit(1000)
|
|
manager.audio_buffer = [b"x" * 100, b"y" * 100]
|
|
|
|
stats = manager.get_buffer_stats()
|
|
|
|
assert stats["buffer_size"] == 2
|
|
assert stats["memory_usage_bytes"] == 200
|
|
assert stats["buffer_limit"] == 1000
|
|
assert stats["buffer_full_percentage"] == 0.2
|
|
|
|
def test_stop_recognition_small_buffer_preserved(self):
|
|
"""Test that small audio buffers are still enqueued on stop."""
|
|
manager = self._create_manager()
|
|
manager.state = RecognitionState.LISTENING
|
|
manager.should_record = True
|
|
manager.audio_buffer = [b"small"]
|
|
manager._recording_segment_has_speech = True
|
|
manager.audio_thread = MagicMock()
|
|
manager.audio_thread.is_alive.return_value = False
|
|
manager.recognition_thread = MagicMock()
|
|
manager.recognition_thread.is_alive.return_value = False
|
|
|
|
with patch.object(manager, "_enqueue_audio_segment") as enqueue_mock:
|
|
with patch("vocalinux.ui.audio_feedback.play_stop_sound"):
|
|
manager.stop_recognition()
|
|
|
|
enqueue_mock.assert_called_once_with([b"small"], release_gate=None)
|
|
assert manager.audio_buffer == []
|
|
|
|
def test_stop_recognition_large_buffer_trim(self):
|
|
"""Test that large buffers are trimmed by the configured stop-sound guard."""
|
|
manager = self._create_manager()
|
|
manager.state = RecognitionState.LISTENING
|
|
manager.should_record = True
|
|
manager.audio_buffer = [b"x" * 100 for _ in range(20)]
|
|
manager._recording_segment_has_speech = True
|
|
manager.audio_thread = MagicMock()
|
|
manager.audio_thread.is_alive.return_value = False
|
|
manager.recognition_thread = MagicMock()
|
|
manager.recognition_thread.is_alive.return_value = False
|
|
|
|
with patch.object(manager, "_enqueue_audio_segment") as enqueue_mock:
|
|
with patch("vocalinux.ui.audio_feedback.play_stop_sound"):
|
|
manager.stop_recognition()
|
|
|
|
trimmed_segment = enqueue_mock.call_args.args[0]
|
|
assert len(trimmed_segment) == 17
|
|
assert len(manager.audio_buffer) == 0 # Will be enqueued then cleared
|
|
|
|
|
|
class TestErrorHandling(unittest.TestCase):
|
|
"""Test error handling in various paths."""
|
|
|
|
def setUp(self):
|
|
"""Set up mocks."""
|
|
self.patches = []
|
|
|
|
def tearDown(self):
|
|
"""Clean up."""
|
|
for p in self.patches:
|
|
p.stop()
|
|
|
|
def test_reconfigure_engine_change(self):
|
|
"""Test reconfiguring to a different engine."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
patch.object(SpeechRecognitionManager, "_init_whisper"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
manager._model_initialized = True
|
|
original_engine = manager.engine
|
|
|
|
manager.reconfigure(engine="whisper", force_download=False)
|
|
|
|
assert manager.engine == "whisper"
|
|
assert manager.engine != original_engine
|
|
|
|
def test_reconfigure_aborts_when_download_cancelled_while_lock_is_busy(self) -> None:
|
|
"""Cancellation while waiting for a transcription lock restores state."""
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
manager._download_claim.acquire()
|
|
manager.engine = "vosk"
|
|
manager._model_lock.acquire()
|
|
manager._model_lock.release()
|
|
|
|
lock_held = threading.Event()
|
|
release_lock = threading.Event()
|
|
|
|
def hold_model_lock() -> None:
|
|
manager._model_lock.acquire()
|
|
lock_held.set()
|
|
release_lock.wait(timeout=2)
|
|
manager._model_lock.release()
|
|
|
|
holder = threading.Thread(target=hold_model_lock)
|
|
holder.start()
|
|
assert lock_held.wait(timeout=2)
|
|
manager.cancel_download()
|
|
|
|
try:
|
|
with patch.object(manager, "_download_selected_model_if_missing"):
|
|
with self.assertRaisesRegex(RuntimeError, "Download cancelled"):
|
|
manager.reconfigure(engine="whisper", force_download=True)
|
|
finally:
|
|
release_lock.set()
|
|
holder.join(timeout=2)
|
|
manager.end_download()
|
|
|
|
assert manager.engine == "vosk"
|
|
assert manager._defer_download is True
|
|
|
|
def test_reconfigure_language_change(self):
|
|
"""Test reconfiguring language triggers restart."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
patch.object(SpeechRecognitionManager, "_init_vosk"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
manager._model_initialized = True
|
|
|
|
manager.reconfigure(language="es", force_download=False)
|
|
|
|
assert manager.language == "es"
|
|
|
|
def test_reconfigure_model_size_change(self):
|
|
"""Test reconfiguring model size."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
patch.object(SpeechRecognitionManager, "_init_vosk"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", model_size="small", defer_download=True)
|
|
|
|
manager.reconfigure(model_size="medium", force_download=False)
|
|
|
|
assert manager.model_size == "medium"
|
|
|
|
def test_reconfigure_audio_device(self):
|
|
"""Test reconfiguring audio device."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
|
|
manager.reconfigure(audio_device_index=2, force_download=False)
|
|
assert manager.audio_device_index == 2
|
|
|
|
manager.reconfigure(audio_device_index=-1, force_download=False)
|
|
assert manager.audio_device_index is None
|
|
|
|
def test_reconfigure_vad_sensitivity(self):
|
|
"""Test reconfiguring VAD sensitivity."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
|
|
manager.reconfigure(vad_sensitivity=4, force_download=False)
|
|
assert manager.vad_sensitivity == 4
|
|
|
|
|
|
class TestTranscriptionEdgeCases(unittest.TestCase):
|
|
"""Test transcription error handling."""
|
|
|
|
def setUp(self):
|
|
"""Set up."""
|
|
self.patches = []
|
|
|
|
def tearDown(self):
|
|
"""Clean up."""
|
|
for p in self.patches:
|
|
p.stop()
|
|
|
|
def _make_manager(self, engine="vosk"):
|
|
"""Helper to create manager."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
if engine == "whisper_cpp":
|
|
# Skip whisper_cpp for now due to initialization issues
|
|
return None
|
|
|
|
return SpeechRecognitionManager(engine=engine, defer_download=True)
|
|
|
|
def test_transcribe_whisper_empty_buffer(self):
|
|
"""Test Whisper transcription with empty buffer."""
|
|
manager = self._make_manager(engine="whisper")
|
|
if manager is None:
|
|
self.skipTest("Manager creation failed")
|
|
|
|
manager.model = MagicMock()
|
|
|
|
result = manager._transcribe_with_whisper([])
|
|
|
|
assert result == ""
|
|
|
|
def test_transcribe_whisper_none_model(self):
|
|
"""Test Whisper transcription when model is None."""
|
|
manager = self._make_manager(engine="whisper")
|
|
if manager is None:
|
|
self.skipTest("Manager creation failed")
|
|
|
|
manager.model = None
|
|
|
|
result = manager._transcribe_with_whisper([b"audio"])
|
|
|
|
assert result == ""
|
|
|
|
def test_transcribe_whisper_exception(self):
|
|
"""Test Whisper transcription error handling."""
|
|
manager = self._make_manager(engine="whisper")
|
|
if manager is None:
|
|
self.skipTest("Manager creation failed")
|
|
|
|
manager.model = MagicMock()
|
|
manager.model.device = "cpu"
|
|
manager.model.transcribe.side_effect = RuntimeError("Transcription failed")
|
|
|
|
with patch("numpy.frombuffer", return_value=MagicMock()):
|
|
with patch("numpy.astype", return_value=MagicMock()):
|
|
result = manager._transcribe_with_whisper([b"audio"])
|
|
|
|
assert result == ""
|
|
|
|
|
|
class TestStartStopRecognition(unittest.TestCase):
|
|
"""Test recognition start/stop flow."""
|
|
|
|
def setUp(self):
|
|
"""Set up patches."""
|
|
self.patches = []
|
|
|
|
def tearDown(self):
|
|
"""Clean up patches."""
|
|
for p in self.patches:
|
|
p.stop()
|
|
|
|
def test_start_recognition_idle_to_listening(self):
|
|
"""Test starting recognition transitions to LISTENING."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
patch("vocalinux.ui.audio_feedback.play_start_sound"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
manager._model_initialized = True
|
|
manager.state = RecognitionState.IDLE
|
|
|
|
assert manager.start_recognition() is True
|
|
|
|
assert manager.state == RecognitionState.LISTENING
|
|
assert manager.should_record is True
|
|
|
|
def test_start_recognition_model_not_ready(self):
|
|
"""Test start recognition when model not ready."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
patch("vocalinux.ui.audio_feedback.play_error_sound"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
manager._model_initialized = False
|
|
manager.state = RecognitionState.IDLE
|
|
|
|
assert manager.start_recognition() is False
|
|
|
|
# Should stay idle if model not ready
|
|
assert manager.state == RecognitionState.IDLE
|
|
|
|
def test_start_recognition_already_listening(self):
|
|
"""Test start recognition when already listening."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
manager._model_initialized = True
|
|
manager.state = RecognitionState.LISTENING
|
|
|
|
assert manager.start_recognition() is False
|
|
|
|
# Should still be listening (no-op)
|
|
assert manager.state == RecognitionState.LISTENING
|
|
|
|
|
|
class TestAudioBufferOperations(unittest.TestCase):
|
|
"""Test audio buffer operations."""
|
|
|
|
def setUp(self):
|
|
"""Set up patches."""
|
|
self.patches = []
|
|
|
|
def tearDown(self):
|
|
"""Clean up patches."""
|
|
for p in self.patches:
|
|
p.stop()
|
|
|
|
def test_buffer_stats_empty_buffer(self):
|
|
"""Test buffer stats with empty buffer."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
manager.audio_buffer = []
|
|
|
|
stats = manager.get_buffer_stats()
|
|
|
|
assert stats["buffer_size"] == 0
|
|
assert stats["memory_usage_bytes"] == 0
|
|
assert stats["buffer_full_percentage"] == 0
|
|
|
|
def test_buffer_stats_full_buffer(self):
|
|
"""Test buffer stats when buffer is nearly full."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
manager.set_buffer_limit(100)
|
|
manager.audio_buffer = [b"x" * 1000 for _ in range(95)]
|
|
|
|
stats = manager.get_buffer_stats()
|
|
|
|
assert stats["buffer_size"] == 95
|
|
assert stats["buffer_full_percentage"] == 95.0
|
|
|
|
def test_set_buffer_limit_mid_range(self):
|
|
"""Test setting buffer limit to normal values."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
manager.set_buffer_limit(1000)
|
|
|
|
assert manager._max_buffer_size == 1000
|
|
|
|
|
|
class TestAudioDeviceReconnection(unittest.TestCase):
|
|
"""Test audio device reconnection logic."""
|
|
|
|
def setUp(self):
|
|
"""Set up patches."""
|
|
self.patches = []
|
|
|
|
def tearDown(self):
|
|
"""Clean up patches."""
|
|
for p in self.patches:
|
|
p.stop()
|
|
|
|
def test_attempt_audio_reconnection_success(self):
|
|
"""Test successful audio reconnection."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
patch(
|
|
"vocalinux.audio.capture._get_supported_channels",
|
|
return_value=1,
|
|
),
|
|
patch(
|
|
"vocalinux.audio.capture._get_supported_sample_rate",
|
|
return_value=16000,
|
|
),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
|
|
# Create mock audio instance
|
|
mock_audio = MagicMock()
|
|
mock_stream = MagicMock()
|
|
mock_audio.open.return_value = mock_stream
|
|
mock_stream.get_read_available.return_value = 1024
|
|
mock_stream.read.return_value = b"audio_data"
|
|
|
|
result = manager._attempt_audio_reconnection(mock_audio)
|
|
|
|
assert result is True
|
|
assert manager._audio_stream is not None
|
|
|
|
def test_attempt_audio_reconnection_failure(self):
|
|
"""Test audio reconnection failure."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
patch(
|
|
"vocalinux.audio.capture._get_supported_channels",
|
|
return_value=1,
|
|
),
|
|
patch(
|
|
"vocalinux.audio.capture._get_supported_sample_rate",
|
|
return_value=16000,
|
|
),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
|
|
# Create mock audio that fails
|
|
mock_audio = MagicMock()
|
|
mock_audio.open.side_effect = IOError("Device error")
|
|
|
|
result = manager._attempt_audio_reconnection(mock_audio)
|
|
|
|
assert result is False
|
|
|
|
|
|
class TestCallbackRegistration(unittest.TestCase):
|
|
"""Test callback registration."""
|
|
|
|
def setUp(self):
|
|
"""Set up patches."""
|
|
self.patches = []
|
|
|
|
def tearDown(self):
|
|
"""Clean up patches."""
|
|
for p in self.patches:
|
|
p.stop()
|
|
|
|
def test_register_text_callback(self):
|
|
"""Test registering text callback."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
|
|
def callback(text):
|
|
pass
|
|
|
|
manager.register_text_callback(callback)
|
|
|
|
assert callback in manager.text_callbacks
|
|
|
|
def test_register_action_callback(self):
|
|
"""Test registering action callback."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
|
|
def action_callback(action):
|
|
pass
|
|
|
|
manager.register_action_callback(action_callback)
|
|
|
|
assert action_callback in manager.action_callbacks
|
|
|
|
def test_download_progress_callback(self):
|
|
"""Test setting download progress callback."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
|
|
def progress_callback(progress, speed, status):
|
|
pass
|
|
|
|
manager._download_progress_callback = progress_callback
|
|
|
|
assert manager._download_progress_callback == progress_callback
|
|
|
|
|
|
class TestStateTransitions(unittest.TestCase):
|
|
"""Test state transition logic."""
|
|
|
|
def setUp(self):
|
|
"""Set up patches."""
|
|
self.patches = []
|
|
|
|
def tearDown(self):
|
|
"""Clean up patches."""
|
|
for p in self.patches:
|
|
p.stop()
|
|
|
|
def test_update_state(self):
|
|
"""Test state update."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
original_state = manager.state
|
|
|
|
manager._update_state(RecognitionState.LISTENING)
|
|
|
|
assert manager.state == RecognitionState.LISTENING
|
|
assert manager.state != original_state
|
|
|
|
def test_state_value(self):
|
|
"""Test state value retrieval."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
manager._update_state(RecognitionState.LISTENING)
|
|
|
|
assert manager.state == RecognitionState.LISTENING
|
|
|
|
def test_should_record_flag(self):
|
|
"""Test should_record flag."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
manager.should_record = True
|
|
|
|
assert manager.should_record is True
|
|
|
|
|
|
class TestModelReadiness(unittest.TestCase):
|
|
"""Test model readiness checks."""
|
|
|
|
def setUp(self):
|
|
"""Set up patches."""
|
|
self.patches = []
|
|
|
|
def tearDown(self):
|
|
"""Clean up patches."""
|
|
for p in self.patches:
|
|
p.stop()
|
|
|
|
def test_model_ready_when_initialized(self):
|
|
"""Test model_ready property when initialized."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
manager._model_initialized = True
|
|
|
|
assert manager.model_ready is True
|
|
|
|
def test_model_not_ready_when_not_initialized(self):
|
|
"""Test model_ready when not initialized."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
manager._model_initialized = False
|
|
|
|
assert manager.model_ready is False
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|