3303d075a6
* fix(audio): resolve device by name to prevent index mismatch crash (#498) Audio device indices can shift between sessions when USB devices are replugged or virtual devices are added/removed. A saved device_index=24 (BRIO webcam) could later point to 'speech-dispatcher-dummy', causing PortAudio to attempt recording from a virtual device and triggering malloc corruption / SIGABRT. Fix has three parts: 1. Store and match by device name, not just index - Config already saves device_name alongside device_index - At recording time, _resolve_device_by_name() enumerates current devices and finds the one matching the saved name - Falls back to stored index, then system default (with warning log) - main.py now passes audio_device_name to SpeechRecognitionManager 2. Filter virtual devices from input device enumeration - _is_virtual_device() detects speech-dispatcher, dummy, null sink, monitor, and ALSA output devices - get_audio_input_devices() skips virtual devices - Settings dialog dropdown only shows real input devices - Debug logging in _record_audio also skips virtual devices 3. Validate resolved device has input channels (from flatpak branch) - Extra safety check: after resolving, verify maxInputChannels > 0 - Falls back to system default if resolved device is invalid - Brings the validation from feature/flatpak-packaging to main Closes #498 * style: apply black formatting to recognition_manager.py * test: add audio_device_name kwarg to mock assertion * test: add coverage for _is_virtual_device, _resolve_device_by_name, device name management * test: fix StopIteration in fallback test by using return_value for mock * ponytail: remove redundant validation and simplify _resolve_device_by_name - Remove ~20 lines of dead validation in _record_audio() — _resolve_device_by_name and _resolve_valid_input_device already check maxInputChannels and virtual devices - Simplify _resolve_device_by_name to just find index by name, then delegate all validation to _resolve_valid_input_device (which already does channel/virtual checks) - Update tests to match the simplified delegation pattern * style(speech): format device fallback warning * test(speech): cover audio device fallback edges * test(speech): cover reconnection device fallback --------- Co-authored-by: Jatin K Malik <jatinkrmalik@users.noreply.github.com>
387 lines
15 KiB
Python
387 lines
15 KiB
Python
"""
|
|
Coverage boost tests targeting major gaps in recognition_manager and ibus_engine.
|
|
|
|
Key focus areas:
|
|
- Model download methods with progress tracking
|
|
- Audio reconnection logic
|
|
- IBus engine utility functions
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import time
|
|
import zipfile as REAL_ZIPFILE # Capture real zipfile before any test mocks it
|
|
from io import BytesIO
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
# Mock GI imports before importing any vocalinux modules that use gi.
|
|
# On CI, real gi/IBus packages are installed; without mocks, importing
|
|
# ibus_engine would connect to a real IBus daemon and hang.
|
|
if "gi" not in sys.modules:
|
|
sys.modules["gi"] = MagicMock()
|
|
if "gi.repository" not in sys.modules:
|
|
sys.modules["gi.repository"] = MagicMock()
|
|
|
|
from vocalinux.speech_recognition.recognition_manager import (
|
|
SpeechRecognitionManager,
|
|
)
|
|
|
|
|
|
def _make_manager(engine="whisper_cpp", **kw):
|
|
"""Create a SpeechRecognitionManager with mocked initialization."""
|
|
with patch.object(SpeechRecognitionManager, "_init_vosk"):
|
|
with patch.object(SpeechRecognitionManager, "_init_whisper"):
|
|
with patch.object(SpeechRecognitionManager, "_init_whispercpp"):
|
|
mgr = SpeechRecognitionManager(
|
|
engine=engine, model_size="small", language="en-us", defer_download=True, **kw
|
|
)
|
|
# Ensure vosk_model_map is set (normally done in _init_vosk)
|
|
if not hasattr(mgr, "vosk_model_map"):
|
|
mgr.vosk_model_map = {
|
|
"small": "model-en-us-0.22-lgraph",
|
|
"medium": "vosk-model-en-us-0.22",
|
|
"large": "vosk-model-en-us-0.22-lgraph",
|
|
}
|
|
return mgr
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def cleanup_sys_modules():
|
|
"""Cleanup sys.modules after each test - full snapshot/restore."""
|
|
# Take a complete snapshot of sys.modules before the test
|
|
saved_modules = dict(sys.modules)
|
|
|
|
yield
|
|
|
|
# Restore sys.modules to exact pre-test state
|
|
added_keys = set(sys.modules.keys()) - set(saved_modules.keys())
|
|
for key in added_keys:
|
|
del sys.modules[key]
|
|
for key, value in saved_modules.items():
|
|
if key not in sys.modules or sys.modules[key] is not value:
|
|
sys.modules[key] = value
|
|
|
|
|
|
class TestDownloadWhispercppModel:
|
|
"""Test _download_whispercpp_model() with runtime import mocking."""
|
|
|
|
def test_download_whispercpp_success_basic(self, tmp_path):
|
|
"""Test successful whisper.cpp model download."""
|
|
manager = _make_manager(engine="whisper_cpp")
|
|
model_file = str(tmp_path / "ggml-small.bin")
|
|
|
|
mock_requests = MagicMock()
|
|
mock_response = MagicMock()
|
|
mock_response.headers = {"content-length": "1000"}
|
|
mock_response.iter_content.return_value = [b"x" * 500, b"y" * 500]
|
|
mock_requests.get.return_value = mock_response
|
|
mock_requests.exceptions.RequestException = Exception
|
|
|
|
with patch.dict("sys.modules", {"requests": mock_requests}):
|
|
with patch(
|
|
"vocalinux.speech_recognition.recognition_manager.get_model_path",
|
|
return_value=model_file,
|
|
):
|
|
manager._download_whispercpp_model()
|
|
|
|
assert os.path.exists(model_file)
|
|
assert os.path.getsize(model_file) == 1000
|
|
|
|
def test_download_whispercpp_progress_callback(self, tmp_path):
|
|
"""Test progress callback is invoked during download."""
|
|
manager = _make_manager(engine="whisper_cpp")
|
|
progress_calls = []
|
|
|
|
def track_progress(progress, speed, status):
|
|
progress_calls.append((progress, speed, status))
|
|
|
|
manager._download_progress_callback = track_progress
|
|
model_file = str(tmp_path / "ggml-small.bin")
|
|
|
|
mock_requests = MagicMock()
|
|
mock_response = MagicMock()
|
|
mock_response.headers = {"content-length": "1000"}
|
|
mock_response.iter_content.return_value = [b"x" * 500, b"y" * 500]
|
|
mock_requests.get.return_value = mock_response
|
|
mock_requests.exceptions.RequestException = Exception
|
|
|
|
with patch.dict("sys.modules", {"requests": mock_requests}):
|
|
with patch(
|
|
"vocalinux.speech_recognition.recognition_manager.get_model_path",
|
|
return_value=model_file,
|
|
):
|
|
with patch("time.time", side_effect=[0, 0.2, 0.4, 0.6]):
|
|
manager._download_whispercpp_model()
|
|
|
|
mock_requests.get.assert_called_once()
|
|
call_args = mock_requests.get.call_args
|
|
assert call_args is not None
|
|
assert len(call_args[0]) > 0 or "url" in call_args[1]
|
|
assert len(progress_calls) >= 1
|
|
|
|
def test_download_whispercpp_no_content_length(self, tmp_path):
|
|
"""Test download when content-length header is missing."""
|
|
manager = _make_manager(engine="whisper_cpp")
|
|
model_file = str(tmp_path / "ggml-small.bin")
|
|
|
|
mock_requests = MagicMock()
|
|
mock_response = MagicMock()
|
|
mock_response.headers = {} # No content-length
|
|
mock_response.iter_content.return_value = [b"x" * 500, b"y" * 500]
|
|
mock_requests.get.return_value = mock_response
|
|
mock_requests.exceptions.RequestException = Exception
|
|
|
|
with patch.dict("sys.modules", {"requests": mock_requests}):
|
|
with patch(
|
|
"vocalinux.speech_recognition.recognition_manager.get_model_path",
|
|
return_value=model_file,
|
|
):
|
|
manager._download_whispercpp_model()
|
|
|
|
assert os.path.exists(model_file)
|
|
|
|
def test_download_whispercpp_request_error(self, tmp_path):
|
|
"""Test download request error handling."""
|
|
manager = _make_manager(engine="whisper_cpp")
|
|
model_file = str(tmp_path / "ggml-small.bin")
|
|
|
|
mock_requests = MagicMock()
|
|
mock_error = Exception("Network error")
|
|
mock_requests.get.side_effect = mock_error
|
|
mock_requests.exceptions.RequestException = Exception
|
|
|
|
with patch.dict("sys.modules", {"requests": mock_requests}):
|
|
with patch(
|
|
"vocalinux.speech_recognition.recognition_manager.get_model_path",
|
|
return_value=model_file,
|
|
):
|
|
with pytest.raises(RuntimeError, match="Failed to download"):
|
|
manager._download_whispercpp_model()
|
|
|
|
|
|
class TestDownloadVoskModel:
|
|
"""Test _download_vosk_model() with runtime import mocking."""
|
|
|
|
def test_download_vosk_progress_callback(self, tmp_path):
|
|
"""Test progress callback during Vosk download."""
|
|
manager = _make_manager(engine="vosk")
|
|
progress_calls = []
|
|
|
|
def track_progress(progress, speed, status):
|
|
progress_calls.append((progress, speed, status))
|
|
|
|
manager._download_progress_callback = track_progress
|
|
|
|
zip_data = BytesIO()
|
|
with REAL_ZIPFILE.ZipFile(zip_data, "w") as zf:
|
|
zf.writestr("model-en-us-0.22-lgraph/am/model.pkl", "x" * 5000)
|
|
zip_bytes = zip_data.getvalue()
|
|
|
|
mock_requests = MagicMock()
|
|
mock_response = MagicMock()
|
|
mock_response.headers = {"content-length": str(len(zip_bytes))}
|
|
mock_response.iter_content.return_value = [zip_bytes]
|
|
mock_requests.get.return_value = mock_response
|
|
mock_requests.exceptions.RequestException = Exception
|
|
|
|
with patch.dict("sys.modules", {"requests": mock_requests}):
|
|
with patch(
|
|
"vocalinux.speech_recognition.recognition_manager.MODELS_DIR", str(tmp_path)
|
|
):
|
|
with patch("time.time", side_effect=[0, 0.2, 0.4, 0.6]):
|
|
manager._download_vosk_model()
|
|
|
|
mock_requests.get.assert_called_once()
|
|
call_args = mock_requests.get.call_args
|
|
assert call_args is not None
|
|
assert len(call_args[0]) > 0 or "url" in call_args[1]
|
|
assert len(progress_calls) >= 1
|
|
|
|
def test_download_vosk_request_error(self, tmp_path):
|
|
"""Test Vosk download request error handling."""
|
|
manager = _make_manager(engine="vosk")
|
|
|
|
mock_requests = MagicMock()
|
|
mock_error = Exception("Network error")
|
|
mock_requests.get.side_effect = mock_error
|
|
mock_requests.exceptions.RequestException = Exception
|
|
|
|
with patch.dict("sys.modules", {"requests": mock_requests}):
|
|
with patch(
|
|
"vocalinux.speech_recognition.recognition_manager.MODELS_DIR", str(tmp_path)
|
|
):
|
|
with pytest.raises(RuntimeError, match="Failed to download"):
|
|
manager._download_vosk_model()
|
|
|
|
|
|
class TestAudioReconnection:
|
|
"""Test audio reconnection logic."""
|
|
|
|
def test_attempt_audio_reconnection_success(self):
|
|
"""Test successful audio reconnection."""
|
|
manager = _make_manager(engine="whisper_cpp")
|
|
|
|
mock_pyaudio_mod = MagicMock()
|
|
mock_pyaudio_mod.paInt16 = 8
|
|
mock_stream = MagicMock()
|
|
mock_stream.read.return_value = b"\x00" * 1024
|
|
mock_audio_instance = MagicMock()
|
|
mock_audio_instance.open.return_value = mock_stream
|
|
|
|
with patch.dict("sys.modules", {"pyaudio": mock_pyaudio_mod}):
|
|
with patch("time.sleep"):
|
|
result = manager._attempt_audio_reconnection(mock_audio_instance)
|
|
|
|
assert result is True
|
|
assert manager._audio_stream == mock_stream
|
|
|
|
def test_attempt_audio_reconnection_falls_back_to_default_resolver(self):
|
|
"""Test reconnection falls back when saved device name/index cannot resolve."""
|
|
manager = _make_manager(engine="whisper_cpp", audio_device_name="Missing Mic")
|
|
|
|
mock_pyaudio_mod = MagicMock()
|
|
mock_pyaudio_mod.paInt16 = 8
|
|
mock_stream = MagicMock()
|
|
mock_stream.read.return_value = b"\x00" * 1024
|
|
mock_audio_instance = MagicMock()
|
|
mock_audio_instance.get_default_input_device_info.return_value = {"index": 0}
|
|
mock_audio_instance.open.return_value = mock_stream
|
|
|
|
with (
|
|
patch.dict("sys.modules", {"pyaudio": mock_pyaudio_mod}),
|
|
patch("time.sleep"),
|
|
patch(
|
|
"vocalinux.speech_recognition.recognition_manager._resolve_device_by_name",
|
|
return_value=None,
|
|
) as mock_resolve_name,
|
|
patch(
|
|
"vocalinux.speech_recognition.recognition_manager._resolve_valid_input_device",
|
|
return_value=1,
|
|
) as mock_resolve_default,
|
|
patch(
|
|
"vocalinux.speech_recognition.recognition_manager._get_supported_channels",
|
|
return_value=1,
|
|
),
|
|
patch(
|
|
"vocalinux.speech_recognition.recognition_manager._get_supported_sample_rate",
|
|
return_value=16000,
|
|
),
|
|
):
|
|
result = manager._attempt_audio_reconnection(mock_audio_instance)
|
|
|
|
assert result is True
|
|
assert manager._audio_stream == mock_stream
|
|
mock_resolve_name.assert_called_once_with(mock_audio_instance, "Missing Mic", None)
|
|
mock_resolve_default.assert_called_once_with(mock_audio_instance, None)
|
|
|
|
def test_attempt_audio_reconnection_no_resolved_device(self):
|
|
"""Test reconnection fails when no resolver finds an input device."""
|
|
manager = _make_manager(engine="whisper_cpp", audio_device_name="Missing Mic")
|
|
|
|
mock_pyaudio_mod = MagicMock()
|
|
mock_audio_instance = MagicMock()
|
|
|
|
with (
|
|
patch.dict("sys.modules", {"pyaudio": mock_pyaudio_mod}),
|
|
patch("time.sleep"),
|
|
patch(
|
|
"vocalinux.speech_recognition.recognition_manager._resolve_device_by_name",
|
|
return_value=None,
|
|
),
|
|
patch(
|
|
"vocalinux.speech_recognition.recognition_manager._resolve_valid_input_device",
|
|
return_value=None,
|
|
),
|
|
):
|
|
result = manager._attempt_audio_reconnection(mock_audio_instance)
|
|
|
|
assert result is False
|
|
mock_audio_instance.open.assert_not_called()
|
|
|
|
def test_attempt_audio_reconnection_max_attempts(self):
|
|
"""Test reconnection stops after max attempts."""
|
|
manager = _make_manager(engine="whisper_cpp")
|
|
manager._reconnection_attempts = manager._max_reconnection_attempts
|
|
|
|
mock_audio_instance = MagicMock()
|
|
|
|
with patch.dict("sys.modules", {"pyaudio": MagicMock()}):
|
|
result = manager._attempt_audio_reconnection(mock_audio_instance)
|
|
|
|
assert result is False
|
|
|
|
def test_attempt_audio_reconnection_open_failure(self):
|
|
"""Test reconnection when stream open fails."""
|
|
manager = _make_manager(engine="whisper_cpp")
|
|
|
|
mock_pyaudio_mod = MagicMock()
|
|
mock_pyaudio_mod.paInt16 = 8
|
|
mock_audio_instance = MagicMock()
|
|
mock_audio_instance.open.side_effect = IOError("Cannot open stream")
|
|
|
|
with patch.dict("sys.modules", {"pyaudio": mock_pyaudio_mod}):
|
|
with patch("time.sleep"):
|
|
result = manager._attempt_audio_reconnection(mock_audio_instance)
|
|
|
|
assert result is False
|
|
|
|
def test_attempt_audio_reconnection_exponential_backoff(self):
|
|
"""Test exponential backoff in reconnection attempts."""
|
|
manager = _make_manager(engine="whisper_cpp")
|
|
manager._reconnection_delay = 0.1
|
|
|
|
mock_pyaudio_mod = MagicMock()
|
|
mock_pyaudio_mod.paInt16 = 8
|
|
mock_stream = MagicMock()
|
|
mock_stream.read.return_value = b"\x00" * 1024
|
|
mock_audio_instance = MagicMock()
|
|
mock_audio_instance.open.return_value = mock_stream
|
|
|
|
sleep_durations = []
|
|
|
|
def track_sleep(duration):
|
|
sleep_durations.append(duration)
|
|
|
|
with patch.dict("sys.modules", {"pyaudio": mock_pyaudio_mod}):
|
|
with patch("time.sleep", side_effect=track_sleep):
|
|
manager._reconnection_attempts = 0
|
|
manager._attempt_audio_reconnection(mock_audio_instance)
|
|
first_delay = sleep_durations[-1]
|
|
|
|
manager._reconnection_attempts = 1
|
|
manager._attempt_audio_reconnection(mock_audio_instance)
|
|
second_delay = sleep_durations[-1]
|
|
|
|
assert second_delay > first_delay
|
|
assert second_delay == first_delay * 2
|
|
|
|
|
|
class TestIBusEngineUtilities:
|
|
"""Test ibus_engine utility functions."""
|
|
|
|
def test_is_ibus_available(self):
|
|
"""Test is_ibus_available() function."""
|
|
from vocalinux.text_injection.ibus_engine import is_ibus_available
|
|
|
|
result = is_ibus_available()
|
|
assert isinstance(result, bool)
|
|
|
|
def test_is_ibus_daemon_running(self):
|
|
"""Test daemon detection when not running."""
|
|
from vocalinux.text_injection.ibus_engine import is_ibus_daemon_running
|
|
|
|
with patch("subprocess.run") as mock_run:
|
|
mock_run.return_value.returncode = 1
|
|
result = is_ibus_daemon_running()
|
|
assert result is False
|
|
|
|
def test_is_ibus_daemon_running_success(self):
|
|
"""Test daemon detection when running."""
|
|
from vocalinux.text_injection.ibus_engine import is_ibus_daemon_running
|
|
|
|
with patch("subprocess.run") as mock_run:
|
|
mock_run.return_value.returncode = 0
|