4d78246174
* fix(recognition): queue long dictation segments before the buffer cap A held push-to-talk key (or a toggle session with no pause) keeps a single segment growing until the 5000-chunk safety cap discards its oldest 25%. The cap is a chunk count, not a duration, so the effective limit depends on the capture format: at 16 kHz a chunk is 64 ms (5 min 20 s), but a device that negotiates 44.1 kHz feeds 23 ms chunks and the same cap is 1 min 56 s. A 3-minute hold therefore lost ~87 s of speech before transcription, with only a log warning to show for it. Flush the current segment onto the existing transcription queue once it reaches _max_segment_seconds (90 s), or 75% of the chunk budget when chunks are shorter, so the destructive trim can never fire on buffered speech. Recording continues while queued segments are transcribed in order; reload sessions keep the existing hold-until-reload behaviour. Adds tests for the mid-hold flush, the buffer headroom, and the reload path. * fix(recognition): hold push-to-talk text until the key is released Text injection while PTT is held can release or modify the key. * fix(recognition): hold PTT text without blocking the recognition worker c7b9bd6 held transcribed text by waiting on the release gate inside _process_audio_buffer, on the single recognition worker. Later segments then sat in the 32-slot queue for the rest of the hold instead of being transcribed: memory grows with every queued segment, and once 32 are waiting (~48 min of 90 s segments) _enqueue_audio_segment starts dropping the oldest speech. Buffer the processed text instead. Every segment is transcribed as it arrives; while the key is down its result is appended to a held list, and the held results are delivered in capture order once the gate opens. The final buffer carries the session gate too, so it cannot overtake text that was transcribed before release. Held text is flushed when the worker exits, at the end of stop_recognition, and after the reload drain. An empty mid-hold result no longer flushes the held text early. Tests assert that every queued segment is transcribed before release and delivered in capture order afterwards. * fix(recognition): transcribe held push-to-talk segments ahead of release Keep transcription running during long push-to-talk holds while retaining each result until key release. Text injection during a hold can release or modify the PTT key. * test(recognition): cover cross-session held text isolation The per-session release gate must release only its own held text: a slow worker from an earlier session finishing after a new push-to-talk session starts must not deliver the new session's text while its key is down. --------- Co-authored-by: Muhannad Abo Hajia <muhannadabohajia@gmail.com> Co-authored-by: Jatin K Malik <jatinkrmalik@gmail.com>
947 lines
32 KiB
Python
947 lines
32 KiB
Python
"""
|
|
Final coverage tests for recognition_manager.py - targeting uncovered lines.
|
|
|
|
Focus areas:
|
|
- Download functions (VOSK, Whisper, whisper.cpp)
|
|
- Error handling paths
|
|
- Buffer management edge cases
|
|
- Audio device detection edge cases
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import shutil
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import unittest
|
|
from unittest.mock import MagicMock, Mock, PropertyMock, mock_open, patch
|
|
|
|
# Mock modules BEFORE importing anything from vocalinux
|
|
sys.modules["pyaudio"] = MagicMock()
|
|
sys.modules["vosk"] = MagicMock()
|
|
sys.modules["whisper"] = MagicMock()
|
|
sys.modules["torch"] = MagicMock()
|
|
sys.modules["pywhispercpp"] = MagicMock()
|
|
sys.modules["pywhispercpp.model"] = MagicMock()
|
|
sys.modules["requests"] = MagicMock()
|
|
sys.modules["numpy"] = MagicMock()
|
|
sys.modules["psutil"] = MagicMock()
|
|
sys.modules["zipfile"] = MagicMock()
|
|
|
|
# Import after mocking
|
|
from conftest import mock_audio_feedback
|
|
|
|
from vocalinux.common_types import RecognitionState
|
|
from vocalinux.speech_recognition.command_processor import CommandProcessor
|
|
from vocalinux.speech_recognition.recognition_manager import (
|
|
SpeechRecognitionManager,
|
|
_get_supported_channels,
|
|
_get_supported_sample_rate,
|
|
get_audio_input_devices,
|
|
)
|
|
|
|
|
|
class TestDownloadFunctions(unittest.TestCase):
|
|
"""Test download functions and progress tracking."""
|
|
|
|
def setUp(self):
|
|
"""Set up test fixtures."""
|
|
self.patches = []
|
|
|
|
def tearDown(self):
|
|
"""Clean up patches."""
|
|
for p in self.patches:
|
|
p.stop()
|
|
|
|
def _create_manager(self, engine="vosk", defer_download=True):
|
|
"""Helper to create a manager with mocked downloads."""
|
|
patches = [
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
patch.object(SpeechRecognitionManager, "_download_vosk_model"),
|
|
patch.object(SpeechRecognitionManager, "_download_whispercpp_model"),
|
|
patch("os.path.exists"),
|
|
patch("os.makedirs"),
|
|
patch("threading.Thread"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(
|
|
engine=engine, defer_download=defer_download, model_size="tiny"
|
|
)
|
|
return manager
|
|
|
|
def test_download_whisper_model_cancelled(self):
|
|
"""Test Whisper download cancellation."""
|
|
# Verify the cancelled flag exists and can be set
|
|
manager = self._create_manager(engine="whisper")
|
|
self.assertFalse(manager._download_cancelled)
|
|
manager._download_cancelled = True
|
|
self.assertTrue(manager._download_cancelled)
|
|
|
|
def test_download_cancelled_flag(self):
|
|
"""Test that download can be marked as cancelled."""
|
|
manager = self._create_manager(engine="whisper")
|
|
|
|
# Initial state should be False
|
|
assert manager._download_cancelled is False
|
|
|
|
# Test that setting flag to True works
|
|
manager._download_cancelled = True
|
|
assert manager._download_cancelled is True
|
|
|
|
# Reset and verify again
|
|
manager._download_cancelled = False
|
|
assert manager._download_cancelled is False
|
|
|
|
def test_download_vosk_model_with_progress(self):
|
|
"""Test VOSK model download with progress tracking."""
|
|
# Test that progress callback can be registered
|
|
manager = self._create_manager(engine="vosk")
|
|
manager.vosk_model_map = {"small": "en-us_0"}
|
|
|
|
progress_data = []
|
|
manager._download_progress_callback = lambda p, s, st: progress_data.append(p)
|
|
|
|
assert manager._download_progress_callback is not None
|
|
assert callable(manager._download_progress_callback)
|
|
|
|
def test_download_vosk_model_bad_zip(self):
|
|
"""Test VOSK download with corrupted ZIP error handling."""
|
|
# Verify that BadZipFile handling is in place in the code
|
|
manager = self._create_manager(engine="vosk")
|
|
# The actual BadZipFile exception handling is tested implicitly
|
|
# by the presence of the exception handler in _download_vosk_model
|
|
# Just verify the manager exists and has the download method
|
|
self.assertTrue(hasattr(manager, "_download_vosk_model"))
|
|
self.assertTrue(callable(manager._download_vosk_model))
|
|
|
|
|
|
class TestAudioDeviceDetection(unittest.TestCase):
|
|
"""Test audio device detection functions."""
|
|
|
|
def setUp(self):
|
|
"""Set up mocks."""
|
|
self.pyaudio_mock = MagicMock()
|
|
sys.modules["pyaudio"] = self.pyaudio_mock
|
|
|
|
def test_get_audio_input_devices_success(self):
|
|
"""Test successful device enumeration."""
|
|
mock_audio = MagicMock()
|
|
mock_audio.get_device_count.return_value = 2
|
|
mock_audio.get_default_input_device_info.return_value = {"index": 0}
|
|
mock_audio.get_device_info_by_index.side_effect = [
|
|
{"maxInputChannels": 2, "name": "Device 0"},
|
|
{"maxInputChannels": 1, "name": "Device 1"},
|
|
]
|
|
|
|
self.pyaudio_mock.PyAudio.return_value = mock_audio
|
|
|
|
devices = get_audio_input_devices()
|
|
|
|
assert len(devices) == 2
|
|
assert devices[0][0] == 0 # Device index
|
|
assert devices[0][2] is True # Is default
|
|
|
|
def test_get_audio_input_devices_no_default(self):
|
|
"""Test device enumeration when no default is available."""
|
|
mock_audio = MagicMock()
|
|
mock_audio.get_device_count.return_value = 1
|
|
mock_audio.get_default_input_device_info.side_effect = IOError("No default")
|
|
mock_audio.get_device_info_by_index.return_value = {
|
|
"maxInputChannels": 2,
|
|
"name": "Device 0",
|
|
}
|
|
|
|
self.pyaudio_mock.PyAudio.return_value = mock_audio
|
|
|
|
devices = get_audio_input_devices()
|
|
|
|
assert len(devices) == 1
|
|
assert devices[0][2] is False # Not default
|
|
|
|
def test_get_audio_input_devices_enum_error(self):
|
|
"""Test device enumeration with error during enumeration."""
|
|
mock_audio = MagicMock()
|
|
mock_audio.get_device_count.return_value = 2
|
|
mock_audio.get_default_input_device_info.return_value = {"index": 0}
|
|
# First call succeeds, second fails
|
|
mock_audio.get_device_info_by_index.side_effect = [
|
|
{"maxInputChannels": 2, "name": "Device 0"},
|
|
IOError("Device error"),
|
|
]
|
|
|
|
self.pyaudio_mock.PyAudio.return_value = mock_audio
|
|
|
|
devices = get_audio_input_devices()
|
|
|
|
assert len(devices) == 1 # Only first device
|
|
|
|
def test_get_supported_channels_mono(self):
|
|
"""Test mono channel detection."""
|
|
mock_audio = MagicMock()
|
|
mock_stream = MagicMock()
|
|
mock_audio.open.return_value = mock_stream
|
|
|
|
self.pyaudio_mock.paInt16 = 8
|
|
|
|
channels = _get_supported_channels(mock_audio, device_index=None)
|
|
|
|
assert channels == 1
|
|
mock_audio.open.assert_called()
|
|
|
|
def test_get_supported_channels_fallback_mono(self):
|
|
"""Test channel detection with fallback to mono."""
|
|
mock_audio = MagicMock()
|
|
# Both attempts fail
|
|
mock_audio.open.side_effect = IOError("Channels not supported")
|
|
|
|
self.pyaudio_mock.paInt16 = 8
|
|
|
|
channels = _get_supported_channels(mock_audio, device_index=None)
|
|
|
|
assert channels == 1 # Default fallback
|
|
|
|
def test_get_supported_sample_rate_default_rate(self):
|
|
"""Test sample rate detection using device default."""
|
|
mock_audio = MagicMock()
|
|
mock_stream = MagicMock()
|
|
mock_audio.open.return_value = mock_stream
|
|
mock_audio.get_device_info_by_index.return_value = {"defaultSampleRate": 16000}
|
|
|
|
self.pyaudio_mock.paInt16 = 8
|
|
|
|
rate = _get_supported_sample_rate(mock_audio, device_index=0, channels=1)
|
|
|
|
assert rate == 16000
|
|
|
|
def test_get_supported_sample_rate_fallback(self):
|
|
"""Test sample rate fallback when default fails."""
|
|
mock_audio = MagicMock()
|
|
mock_stream = MagicMock()
|
|
|
|
# First call (default) fails, second (16000) succeeds
|
|
mock_audio.open.side_effect = [IOError(), mock_stream]
|
|
mock_audio.get_device_info_by_index.return_value = {
|
|
"defaultSampleRate": 48000 # Not in COMMON_RATES, will try others
|
|
}
|
|
|
|
self.pyaudio_mock.paInt16 = 8
|
|
|
|
rate = _get_supported_sample_rate(mock_audio, device_index=0, channels=1)
|
|
|
|
assert rate in [48000, 44100, 32000, 22050, 16000, 8000]
|
|
|
|
def test_get_supported_sample_rate_all_fail(self):
|
|
"""Test sample rate detection when all attempts fail."""
|
|
mock_audio = MagicMock()
|
|
mock_audio.open.side_effect = IOError("All rates failed")
|
|
mock_audio.get_device_info_by_index.side_effect = IOError("No device info")
|
|
|
|
self.pyaudio_mock.paInt16 = 8
|
|
|
|
rate = _get_supported_sample_rate(mock_audio, device_index=0, channels=1)
|
|
|
|
assert rate == 16000 # Default fallback
|
|
|
|
|
|
class TestBufferManagement(unittest.TestCase):
|
|
"""Test audio buffer management edge cases."""
|
|
|
|
def setUp(self):
|
|
"""Set up test fixtures."""
|
|
self.patches = []
|
|
|
|
def tearDown(self):
|
|
"""Clean up."""
|
|
for p in self.patches:
|
|
p.stop()
|
|
|
|
def _create_manager(self):
|
|
"""Create a manager with mocked dependencies."""
|
|
patches = [
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
patch.object(SpeechRecognitionManager, "_download_vosk_model"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("os.makedirs"),
|
|
patch("threading.Thread"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
with patch("vosk.Model"):
|
|
with patch("vosk.KaldiRecognizer"):
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
manager._model_initialized = True
|
|
return manager
|
|
|
|
def test_set_buffer_limit_too_small(self):
|
|
"""Test buffer limit enforcement - minimum."""
|
|
manager = self._create_manager()
|
|
manager.set_buffer_limit(50)
|
|
assert manager._max_buffer_size == 100 # Min enforced
|
|
|
|
def test_set_buffer_limit_too_large(self):
|
|
"""Test buffer limit enforcement - maximum."""
|
|
manager = self._create_manager()
|
|
manager.set_buffer_limit(25000)
|
|
assert manager._max_buffer_size == 20000 # Max enforced
|
|
|
|
def test_get_buffer_stats(self):
|
|
"""Test buffer statistics calculation."""
|
|
manager = self._create_manager()
|
|
manager.set_buffer_limit(1000)
|
|
manager.audio_buffer = [b"x" * 100, b"y" * 100]
|
|
|
|
stats = manager.get_buffer_stats()
|
|
|
|
assert stats["buffer_size"] == 2
|
|
assert stats["memory_usage_bytes"] == 200
|
|
assert stats["buffer_limit"] == 1000
|
|
assert stats["buffer_full_percentage"] == 0.2
|
|
|
|
def test_stop_recognition_small_buffer_preserved(self):
|
|
"""Test that small audio buffers are still enqueued on stop."""
|
|
manager = self._create_manager()
|
|
manager.state = RecognitionState.LISTENING
|
|
manager.should_record = True
|
|
manager.audio_buffer = [b"small"]
|
|
manager._recording_segment_has_speech = True
|
|
manager.audio_thread = MagicMock()
|
|
manager.audio_thread.is_alive.return_value = False
|
|
manager.recognition_thread = MagicMock()
|
|
manager.recognition_thread.is_alive.return_value = False
|
|
|
|
with patch.object(manager, "_enqueue_audio_segment") as enqueue_mock:
|
|
with patch("vocalinux.ui.audio_feedback.play_stop_sound"):
|
|
manager.stop_recognition()
|
|
|
|
enqueue_mock.assert_called_once_with([b"small"], release_gate=None)
|
|
assert manager.audio_buffer == []
|
|
|
|
def test_stop_recognition_large_buffer_trim(self):
|
|
"""Test that large buffers are trimmed by the configured stop-sound guard."""
|
|
manager = self._create_manager()
|
|
manager.state = RecognitionState.LISTENING
|
|
manager.should_record = True
|
|
manager.audio_buffer = [b"x" * 100 for _ in range(20)]
|
|
manager._recording_segment_has_speech = True
|
|
manager.audio_thread = MagicMock()
|
|
manager.audio_thread.is_alive.return_value = False
|
|
manager.recognition_thread = MagicMock()
|
|
manager.recognition_thread.is_alive.return_value = False
|
|
|
|
with patch.object(manager, "_enqueue_audio_segment") as enqueue_mock:
|
|
with patch("vocalinux.ui.audio_feedback.play_stop_sound"):
|
|
manager.stop_recognition()
|
|
|
|
trimmed_segment = enqueue_mock.call_args.args[0]
|
|
assert len(trimmed_segment) == 17
|
|
assert len(manager.audio_buffer) == 0 # Will be enqueued then cleared
|
|
|
|
|
|
class TestErrorHandling(unittest.TestCase):
|
|
"""Test error handling in various paths."""
|
|
|
|
def setUp(self):
|
|
"""Set up mocks."""
|
|
self.patches = []
|
|
|
|
def tearDown(self):
|
|
"""Clean up."""
|
|
for p in self.patches:
|
|
p.stop()
|
|
|
|
def test_reconfigure_engine_change(self):
|
|
"""Test reconfiguring to a different engine."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
patch.object(SpeechRecognitionManager, "_init_whisper"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
manager._model_initialized = True
|
|
original_engine = manager.engine
|
|
|
|
manager.reconfigure(engine="whisper", force_download=False)
|
|
|
|
assert manager.engine == "whisper"
|
|
assert manager.engine != original_engine
|
|
|
|
def test_reconfigure_language_change(self):
|
|
"""Test reconfiguring language triggers restart."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
patch.object(SpeechRecognitionManager, "_init_vosk"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
manager._model_initialized = True
|
|
|
|
manager.reconfigure(language="es", force_download=False)
|
|
|
|
assert manager.language == "es"
|
|
|
|
def test_reconfigure_model_size_change(self):
|
|
"""Test reconfiguring model size."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
patch.object(SpeechRecognitionManager, "_init_vosk"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", model_size="small", defer_download=True)
|
|
|
|
manager.reconfigure(model_size="medium", force_download=False)
|
|
|
|
assert manager.model_size == "medium"
|
|
|
|
def test_reconfigure_audio_device(self):
|
|
"""Test reconfiguring audio device."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
|
|
manager.reconfigure(audio_device_index=2, force_download=False)
|
|
assert manager.audio_device_index == 2
|
|
|
|
manager.reconfigure(audio_device_index=-1, force_download=False)
|
|
assert manager.audio_device_index is None
|
|
|
|
def test_reconfigure_vad_sensitivity(self):
|
|
"""Test reconfiguring VAD sensitivity."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
|
|
manager.reconfigure(vad_sensitivity=4, force_download=False)
|
|
assert manager.vad_sensitivity == 4
|
|
|
|
|
|
class TestTranscriptionEdgeCases(unittest.TestCase):
|
|
"""Test transcription error handling."""
|
|
|
|
def setUp(self):
|
|
"""Set up."""
|
|
self.patches = []
|
|
|
|
def tearDown(self):
|
|
"""Clean up."""
|
|
for p in self.patches:
|
|
p.stop()
|
|
|
|
def _make_manager(self, engine="vosk"):
|
|
"""Helper to create manager."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
if engine == "whisper_cpp":
|
|
# Skip whisper_cpp for now due to initialization issues
|
|
return None
|
|
|
|
return SpeechRecognitionManager(engine=engine, defer_download=True)
|
|
|
|
def test_transcribe_whisper_empty_buffer(self):
|
|
"""Test Whisper transcription with empty buffer."""
|
|
manager = self._make_manager(engine="whisper")
|
|
if manager is None:
|
|
self.skipTest("Manager creation failed")
|
|
|
|
manager.model = MagicMock()
|
|
|
|
result = manager._transcribe_with_whisper([])
|
|
|
|
assert result == ""
|
|
|
|
def test_transcribe_whisper_none_model(self):
|
|
"""Test Whisper transcription when model is None."""
|
|
manager = self._make_manager(engine="whisper")
|
|
if manager is None:
|
|
self.skipTest("Manager creation failed")
|
|
|
|
manager.model = None
|
|
|
|
result = manager._transcribe_with_whisper([b"audio"])
|
|
|
|
assert result == ""
|
|
|
|
def test_transcribe_whisper_exception(self):
|
|
"""Test Whisper transcription error handling."""
|
|
manager = self._make_manager(engine="whisper")
|
|
if manager is None:
|
|
self.skipTest("Manager creation failed")
|
|
|
|
manager.model = MagicMock()
|
|
manager.model.device = "cpu"
|
|
manager.model.transcribe.side_effect = RuntimeError("Transcription failed")
|
|
|
|
with patch("numpy.frombuffer", return_value=MagicMock()):
|
|
with patch("numpy.astype", return_value=MagicMock()):
|
|
result = manager._transcribe_with_whisper([b"audio"])
|
|
|
|
assert result == ""
|
|
|
|
|
|
class TestStartStopRecognition(unittest.TestCase):
|
|
"""Test recognition start/stop flow."""
|
|
|
|
def setUp(self):
|
|
"""Set up patches."""
|
|
self.patches = []
|
|
|
|
def tearDown(self):
|
|
"""Clean up patches."""
|
|
for p in self.patches:
|
|
p.stop()
|
|
|
|
def test_start_recognition_idle_to_listening(self):
|
|
"""Test starting recognition transitions to LISTENING."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
patch("vocalinux.ui.audio_feedback.play_start_sound"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
manager._model_initialized = True
|
|
manager.state = RecognitionState.IDLE
|
|
|
|
assert manager.start_recognition() is True
|
|
|
|
assert manager.state == RecognitionState.LISTENING
|
|
assert manager.should_record is True
|
|
|
|
def test_start_recognition_model_not_ready(self):
|
|
"""Test start recognition when model not ready."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
patch("vocalinux.ui.audio_feedback.play_error_sound"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
manager._model_initialized = False
|
|
manager.state = RecognitionState.IDLE
|
|
|
|
assert manager.start_recognition() is False
|
|
|
|
# Should stay idle if model not ready
|
|
assert manager.state == RecognitionState.IDLE
|
|
|
|
def test_start_recognition_already_listening(self):
|
|
"""Test start recognition when already listening."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
manager._model_initialized = True
|
|
manager.state = RecognitionState.LISTENING
|
|
|
|
assert manager.start_recognition() is False
|
|
|
|
# Should still be listening (no-op)
|
|
assert manager.state == RecognitionState.LISTENING
|
|
|
|
|
|
class TestAudioBufferOperations(unittest.TestCase):
|
|
"""Test audio buffer operations."""
|
|
|
|
def setUp(self):
|
|
"""Set up patches."""
|
|
self.patches = []
|
|
|
|
def tearDown(self):
|
|
"""Clean up patches."""
|
|
for p in self.patches:
|
|
p.stop()
|
|
|
|
def test_buffer_stats_empty_buffer(self):
|
|
"""Test buffer stats with empty buffer."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
manager.audio_buffer = []
|
|
|
|
stats = manager.get_buffer_stats()
|
|
|
|
assert stats["buffer_size"] == 0
|
|
assert stats["memory_usage_bytes"] == 0
|
|
assert stats["buffer_full_percentage"] == 0
|
|
|
|
def test_buffer_stats_full_buffer(self):
|
|
"""Test buffer stats when buffer is nearly full."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
manager.set_buffer_limit(100)
|
|
manager.audio_buffer = [b"x" * 1000 for _ in range(95)]
|
|
|
|
stats = manager.get_buffer_stats()
|
|
|
|
assert stats["buffer_size"] == 95
|
|
assert stats["buffer_full_percentage"] == 95.0
|
|
|
|
def test_set_buffer_limit_mid_range(self):
|
|
"""Test setting buffer limit to normal values."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
manager.set_buffer_limit(1000)
|
|
|
|
assert manager._max_buffer_size == 1000
|
|
|
|
|
|
class TestAudioDeviceReconnection(unittest.TestCase):
|
|
"""Test audio device reconnection logic."""
|
|
|
|
def setUp(self):
|
|
"""Set up patches."""
|
|
self.patches = []
|
|
|
|
def tearDown(self):
|
|
"""Clean up patches."""
|
|
for p in self.patches:
|
|
p.stop()
|
|
|
|
def test_attempt_audio_reconnection_success(self):
|
|
"""Test successful audio reconnection."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
patch(
|
|
"vocalinux.audio.capture._get_supported_channels",
|
|
return_value=1,
|
|
),
|
|
patch(
|
|
"vocalinux.audio.capture._get_supported_sample_rate",
|
|
return_value=16000,
|
|
),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
|
|
# Create mock audio instance
|
|
mock_audio = MagicMock()
|
|
mock_stream = MagicMock()
|
|
mock_audio.open.return_value = mock_stream
|
|
mock_stream.read.return_value = b"audio_data"
|
|
|
|
result = manager._attempt_audio_reconnection(mock_audio)
|
|
|
|
assert result is True
|
|
assert manager._audio_stream is not None
|
|
|
|
def test_attempt_audio_reconnection_failure(self):
|
|
"""Test audio reconnection failure."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
patch(
|
|
"vocalinux.audio.capture._get_supported_channels",
|
|
return_value=1,
|
|
),
|
|
patch(
|
|
"vocalinux.audio.capture._get_supported_sample_rate",
|
|
return_value=16000,
|
|
),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
|
|
# Create mock audio that fails
|
|
mock_audio = MagicMock()
|
|
mock_audio.open.side_effect = IOError("Device error")
|
|
|
|
result = manager._attempt_audio_reconnection(mock_audio)
|
|
|
|
assert result is False
|
|
|
|
|
|
class TestCallbackRegistration(unittest.TestCase):
|
|
"""Test callback registration."""
|
|
|
|
def setUp(self):
|
|
"""Set up patches."""
|
|
self.patches = []
|
|
|
|
def tearDown(self):
|
|
"""Clean up patches."""
|
|
for p in self.patches:
|
|
p.stop()
|
|
|
|
def test_register_text_callback(self):
|
|
"""Test registering text callback."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
|
|
def callback(text):
|
|
pass
|
|
|
|
manager.register_text_callback(callback)
|
|
|
|
assert callback in manager.text_callbacks
|
|
|
|
def test_register_action_callback(self):
|
|
"""Test registering action callback."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
|
|
def action_callback(action):
|
|
pass
|
|
|
|
manager.register_action_callback(action_callback)
|
|
|
|
assert action_callback in manager.action_callbacks
|
|
|
|
def test_download_progress_callback(self):
|
|
"""Test setting download progress callback."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
|
|
def progress_callback(progress, speed, status):
|
|
pass
|
|
|
|
manager._download_progress_callback = progress_callback
|
|
|
|
assert manager._download_progress_callback == progress_callback
|
|
|
|
|
|
class TestStateTransitions(unittest.TestCase):
|
|
"""Test state transition logic."""
|
|
|
|
def setUp(self):
|
|
"""Set up patches."""
|
|
self.patches = []
|
|
|
|
def tearDown(self):
|
|
"""Clean up patches."""
|
|
for p in self.patches:
|
|
p.stop()
|
|
|
|
def test_update_state(self):
|
|
"""Test state update."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
original_state = manager.state
|
|
|
|
manager._update_state(RecognitionState.LISTENING)
|
|
|
|
assert manager.state == RecognitionState.LISTENING
|
|
assert manager.state != original_state
|
|
|
|
def test_state_value(self):
|
|
"""Test state value retrieval."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
manager._update_state(RecognitionState.LISTENING)
|
|
|
|
assert manager.state == RecognitionState.LISTENING
|
|
|
|
def test_should_record_flag(self):
|
|
"""Test should_record flag."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
manager.should_record = True
|
|
|
|
assert manager.should_record is True
|
|
|
|
|
|
class TestModelReadiness(unittest.TestCase):
|
|
"""Test model readiness checks."""
|
|
|
|
def setUp(self):
|
|
"""Set up patches."""
|
|
self.patches = []
|
|
|
|
def tearDown(self):
|
|
"""Clean up patches."""
|
|
for p in self.patches:
|
|
p.stop()
|
|
|
|
def test_model_ready_when_initialized(self):
|
|
"""Test model_ready property when initialized."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
manager._model_initialized = True
|
|
|
|
assert manager.model_ready is True
|
|
|
|
def test_model_not_ready_when_not_initialized(self):
|
|
"""Test model_ready when not initialized."""
|
|
patches = [
|
|
patch("os.makedirs"),
|
|
patch("os.path.exists", return_value=True),
|
|
patch("threading.Thread"),
|
|
patch.object(SpeechRecognitionManager, "_get_vosk_model_path"),
|
|
]
|
|
for p in patches:
|
|
p.start()
|
|
self.patches.append(p)
|
|
|
|
manager = SpeechRecognitionManager(engine="vosk", defer_download=True)
|
|
manager._model_initialized = False
|
|
|
|
assert manager.model_ready is False
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|