Jatin K MalikandGitHub
be0614010d
* refactor: improve code quality across multiple dimensions Code quality improvements identified and driven by desloppify analysis: **action_handler.py** - Replace 9 repetitive shortcut forwarding methods with data-driven _SHORTCUT_ACTIONS class-level dispatch table - Add _make_shortcut_handler() factory for shortcut closures - Add explicit type annotation for action_handlers dict **config_manager.py** - Move 'import copy' from __init__ to module level (stdlib best practice) - Add _VALID_SECTIONS frozenset for section name validation in set() - Narrow broad Exception catch to (KeyError, TypeError) in set() - Add deprecation warning to save_settings() guiding toward save_config() **tray_indicator.py** - Consolidate all imports at module top level - Remove noqa E402 suppressions for mid-file imports - Move ResourceManager import alongside other local imports **recognition_manager.py** - Narrow broad 'except Exception' to 'except OSError' for audio device enumeration errors - Narrow 'except Exception' to 'except (FileNotFoundError, OSError)' for notification subprocess errors **Tests** - Update test assertions to use specific exception types matching narrowed catches (OSError, FileNotFoundError instead of Exception/RuntimeError) * refactor: improve type safety, callback contracts, and method decomposition **main.py** — Mid-level elegance - Document callback wiring contract: text_callback, action_callback, state_callback with threading context, argument types, and behavior - Add return type annotations to nested callback functions - Reorder: define callbacks before registration for readability **config_manager.py** — Type safety, error consistency - Add typed accessors: get_str(), get_bool(), get_int(), get_float(), get_optional_int() eliminating Any returns for common access patterns - Narrow load_config exception to (json.JSONDecodeError, OSError) - Narrow save_config exception to (OSError, TypeError) **recognition_manager.py** — Low-level elegance - Decompose _init_whispercpp (141 lines) into three focused methods: - _init_whispercpp: validation and download gating (35 lines) - _load_whispercpp_model: backend detection and model loading (55 lines) - _handle_gpu_fallback: GPU incompatibility recovery (40 lines) - Narrow exception catch from broad Exception to specific types **tray_indicator.py, settings_dialog.py** — Type safety - Migrate config access from untyped get() to typed get_str(), get_bool(), get_optional_int() across 8 call sites * refactor: narrow 11 more broad Exception catches to specific types Systematically replaced broad 'except Exception' with specific exception types across recognition_manager.py: - test_audio_input: (OSError, ValueError) for audio read errors - test_audio_input outer: (OSError, ValueError, RuntimeError) for unexpected - _init_whisper: (RuntimeError, OSError) for engine init failures - _transcribe_with_whisper: (RuntimeError, OSError, ValueError) for transcription - _download_whispercpp_model: (OSError, RuntimeError, ValueError) for downloads - _download_vosk_model: (OSError, RuntimeError, ValueError) for downloads - _download_whisper_model: (OSError, RuntimeError, ValueError) for downloads This brings the total narrowed exception catches to 18+ across the codebase, improving error consistency and making failure modes more predictable. * style: fix black formatting issues Auto-format with black to fix CI lint failures in recognition_manager.py, settings_dialog.py, and test_settings_shortcuts.py. * test: add coverage for typed config accessors Add TestTypedAccessors class with 15 tests covering: - get_str(): string values, defaults, non-string conversion, None handling - get_bool(): boolean values, defaults - get_int(): integer values, invalid input fallback, missing key defaults - get_float(): float values, invalid input fallback - get_optional_int(): int values, None values, missing keys, invalid input Fixes codecov/patch coverage requirement for new typed accessor methods.
1098 lines
42 KiB
Python
1098 lines
42 KiB
Python
"""
|
|
Tests for the speech recognition manager.
|
|
"""
|
|
|
|
import sys
|
|
import unittest
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
# Mock modules before importing any modules that might use them
|
|
sys.modules["vosk"] = MagicMock()
|
|
sys.modules["whisper"] = MagicMock()
|
|
sys.modules["requests"] = MagicMock()
|
|
sys.modules["pyaudio"] = MagicMock()
|
|
sys.modules["wave"] = MagicMock()
|
|
sys.modules["tempfile"] = MagicMock()
|
|
sys.modules["tqdm"] = MagicMock()
|
|
sys.modules["numpy"] = MagicMock()
|
|
sys.modules["zipfile"] = MagicMock()
|
|
|
|
# Import the shared mock from conftest
|
|
from conftest import mock_audio_feedback # noqa: E402
|
|
|
|
# Update import paths to use the new package structure
|
|
from vocalinux.common_types import RecognitionState # noqa: E402
|
|
from vocalinux.speech_recognition.command_processor import CommandProcessor # noqa: E402
|
|
from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager # noqa: E402
|
|
|
|
|
|
class TestSpeechRecognition(unittest.TestCase):
|
|
"""Test cases for the speech recognition functionality."""
|
|
|
|
def setUp(self):
|
|
"""Set up for tests."""
|
|
# Create patches for our mocks
|
|
self.mockKaldi = patch.object(sys.modules["vosk"], "KaldiRecognizer")
|
|
self.mockModel = patch.object(sys.modules["vosk"], "Model")
|
|
self.mockMakeDirs = patch("os.makedirs")
|
|
self.mockThread = patch("threading.Thread")
|
|
self.mockPath = patch.object(SpeechRecognitionManager, "_get_vosk_model_path")
|
|
self.mockDownload = patch.object(SpeechRecognitionManager, "_download_vosk_model")
|
|
self.mockCmdProcessor = patch.object(CommandProcessor, "process_text")
|
|
|
|
# Start all patches
|
|
self.kaldiMock = self.mockKaldi.start()
|
|
self.modelMock = self.mockModel.start()
|
|
self.makeDirsMock = self.mockMakeDirs.start()
|
|
self.threadMock = self.mockThread.start()
|
|
self.pathMock = self.mockPath.start()
|
|
self.downloadMock = self.mockDownload.start()
|
|
self.cmdProcessorMock = self.mockCmdProcessor.start()
|
|
|
|
# Set up return values
|
|
self.recognizerMock = MagicMock()
|
|
self.kaldiMock.return_value = self.recognizerMock
|
|
self.pathMock.return_value = "/mock/path/vosk-model"
|
|
self.threadInstance = MagicMock()
|
|
self.threadMock.return_value = self.threadInstance
|
|
self.cmdProcessorMock.return_value = ("processed text", ["action1"])
|
|
|
|
# Critical: Set FinalResult to return valid JSON string
|
|
self.recognizerMock.FinalResult.return_value = '{"text": "test transcription"}'
|
|
|
|
# Reset audio feedback mocks before each test
|
|
mock_audio_feedback.play_start_sound.reset_mock()
|
|
mock_audio_feedback.play_stop_sound.reset_mock()
|
|
mock_audio_feedback.play_error_sound.reset_mock()
|
|
|
|
# Patch os.makedirs to avoid creating directories
|
|
self.patcher_makedirs = patch("os.makedirs")
|
|
self.mock_makedirs = self.patcher_makedirs.start()
|
|
|
|
# Patch os.path.exists to return True for any path
|
|
self.patcher_exists = patch("os.path.exists", return_value=True)
|
|
self.mock_exists = self.patcher_exists.start()
|
|
|
|
# Patch os.unlink to avoid removing files
|
|
self.patcher_unlink = patch("os.unlink")
|
|
self.mock_unlink = self.patcher_unlink.start()
|
|
|
|
# Patch tempfile.NamedTemporaryFile for whisper tests
|
|
self.patcher_temp = patch("tempfile.NamedTemporaryFile")
|
|
self.mock_temp = self.patcher_temp.start()
|
|
self.mock_temp_file = MagicMock()
|
|
self.mock_temp_file.__enter__ = MagicMock(return_value=self.mock_temp_file)
|
|
self.mock_temp_file.__exit__ = MagicMock(return_value=None)
|
|
self.mock_temp_file.name = "/tmp/test.wav"
|
|
self.mock_temp.return_value = self.mock_temp_file
|
|
|
|
# Patch wave.open for whisper tests
|
|
self.patcher_wave = patch("wave.open")
|
|
self.mock_wave = self.patcher_wave.start()
|
|
self.mock_wave_file = MagicMock()
|
|
self.mock_wave_file.__enter__ = MagicMock(return_value=self.mock_wave_file)
|
|
self.mock_wave_file.__exit__ = MagicMock(return_value=None)
|
|
self.mock_wave.return_value = self.mock_wave_file
|
|
|
|
def tearDown(self):
|
|
"""Clean up after tests."""
|
|
# Stop all patches
|
|
self.mockKaldi.stop()
|
|
self.mockModel.stop()
|
|
self.mockMakeDirs.stop()
|
|
self.mockThread.stop()
|
|
self.mockPath.stop()
|
|
self.mockDownload.stop()
|
|
self.mockCmdProcessor.stop()
|
|
|
|
self.patcher_makedirs.stop()
|
|
self.patcher_exists.stop()
|
|
self.patcher_unlink.stop()
|
|
self.patcher_temp.stop()
|
|
self.patcher_wave.stop()
|
|
|
|
def test_init(self):
|
|
"""Test initialization with different engines."""
|
|
# Test VOSK initialization
|
|
manager = SpeechRecognitionManager(engine="vosk", model_size="small")
|
|
|
|
# Verify initial state
|
|
self.assertEqual(manager.state, RecognitionState.IDLE)
|
|
self.assertEqual(manager.engine, "vosk")
|
|
self.assertEqual(manager.model_size, "small")
|
|
self.assertFalse(manager.should_record)
|
|
self.assertEqual(manager.audio_buffer, [])
|
|
|
|
# Verify VOSK model was initialized
|
|
self.modelMock.assert_called_once()
|
|
self.kaldiMock.assert_called_once()
|
|
|
|
# Test invalid engine
|
|
with self.assertRaises(ValueError):
|
|
SpeechRecognitionManager(engine="invalid")
|
|
|
|
def test_register_callbacks(self):
|
|
"""Test callback registration."""
|
|
manager = SpeechRecognitionManager(engine="vosk")
|
|
|
|
# Create mock callbacks
|
|
text_callback = MagicMock()
|
|
state_callback = MagicMock()
|
|
action_callback = MagicMock()
|
|
|
|
# Register callbacks
|
|
manager.register_text_callback(text_callback)
|
|
manager.register_state_callback(state_callback)
|
|
manager.register_action_callback(action_callback)
|
|
|
|
# Verify callbacks were registered
|
|
self.assertEqual(manager.text_callbacks, [text_callback])
|
|
self.assertEqual(manager.state_callbacks, [state_callback])
|
|
self.assertEqual(manager.action_callbacks, [action_callback])
|
|
|
|
# Test state update
|
|
manager._update_state(RecognitionState.LISTENING)
|
|
self.assertEqual(manager.state, RecognitionState.LISTENING)
|
|
state_callback.assert_called_once_with(RecognitionState.LISTENING)
|
|
|
|
def test_process_buffer(self):
|
|
"""Test processing audio buffer."""
|
|
# Setup for test
|
|
manager = SpeechRecognitionManager(engine="vosk")
|
|
|
|
# Register callbacks
|
|
text_callback = MagicMock()
|
|
action_callback = MagicMock()
|
|
manager.register_text_callback(text_callback)
|
|
manager.register_action_callback(action_callback)
|
|
|
|
# Setup audio buffer
|
|
manager.audio_buffer = [b"data1", b"data2"]
|
|
|
|
# Process buffer
|
|
manager._process_final_buffer()
|
|
|
|
# Verify Vosk methods were called
|
|
self.recognizerMock.AcceptWaveform.assert_any_call(b"data1")
|
|
self.recognizerMock.AcceptWaveform.assert_any_call(b"data2")
|
|
self.recognizerMock.FinalResult.assert_called_once()
|
|
|
|
# Verify command processor was called
|
|
self.cmdProcessorMock.assert_called_once_with("test transcription")
|
|
|
|
# Verify callbacks were called
|
|
text_callback.assert_called_once_with("processed text")
|
|
action_callback.assert_called_once_with("action1")
|
|
|
|
def test_start_stop_recognition(self):
|
|
"""Test starting and stopping recognition."""
|
|
manager = SpeechRecognitionManager(engine="vosk")
|
|
|
|
# Start recognition
|
|
manager.start_recognition()
|
|
|
|
# Verify state and calls
|
|
self.assertEqual(manager.state, RecognitionState.LISTENING)
|
|
self.assertTrue(manager.should_record)
|
|
mock_audio_feedback.play_start_sound.assert_called_once()
|
|
self.assertEqual(self.threadMock.call_count, 2)
|
|
self.threadInstance.start.assert_called()
|
|
|
|
# Reset mocks
|
|
self.threadMock.reset_mock()
|
|
self.threadInstance.reset_mock()
|
|
|
|
# Already listening, should not start again
|
|
manager.start_recognition()
|
|
self.threadMock.assert_not_called()
|
|
|
|
# Stop recognition
|
|
manager.audio_thread = self.threadInstance
|
|
manager.recognition_thread = self.threadInstance
|
|
manager.stop_recognition()
|
|
|
|
# Verify stopped state
|
|
self.assertEqual(manager.state, RecognitionState.IDLE)
|
|
self.assertFalse(manager.should_record)
|
|
mock_audio_feedback.play_stop_sound.assert_called_once()
|
|
self.threadInstance.join.assert_called()
|
|
|
|
def test_whisper_engine(self):
|
|
"""Test initialization and usage with Whisper engine."""
|
|
# Setup Whisper and torch mocks
|
|
whisper_mock = MagicMock()
|
|
torch_mock = MagicMock()
|
|
whisper_mock.load_model = MagicMock()
|
|
torch_mock.cuda.is_available.return_value = False
|
|
model_mock = MagicMock()
|
|
whisper_mock.load_model.return_value = model_mock
|
|
|
|
# Patch both whisper and torch modules
|
|
with patch.dict("sys.modules", {"whisper": whisper_mock, "torch": torch_mock}):
|
|
# Create manager with Whisper engine
|
|
manager = SpeechRecognitionManager(engine="whisper", model_size="medium")
|
|
|
|
# Verify Whisper was initialized
|
|
self.assertEqual(manager.engine, "whisper")
|
|
self.assertEqual(manager.model_size, "medium")
|
|
whisper_mock.load_model.assert_called_once()
|
|
|
|
# Instead of calling the actual _process_final_buffer method which does file operations,
|
|
# let's test the Whisper functionality by directly mocking that method
|
|
original_process = manager._process_final_buffer
|
|
|
|
# Replace _process_final_buffer with our own implementation for testing
|
|
def mock_process():
|
|
# Skip file operations, just simulate the Whisper transcription directly
|
|
# Mock the Whisper result and process it
|
|
processed_text, actions = self.cmdProcessorMock("whisper test")
|
|
# Call the callbacks
|
|
for callback in manager.text_callbacks:
|
|
callback(processed_text)
|
|
for callback in manager.action_callbacks:
|
|
for action in actions:
|
|
callback(action)
|
|
|
|
# Replace the method with our mock implementation
|
|
manager._process_final_buffer = mock_process
|
|
|
|
# Register callbacks
|
|
text_callback = MagicMock()
|
|
manager.register_text_callback(text_callback)
|
|
|
|
# Mock command processor
|
|
self.cmdProcessorMock.return_value = ("processed whisper", [])
|
|
|
|
# Call the mocked process method
|
|
manager._process_final_buffer()
|
|
|
|
# Verify callback was called
|
|
text_callback.assert_called_with("processed whisper")
|
|
|
|
# Restore the original method
|
|
manager._process_final_buffer = original_process
|
|
|
|
def test_vosk_model_path(self):
|
|
"""Test model path generation."""
|
|
# Disable our path mock to test actual implementation
|
|
self.mockPath.stop()
|
|
|
|
# Test with different model sizes
|
|
manager = SpeechRecognitionManager(engine="vosk", model_size="small")
|
|
path = manager._get_vosk_model_path()
|
|
self.assertIn("small", path)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", model_size="medium")
|
|
path = manager._get_vosk_model_path()
|
|
self.assertIn("0.22", path)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", model_size="large")
|
|
path = manager._get_vosk_model_path()
|
|
self.assertIn("0.22", path) # Updated to match actual implementation
|
|
|
|
# Restart our mock
|
|
self.pathMock = self.mockPath.start()
|
|
self.pathMock.return_value = "/mock/path/vosk-model"
|
|
|
|
def test_configure(self):
|
|
"""Test configuration method."""
|
|
manager = SpeechRecognitionManager(engine="vosk")
|
|
|
|
# Default values
|
|
self.assertEqual(manager.vad_sensitivity, 3)
|
|
self.assertEqual(manager.silence_timeout, 2.0)
|
|
|
|
# Configure with valid values
|
|
manager.reconfigure(vad_sensitivity=4, silence_timeout=1.5)
|
|
self.assertEqual(manager.vad_sensitivity, 4)
|
|
self.assertEqual(manager.silence_timeout, 1.5)
|
|
|
|
# Test bounds checking
|
|
manager.reconfigure(vad_sensitivity=10, silence_timeout=10.0)
|
|
self.assertEqual(manager.vad_sensitivity, 5) # Max is 5
|
|
self.assertEqual(manager.silence_timeout, 5.0) # Max is 5.0
|
|
|
|
manager.reconfigure(vad_sensitivity=0, silence_timeout=0.0)
|
|
self.assertEqual(manager.vad_sensitivity, 1) # Min is 1
|
|
self.assertEqual(manager.silence_timeout, 0.5) # Min is 0.5
|
|
|
|
def test_unregister_text_callback(self):
|
|
"""Test unregistering text callbacks."""
|
|
manager = SpeechRecognitionManager(engine="vosk")
|
|
|
|
callback = MagicMock()
|
|
manager.register_text_callback(callback)
|
|
self.assertIn(callback, manager.text_callbacks)
|
|
|
|
manager.unregister_text_callback(callback)
|
|
self.assertNotIn(callback, manager.text_callbacks)
|
|
|
|
# Unregistering non-existent callback should not raise
|
|
manager.unregister_text_callback(callback)
|
|
|
|
def test_get_set_text_callbacks(self):
|
|
"""Test getting and setting text callbacks."""
|
|
manager = SpeechRecognitionManager(engine="vosk")
|
|
|
|
callback1 = MagicMock()
|
|
callback2 = MagicMock()
|
|
|
|
manager.register_text_callback(callback1)
|
|
callbacks = manager.get_text_callbacks()
|
|
self.assertEqual(callbacks, [callback1])
|
|
|
|
# Verify it's a copy
|
|
callbacks.append(callback2)
|
|
self.assertEqual(len(manager.text_callbacks), 1)
|
|
|
|
# Set callbacks
|
|
manager.set_text_callbacks([callback1, callback2])
|
|
self.assertEqual(len(manager.text_callbacks), 2)
|
|
|
|
def test_audio_level_callbacks(self):
|
|
"""Test audio level callback registration."""
|
|
manager = SpeechRecognitionManager(engine="vosk")
|
|
|
|
callback = MagicMock()
|
|
manager.register_audio_level_callback(callback)
|
|
self.assertIn(callback, manager._audio_level_callbacks)
|
|
|
|
manager.unregister_audio_level_callback(callback)
|
|
self.assertNotIn(callback, manager._audio_level_callbacks)
|
|
|
|
# Unregistering non-existent callback should not raise
|
|
manager.unregister_audio_level_callback(callback)
|
|
|
|
def test_audio_device_management(self):
|
|
"""Test audio device get/set."""
|
|
manager = SpeechRecognitionManager(engine="vosk")
|
|
|
|
# Default is None
|
|
self.assertIsNone(manager.get_audio_device())
|
|
|
|
# Set device
|
|
manager.set_audio_device(2)
|
|
self.assertEqual(manager.get_audio_device(), 2)
|
|
|
|
# Set back to None
|
|
manager.set_audio_device(None)
|
|
self.assertIsNone(manager.get_audio_device())
|
|
|
|
def test_get_last_audio_level(self):
|
|
"""Test getting last audio level."""
|
|
manager = SpeechRecognitionManager(engine="vosk")
|
|
|
|
# Initially 0
|
|
self.assertEqual(manager.get_last_audio_level(), 0.0)
|
|
|
|
# Set a value directly
|
|
manager._last_audio_level = 50.5
|
|
self.assertEqual(manager.get_last_audio_level(), 50.5)
|
|
|
|
def test_model_ready_property(self):
|
|
"""Test model_ready property."""
|
|
manager = SpeechRecognitionManager(engine="vosk")
|
|
|
|
# Should be ready after init with mocked model
|
|
self.assertTrue(manager.model_ready)
|
|
|
|
# Set model to None
|
|
manager.model = None
|
|
self.assertFalse(manager.model_ready)
|
|
|
|
# Set initialized to False
|
|
manager.model = MagicMock()
|
|
manager._model_initialized = False
|
|
self.assertFalse(manager.model_ready)
|
|
|
|
def test_download_progress_callback(self):
|
|
"""Test download progress callback setting."""
|
|
manager = SpeechRecognitionManager(engine="vosk")
|
|
|
|
callback = MagicMock()
|
|
manager.set_download_progress_callback(callback)
|
|
self.assertEqual(manager._download_progress_callback, callback)
|
|
|
|
# Clear callback
|
|
manager.set_download_progress_callback(None)
|
|
self.assertIsNone(manager._download_progress_callback)
|
|
|
|
def test_cancel_download(self):
|
|
"""Test download cancellation."""
|
|
manager = SpeechRecognitionManager(engine="vosk")
|
|
|
|
self.assertFalse(manager._download_cancelled)
|
|
manager.cancel_download()
|
|
self.assertTrue(manager._download_cancelled)
|
|
|
|
def test_reconfigure_audio_device(self):
|
|
"""Test reconfiguring audio device index."""
|
|
manager = SpeechRecognitionManager(engine="vosk")
|
|
|
|
# Set device via reconfigure
|
|
manager.reconfigure(audio_device_index=1)
|
|
self.assertEqual(manager.audio_device_index, 1)
|
|
|
|
# Clear with -1
|
|
manager.reconfigure(audio_device_index=-1)
|
|
self.assertIsNone(manager.audio_device_index)
|
|
|
|
def test_reconfigure_engine_change(self):
|
|
"""Test reconfiguring to a different engine."""
|
|
# Setup Whisper and torch mocks
|
|
whisper_mock = MagicMock()
|
|
torch_mock = MagicMock()
|
|
whisper_mock.load_model = MagicMock()
|
|
torch_mock.cuda.is_available.return_value = False
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk")
|
|
self.assertEqual(manager.engine, "vosk")
|
|
|
|
with patch.dict("sys.modules", {"whisper": whisper_mock, "torch": torch_mock}):
|
|
manager.reconfigure(engine="whisper", force_download=False)
|
|
self.assertEqual(manager.engine, "whisper")
|
|
|
|
def test_reconfigure_model_size_change(self):
|
|
"""Test reconfiguring model size."""
|
|
manager = SpeechRecognitionManager(engine="vosk", model_size="small")
|
|
self.assertEqual(manager.model_size, "small")
|
|
|
|
manager.reconfigure(model_size="medium", force_download=False)
|
|
self.assertEqual(manager.model_size, "medium")
|
|
|
|
def test_reconfigure_language_change(self):
|
|
"""Test reconfiguring language."""
|
|
manager = SpeechRecognitionManager(engine="vosk", language="en-us")
|
|
self.assertEqual(manager.language, "en-us")
|
|
|
|
manager.reconfigure(language="de", force_download=False)
|
|
self.assertEqual(manager.language, "de")
|
|
|
|
def test_start_recognition_not_ready(self):
|
|
"""Test starting recognition when model is not ready."""
|
|
manager = SpeechRecognitionManager(engine="vosk")
|
|
manager._model_initialized = False
|
|
manager.model = None
|
|
|
|
# Should not start and should play error sound
|
|
manager.start_recognition()
|
|
self.assertEqual(manager.state, RecognitionState.IDLE)
|
|
mock_audio_feedback.play_error_sound.assert_called()
|
|
|
|
def test_stop_recognition_when_idle(self):
|
|
"""Test stopping recognition when already idle."""
|
|
manager = SpeechRecognitionManager(engine="vosk")
|
|
self.assertEqual(manager.state, RecognitionState.IDLE)
|
|
|
|
# Should do nothing
|
|
manager.stop_recognition()
|
|
self.assertEqual(manager.state, RecognitionState.IDLE)
|
|
|
|
def test_process_empty_buffer(self):
|
|
"""Test processing empty buffer does nothing."""
|
|
manager = SpeechRecognitionManager(engine="vosk")
|
|
manager.audio_buffer = []
|
|
|
|
# Should return without error
|
|
manager._process_final_buffer()
|
|
|
|
# Recognizer should not be called
|
|
self.recognizerMock.AcceptWaveform.assert_not_called()
|
|
|
|
def test_process_buffer_unknown_engine(self):
|
|
"""Test processing buffer with unknown engine."""
|
|
manager = SpeechRecognitionManager(engine="vosk")
|
|
manager.engine = "unknown"
|
|
manager.audio_buffer = [b"data"]
|
|
|
|
# Should log error but not crash
|
|
manager._process_final_buffer()
|
|
|
|
def test_init_with_kwargs(self):
|
|
"""Test initialization with additional kwargs."""
|
|
manager = SpeechRecognitionManager(
|
|
engine="vosk", vad_sensitivity=4, silence_timeout=1.5, audio_device_index=2
|
|
)
|
|
|
|
self.assertEqual(manager.vad_sensitivity, 4)
|
|
self.assertEqual(manager.silence_timeout, 1.5)
|
|
self.assertEqual(manager.audio_device_index, 2)
|
|
|
|
|
|
class TestModuleLevelFunctions(unittest.TestCase):
|
|
"""Test module-level functions in recognition_manager."""
|
|
|
|
def setUp(self):
|
|
"""Set up patches."""
|
|
self.patcher_makedirs = patch("os.makedirs")
|
|
self.mock_makedirs = self.patcher_makedirs.start()
|
|
|
|
def tearDown(self):
|
|
"""Clean up patches."""
|
|
self.patcher_makedirs.stop()
|
|
|
|
def test_get_audio_input_devices(self):
|
|
"""Test getting audio input devices."""
|
|
from vocalinux.speech_recognition import recognition_manager
|
|
|
|
# Mock PyAudio
|
|
mock_pyaudio_instance = MagicMock()
|
|
mock_pyaudio_instance.get_default_input_device_info.return_value = {"index": 0}
|
|
mock_pyaudio_instance.get_device_count.return_value = 2
|
|
mock_pyaudio_instance.get_device_info_by_index.side_effect = [
|
|
{"name": "Built-in Mic", "maxInputChannels": 1},
|
|
{"name": "USB Mic", "maxInputChannels": 2},
|
|
]
|
|
|
|
mock_pyaudio = MagicMock()
|
|
mock_pyaudio.PyAudio.return_value = mock_pyaudio_instance
|
|
|
|
with patch.dict(sys.modules, {"pyaudio": mock_pyaudio}):
|
|
# Need to reload to pick up the mock
|
|
devices = recognition_manager.get_audio_input_devices()
|
|
|
|
# With mocking, it may return empty if pyaudio was already imported
|
|
# The important thing is it doesn't crash
|
|
|
|
def test_get_audio_input_devices_no_default(self):
|
|
"""Test getting audio devices when no default is set."""
|
|
from vocalinux.speech_recognition import recognition_manager
|
|
|
|
mock_pyaudio_instance = MagicMock()
|
|
mock_pyaudio_instance.get_default_input_device_info.side_effect = IOError("No default")
|
|
mock_pyaudio_instance.get_device_count.return_value = 1
|
|
mock_pyaudio_instance.get_device_info_by_index.return_value = {
|
|
"name": "Mic",
|
|
"maxInputChannels": 1,
|
|
}
|
|
|
|
mock_pyaudio = MagicMock()
|
|
mock_pyaudio.PyAudio.return_value = mock_pyaudio_instance
|
|
|
|
with patch.dict(sys.modules, {"pyaudio": mock_pyaudio}):
|
|
devices = recognition_manager.get_audio_input_devices()
|
|
|
|
def test_show_notification(self):
|
|
"""Test _show_notification helper function."""
|
|
from vocalinux.speech_recognition.recognition_manager import _show_notification
|
|
|
|
with patch("subprocess.Popen") as mock_popen:
|
|
_show_notification("Test Title", "Test Message")
|
|
mock_popen.assert_called_once()
|
|
|
|
def test_show_notification_error(self):
|
|
"""Test _show_notification handles errors gracefully."""
|
|
from vocalinux.speech_recognition.recognition_manager import _show_notification
|
|
|
|
with patch("subprocess.Popen", side_effect=FileNotFoundError("notify-send not found")):
|
|
# Should not raise
|
|
_show_notification("Test Title", "Test Message")
|
|
|
|
def test_test_audio_input_success(self):
|
|
"""Test test_audio_input with successful recording."""
|
|
from vocalinux.speech_recognition import recognition_manager
|
|
|
|
# Create mock for numpy
|
|
mock_np = MagicMock()
|
|
mock_np.frombuffer.return_value = MagicMock()
|
|
mock_np.frombuffer.return_value.max.return_value = 1000
|
|
mock_np.frombuffer.return_value.mean.return_value = 500
|
|
mock_np.abs.return_value = mock_np.frombuffer.return_value
|
|
mock_np.int16 = "int16"
|
|
|
|
# Create mock for PyAudio
|
|
mock_stream = MagicMock()
|
|
mock_stream.read.return_value = b"\x00" * 2048
|
|
|
|
mock_pyaudio_instance = MagicMock()
|
|
mock_pyaudio_instance.get_device_info_by_index.return_value = {
|
|
"name": "Test Mic",
|
|
"defaultSampleRate": 16000,
|
|
}
|
|
mock_pyaudio_instance.open.return_value = mock_stream
|
|
mock_pyaudio_instance.paInt16 = 8
|
|
|
|
mock_pyaudio = MagicMock()
|
|
mock_pyaudio.PyAudio.return_value = mock_pyaudio_instance
|
|
mock_pyaudio.paInt16 = 8
|
|
|
|
with patch.dict(sys.modules, {"pyaudio": mock_pyaudio, "numpy": mock_np}):
|
|
result = recognition_manager.test_audio_input(device_index=0, duration=0.1)
|
|
# The function may return early due to mocking complexity
|
|
# Just verify no crash
|
|
|
|
def test_test_audio_input_import_error(self):
|
|
"""Test test_audio_input when pyaudio is not available."""
|
|
from vocalinux.speech_recognition import recognition_manager
|
|
|
|
# Save original
|
|
original_pyaudio = sys.modules.get("pyaudio")
|
|
|
|
# Remove pyaudio from modules to simulate import error
|
|
# This is tricky because the import happens inside the function
|
|
# We'll patch the import itself
|
|
with patch.dict(sys.modules, {"pyaudio": None}):
|
|
# Force import error by making pyaudio None
|
|
pass # The actual test would need to reload the module
|
|
|
|
|
|
class TestALSAErrorHandler(unittest.TestCase):
|
|
"""Tests for ALSA error handler setup."""
|
|
|
|
def test_setup_alsa_error_handler_success(self):
|
|
"""Test ALSA error handler setup when ALSA is available."""
|
|
from vocalinux.speech_recognition import recognition_manager
|
|
|
|
# The function is called at module load, so it was already executed
|
|
# Just verify the module loaded without crashing
|
|
self.assertTrue(hasattr(recognition_manager, "_alsa_handler"))
|
|
|
|
def test_setup_alsa_error_handler_oserror(self):
|
|
"""Test ALSA error handler returns None when ALSA not available."""
|
|
from vocalinux.speech_recognition.recognition_manager import _setup_alsa_error_handler
|
|
|
|
with patch("ctypes.CDLL", side_effect=OSError("ALSA not available")):
|
|
result = _setup_alsa_error_handler()
|
|
self.assertIsNone(result)
|
|
|
|
def test_setup_alsa_error_handler_attribute_error(self):
|
|
"""Test ALSA error handler returns None on AttributeError."""
|
|
from vocalinux.speech_recognition.recognition_manager import _setup_alsa_error_handler
|
|
|
|
mock_lib = MagicMock()
|
|
del mock_lib.snd_lib_error_set_handler # Remove the attribute
|
|
|
|
with patch("ctypes.CDLL", return_value=mock_lib):
|
|
result = _setup_alsa_error_handler()
|
|
self.assertIsNone(result)
|
|
|
|
|
|
class TestVoskModelPath(unittest.TestCase):
|
|
"""Tests for VOSK model path resolution."""
|
|
|
|
def setUp(self):
|
|
"""Set up patches for VOSK tests."""
|
|
self.patcher_makedirs = patch("os.makedirs")
|
|
self.mock_makedirs = self.patcher_makedirs.start()
|
|
|
|
def tearDown(self):
|
|
"""Clean up patches."""
|
|
self.patcher_makedirs.stop()
|
|
|
|
def test_get_vosk_model_path_from_system_dirs(self):
|
|
"""Test finding VOSK model in system directories."""
|
|
from vocalinux.speech_recognition.recognition_manager import (
|
|
MODELS_DIR,
|
|
SYSTEM_MODELS_DIRS,
|
|
SpeechRecognitionManager,
|
|
)
|
|
|
|
# Mock sys.modules for vosk
|
|
mock_vosk = MagicMock()
|
|
mock_vosk.Model = MagicMock()
|
|
mock_vosk.KaldiRecognizer = MagicMock()
|
|
|
|
# We need to mock os.path.exists to return True for system paths
|
|
with patch.dict(sys.modules, {"vosk": mock_vosk}):
|
|
with patch("os.path.exists") as mock_exists:
|
|
# Make user dir not exist but system dir exist
|
|
def exists_side_effect(path):
|
|
if SYSTEM_MODELS_DIRS[0] in path:
|
|
return True
|
|
return False
|
|
|
|
mock_exists.side_effect = exists_side_effect
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk")
|
|
# Method already called in init, test passes if no exception
|
|
|
|
|
|
class TestInitializationEdgeCases(unittest.TestCase):
|
|
"""Test initialization edge cases."""
|
|
|
|
def setUp(self):
|
|
"""Set up patches."""
|
|
self.patcher_makedirs = patch("os.makedirs")
|
|
self.mock_makedirs = self.patcher_makedirs.start()
|
|
self.patcher_exists = patch("os.path.exists", return_value=True)
|
|
self.mock_exists = self.patcher_exists.start()
|
|
|
|
def tearDown(self):
|
|
"""Clean up patches."""
|
|
self.patcher_makedirs.stop()
|
|
self.patcher_exists.stop()
|
|
|
|
def test_init_vosk_import_error(self):
|
|
"""Test VOSK initialization when vosk module cannot be imported."""
|
|
# This is tricky to test since vosk is already mocked at module level
|
|
# We test that the error handling code path exists
|
|
pass
|
|
|
|
def test_init_vosk_with_preinstalled_marker(self):
|
|
"""Test detecting installer-provided model."""
|
|
from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager
|
|
|
|
mock_vosk = MagicMock()
|
|
mock_vosk.Model = MagicMock()
|
|
mock_vosk.KaldiRecognizer = MagicMock()
|
|
|
|
with patch.dict(sys.modules, {"vosk": mock_vosk}):
|
|
with patch("os.path.exists") as mock_exists:
|
|
|
|
def exists_check(path):
|
|
if ".vocalinux_preinstalled" in path:
|
|
return True
|
|
return True
|
|
|
|
mock_exists.side_effect = exists_check
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk")
|
|
self.assertTrue(manager._model_initialized)
|
|
|
|
def test_init_vosk_model_not_found_deferred(self):
|
|
"""Test VOSK init with missing model and deferred download."""
|
|
from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager
|
|
|
|
mock_vosk = MagicMock()
|
|
mock_vosk.Model = MagicMock()
|
|
mock_vosk.KaldiRecognizer = MagicMock()
|
|
|
|
with patch.dict(sys.modules, {"vosk": mock_vosk}):
|
|
with patch("os.path.exists", return_value=False):
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
self.assertFalse(manager._model_initialized)
|
|
|
|
|
|
class TestRecognitionManagerMethods(unittest.TestCase):
|
|
"""Test additional SpeechRecognitionManager methods."""
|
|
|
|
def setUp(self):
|
|
"""Set up patches."""
|
|
self.patcher_makedirs = patch("os.makedirs")
|
|
self.mock_makedirs = self.patcher_makedirs.start()
|
|
self.patcher_exists = patch("os.path.exists", return_value=True)
|
|
self.mock_exists = self.patcher_exists.start()
|
|
|
|
mock_vosk = MagicMock()
|
|
mock_vosk.Model = MagicMock()
|
|
mock_vosk.KaldiRecognizer = MagicMock()
|
|
|
|
self.patcher_vosk = patch.dict(sys.modules, {"vosk": mock_vosk})
|
|
self.patcher_vosk.start()
|
|
|
|
def tearDown(self):
|
|
"""Clean up patches."""
|
|
self.patcher_makedirs.stop()
|
|
self.patcher_exists.stop()
|
|
self.patcher_vosk.stop()
|
|
|
|
def test_model_ready_property_true(self):
|
|
"""Test model_ready property when model is initialized."""
|
|
from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk")
|
|
manager._model_initialized = True
|
|
manager.model = MagicMock()
|
|
|
|
self.assertTrue(manager.model_ready)
|
|
|
|
def test_model_ready_property_false_not_initialized(self):
|
|
"""Test model_ready property when not initialized."""
|
|
from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk")
|
|
manager._model_initialized = False
|
|
|
|
self.assertFalse(manager.model_ready)
|
|
|
|
def test_model_ready_property_false_no_model(self):
|
|
"""Test model_ready property when model is None."""
|
|
from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk")
|
|
manager._model_initialized = True
|
|
manager.model = None
|
|
|
|
self.assertFalse(manager.model_ready)
|
|
|
|
def test_update_state(self):
|
|
"""Test _update_state method."""
|
|
from vocalinux.common_types import RecognitionState
|
|
from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk")
|
|
|
|
callback = MagicMock()
|
|
manager.register_state_callback(callback)
|
|
|
|
manager._update_state(RecognitionState.LISTENING)
|
|
|
|
self.assertEqual(manager.state, RecognitionState.LISTENING)
|
|
callback.assert_called_once_with(RecognitionState.LISTENING)
|
|
|
|
def test_update_state_multiple_callbacks(self):
|
|
"""Test _update_state with multiple callbacks."""
|
|
from vocalinux.common_types import RecognitionState
|
|
from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk")
|
|
|
|
callback1 = MagicMock()
|
|
callback2 = MagicMock()
|
|
manager.register_state_callback(callback1)
|
|
manager.register_state_callback(callback2)
|
|
|
|
manager._update_state(RecognitionState.PROCESSING)
|
|
|
|
callback1.assert_called_once_with(RecognitionState.PROCESSING)
|
|
callback2.assert_called_once_with(RecognitionState.PROCESSING)
|
|
|
|
def test_set_download_progress_callback(self):
|
|
"""Test setting download progress callback."""
|
|
from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk")
|
|
|
|
callback = MagicMock()
|
|
manager.set_download_progress_callback(callback)
|
|
|
|
self.assertEqual(manager._download_progress_callback, callback)
|
|
|
|
def test_cancel_download(self):
|
|
"""Test cancel download sets flag."""
|
|
from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk")
|
|
|
|
self.assertFalse(manager._download_cancelled)
|
|
manager.cancel_download()
|
|
self.assertTrue(manager._download_cancelled)
|
|
|
|
def test_get_vosk_model_path_small(self):
|
|
"""Test _get_vosk_model_path for small model."""
|
|
from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", model_size="small")
|
|
|
|
path = manager._get_vosk_model_path()
|
|
self.assertIn("small", path.lower())
|
|
|
|
def test_get_vosk_model_path_medium(self):
|
|
"""Test _get_vosk_model_path for medium model."""
|
|
from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", model_size="medium")
|
|
|
|
path = manager._get_vosk_model_path()
|
|
# Medium model path should be valid
|
|
self.assertIsNotNone(path)
|
|
|
|
def test_get_text_callbacks(self):
|
|
"""Test getting text callbacks returns a copy."""
|
|
from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk")
|
|
|
|
callback = MagicMock()
|
|
manager.register_text_callback(callback)
|
|
|
|
callbacks = manager.get_text_callbacks()
|
|
self.assertEqual(callbacks, [callback])
|
|
|
|
# Modifying returned list should not affect original
|
|
callbacks.append(MagicMock())
|
|
self.assertEqual(len(manager.text_callbacks), 1)
|
|
|
|
def test_set_text_callbacks(self):
|
|
"""Test setting text callbacks replaces all."""
|
|
from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk")
|
|
|
|
callback1 = MagicMock()
|
|
callback2 = MagicMock()
|
|
|
|
manager.set_text_callbacks([callback1, callback2])
|
|
|
|
self.assertEqual(len(manager.text_callbacks), 2)
|
|
self.assertIn(callback1, manager.text_callbacks)
|
|
self.assertIn(callback2, manager.text_callbacks)
|
|
|
|
|
|
class TestProcessFinalBuffer(unittest.TestCase):
|
|
"""Test _process_final_buffer method."""
|
|
|
|
def setUp(self):
|
|
"""Set up patches."""
|
|
self.patcher_makedirs = patch("os.makedirs")
|
|
self.mock_makedirs = self.patcher_makedirs.start()
|
|
self.patcher_exists = patch("os.path.exists", return_value=True)
|
|
self.mock_exists = self.patcher_exists.start()
|
|
|
|
# Mock vosk
|
|
self.mock_vosk = MagicMock()
|
|
self.mock_recognizer = MagicMock()
|
|
self.mock_vosk.Model.return_value = MagicMock()
|
|
self.mock_vosk.KaldiRecognizer.return_value = self.mock_recognizer
|
|
|
|
self.patcher_vosk = patch.dict(sys.modules, {"vosk": self.mock_vosk})
|
|
self.patcher_vosk.start()
|
|
|
|
def tearDown(self):
|
|
"""Clean up patches."""
|
|
self.patcher_makedirs.stop()
|
|
self.patcher_exists.stop()
|
|
self.patcher_vosk.stop()
|
|
|
|
def test_process_final_buffer_empty(self):
|
|
"""Test processing empty buffer."""
|
|
from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk")
|
|
manager.audio_buffer = []
|
|
|
|
# Should return without error
|
|
manager._process_final_buffer()
|
|
|
|
# Recognizer should not be called
|
|
self.mock_recognizer.AcceptWaveform.assert_not_called()
|
|
|
|
def test_process_final_buffer_vosk(self):
|
|
"""Test processing buffer with vosk."""
|
|
from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk")
|
|
manager.audio_buffer = [b"data1", b"data2"]
|
|
|
|
# Mock recognizer result
|
|
self.mock_recognizer.FinalResult.return_value = '{"text": "hello world"}'
|
|
|
|
# Register callbacks
|
|
text_callback = MagicMock()
|
|
action_callback = MagicMock()
|
|
manager.register_text_callback(text_callback)
|
|
manager.register_action_callback(action_callback)
|
|
|
|
# Mock command processor
|
|
with patch.object(manager.command_processor, "process_text") as mock_process:
|
|
mock_process.return_value = ("processed hello world", [])
|
|
manager._process_final_buffer()
|
|
|
|
text_callback.assert_called_once_with("processed hello world")
|
|
|
|
def test_process_final_buffer_unknown_engine(self):
|
|
"""Test processing buffer with unknown engine."""
|
|
from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk")
|
|
manager.engine = "unknown"
|
|
manager.audio_buffer = [b"data"]
|
|
|
|
# Should log error but not crash
|
|
manager._process_final_buffer()
|
|
|
|
def test_process_final_buffer_with_actions(self):
|
|
"""Test processing buffer returns actions."""
|
|
from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk")
|
|
manager.audio_buffer = [b"data"]
|
|
|
|
# Mock recognizer result
|
|
self.mock_recognizer.FinalResult.return_value = '{"text": "delete that"}'
|
|
|
|
# Register callbacks
|
|
action_callback = MagicMock()
|
|
manager.register_action_callback(action_callback)
|
|
|
|
# Mock command processor to return an action
|
|
with patch.object(manager.command_processor, "process_text") as mock_process:
|
|
mock_process.return_value = ("", ["delete_last"])
|
|
manager._process_final_buffer()
|
|
|
|
action_callback.assert_called_once_with("delete_last")
|
|
|
|
def test_process_final_buffer_empty_processed_text(self):
|
|
"""Test processing when processed text is empty."""
|
|
from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk")
|
|
manager.audio_buffer = [b"data"]
|
|
|
|
# Mock recognizer result
|
|
self.mock_recognizer.FinalResult.return_value = '{"text": "silence"}'
|
|
|
|
# Register callbacks
|
|
text_callback = MagicMock()
|
|
manager.register_text_callback(text_callback)
|
|
|
|
# Mock command processor to return empty text
|
|
with patch.object(manager.command_processor, "process_text") as mock_process:
|
|
mock_process.return_value = ("", [])
|
|
manager._process_final_buffer()
|
|
|
|
# Text callback should not be called for empty text
|
|
text_callback.assert_not_called()
|
|
|
|
|
|
class TestReconfigureMethod(unittest.TestCase):
|
|
"""Test reconfigure method."""
|
|
|
|
def setUp(self):
|
|
"""Set up patches."""
|
|
self.patcher_makedirs = patch("os.makedirs")
|
|
self.mock_makedirs = self.patcher_makedirs.start()
|
|
self.patcher_exists = patch("os.path.exists", return_value=True)
|
|
self.mock_exists = self.patcher_exists.start()
|
|
|
|
mock_vosk = MagicMock()
|
|
mock_vosk.Model = MagicMock()
|
|
mock_vosk.KaldiRecognizer = MagicMock()
|
|
|
|
self.patcher_vosk = patch.dict(sys.modules, {"vosk": mock_vosk})
|
|
self.patcher_vosk.start()
|
|
|
|
def tearDown(self):
|
|
"""Clean up patches."""
|
|
self.patcher_makedirs.stop()
|
|
self.patcher_exists.stop()
|
|
self.patcher_vosk.stop()
|
|
|
|
def test_reconfigure_vad_sensitivity_bounds(self):
|
|
"""Test VAD sensitivity is bounded to valid range."""
|
|
from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk")
|
|
|
|
# Test upper bound
|
|
manager.reconfigure(vad_sensitivity=10)
|
|
self.assertEqual(manager.vad_sensitivity, 5)
|
|
|
|
# Test lower bound
|
|
manager.reconfigure(vad_sensitivity=0)
|
|
self.assertEqual(manager.vad_sensitivity, 1)
|
|
|
|
def test_reconfigure_silence_timeout_bounds(self):
|
|
"""Test silence timeout is bounded to valid range."""
|
|
from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk")
|
|
|
|
# Test upper bound
|
|
manager.reconfigure(silence_timeout=10.0)
|
|
self.assertEqual(manager.silence_timeout, 5.0)
|
|
|
|
# Test lower bound
|
|
manager.reconfigure(silence_timeout=0.1)
|
|
self.assertEqual(manager.silence_timeout, 0.5)
|
|
|
|
def test_reconfigure_audio_device_clear(self):
|
|
"""Test clearing audio device with -1."""
|
|
from vocalinux.speech_recognition.recognition_manager import SpeechRecognitionManager
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk")
|
|
manager.audio_device_index = 1
|
|
|
|
manager.reconfigure(audio_device_index=-1)
|
|
self.assertIsNone(manager.audio_device_index)
|