5e531cbac0
* refactor(audio): extract capture source from recognition manager Move audio capture out of the silence-segmented dictation loop into vocalinux.audio.capture: a reusable PortAudioCaptureSource that owns device resolution, format negotiation, downmixing, resampling, and stream lifecycle (open/read/reopen/close). The dictation loop now consumes 16 kHz mono chunks and keeps only segmentation policy (VAD, silence counters, buffer limits, reconnection backoff). This unblocks alternate capture sources (PipeWire system-audio capture for #751, tdrz long-form output for #756) which need the same capture mechanics without dictation's silence segmentation. Behavior is identical: the same helpers run in the same order, moved verbatim. They are re-exported from recognition_manager for existing imports, and a CaptureSource protocol is declared in common_types.py. Tests repoint their patches of the moved helpers to the new module. Closes #760 * fix(audio): annotate capture API and keep a superseded session's source * fix(audio): reset retry counters when a new capture session starts * fix(audio): keep last-error timestamp across capture sessions The new-session reset wiped _last_audio_error_time, so an error within five seconds of the previous session's error always took the reconnect path instead of failing fast. Only the attempt count is per-session; the timestamp is also the rapid-failure throttle. --------- Co-authored-by: Jatin K Malik <jatinkrmalik@gmail.com>
1364 lines
54 KiB
Python
1364 lines
54 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()
|
|
|
|
# os.makedirs is already patched above as mockMakeDirs. Patching it a
|
|
# second time here and then stopping the two out of order (this one last)
|
|
# restored the *first* mock into the os module, leaking a fake makedirs
|
|
# into every test that ran after this class.
|
|
self.mock_makedirs = self.makeDirsMock
|
|
|
|
# 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_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
|
|
self.assertFalse(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)
|
|
|
|
def test_init_with_device_name(self):
|
|
"""Test initialization with audio_device_name kwarg."""
|
|
manager = SpeechRecognitionManager(
|
|
engine="vosk", audio_device_index=3, audio_device_name="USB Mic"
|
|
)
|
|
|
|
self.assertEqual(manager.audio_device_index, 3)
|
|
self.assertEqual(manager.get_audio_device_name(), "USB Mic")
|
|
|
|
def test_set_audio_device_with_name(self):
|
|
"""Test set_audio_device stores both index and name."""
|
|
manager = SpeechRecognitionManager(engine="vosk")
|
|
|
|
manager.set_audio_device(5, "BRIO Webcam")
|
|
self.assertEqual(manager.get_audio_device(), 5)
|
|
self.assertEqual(manager.get_audio_device_name(), "BRIO Webcam")
|
|
|
|
def test_set_audio_device_clears_name(self):
|
|
"""Test set_audio_device(None, None) clears both."""
|
|
manager = SpeechRecognitionManager(engine="vosk", audio_device_name="Mic")
|
|
|
|
manager.set_audio_device(None, None)
|
|
self.assertIsNone(manager.get_audio_device())
|
|
self.assertIsNone(manager.get_audio_device_name())
|
|
|
|
def test_reconfigure_with_device_name(self):
|
|
"""Test reconfigure accepts audio_device_name."""
|
|
manager = SpeechRecognitionManager(engine="vosk")
|
|
|
|
manager.reconfigure(audio_device_index=2, audio_device_name="Headset")
|
|
self.assertEqual(manager.audio_device_index, 2)
|
|
self.assertEqual(manager.get_audio_device_name(), "Headset")
|
|
|
|
|
|
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_get_audio_input_devices_skips_unreadable_device(self):
|
|
"""Test unreadable devices are skipped during enumeration."""
|
|
from vocalinux.speech_recognition import recognition_manager
|
|
|
|
mock_pyaudio_instance = MagicMock()
|
|
mock_pyaudio_instance.get_default_input_device_info.return_value = {"index": 1}
|
|
mock_pyaudio_instance.get_device_count.return_value = 2
|
|
mock_pyaudio_instance.get_device_info_by_index.side_effect = [
|
|
IOError("device disappeared"),
|
|
{"name": "USB 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()
|
|
|
|
self.assertEqual(devices, [(1, "USB Mic", True)])
|
|
|
|
def test_get_audio_input_devices_skips_output_only_device(self):
|
|
"""Test output-only devices are skipped during enumeration."""
|
|
from vocalinux.speech_recognition import recognition_manager
|
|
|
|
mock_pyaudio_instance = MagicMock()
|
|
mock_pyaudio_instance.get_default_input_device_info.return_value = {"index": 1}
|
|
mock_pyaudio_instance.get_device_count.return_value = 2
|
|
mock_pyaudio_instance.get_device_info_by_index.side_effect = [
|
|
{"name": "HDMI Output", "maxInputChannels": 0},
|
|
{"name": "USB 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()
|
|
|
|
self.assertEqual(devices, [(1, "USB Mic", True)])
|
|
|
|
def test_get_audio_input_devices_import_error(self):
|
|
"""Test missing PyAudio returns an empty device list."""
|
|
from vocalinux.speech_recognition import recognition_manager
|
|
|
|
with patch.dict(sys.modules, {"pyaudio": None}):
|
|
devices = recognition_manager.get_audio_input_devices()
|
|
|
|
self.assertEqual(devices, [])
|
|
|
|
def test_get_audio_input_devices_pyaudio_error(self):
|
|
"""Test PyAudio enumeration errors return an empty device list."""
|
|
from vocalinux.speech_recognition import recognition_manager
|
|
|
|
mock_pyaudio = MagicMock()
|
|
mock_pyaudio.PyAudio.side_effect = OSError("boom")
|
|
|
|
with patch.dict(sys.modules, {"pyaudio": mock_pyaudio}):
|
|
devices = recognition_manager.get_audio_input_devices()
|
|
|
|
self.assertEqual(devices, [])
|
|
|
|
def test_is_virtual_device(self):
|
|
"""Test _is_virtual_device detects known virtual device patterns."""
|
|
from vocalinux.speech_recognition.recognition_manager import _is_virtual_device
|
|
|
|
# Virtual devices
|
|
self.assertTrue(_is_virtual_device("speech-dispatcher-dummy"))
|
|
self.assertTrue(_is_virtual_device("speech-dispatcher-generic"))
|
|
self.assertTrue(_is_virtual_device("Dummy Output"))
|
|
self.assertTrue(_is_virtual_device("Null Sink"))
|
|
self.assertTrue(_is_virtual_device("Monitor of Built-in Audio"))
|
|
self.assertTrue(_is_virtual_device("alsa_output.pci-0000"))
|
|
|
|
# Real devices
|
|
self.assertFalse(_is_virtual_device("BRIO Ultra HD Webcam"))
|
|
self.assertFalse(_is_virtual_device("Built-in Microphone"))
|
|
self.assertFalse(_is_virtual_device("USB Headset"))
|
|
|
|
# Edge cases
|
|
self.assertFalse(_is_virtual_device(""))
|
|
self.assertFalse(_is_virtual_device(None))
|
|
|
|
def test_is_virtual_device_case_insensitive(self):
|
|
"""Test _is_virtual_device is case-insensitive."""
|
|
from vocalinux.speech_recognition.recognition_manager import _is_virtual_device
|
|
|
|
self.assertTrue(_is_virtual_device("SPEECH-DISPATCHER-DUMMY"))
|
|
self.assertTrue(_is_virtual_device("NULL SINK"))
|
|
self.assertTrue(_is_virtual_device("Monitor Of Something"))
|
|
|
|
def test_get_audio_input_devices_filters_virtual(self):
|
|
"""Test that get_audio_input_devices excludes virtual devices."""
|
|
from vocalinux.speech_recognition import recognition_manager
|
|
|
|
mock_pyaudio_instance = MagicMock()
|
|
mock_pyaudio_instance.get_default_input_device_info.return_value = {"index": 0}
|
|
mock_pyaudio_instance.get_device_count.return_value = 3
|
|
mock_pyaudio_instance.get_device_info_by_index.side_effect = [
|
|
{"name": "Real Mic", "maxInputChannels": 2},
|
|
{"name": "speech-dispatcher-dummy", "maxInputChannels": 2},
|
|
{"name": "USB Headset", "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()
|
|
|
|
# Should only have 2 devices (virtual one filtered out)
|
|
device_names = [d[1] for d in devices]
|
|
self.assertIn("Real Mic", device_names)
|
|
self.assertIn("USB Headset", device_names)
|
|
self.assertNotIn("speech-dispatcher-dummy", device_names)
|
|
|
|
@patch("vocalinux.audio.capture._resolve_valid_input_device")
|
|
def test_resolve_device_by_name_found(self, mock_valid):
|
|
"""Test _resolve_device_by_name finds a matching device and delegates validation."""
|
|
from vocalinux.speech_recognition.recognition_manager import _resolve_device_by_name
|
|
|
|
mock_valid.return_value = 1
|
|
mock_audio = MagicMock()
|
|
mock_audio.get_device_count.return_value = 3
|
|
mock_audio.get_device_info_by_index.side_effect = [
|
|
{"name": "Built-in Mic", "maxInputChannels": 1},
|
|
{"name": "BRIO Webcam", "maxInputChannels": 2},
|
|
{"name": "speech-dispatcher-dummy", "maxInputChannels": 2},
|
|
]
|
|
|
|
result = _resolve_device_by_name(mock_audio, "BRIO Webcam")
|
|
self.assertEqual(result, 1)
|
|
mock_valid.assert_called_once_with(mock_audio, 1)
|
|
|
|
@patch("vocalinux.audio.capture._resolve_valid_input_device")
|
|
def test_resolve_device_by_name_not_found_with_fallback(self, mock_valid):
|
|
"""Test _resolve_device_by_name falls back to index when name not found."""
|
|
from vocalinux.speech_recognition.recognition_manager import _resolve_device_by_name
|
|
|
|
mock_valid.return_value = 0
|
|
mock_audio = MagicMock()
|
|
mock_audio.get_device_count.return_value = 2
|
|
mock_audio.get_device_info_by_index.return_value = {
|
|
"name": "Mic",
|
|
"maxInputChannels": 1,
|
|
}
|
|
|
|
result = _resolve_device_by_name(mock_audio, "Missing Device", fallback_index=0)
|
|
self.assertEqual(result, 0)
|
|
mock_valid.assert_called_once_with(mock_audio, 0)
|
|
|
|
@patch("vocalinux.audio.capture._resolve_valid_input_device")
|
|
def test_resolve_device_by_name_skips_unreadable_device(self, mock_valid):
|
|
"""Test _resolve_device_by_name skips devices that cannot be read."""
|
|
from vocalinux.speech_recognition.recognition_manager import _resolve_device_by_name
|
|
|
|
mock_valid.return_value = 1
|
|
mock_audio = MagicMock()
|
|
mock_audio.get_device_count.return_value = 2
|
|
mock_audio.get_device_info_by_index.side_effect = [
|
|
OSError("device disappeared"),
|
|
{"name": "USB Mic", "maxInputChannels": 1},
|
|
]
|
|
|
|
result = _resolve_device_by_name(mock_audio, "USB Mic")
|
|
self.assertEqual(result, 1)
|
|
mock_valid.assert_called_once_with(mock_audio, 1)
|
|
|
|
@patch("vocalinux.audio.capture._resolve_valid_input_device")
|
|
def test_resolve_device_by_name_ignores_non_dict_device_info(self, mock_valid):
|
|
"""Test _resolve_device_by_name ignores malformed device info."""
|
|
from vocalinux.speech_recognition.recognition_manager import _resolve_device_by_name
|
|
|
|
mock_valid.return_value = 0
|
|
mock_audio = MagicMock()
|
|
mock_audio.get_device_count.return_value = 1
|
|
mock_audio.get_device_info_by_index.return_value = None
|
|
|
|
result = _resolve_device_by_name(mock_audio, "USB Mic", fallback_index=0)
|
|
self.assertEqual(result, 0)
|
|
mock_valid.assert_called_once_with(mock_audio, 0)
|
|
|
|
@patch("vocalinux.audio.capture._resolve_valid_input_device")
|
|
def test_resolve_device_by_name_not_found_no_fallback(self, mock_valid):
|
|
"""Test _resolve_device_by_name returns None when name not found and no fallback."""
|
|
from vocalinux.speech_recognition.recognition_manager import _resolve_device_by_name
|
|
|
|
mock_valid.return_value = None
|
|
mock_audio = MagicMock()
|
|
mock_audio.get_device_count.return_value = 1
|
|
mock_audio.get_device_info_by_index.return_value = {
|
|
"name": "Mic",
|
|
"maxInputChannels": 1,
|
|
}
|
|
|
|
result = _resolve_device_by_name(mock_audio, "Missing Device")
|
|
self.assertIsNone(result)
|
|
|
|
@patch("vocalinux.audio.capture._resolve_valid_input_device")
|
|
def test_resolve_device_by_name_empty_name_with_fallback(self, mock_valid):
|
|
"""Test _resolve_device_by_name with empty name uses fallback index."""
|
|
from vocalinux.speech_recognition.recognition_manager import _resolve_device_by_name
|
|
|
|
mock_valid.return_value = 0
|
|
mock_audio = MagicMock()
|
|
|
|
result = _resolve_device_by_name(mock_audio, None, fallback_index=0)
|
|
self.assertEqual(result, 0)
|
|
mock_valid.assert_called_once_with(mock_audio, 0)
|
|
|
|
@patch("vocalinux.audio.capture._resolve_valid_input_device")
|
|
def test_resolve_device_by_name_empty_name_no_fallback(self, mock_valid):
|
|
"""Test _resolve_device_by_name with empty name and no fallback returns None."""
|
|
from vocalinux.speech_recognition.recognition_manager import _resolve_device_by_name
|
|
|
|
mock_valid.return_value = None
|
|
mock_audio = MagicMock()
|
|
result = _resolve_device_by_name(mock_audio, None)
|
|
self.assertIsNone(result)
|
|
|
|
@patch("vocalinux.audio.capture._resolve_valid_input_device")
|
|
def test_resolve_device_by_name_get_count_error(self, mock_valid):
|
|
"""Test _resolve_device_by_name handles get_device_count errors."""
|
|
from vocalinux.speech_recognition.recognition_manager import _resolve_device_by_name
|
|
|
|
mock_valid.return_value = 0
|
|
mock_audio = MagicMock()
|
|
mock_audio.get_device_count.side_effect = IOError("boom")
|
|
|
|
result = _resolve_device_by_name(mock_audio, "Some Device", fallback_index=0)
|
|
self.assertEqual(result, 0)
|
|
mock_valid.assert_called_once_with(mock_audio, 0)
|
|
|
|
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)
|