Jatin K MalikandGitHub
2152dbb90d
* fix(whispercpp): skip unsupported context_params on pywhispercpp 1.4 Fixes #625 * test(whispercpp): cover context_params signature inspection failure
1382 lines
58 KiB
Python
1382 lines
58 KiB
Python
"""
|
|
Extra tests for recognition_manager.py to increase coverage.
|
|
|
|
Focuses on uncovered lines in:
|
|
- _init_vosk() error handling and model path resolution
|
|
- _init_whisper() model validation and error paths
|
|
- _init_whispercpp() backend detection and GPU fallback
|
|
- Download functions with progress tracking and error handling
|
|
- Transcription with model lock and error conditions
|
|
- start_recognition() and stop_recognition() flows
|
|
- Audio device detection and sample rate negotiation
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import queue
|
|
import struct
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import unittest
|
|
from unittest import mock
|
|
from unittest.mock import MagicMock, Mock, call, patch
|
|
|
|
import pytest
|
|
|
|
from vocalinux.common_types import RecognitionState
|
|
from vocalinux.speech_recognition.recognition_manager import (
|
|
SpeechRecognitionManager,
|
|
_filter_non_speech,
|
|
_get_supported_channels,
|
|
_get_supported_sample_rate,
|
|
_is_bluetooth_device,
|
|
_open_capture_stream,
|
|
_safe_close_stream,
|
|
get_audio_input_devices,
|
|
test_audio_input,
|
|
)
|
|
|
|
|
|
def _make_manager(engine="whisper_cpp", **kw):
|
|
"""Helper to create a manager with all init methods patched."""
|
|
with patch.object(SpeechRecognitionManager, "_init_vosk"):
|
|
with patch.object(SpeechRecognitionManager, "_init_whisper"):
|
|
with patch.object(SpeechRecognitionManager, "_init_whispercpp"):
|
|
return SpeechRecognitionManager(
|
|
engine=engine, model_size="small", language="en-us", defer_download=True, **kw
|
|
)
|
|
|
|
|
|
class TestAudioDeviceDetection(unittest.TestCase):
|
|
"""Test audio device enumeration functions."""
|
|
|
|
def test_is_bluetooth_device_detection(self):
|
|
"""Bluetooth headset/mic names should be recognized (Issue #567)."""
|
|
assert _is_bluetooth_device("Bluetooth internal capture stream for HUAWEI FreeBuds Pro 3")
|
|
assert _is_bluetooth_device("bluez_source.XX_XX_XX")
|
|
assert _is_bluetooth_device("Hands-Free AG Speech")
|
|
assert not _is_bluetooth_device("WH-1000XM5 Headset") # no bluetooth marker
|
|
assert not _is_bluetooth_device("USB Audio Device")
|
|
assert not _is_bluetooth_device(None)
|
|
assert not _is_bluetooth_device("")
|
|
|
|
def test_safe_close_stream_stops_before_close(self):
|
|
"""PortAudio streams must be stopped before close to avoid heap corruption."""
|
|
mock_stream = MagicMock()
|
|
_safe_close_stream(mock_stream)
|
|
mock_stream.stop_stream.assert_called_once()
|
|
mock_stream.close.assert_called_once()
|
|
# stop must happen before close
|
|
assert mock_stream.mock_calls.index(mock.call.stop_stream()) < mock_stream.mock_calls.index(
|
|
mock.call.close()
|
|
)
|
|
|
|
def test_safe_close_stream_tolerates_errors(self):
|
|
"""Cleanup must not raise even if stop/close fail."""
|
|
mock_stream = MagicMock()
|
|
mock_stream.stop_stream.side_effect = OSError("already stopped")
|
|
mock_stream.close.side_effect = OSError("already closed")
|
|
_safe_close_stream(mock_stream) # should not raise
|
|
_safe_close_stream(None) # should not raise
|
|
|
|
def test_open_capture_stream_returns_live_stream(self):
|
|
"""Negotiation must return the opened stream, never close-and-reopen."""
|
|
mock_audio = MagicMock()
|
|
mock_stream = MagicMock()
|
|
mock_audio.open.return_value = mock_stream
|
|
mock_audio.get_device_info_by_index.return_value = {
|
|
"name": "USB Mic",
|
|
"defaultSampleRate": 48000,
|
|
"maxInputChannels": 1,
|
|
}
|
|
mock_pyaudio = MagicMock(paInt16=8)
|
|
|
|
with patch.dict("sys.modules", {"pyaudio": mock_pyaudio}):
|
|
channels, rate, stream = _open_capture_stream(mock_audio, 0)
|
|
assert channels == 1
|
|
assert rate == 48000
|
|
assert stream is mock_stream
|
|
assert mock_audio.open.call_count == 1
|
|
# The negotiated stream is handed to the caller live.
|
|
mock_stream.stop_stream.assert_not_called()
|
|
mock_stream.close.assert_not_called()
|
|
|
|
def test_open_capture_stream_bluetooth_single_open(self):
|
|
"""Bluetooth SCO capture must be opened exactly once (Issue #567).
|
|
|
|
The previous channels-then-rate-then-real-open flow opened the SCO
|
|
capture stream three times in under a second, which aborts with
|
|
malloc heap corruption on devices like the HUAWEI FreeBuds Pro 3.
|
|
"""
|
|
mock_audio = MagicMock()
|
|
mock_stream = MagicMock()
|
|
mock_audio.open.return_value = mock_stream
|
|
mock_audio.get_device_info_by_index.return_value = {
|
|
"name": "Bluetooth internal capture stream for HUAWEI FreeBuds Pro 3",
|
|
"defaultSampleRate": 16000,
|
|
"maxInputChannels": 1,
|
|
}
|
|
mock_pyaudio = MagicMock(paInt16=8)
|
|
|
|
with patch.dict("sys.modules", {"pyaudio": mock_pyaudio}):
|
|
channels, rate, stream = _open_capture_stream(mock_audio, 14)
|
|
assert channels == 1
|
|
assert rate == 16000
|
|
assert stream is mock_stream
|
|
assert mock_audio.open.call_count == 1
|
|
|
|
def test_open_capture_stream_mono_device_never_probes_stereo(self):
|
|
"""Mono-only devices must never be opened with 2 channels.
|
|
|
|
Opening with more channels than the device supports is itself a
|
|
PortAudio/ALSA heap-corruption trigger.
|
|
"""
|
|
mock_audio = MagicMock()
|
|
mock_audio.open.side_effect = IOError("cannot open")
|
|
mock_audio.get_device_info_by_index.return_value = {
|
|
"name": "Bluetooth internal capture stream",
|
|
"defaultSampleRate": 16000,
|
|
"maxInputChannels": 1,
|
|
}
|
|
mock_pyaudio = MagicMock(paInt16=8)
|
|
|
|
with patch.dict("sys.modules", {"pyaudio": mock_pyaudio}):
|
|
with patch("vocalinux.speech_recognition.recognition_manager.time.sleep"):
|
|
channels, rate, stream = _open_capture_stream(mock_audio, 14)
|
|
assert stream is None
|
|
assert (channels, rate) == (1, 16000)
|
|
assert all(call.kwargs.get("channels") == 1 for call in mock_audio.open.call_args_list)
|
|
|
|
def test_open_capture_stream_failure_returns_none_stream(self):
|
|
"""Total negotiation failure returns (1, 16000, None) for caller fallback."""
|
|
mock_audio = MagicMock()
|
|
mock_audio.open.side_effect = IOError("no device")
|
|
mock_audio.get_device_info_by_index.side_effect = IOError("gone")
|
|
mock_pyaudio = MagicMock(paInt16=8)
|
|
|
|
with patch.dict("sys.modules", {"pyaudio": mock_pyaudio}):
|
|
channels, rate, stream = _open_capture_stream(mock_audio, 3)
|
|
assert (channels, rate, stream) == (1, 16000, None)
|
|
|
|
def test_test_audio_input_single_portaudio_open(self):
|
|
"""The mic test must perform exactly ONE PortAudio open (Issue #567)."""
|
|
from vocalinux.speech_recognition.recognition_manager import test_audio_input
|
|
|
|
mock_stream = MagicMock()
|
|
mock_stream.read.return_value = b"\x00\x01" * 1024
|
|
mock_audio = MagicMock()
|
|
mock_audio.open.return_value = mock_stream
|
|
mock_audio.get_device_info_by_index.return_value = {
|
|
"index": 14,
|
|
"name": "Bluetooth internal capture stream for HUAWEI FreeBuds Pro 3",
|
|
"defaultSampleRate": 16000,
|
|
"maxInputChannels": 1,
|
|
}
|
|
mock_pyaudio = MagicMock(paInt16=8)
|
|
mock_pyaudio.PyAudio.return_value = mock_audio
|
|
|
|
with patch.dict("sys.modules", {"pyaudio": mock_pyaudio}):
|
|
# Other test modules poison sys.modules["numpy"] with a MagicMock
|
|
# at import time; drop it so test_audio_input gets real numpy.
|
|
# patch.dict restores the original sys.modules afterwards.
|
|
if isinstance(sys.modules.get("numpy"), MagicMock):
|
|
del sys.modules["numpy"]
|
|
result = test_audio_input(device_index=14, duration=0.1)
|
|
|
|
assert result["success"] is True
|
|
assert mock_audio.open.call_count == 1
|
|
# The stream that was read from is the negotiated stream itself.
|
|
assert mock_stream.read.called
|
|
mock_stream.stop_stream.assert_called_once()
|
|
mock_stream.close.assert_called_once()
|
|
|
|
def test_test_audio_input_negotiation_fallback_open(self):
|
|
"""When negotiation returns no stream, mic test falls back to a plain open."""
|
|
mock_stream = MagicMock()
|
|
mock_stream.read.return_value = b"\x00\x01" * 1024
|
|
mock_audio = MagicMock()
|
|
mock_audio.open.return_value = mock_stream
|
|
mock_audio.get_device_info_by_index.return_value = {
|
|
"index": 14,
|
|
"name": "Bluetooth internal capture stream",
|
|
"defaultSampleRate": 16000,
|
|
"maxInputChannels": 1,
|
|
}
|
|
mock_pyaudio = MagicMock(paInt16=8)
|
|
mock_pyaudio.PyAudio.return_value = mock_audio
|
|
|
|
with patch.dict("sys.modules", {"pyaudio": mock_pyaudio}):
|
|
if isinstance(sys.modules.get("numpy"), MagicMock):
|
|
del sys.modules["numpy"]
|
|
with patch(
|
|
"vocalinux.speech_recognition.recognition_manager._open_capture_stream",
|
|
return_value=(1, 16000, None),
|
|
):
|
|
result = test_audio_input(device_index=14, duration=0.1)
|
|
|
|
assert result["success"] is True
|
|
mock_audio.open.assert_called_once()
|
|
|
|
def test_test_audio_input_negotiation_and_fallback_both_fail(self):
|
|
"""When negotiation and fallback open both fail, return a clear error."""
|
|
mock_audio = MagicMock()
|
|
mock_audio.open.side_effect = IOError("device busy")
|
|
mock_audio.get_device_info_by_index.return_value = {
|
|
"index": 14,
|
|
"name": "Bluetooth internal capture stream",
|
|
"defaultSampleRate": 16000,
|
|
"maxInputChannels": 1,
|
|
}
|
|
mock_pyaudio = MagicMock(paInt16=8)
|
|
mock_pyaudio.PyAudio.return_value = mock_audio
|
|
|
|
with patch.dict("sys.modules", {"pyaudio": mock_pyaudio}):
|
|
with patch(
|
|
"vocalinux.speech_recognition.recognition_manager._open_capture_stream",
|
|
return_value=(1, 16000, None),
|
|
):
|
|
result = test_audio_input(device_index=14, duration=0.1)
|
|
|
|
assert result["success"] is False
|
|
assert "Cannot open audio stream" in result["error"]
|
|
|
|
def test_get_supported_sample_rate_bluetooth_settle_on_failure(self):
|
|
"""Bluetooth devices should pause between failed sample-rate probes."""
|
|
mock_audio = MagicMock()
|
|
mock_audio.open.side_effect = IOError("busy")
|
|
mock_audio.get_device_info_by_index.return_value = {
|
|
"name": "Bluetooth internal capture stream",
|
|
"defaultSampleRate": 48000,
|
|
}
|
|
mock_pyaudio = MagicMock(paInt16=8)
|
|
|
|
with patch.dict("sys.modules", {"pyaudio": mock_pyaudio}):
|
|
with patch("vocalinux.speech_recognition.recognition_manager.time.sleep") as mock_sleep:
|
|
rate = _get_supported_sample_rate(mock_audio, 0, channels=1)
|
|
|
|
assert rate == 16000
|
|
assert mock_sleep.called
|
|
|
|
def test_open_capture_stream_logs_channel_rejection(self):
|
|
"""Invalid-channel errors during negotiation should be logged distinctly."""
|
|
mock_audio = MagicMock()
|
|
mock_audio.open.side_effect = IOError("[Errno -9998] Invalid number of channels")
|
|
mock_audio.get_device_info_by_index.return_value = {
|
|
"name": "USB Mic",
|
|
"defaultSampleRate": 48000,
|
|
"maxInputChannels": 2,
|
|
}
|
|
mock_pyaudio = MagicMock(paInt16=8)
|
|
|
|
with patch.dict("sys.modules", {"pyaudio": mock_pyaudio}):
|
|
channels, rate, stream = _open_capture_stream(mock_audio, 0)
|
|
|
|
assert (channels, rate, stream) == (1, 16000, None)
|
|
assert mock_audio.open.call_count >= 1
|
|
|
|
def test_get_supported_channels_mono_success(self):
|
|
"""Test mono channel support detection."""
|
|
mock_audio = MagicMock()
|
|
mock_stream = MagicMock()
|
|
mock_audio.open.return_value = mock_stream
|
|
mock_pyaudio = MagicMock(paInt16=8)
|
|
|
|
with patch.dict("sys.modules", {"pyaudio": mock_pyaudio}):
|
|
channels = _get_supported_channels(mock_audio, 0)
|
|
assert channels == 1
|
|
|
|
def test_get_supported_channels_stereo_fallback(self):
|
|
"""Test fallback to stereo when mono fails."""
|
|
mock_audio = MagicMock()
|
|
mock_stream = MagicMock()
|
|
|
|
# First call (mono) fails, second (stereo) succeeds
|
|
def open_side_effect(**kwargs):
|
|
if kwargs.get("channels") == 1:
|
|
raise IOError("invalid number of channels")
|
|
return mock_stream
|
|
|
|
mock_audio.open.side_effect = open_side_effect
|
|
mock_pyaudio = MagicMock(paInt16=8)
|
|
|
|
with patch.dict("sys.modules", {"pyaudio": mock_pyaudio}):
|
|
channels = _get_supported_channels(mock_audio, 0)
|
|
assert channels == 2
|
|
|
|
def test_get_supported_channels_all_fail(self):
|
|
"""Test fallback to mono when all channels fail."""
|
|
mock_audio = MagicMock()
|
|
mock_audio.open.side_effect = IOError("unsupported operation")
|
|
mock_pyaudio = MagicMock(paInt16=8)
|
|
|
|
with patch.dict("sys.modules", {"pyaudio": mock_pyaudio}):
|
|
channels = _get_supported_channels(mock_audio, None)
|
|
assert channels == 1
|
|
|
|
def test_get_supported_channels_48khz_only_device(self):
|
|
"""Test channel detection on 48kHz-only pro audio devices (Issue #340).
|
|
|
|
Professional audio interfaces (MUPRO, Vocaster, etc.) only support 48kHz
|
|
and reject 16kHz probes with misleading "Invalid number of channels" error.
|
|
The fix should use the device's defaultSampleRate for channel probing.
|
|
"""
|
|
mock_audio = MagicMock()
|
|
mock_stream = MagicMock()
|
|
|
|
def open_side_effect(**kwargs):
|
|
rate = kwargs.get("rate")
|
|
channels = kwargs.get("channels")
|
|
|
|
# Simulate 48kHz-only device: 16kHz fails, 48kHz succeeds
|
|
if rate == 16000:
|
|
raise IOError("[Errno -9998] Invalid number of channels")
|
|
elif rate == 48000 and channels == 1:
|
|
return mock_stream
|
|
else:
|
|
raise IOError("Unsupported configuration")
|
|
|
|
mock_audio.open.side_effect = open_side_effect
|
|
mock_audio.get_device_info_by_index.return_value = {"defaultSampleRate": 48000}
|
|
mock_pyaudio = MagicMock(paInt16=8)
|
|
|
|
with patch.dict("sys.modules", {"pyaudio": mock_pyaudio}):
|
|
channels = _get_supported_channels(mock_audio, 0)
|
|
assert channels == 1
|
|
# Verify that it tried using the device's default rate (48000)
|
|
assert any(
|
|
call[1].get("rate") == 48000 for call in mock_audio.open.call_args_list
|
|
), "Should probe using device's defaultSampleRate (48000)"
|
|
|
|
def test_get_supported_channels_default_rate_fails_fallback(self):
|
|
"""Test fallback when device's default rate fails during channel probing."""
|
|
mock_audio = MagicMock()
|
|
mock_stream = MagicMock()
|
|
|
|
def open_side_effect(**kwargs):
|
|
rate = kwargs.get("rate")
|
|
channels = kwargs.get("channels")
|
|
|
|
# Default rate (48000) fails, but 44100 works
|
|
if rate == 48000:
|
|
raise IOError("Device busy")
|
|
elif rate == 44100 and channels == 1:
|
|
return mock_stream
|
|
else:
|
|
raise IOError("Unsupported")
|
|
|
|
mock_audio.open.side_effect = open_side_effect
|
|
mock_audio.get_device_info_by_index.return_value = {"defaultSampleRate": 48000}
|
|
mock_pyaudio = MagicMock(paInt16=8)
|
|
|
|
with patch.dict("sys.modules", {"pyaudio": mock_pyaudio}):
|
|
channels = _get_supported_channels(mock_audio, 0)
|
|
assert channels == 1
|
|
|
|
def test_get_supported_channels_no_device_info(self):
|
|
"""Test channel probing when device info is unavailable."""
|
|
mock_audio = MagicMock()
|
|
mock_stream = MagicMock()
|
|
|
|
def open_side_effect(**kwargs):
|
|
# Works with 44100Hz mono
|
|
if kwargs.get("rate") == 44100 and kwargs.get("channels") == 1:
|
|
return mock_stream
|
|
raise IOError("Unsupported")
|
|
|
|
mock_audio.open.side_effect = open_side_effect
|
|
mock_audio.get_device_info_by_index.side_effect = IOError("Device not found")
|
|
mock_pyaudio = MagicMock(paInt16=8)
|
|
|
|
with patch.dict("sys.modules", {"pyaudio": mock_pyaudio}):
|
|
channels = _get_supported_channels(mock_audio, 0)
|
|
assert channels == 1 # Should fallback through common rates and find 44100
|
|
|
|
def test_get_supported_sample_rate_default_rate_works(self):
|
|
"""Test using device's default sample rate."""
|
|
mock_audio = MagicMock()
|
|
mock_stream = MagicMock()
|
|
mock_audio.open.return_value = mock_stream
|
|
mock_audio.get_device_info_by_index.return_value = {"defaultSampleRate": 48000}
|
|
mock_pyaudio = MagicMock(paInt16=8)
|
|
|
|
with patch.dict("sys.modules", {"pyaudio": mock_pyaudio}):
|
|
rate = _get_supported_sample_rate(mock_audio, 0, 1)
|
|
assert rate == 48000
|
|
|
|
def test_get_supported_sample_rate_default_rate_fails(self):
|
|
"""Test fallback from device default rate."""
|
|
mock_audio = MagicMock()
|
|
mock_stream = MagicMock()
|
|
|
|
def open_side_effect(**kwargs):
|
|
if kwargs.get("rate") == 48000:
|
|
raise IOError("unsupported rate")
|
|
return mock_stream
|
|
|
|
mock_audio.open.side_effect = open_side_effect
|
|
mock_audio.get_device_info_by_index.return_value = {"defaultSampleRate": 48000}
|
|
mock_pyaudio = MagicMock(paInt16=8)
|
|
|
|
with patch.dict("sys.modules", {"pyaudio": mock_pyaudio}):
|
|
rate = _get_supported_sample_rate(mock_audio, 0, 1)
|
|
assert rate == 44100 # First fallback rate
|
|
|
|
def test_get_supported_sample_rate_all_fail(self):
|
|
"""Test fallback to default rate when all fail."""
|
|
mock_audio = MagicMock()
|
|
mock_audio.open.side_effect = IOError("all fail")
|
|
mock_audio.get_device_info_by_index.side_effect = IOError("no device info")
|
|
mock_pyaudio = MagicMock(paInt16=8)
|
|
|
|
with patch.dict("sys.modules", {"pyaudio": mock_pyaudio}):
|
|
rate = _get_supported_sample_rate(mock_audio, None, 1)
|
|
assert rate == 16000 # Default fallback
|
|
|
|
|
|
class TestRecordAudioNegotiationFallback(unittest.TestCase):
|
|
"""Cover _record_audio fallback when stream negotiation returns None."""
|
|
|
|
def test_record_audio_falls_back_when_negotiation_returns_none(self):
|
|
"""Negotiation failure must fall back to a plain PortAudio open."""
|
|
manager = _make_manager(engine="whisper_cpp")
|
|
manager.should_record = True
|
|
manager.state = RecognitionState.LISTENING
|
|
manager.silence_timeout = 0.05
|
|
manager._silero_vad = None
|
|
|
|
mock_stream = MagicMock()
|
|
|
|
def _read_once(*_args, **_kwargs):
|
|
manager.should_record = False
|
|
return b"\x00" * 2048
|
|
|
|
mock_stream.read.side_effect = _read_once
|
|
|
|
mock_audio = MagicMock()
|
|
mock_audio.get_device_count.return_value = 1
|
|
mock_audio.get_device_info_by_index.return_value = {
|
|
"index": 0,
|
|
"name": "test mic",
|
|
"maxInputChannels": 1,
|
|
}
|
|
mock_audio.get_default_input_device_info.return_value = {"index": 0}
|
|
mock_audio.open.return_value = mock_stream
|
|
|
|
mock_pyaudio = MagicMock(paInt16=8)
|
|
mock_pyaudio.PyAudio.return_value = mock_audio
|
|
|
|
with (
|
|
patch.dict("sys.modules", {"pyaudio": mock_pyaudio}),
|
|
patch(
|
|
"vocalinux.speech_recognition.recognition_manager._resolve_device_by_name",
|
|
return_value=0,
|
|
),
|
|
patch(
|
|
"vocalinux.speech_recognition.recognition_manager._open_capture_stream",
|
|
return_value=(1, 16000, None),
|
|
),
|
|
patch(
|
|
"vocalinux.speech_recognition.recognition_manager.play_error_sound",
|
|
),
|
|
):
|
|
if isinstance(sys.modules.get("numpy"), MagicMock):
|
|
del sys.modules["numpy"]
|
|
manager._record_audio()
|
|
|
|
mock_audio.open.assert_called_once()
|
|
|
|
|
|
class TestFilterNonSpeech(unittest.TestCase):
|
|
"""Test the _filter_non_speech function."""
|
|
|
|
def test_filter_non_speech_empty(self):
|
|
"""Test filtering empty string."""
|
|
result = _filter_non_speech("")
|
|
assert result == ""
|
|
|
|
def test_filter_non_speech_normal(self):
|
|
"""Test filtering normal text."""
|
|
result = _filter_non_speech("hello world")
|
|
assert result == "hello world"
|
|
|
|
def test_filter_non_speech_with_special_tokens(self):
|
|
"""Test filtering text with special tokens."""
|
|
result = _filter_non_speech("[BLANK_AUDIO]")
|
|
assert result == ""
|
|
|
|
def test_filter_non_speech_mixed(self):
|
|
"""Test filtering mixed content."""
|
|
result = _filter_non_speech("hello [BLANK_AUDIO] world")
|
|
assert "hello" in result or result == ""
|
|
|
|
|
|
class TestVoskInitialization(unittest.TestCase):
|
|
"""Test VOSK engine initialization."""
|
|
|
|
def test_init_vosk_model_not_found_deferred(self):
|
|
"""Test VOSK initialization with deferred download."""
|
|
manager = _make_manager(engine="vosk")
|
|
manager.language = "en-us"
|
|
manager._defer_download = True
|
|
manager.model_size = "small"
|
|
|
|
mock_vosk = MagicMock()
|
|
with patch(
|
|
"vocalinux.speech_recognition.recognition_manager.VOSK_MODEL_INFO",
|
|
{
|
|
"small": {"languages": {"en-us": "model-name"}},
|
|
"medium": {"languages": {"en-us": "model-name"}},
|
|
"large": {"languages": {"en-us": "model-name"}},
|
|
},
|
|
):
|
|
with patch("os.path.exists", return_value=False):
|
|
with patch.object(manager, "_get_vosk_model_path", return_value="/fake/path"):
|
|
with patch.dict("sys.modules", {"vosk": mock_vosk}):
|
|
manager._init_vosk()
|
|
assert manager._model_initialized is False
|
|
|
|
def test_init_vosk_import_error(self):
|
|
"""Test VOSK initialization when import fails."""
|
|
manager = _make_manager(engine="vosk")
|
|
manager.language = "en-us"
|
|
manager.model_size = "small"
|
|
|
|
with patch(
|
|
"vocalinux.speech_recognition.recognition_manager.VOSK_MODEL_INFO",
|
|
{
|
|
"small": {"languages": {"en-us": "model-name"}},
|
|
"medium": {"languages": {"en-us": "model-name"}},
|
|
"large": {"languages": {"en-us": "model-name"}},
|
|
},
|
|
):
|
|
with patch.object(manager, "_get_vosk_model_path", return_value="/fake/path"):
|
|
with patch("os.path.exists", return_value=False):
|
|
with patch.dict("sys.modules", {"vosk": None}):
|
|
with pytest.raises(ImportError):
|
|
manager._init_vosk()
|
|
assert manager.state == RecognitionState.ERROR
|
|
|
|
def test_init_vosk_preinstalled_model(self):
|
|
"""Test VOSK initialization with pre-installed model."""
|
|
manager = _make_manager(engine="vosk")
|
|
manager.language = "en-us"
|
|
manager.model_size = "small"
|
|
|
|
mock_vosk = MagicMock()
|
|
mock_model = MagicMock()
|
|
mock_recognizer = MagicMock()
|
|
mock_vosk.Model = MagicMock(return_value=mock_model)
|
|
mock_vosk.KaldiRecognizer = MagicMock(return_value=mock_recognizer)
|
|
|
|
with patch(
|
|
"vocalinux.speech_recognition.recognition_manager.VOSK_MODEL_INFO",
|
|
{
|
|
"small": {"languages": {"en-us": "model-name"}},
|
|
"medium": {"languages": {"en-us": "model-name"}},
|
|
"large": {"languages": {"en-us": "model-name"}},
|
|
},
|
|
):
|
|
with patch(
|
|
"vocalinux.speech_recognition.recognition_manager.SYSTEM_MODELS_DIRS",
|
|
["/usr/share/vocalinux"],
|
|
):
|
|
with patch.object(
|
|
manager, "_get_vosk_model_path", return_value="/usr/share/vocalinux/model"
|
|
):
|
|
with patch("os.path.exists", return_value=True):
|
|
with patch.dict("sys.modules", {"vosk": mock_vosk}):
|
|
manager._init_vosk()
|
|
assert manager._model_initialized is True
|
|
assert manager.model == mock_model
|
|
|
|
|
|
class TestWhisperInitialization(unittest.TestCase):
|
|
"""Test Whisper engine initialization."""
|
|
|
|
def test_init_whisper_invalid_model_size(self):
|
|
"""Test Whisper with invalid model size."""
|
|
manager = _make_manager(engine="whisper")
|
|
manager.model_size = "invalid"
|
|
manager._defer_download = True
|
|
|
|
mock_whisper = MagicMock()
|
|
mock_torch = MagicMock()
|
|
|
|
with patch.dict("sys.modules", {"whisper": mock_whisper, "torch": mock_torch}):
|
|
with patch("os.path.exists", return_value=False):
|
|
manager._init_whisper()
|
|
assert manager.model_size == "base" # Should be corrected
|
|
assert manager._model_initialized is False
|
|
|
|
def test_init_whisper_model_exists(self):
|
|
"""Test Whisper when model already exists."""
|
|
manager = _make_manager(engine="whisper")
|
|
manager.model_size = "tiny"
|
|
|
|
mock_whisper = MagicMock()
|
|
mock_torch = MagicMock()
|
|
mock_model = MagicMock()
|
|
mock_whisper.load_model.return_value = mock_model
|
|
mock_torch.cuda.is_available.return_value = False
|
|
|
|
with patch.dict("sys.modules", {"whisper": mock_whisper, "torch": mock_torch}):
|
|
with patch("os.path.exists", return_value=True):
|
|
manager._init_whisper()
|
|
assert manager._model_initialized is True
|
|
assert manager.model == mock_model
|
|
|
|
def test_init_whisper_import_error(self):
|
|
"""Test Whisper initialization when import fails."""
|
|
manager = _make_manager(engine="whisper")
|
|
|
|
with patch.dict("sys.modules", {"whisper": None, "torch": None}):
|
|
with pytest.raises(ImportError):
|
|
manager._init_whisper()
|
|
assert manager.state == RecognitionState.ERROR
|
|
|
|
def test_init_whisper_runtime_error(self):
|
|
"""Test Whisper initialization with runtime error."""
|
|
manager = _make_manager(engine="whisper")
|
|
manager._defer_download = False
|
|
|
|
mock_whisper = MagicMock()
|
|
mock_torch = MagicMock()
|
|
|
|
with patch.dict("sys.modules", {"whisper": mock_whisper, "torch": mock_torch}):
|
|
with patch("os.path.exists", return_value=False):
|
|
with patch.object(
|
|
manager, "_download_whisper_model", side_effect=RuntimeError("Download failed")
|
|
):
|
|
with pytest.raises(RuntimeError):
|
|
manager._init_whisper()
|
|
|
|
|
|
class TestWhispercppInitialization(unittest.TestCase):
|
|
"""Test whisper.cpp engine initialization."""
|
|
|
|
def test_init_whispercpp_invalid_model_size(self):
|
|
"""Test whisper.cpp with invalid model size."""
|
|
manager = _make_manager(engine="whisper_cpp")
|
|
manager.model_size = "invalid"
|
|
manager._defer_download = True
|
|
|
|
mock_pywhispercpp = MagicMock()
|
|
|
|
with patch(
|
|
"vocalinux.speech_recognition.recognition_manager.WHISPERCPP_MODEL_INFO",
|
|
{"tiny": {}, "base": {}},
|
|
):
|
|
with patch(
|
|
"vocalinux.speech_recognition.recognition_manager.get_model_path",
|
|
return_value="/fake/path",
|
|
):
|
|
with patch("os.path.exists", return_value=False):
|
|
with patch.dict(
|
|
"sys.modules",
|
|
{
|
|
"pywhispercpp": mock_pywhispercpp,
|
|
"pywhispercpp.model": mock_pywhispercpp,
|
|
},
|
|
):
|
|
manager._init_whispercpp()
|
|
assert manager.model_size == "tiny" # Should be corrected
|
|
assert manager._model_initialized is False
|
|
|
|
def test_init_whispercpp_gpu_fallback(self):
|
|
"""Test whisper.cpp GPU fallback to CPU."""
|
|
manager = _make_manager(engine="whisper_cpp")
|
|
manager.model_size = "tiny"
|
|
|
|
mock_pywhispercpp = MagicMock()
|
|
mock_model_success = MagicMock()
|
|
|
|
# Setup module hierarchy
|
|
model_class = MagicMock(
|
|
side_effect=[RuntimeError("16-bit storage not supported"), mock_model_success]
|
|
)
|
|
mock_pywhispercpp.Model = model_class
|
|
mock_pywhispercpp.model.Model = model_class
|
|
|
|
# Mock the imported functions from whispercpp_model_info
|
|
mock_psutil = MagicMock()
|
|
mock_psutil.virtual_memory.return_value.total = 8 * 1024 * 1024 * 1024
|
|
|
|
with patch(
|
|
"vocalinux.speech_recognition.recognition_manager.WHISPERCPP_MODEL_INFO", {"tiny": {}}
|
|
):
|
|
with patch(
|
|
"vocalinux.speech_recognition.recognition_manager.get_model_path",
|
|
return_value="/fake/model.bin",
|
|
):
|
|
with patch("os.path.getsize", return_value=100000000): # Mock file size
|
|
with patch("os.path.exists", return_value=True):
|
|
with patch.dict(
|
|
"sys.modules",
|
|
{
|
|
"pywhispercpp": mock_pywhispercpp,
|
|
"pywhispercpp.model": mock_pywhispercpp,
|
|
"psutil": mock_psutil,
|
|
},
|
|
):
|
|
# Patch the imports that happen inside _init_whispercpp
|
|
import vocalinux.utils.whispercpp_model_info as whispercpp_info
|
|
|
|
with patch.object(
|
|
whispercpp_info,
|
|
"detect_compute_backend",
|
|
return_value=(MagicMock(), "test"),
|
|
):
|
|
with patch.object(
|
|
whispercpp_info,
|
|
"get_backend_display_name",
|
|
return_value="Vulkan",
|
|
):
|
|
with patch(
|
|
"vocalinux.speech_recognition.recognition_manager._show_notification"
|
|
):
|
|
with patch.dict("os.environ", {}, clear=True):
|
|
manager._init_whispercpp()
|
|
assert manager._model_initialized is True
|
|
|
|
def test_init_whispercpp_import_error(self):
|
|
"""Test whisper.cpp when import fails."""
|
|
manager = _make_manager(engine="whisper_cpp")
|
|
|
|
with patch.dict("sys.modules", {"pywhispercpp": None, "pywhispercpp.model": None}):
|
|
with pytest.raises(ImportError):
|
|
manager._init_whispercpp()
|
|
assert manager.state == RecognitionState.ERROR
|
|
|
|
def test_init_whispercpp_model_file_not_found(self):
|
|
"""Test whisper.cpp when model file is missing."""
|
|
manager = _make_manager(engine="whisper_cpp")
|
|
manager._defer_download = False
|
|
manager.model_size = "tiny"
|
|
|
|
mock_pywhispercpp = MagicMock()
|
|
|
|
with patch(
|
|
"vocalinux.speech_recognition.recognition_manager.WHISPERCPP_MODEL_INFO",
|
|
{"tiny": {"url": "http://example.com/model"}},
|
|
):
|
|
with patch(
|
|
"vocalinux.speech_recognition.recognition_manager.get_model_path",
|
|
return_value="/fake/model.bin",
|
|
):
|
|
with patch("os.path.exists", return_value=False):
|
|
with patch.dict(
|
|
"sys.modules",
|
|
{
|
|
"pywhispercpp": mock_pywhispercpp,
|
|
"pywhispercpp.model": mock_pywhispercpp,
|
|
},
|
|
):
|
|
with patch.object(
|
|
manager,
|
|
"_download_whispercpp_model",
|
|
side_effect=Exception("Download failed"),
|
|
):
|
|
with pytest.raises(Exception):
|
|
manager._init_whispercpp()
|
|
|
|
|
|
class TestWhispercppGpuDeviceSelection(unittest.TestCase):
|
|
"""Test whisper.cpp GPU device selection logic."""
|
|
|
|
class ContextParamsModel:
|
|
calls = []
|
|
|
|
def __init__(self, model_path, context_params=None, **kwargs):
|
|
call_kwargs = dict(kwargs)
|
|
if context_params is not None:
|
|
call_kwargs["context_params"] = context_params
|
|
self.__class__.calls.append((model_path, call_kwargs))
|
|
|
|
class OldSignatureModel:
|
|
calls = []
|
|
|
|
def __init__(self, model_path, **kwargs):
|
|
self.__class__.calls.append((model_path, dict(kwargs)))
|
|
|
|
class AttributeErrorContextParamsModel:
|
|
calls = []
|
|
|
|
def __init__(self, model_path, context_params=None, **kwargs):
|
|
call_kwargs = dict(kwargs)
|
|
if context_params is not None:
|
|
call_kwargs["context_params"] = context_params
|
|
self.__class__.calls.append((model_path, call_kwargs))
|
|
if context_params is not None:
|
|
raise AttributeError(
|
|
"'whisper_full_params' object has no attribute 'context_params'"
|
|
)
|
|
|
|
def test_gpu_device_auto_select_discrete(self):
|
|
"""Test auto-selection of discrete GPU when configured as None."""
|
|
manager = _make_manager(engine="whisper_cpp", whispercpp_gpu_device=None)
|
|
manager.model_size = "tiny"
|
|
|
|
mock_pywhispercpp = MagicMock()
|
|
self.ContextParamsModel.calls = []
|
|
mock_pywhispercpp.Model = self.ContextParamsModel
|
|
mock_pywhispercpp.model.Model = self.ContextParamsModel
|
|
|
|
mock_psutil = MagicMock()
|
|
mock_psutil.virtual_memory.return_value.total = 8 * 1024 * 1024 * 1024
|
|
|
|
vulkan_devices = [
|
|
{"index": 0, "name": "Intel UHD 630", "device_type": "integrated"},
|
|
{"index": 1, "name": "NVIDIA RTX 2060", "device_type": "discrete"},
|
|
]
|
|
|
|
with patch(
|
|
"vocalinux.speech_recognition.recognition_manager.WHISPERCPP_MODEL_INFO", {"tiny": {}}
|
|
):
|
|
with patch(
|
|
"vocalinux.speech_recognition.recognition_manager.get_model_path",
|
|
return_value="/fake/model.bin",
|
|
):
|
|
with patch("os.path.getsize", return_value=100000000):
|
|
with patch("os.path.exists", return_value=True):
|
|
with patch.dict(
|
|
"sys.modules",
|
|
{
|
|
"pywhispercpp": mock_pywhispercpp,
|
|
"pywhispercpp.model": mock_pywhispercpp,
|
|
"psutil": mock_psutil,
|
|
},
|
|
):
|
|
import vocalinux.utils.whispercpp_model_info as whispercpp_info
|
|
|
|
with patch.object(
|
|
whispercpp_info,
|
|
"detect_compute_backend",
|
|
return_value=(whispercpp_info.ComputeBackend.VULKAN, "Vulkan GPU"),
|
|
):
|
|
with patch.object(
|
|
whispercpp_info,
|
|
"get_backend_display_name",
|
|
return_value="Vulkan",
|
|
):
|
|
with patch.object(
|
|
whispercpp_info,
|
|
"detect_vulkan_devices",
|
|
return_value=vulkan_devices,
|
|
):
|
|
with patch.object(
|
|
whispercpp_info,
|
|
"_prefer_discrete_vulkan_device",
|
|
return_value=1,
|
|
):
|
|
manager._init_whispercpp()
|
|
assert self.ContextParamsModel.calls[-1][1].get(
|
|
"context_params"
|
|
) == {"gpu_device": 1}
|
|
|
|
def test_gpu_device_explicit_index(self):
|
|
"""Test explicit GPU device index selection."""
|
|
manager = _make_manager(engine="whisper_cpp", whispercpp_gpu_device=0)
|
|
manager.model_size = "tiny"
|
|
|
|
mock_pywhispercpp = MagicMock()
|
|
self.ContextParamsModel.calls = []
|
|
mock_pywhispercpp.Model = self.ContextParamsModel
|
|
mock_pywhispercpp.model.Model = self.ContextParamsModel
|
|
|
|
mock_psutil = MagicMock()
|
|
mock_psutil.virtual_memory.return_value.total = 8 * 1024 * 1024 * 1024
|
|
|
|
vulkan_devices = [
|
|
{"index": 0, "name": "Intel UHD 630", "device_type": "integrated"},
|
|
{"index": 1, "name": "NVIDIA RTX 2060", "device_type": "discrete"},
|
|
]
|
|
|
|
with patch(
|
|
"vocalinux.speech_recognition.recognition_manager.WHISPERCPP_MODEL_INFO", {"tiny": {}}
|
|
):
|
|
with patch(
|
|
"vocalinux.speech_recognition.recognition_manager.get_model_path",
|
|
return_value="/fake/model.bin",
|
|
):
|
|
with patch("os.path.getsize", return_value=100000000):
|
|
with patch("os.path.exists", return_value=True):
|
|
with patch.dict(
|
|
"sys.modules",
|
|
{
|
|
"pywhispercpp": mock_pywhispercpp,
|
|
"pywhispercpp.model": mock_pywhispercpp,
|
|
"psutil": mock_psutil,
|
|
},
|
|
):
|
|
import vocalinux.utils.whispercpp_model_info as whispercpp_info
|
|
|
|
with patch.object(
|
|
whispercpp_info,
|
|
"detect_compute_backend",
|
|
return_value=(whispercpp_info.ComputeBackend.VULKAN, "Vulkan GPU"),
|
|
):
|
|
with patch.object(
|
|
whispercpp_info,
|
|
"get_backend_display_name",
|
|
return_value="Vulkan",
|
|
):
|
|
with patch.object(
|
|
whispercpp_info,
|
|
"detect_vulkan_devices",
|
|
return_value=vulkan_devices,
|
|
):
|
|
manager._init_whispercpp()
|
|
assert self.ContextParamsModel.calls[-1][1].get(
|
|
"context_params"
|
|
) == {"gpu_device": 0}
|
|
|
|
def test_gpu_device_skips_context_params_for_old_pywhispercpp_signature(self):
|
|
"""Old pywhispercpp signatures must not receive context_params."""
|
|
manager = _make_manager(engine="whisper_cpp", whispercpp_gpu_device=1)
|
|
manager.model_size = "tiny"
|
|
|
|
mock_pywhispercpp = MagicMock()
|
|
self.OldSignatureModel.calls = []
|
|
mock_pywhispercpp.Model = self.OldSignatureModel
|
|
|
|
with patch.dict(
|
|
"sys.modules",
|
|
{
|
|
"pywhispercpp": mock_pywhispercpp,
|
|
"pywhispercpp.model": mock_pywhispercpp,
|
|
},
|
|
):
|
|
result = manager._load_model_with_compatible_params("/fake/model.bin", {}, gpu_device=1)
|
|
assert isinstance(result, self.OldSignatureModel)
|
|
assert len(self.OldSignatureModel.calls) == 1
|
|
assert "context_params" not in self.OldSignatureModel.calls[0][1]
|
|
|
|
def test_gpu_device_skips_context_params_when_signature_inspection_fails(self):
|
|
"""If Model.__init__ cannot be inspected, skip context_params safely."""
|
|
manager = _make_manager(engine="whisper_cpp", whispercpp_gpu_device=1)
|
|
manager.model_size = "tiny"
|
|
|
|
mock_pywhispercpp = MagicMock()
|
|
self.OldSignatureModel.calls = []
|
|
mock_pywhispercpp.Model = self.OldSignatureModel
|
|
|
|
with patch.dict(
|
|
"sys.modules",
|
|
{
|
|
"pywhispercpp": mock_pywhispercpp,
|
|
"pywhispercpp.model": mock_pywhispercpp,
|
|
},
|
|
):
|
|
with patch(
|
|
"vocalinux.speech_recognition.recognition_manager.inspect.signature",
|
|
side_effect=ValueError("no signature"),
|
|
):
|
|
result = manager._load_model_with_compatible_params(
|
|
"/fake/model.bin", {}, gpu_device=1
|
|
)
|
|
|
|
assert isinstance(result, self.OldSignatureModel)
|
|
assert len(self.OldSignatureModel.calls) == 1
|
|
assert "context_params" not in self.OldSignatureModel.calls[0][1]
|
|
|
|
def test_gpu_device_context_params_fallback_on_attribute_error(self):
|
|
"""AttributeError from pywhispercpp context_params is retried without it."""
|
|
manager = _make_manager(engine="whisper_cpp", whispercpp_gpu_device=1)
|
|
manager.model_size = "tiny"
|
|
|
|
mock_pywhispercpp = MagicMock()
|
|
self.AttributeErrorContextParamsModel.calls = []
|
|
mock_pywhispercpp.Model = self.AttributeErrorContextParamsModel
|
|
|
|
with patch.dict(
|
|
"sys.modules",
|
|
{
|
|
"pywhispercpp": mock_pywhispercpp,
|
|
"pywhispercpp.model": mock_pywhispercpp,
|
|
},
|
|
):
|
|
result = manager._load_model_with_compatible_params("/fake/model.bin", {}, gpu_device=1)
|
|
assert isinstance(result, self.AttributeErrorContextParamsModel)
|
|
assert len(self.AttributeErrorContextParamsModel.calls) == 2
|
|
assert self.AttributeErrorContextParamsModel.calls[0][1].get("context_params") == {
|
|
"gpu_device": 1
|
|
}
|
|
assert "context_params" not in self.AttributeErrorContextParamsModel.calls[1][1]
|
|
|
|
def test_gpu_device_not_set_for_cpu_backend(self):
|
|
"""Test that GPU device selection is skipped on CPU backend."""
|
|
manager = _make_manager(engine="whisper_cpp", whispercpp_gpu_device=None)
|
|
manager.model_size = "tiny"
|
|
|
|
mock_pywhispercpp = MagicMock()
|
|
mock_model = MagicMock()
|
|
mock_pywhispercpp.Model = MagicMock(return_value=mock_model)
|
|
mock_pywhispercpp.model.Model = MagicMock(return_value=mock_model)
|
|
|
|
mock_psutil = MagicMock()
|
|
mock_psutil.virtual_memory.return_value.total = 8 * 1024 * 1024 * 1024
|
|
|
|
with patch(
|
|
"vocalinux.speech_recognition.recognition_manager.WHISPERCPP_MODEL_INFO", {"tiny": {}}
|
|
):
|
|
with patch(
|
|
"vocalinux.speech_recognition.recognition_manager.get_model_path",
|
|
return_value="/fake/model.bin",
|
|
):
|
|
with patch("os.path.getsize", return_value=100000000):
|
|
with patch("os.path.exists", return_value=True):
|
|
with patch.dict(
|
|
"sys.modules",
|
|
{
|
|
"pywhispercpp": mock_pywhispercpp,
|
|
"pywhispercpp.model": mock_pywhispercpp,
|
|
"psutil": mock_psutil,
|
|
},
|
|
):
|
|
import vocalinux.utils.whispercpp_model_info as whispercpp_info
|
|
|
|
with patch.object(
|
|
whispercpp_info,
|
|
"detect_compute_backend",
|
|
return_value=(whispercpp_info.ComputeBackend.CPU, "CPU"),
|
|
):
|
|
with patch.object(
|
|
whispercpp_info,
|
|
"get_backend_display_name",
|
|
return_value="CPU",
|
|
):
|
|
manager._init_whispercpp()
|
|
assert (
|
|
"context_params"
|
|
not in mock_pywhispercpp.Model.call_args.kwargs
|
|
)
|
|
|
|
|
|
class TestTranscription(unittest.TestCase):
|
|
"""Test transcription methods."""
|
|
|
|
def test_transcribe_with_whisper_empty_buffer(self):
|
|
"""Test Whisper transcription with empty buffer."""
|
|
manager = _make_manager(engine="whisper")
|
|
manager.model = MagicMock()
|
|
|
|
result = manager._transcribe_with_whisper([])
|
|
assert result == ""
|
|
|
|
def test_transcribe_with_whisper_none_model(self):
|
|
"""Test Whisper transcription when model is None."""
|
|
manager = _make_manager(engine="whisper")
|
|
manager.model = None
|
|
|
|
audio_buffer = [b"\x00\x00\x00\x00"]
|
|
result = manager._transcribe_with_whisper(audio_buffer)
|
|
assert result == ""
|
|
|
|
def test_transcribe_with_whisper_success(self):
|
|
"""Test successful Whisper transcription."""
|
|
manager = _make_manager(engine="whisper")
|
|
|
|
# Create mock that behaves like a torch device
|
|
mock_device = MagicMock()
|
|
mock_device.__ne__ = MagicMock(return_value=True) # device != torch.device("cpu")
|
|
|
|
mock_model = MagicMock()
|
|
mock_model.device = mock_device
|
|
mock_model.transcribe.return_value = {"text": "hello world"}
|
|
manager.model = mock_model
|
|
manager.language = "en-us"
|
|
|
|
audio_buffer = [b"\x00\x00" for _ in range(16000)] # 1 second of audio
|
|
|
|
mock_np = MagicMock()
|
|
mock_audio_data = MagicMock()
|
|
mock_audio_float = MagicMock()
|
|
|
|
mock_np.frombuffer.return_value = mock_audio_data
|
|
mock_audio_data.astype.return_value = mock_audio_float
|
|
|
|
mock_torch = MagicMock()
|
|
|
|
with patch.dict("sys.modules", {"numpy": mock_np, "torch": mock_torch}):
|
|
result = manager._transcribe_with_whisper(audio_buffer)
|
|
assert result == "hello world"
|
|
|
|
def test_transcribe_with_whispercpp_empty_buffer(self):
|
|
"""Test whisper.cpp transcription with empty buffer."""
|
|
manager = _make_manager(engine="whisper_cpp")
|
|
manager.model = MagicMock()
|
|
|
|
result = manager._transcribe_with_whispercpp([])
|
|
assert result == ""
|
|
|
|
def test_transcribe_with_whispercpp_none_model(self):
|
|
"""Test whisper.cpp transcription when model is None."""
|
|
manager = _make_manager(engine="whisper_cpp")
|
|
manager.model = None
|
|
|
|
audio_buffer = [b"\x00\x00\x00\x00"]
|
|
result = manager._transcribe_with_whispercpp(audio_buffer)
|
|
assert result == ""
|
|
|
|
def test_transcribe_with_whispercpp_success(self):
|
|
"""Test successful whisper.cpp transcription."""
|
|
manager = _make_manager(engine="whisper_cpp")
|
|
|
|
# Mock segment with text attribute
|
|
mock_segment = MagicMock()
|
|
mock_segment.text = "test result"
|
|
|
|
mock_model = MagicMock()
|
|
mock_model.transcribe.return_value = [mock_segment]
|
|
manager.model = mock_model
|
|
manager.language = "en-us"
|
|
|
|
audio_buffer = [b"\x00\x00\x00\x00" * 16000]
|
|
|
|
mock_np = MagicMock()
|
|
mock_np.frombuffer.return_value = MagicMock()
|
|
mock_np.frombuffer.return_value.astype.return_value = MagicMock()
|
|
|
|
with patch.dict("sys.modules", {"numpy": mock_np, "np": mock_np}):
|
|
result = manager._transcribe_with_whispercpp(audio_buffer)
|
|
assert result == "test result"
|
|
|
|
|
|
class TestStartStopRecognition(unittest.TestCase):
|
|
"""Test start_recognition and stop_recognition flows."""
|
|
|
|
def test_start_recognition_invalid_state(self):
|
|
"""Test starting recognition when not IDLE."""
|
|
manager = _make_manager()
|
|
manager.state = RecognitionState.LISTENING
|
|
|
|
with patch("vocalinux.speech_recognition.recognition_manager.play_error_sound"):
|
|
manager.start_recognition()
|
|
# Should return early without starting threads
|
|
|
|
def test_start_recognition_model_not_ready(self):
|
|
"""Test starting recognition when model not ready."""
|
|
manager = _make_manager()
|
|
manager.state = RecognitionState.IDLE
|
|
manager._model_initialized = False
|
|
|
|
with patch("vocalinux.speech_recognition.recognition_manager.play_error_sound"):
|
|
with patch("vocalinux.speech_recognition.recognition_manager._show_notification"):
|
|
manager.start_recognition()
|
|
assert manager.state == RecognitionState.IDLE # Should not change
|
|
|
|
def test_start_recognition_success(self):
|
|
"""Test successful recognition start."""
|
|
manager = _make_manager()
|
|
manager.state = RecognitionState.IDLE
|
|
manager._model_initialized = True
|
|
manager.model = MagicMock()
|
|
|
|
with patch("vocalinux.speech_recognition.recognition_manager.play_start_sound"):
|
|
with patch.object(manager, "_record_audio"):
|
|
with patch.object(manager, "_perform_recognition"):
|
|
manager.start_recognition()
|
|
assert manager.state == RecognitionState.LISTENING
|
|
assert manager.should_record is True
|
|
|
|
def test_stop_recognition_when_idle(self):
|
|
"""Test stopping recognition when already idle."""
|
|
manager = _make_manager()
|
|
manager.state = RecognitionState.IDLE
|
|
|
|
manager.stop_recognition()
|
|
# Should return early
|
|
assert manager.state == RecognitionState.IDLE
|
|
|
|
def test_stop_recognition_with_threads(self):
|
|
"""Test stopping recognition with active threads."""
|
|
manager = _make_manager()
|
|
manager.state = RecognitionState.LISTENING
|
|
manager.should_record = True
|
|
manager.audio_buffer = [b"\x00\x00" for _ in range(20)]
|
|
manager._recording_segment_has_speech = True
|
|
|
|
# Create dummy threads
|
|
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("vocalinux.speech_recognition.recognition_manager.play_stop_sound"):
|
|
with patch.object(manager, "_signal_recognition_stop"):
|
|
with patch.object(manager, "_enqueue_audio_segment") as enqueue_mock:
|
|
manager.stop_recognition()
|
|
assert manager.should_record is False
|
|
assert len(enqueue_mock.call_args.args[0]) == 17
|
|
assert manager._recording_segment_has_speech is False
|
|
|
|
def test_stop_recognition_drops_final_buffer_without_speech(self):
|
|
"""Final silence-only buffers should not be queued for transcription."""
|
|
manager = _make_manager()
|
|
manager.state = RecognitionState.LISTENING
|
|
manager.should_record = True
|
|
manager.audio_buffer = [b"\x00\x00" for _ in range(20)]
|
|
manager._recording_segment_has_speech = False
|
|
|
|
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("vocalinux.speech_recognition.recognition_manager.play_stop_sound"):
|
|
with patch.object(manager, "_signal_recognition_stop"):
|
|
with patch.object(manager, "_enqueue_audio_segment") as enqueue_mock:
|
|
manager.stop_recognition()
|
|
enqueue_mock.assert_not_called()
|
|
assert manager.audio_buffer == []
|
|
|
|
|
|
class TestDownloads(unittest.TestCase):
|
|
"""Test model download functions."""
|
|
|
|
def test_download_whispercpp_model_success(self):
|
|
"""Test successful whisper.cpp model download - skipped (requires requests)."""
|
|
# This test requires requests library which may not be available
|
|
# The actual download logic is tested through integration tests
|
|
pass
|
|
|
|
def test_download_whispercpp_model_cancelled(self):
|
|
"""Test cancelled whisper.cpp model download - skipped (requires requests)."""
|
|
# This test requires requests library which may not be available
|
|
# The actual download logic is tested through integration tests
|
|
pass
|
|
|
|
def test_download_vosk_model_success(self):
|
|
"""Test successful VOSK model download - skipped (requires requests)."""
|
|
# This test requires requests library which may not be available
|
|
# The actual download logic is tested through integration tests
|
|
pass
|
|
|
|
def test_download_vosk_model_bad_zip(self):
|
|
"""Test VOSK download with corrupted zip file - skipped (requires requests)."""
|
|
# This test requires requests library which may not be available
|
|
# The actual download logic is tested through integration tests
|
|
pass
|
|
|
|
|
|
class TestReconfiguration(unittest.TestCase):
|
|
"""Test reconfiguration and model switching."""
|
|
|
|
def test_reconfigure_engine(self):
|
|
"""Test reconfiguring to a different engine."""
|
|
manager = _make_manager(engine="vosk")
|
|
manager.engine = "vosk"
|
|
|
|
with patch.object(manager, "_init_whisper"):
|
|
with patch.object(manager, "stop_recognition"):
|
|
manager.reconfigure(engine="whisper")
|
|
assert manager.engine == "whisper"
|
|
|
|
|
|
class TestEnqueueAudioSegment(unittest.TestCase):
|
|
"""Direct tests for _enqueue_audio_segment (not mocked)."""
|
|
|
|
def test_enqueue_empty_buffer_returns_early(self):
|
|
"""Empty buffer should return without enqueuing anything."""
|
|
manager = _make_manager()
|
|
manager._segment_queue = queue.Queue()
|
|
manager._enqueue_audio_segment([])
|
|
assert manager._segment_queue.empty()
|
|
|
|
def test_enqueue_puts_segment(self):
|
|
"""Non-empty buffer should be copied and enqueued."""
|
|
manager = _make_manager()
|
|
manager._segment_queue = queue.Queue()
|
|
buf = [b"\x00\x01", b"\x02\x03"]
|
|
manager._enqueue_audio_segment(buf)
|
|
assert not manager._segment_queue.empty()
|
|
assert manager._segment_queue.get_nowait() == buf
|
|
|
|
def test_enqueue_drops_oldest_when_queue_full(self):
|
|
"""When queue is full, oldest item is dropped and new one inserted."""
|
|
manager = _make_manager()
|
|
manager._segment_queue = queue.Queue(maxsize=1)
|
|
manager._segment_queue.put_nowait([b"old"])
|
|
manager._enqueue_audio_segment([b"new"])
|
|
assert manager._segment_queue.get_nowait() == [b"new"]
|
|
|
|
|
|
class TestPerformRecognition(unittest.TestCase):
|
|
"""Direct tests for _perform_recognition (not mocked)."""
|
|
|
|
def _run_recognition(self, manager, pre_hook=None):
|
|
"""Start _perform_recognition in a thread and return the thread."""
|
|
t = threading.Thread(target=manager._perform_recognition)
|
|
t.start()
|
|
if pre_hook:
|
|
pre_hook(manager)
|
|
return t
|
|
|
|
def test_processes_segment_and_exits_on_none_signal(self):
|
|
"""Feed a segment, then a None signal; thread processes and exits."""
|
|
manager = _make_manager()
|
|
manager._segment_queue = queue.Queue()
|
|
manager.should_record = True
|
|
|
|
with patch.object(manager, "_process_audio_buffer") as mock_proc:
|
|
with patch.object(manager, "_update_state"):
|
|
t = self._run_recognition(manager)
|
|
manager._segment_queue.put([b"\x00\x01"])
|
|
manager.should_record = False
|
|
manager._segment_queue.put(None)
|
|
t.join(timeout=2)
|
|
assert not t.is_alive()
|
|
assert mock_proc.called
|
|
|
|
def test_drains_remaining_items_on_none_signal(self):
|
|
"""None signal drains any remaining queued segments before exit."""
|
|
manager = _make_manager()
|
|
manager._segment_queue = queue.Queue()
|
|
manager.should_record = True
|
|
|
|
with patch.object(manager, "_process_audio_buffer") as mock_proc:
|
|
with patch.object(manager, "_update_state"):
|
|
t = self._run_recognition(manager)
|
|
manager._segment_queue.put([b"first"])
|
|
manager._segment_queue.put([b"second"])
|
|
manager.should_record = False
|
|
manager._segment_queue.put(None)
|
|
t.join(timeout=2)
|
|
assert not t.is_alive()
|
|
assert mock_proc.call_count == 2
|
|
|
|
def test_exits_when_idle_and_queue_empty(self):
|
|
"""When not recording and queue empty, thread exits after brief wait."""
|
|
manager = _make_manager()
|
|
manager._segment_queue = queue.Queue()
|
|
manager.should_record = False
|
|
|
|
with patch.object(manager, "_process_audio_buffer") as mock_proc:
|
|
with patch.object(manager, "_update_state"):
|
|
t = self._run_recognition(manager)
|
|
t.join(timeout=2)
|
|
assert not t.is_alive()
|
|
assert not mock_proc.called
|
|
|
|
def test_queue_timeout_while_recording(self):
|
|
"""While recording, an empty-queue timeout causes the loop to continue."""
|
|
manager = _make_manager()
|
|
manager._segment_queue = queue.Queue()
|
|
manager.should_record = True
|
|
|
|
with patch.object(manager, "_process_audio_buffer") as mock_proc:
|
|
with patch.object(manager, "_update_state"):
|
|
t = self._run_recognition(manager)
|
|
# Allow at least one 0.1 s timeout to fire
|
|
time.sleep(0.25)
|
|
manager.should_record = False
|
|
manager._segment_queue.put(None)
|
|
t.join(timeout=2)
|
|
assert not t.is_alive()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__, "-v"])
|