xavierk/voculinux
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

Fix: Audio device index mismatch causing crash on recording start (#498) (#499)

Browse Source 
* 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>
This commit is contained in:
Jatin K Malik
2026-07-09 16:37:45 -07:00
committed by GitHub
co-authored by Jatin K Malik
parent 8f32f01f0e
commit 3303d075a6
6 changed files with 457 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/vocalinux/main.py
+6 -1
View File
@@ -354,12 +354,16 @@ def main():
stop_sound_guard_ms = saved_settings.get("stop_sound_guard_ms", 200)
voice_commands_enabled = saved_settings.get("voice_commands_enabled") # None = auto
audio_device_index = audio_settings.get("device_index", None)
audio_device_name = audio_settings.get("device_name", None)
advanced_settings = config_manager.get_settings().get("advanced", {})
logger.info(f"Final settings: engine={engine}, language={language}, model={model_size}")
if audio_device_index is not None:
logger.info(f"Using audio device index={audio_device_index} (from saved config)")
logger.info(
f"Using audio device index={audio_device_index} "
f"(name={audio_device_name}, from saved config)"
)
# Initialize main components
logger.info("Initializing Vocalinux...")
@@ -375,6 +379,7 @@ def main():
stop_sound_guard_ms=stop_sound_guard_ms,
voice_commands_enabled=voice_commands_enabled,
audio_device_index=audio_device_index,
audio_device_name=audio_device_name,
whispercpp_no_timestamps=advanced_settings.get("whispercpp_no_timestamps", True),
whispercpp_no_context=advanced_settings.get("whispercpp_no_context", True),
whispercpp_initial_prompt=advanced_settings.get("whispercpp_initial_prompt", ""),
src/vocalinux/speech_recognition/recognition_manager.py
+103 -15
View File
@@ -180,9 +180,36 @@ def _preload_pywhispercpp_shared_libraries() -> None:
)
def _is_virtual_device(device_name: str) -> bool:
"""Check if a device name corresponds to a virtual audio device.
Virtual devices (e.g. speech-dispatcher-dummy, PulseAudio monitors,
PipeWire null sinks) cannot be used for recording and may cause crashes
if PortAudio tries to open them.
Args:
device_name: The device name to check.
Returns:
True if the device appears to be virtual, False otherwise.
"""
if not device_name:
return False
name_lower = device_name.lower()
virtual_patterns = [
"speech-dispatcher",
"speechdispatcher",
"dummy",
"null sink",
"monitor of",
"alsa_output", # ALSA output devices exposed as monitors
]
return any(pattern in name_lower for pattern in virtual_patterns)
def get_audio_input_devices() -> list:
"""
Get a list of available audio input devices.
Get a list of available audio input devices, excluding virtual devices.
Returns:
List of tuples: (device_index, device_name, is_default)
@@ -206,6 +233,10 @@ def get_audio_input_devices() -> list:
# Only include devices that have input channels
if info.get("maxInputChannels", 0) > 0:
name = info.get("name", f"Device {i}")
# Skip virtual devices that can cause crashes
if _is_virtual_device(name):
logger.debug(f"Filtering virtual device [{i}]: {name}")
continue
is_default = i == default_input_device
devices.append((i, name, is_default))
except (IOError, OSError):
@@ -220,6 +251,37 @@ def get_audio_input_devices() -> list:
return devices
def _resolve_device_by_name(
audio, device_name: Optional[str], fallback_index: Optional[int] = None
) -> Optional[int]:
"""Resolve a device index by name, falling back to index, then system default.
Device indices shift when USB devices are replugged or virtual devices are
added/removed. Name matching is more stable.
"""
if not device_name:
return _resolve_valid_input_device(audio, fallback_index)
try:
device_count = int(audio.get_device_count())
except (IOError, OSError, TypeError, ValueError, AttributeError):
return _resolve_valid_input_device(audio, fallback_index)
# ponytail: just find the index by name, delegate validation to _resolve_valid_input_device
for i in range(device_count):
try:
info = audio.get_device_info_by_index(i)
except (IOError, OSError, TypeError, ValueError, AttributeError):
continue
if isinstance(info, dict) and info.get("name", "") == device_name:
return _resolve_valid_input_device(audio, i)
logger.warning(
f"Audio device '{device_name}' not found, falling back to index {fallback_index}"
)
return _resolve_valid_input_device(audio, fallback_index)
def _resolve_valid_input_device(audio, preferred_index: Optional[int] = None) -> Optional[int]:
"""Resolve a valid audio input device, skipping output-only devices (e.g. HDMI).
@@ -765,6 +827,7 @@ class SpeechRecognitionManager:
# Audio device selection (None means use system default)
self.audio_device_index = kwargs.get("audio_device_index", None)
self.audio_device_name = kwargs.get("audio_device_name", None)
# whisper.cpp advanced parameters
self.whispercpp_no_timestamps = kwargs.get("whispercpp_no_timestamps", True)
@@ -2026,21 +2089,30 @@ class SpeechRecognitionManager:
except ValueError:
pass
def set_audio_device(self, device_index: Optional[int]):
def set_audio_device(self, device_index: Optional[int], device_name: Optional[str] = None):
"""
Set the audio input device to use.
Args:
device_index: The device index to use, or None for system default
device_name: The device name (for stable re-resolution on next recording)
"""
if device_index != self.audio_device_index:
logger.info(f"Audio device changed from {self.audio_device_index} to {device_index}")
self.audio_device_index = device_index
if device_index != self.audio_device_index or device_name != self.audio_device_name:
logger.info(
f"Audio device changed from {self.audio_device_index} "
f"to {device_index} (name: {device_name})"
)
self.audio_device_index = device_index
self.audio_device_name = device_name
def get_audio_device(self) -> Optional[int]:
"""Get the currently configured audio device index."""
return self.audio_device_index
def get_audio_device_name(self) -> Optional[str]:
"""Get the currently configured audio device name."""
return self.audio_device_name
def get_last_audio_level(self) -> float:
"""Get the last recorded audio level (0-100)."""
return self._last_audio_level
@@ -2199,8 +2271,14 @@ class SpeechRecognitionManager:
self._pyaudio_instance = pyaudio.PyAudio()
audio = self._pyaudio_instance
# Resolve a valid input device — skip output-only devices (e.g. HDMI)
resolved_device_index = _resolve_valid_input_device(audio, self.audio_device_index)
# Resolve the input device by name first (indices can shift between
# sessions due to USB replugging or virtual devices being added).
# Fall back to the stored index, then to the system default.
resolved_device_index = _resolve_device_by_name(
audio, self.audio_device_name, self.audio_device_index
)
if resolved_device_index is None:
resolved_device_index = _resolve_valid_input_device(audio, None)
if resolved_device_index is None:
logger.error("No audio input devices found with input channels.")
logger.error(
@@ -2211,15 +2289,16 @@ class SpeechRecognitionManager:
self._update_state(RecognitionState.ERROR)
return
# Log available devices for debugging
# Log available devices for debugging (skip virtual devices)
logger.debug("Available audio input devices:")
for i in range(audio.get_device_count()):
try:
info = audio.get_device_info_by_index(i)
if info.get("maxInputChannels", 0) > 0:
logger.debug(
f" [{i}] {info.get('name')} (inputs: {info.get('maxInputChannels')})"
)
name = info.get("name", "")
if _is_virtual_device(name):
continue
logger.debug(f" [{i}] {name} (inputs: {info.get('maxInputChannels')})")
except (IOError, OSError):
continue
@@ -2658,6 +2737,7 @@ class SpeechRecognitionManager:
vad_sensitivity: Optional[int] = None,
silence_timeout: Optional[float] = None,
audio_device_index: Optional[int] = None,
audio_device_name: Optional[str] = None,
force_download: bool = True,
**kwargs, # Allow for future expansion
):
@@ -2671,10 +2751,13 @@ class SpeechRecognitionManager:
vad_sensitivity: New VAD sensitivity (for VOSK).
silence_timeout: New silence timeout (for VOSK).
audio_device_index: Audio input device index (None for default, -1 to clear).
audio_device_name: Audio device name for stable re-resolution.
force_download: If True, download missing models (default: True for UI-triggered reconfigures).
"""
logger.info(
f"Reconfiguring speech engine. New settings: engine={engine}, model_size={model_size}, language={language}, vad={vad_sensitivity}, silence={silence_timeout}, audio_device={audio_device_index}"
f"Reconfiguring speech engine. New settings: engine={engine}, model_size={model_size}, "
f"language={language}, vad={vad_sensitivity}, silence={silence_timeout}, "
f"audio_device={audio_device_index}, audio_device_name={audio_device_name}"
)
restart_needed = False
@@ -2704,8 +2787,11 @@ class SpeechRecognitionManager:
if audio_device_index is not None:
if audio_device_index == -1:
self.audio_device_index = None
self.audio_device_name = None
else:
self.audio_device_index = audio_device_index
if audio_device_name is not None:
self.audio_device_name = audio_device_name
if "voice_commands_enabled" in kwargs:
self._voice_commands_preference = kwargs.get("voice_commands_enabled")
@@ -2828,10 +2914,12 @@ class SpeechRecognitionManager:
except Exception as e:
logger.debug(f"Error closing old audio stream: {e}")
# Resolve a valid input device — skip output-only devices (e.g. HDMI)
resolved_device_index = _resolve_valid_input_device(
audio_instance, self.audio_device_index
# Resolve a valid input device — by name first, then by index
resolved_device_index = _resolve_device_by_name(
audio_instance, self.audio_device_name, self.audio_device_index
)
if resolved_device_index is None:
resolved_device_index = _resolve_valid_input_device(audio_instance, None)
if resolved_device_index is None:
logger.error("Reconnection failed: no input devices available.")
return False
src/vocalinux/ui/settings_dialog.py
+19 -5
View File
@@ -3251,13 +3251,27 @@ For now, the engine has been reverted to VOSK."""
self.audio_device_combo.append(str(device_index), label)
saved_device = self.config_manager.get_optional_int("audio", "device_index", None)
saved_device_name = self.config_manager.get("audio", "device_name", None)
if saved_device is None:
self.audio_device_combo.set_active_id("-1")
else:
if not self.audio_device_combo.set_active_id(str(saved_device)):
logger.warning(f"Saved audio device {saved_device} no longer available")
self.audio_device_combo.set_active_id("-1")
# Try to match by saved device name first (more stable across reboots)
matched = False
if saved_device_name:
for idx, name, _ in devices:
if name == saved_device_name:
self.audio_device_combo.set_active_id(str(idx))
matched = True
break
if not matched:
# Fall back to stored index
if not self.audio_device_combo.set_active_id(str(saved_device)):
logger.warning(
f"Saved audio device {saved_device} "
f"(name: {saved_device_name}) no longer available"
)
self.audio_device_combo.set_active_id("-1")
logger.info(f"Found {len(devices)} audio input devices")
@@ -3288,9 +3302,9 @@ For now, the engine has been reverted to VOSK."""
self.config_manager.save_settings()
if device_index == -1:
self.speech_engine.set_audio_device(None)
self.speech_engine.set_audio_device(None, None)
else:
self.speech_engine.set_audio_device(device_index)
self.speech_engine.set_audio_device(device_index, device_name)
logger.info(f"Audio device changed to: [{device_index}] {device_name}")
self.audio_test_status.set_markup(f"<i>Selected: {device_name}</i>")
tests/test_main.py
+1
View File
@@ -231,6 +231,7 @@ class TestMainModule(unittest.TestCase):
stop_sound_guard_ms=200,
voice_commands_enabled=None,
audio_device_index=None,
audio_device_name=None,
whispercpp_no_timestamps=True,
whispercpp_no_context=True,
whispercpp_initial_prompt="",
tests/test_recognition_manager.py
+265
View File
@@ -518,6 +518,39 @@ class TestSpeechRecognition(unittest.TestCase):
self.assertEqual(manager.silence_timeout, 1.5)
self.assertEqual(manager.audio_device_index, 2)
def test_init_with_device_name(self):
"""Test initialization with audio_device_name kwarg."""
manager = SpeechRecognitionManager(
engine="vosk", audio_device_index=3, audio_device_name="USB Mic"
)
self.assertEqual(manager.audio_device_index, 3)
self.assertEqual(manager.get_audio_device_name(), "USB Mic")
def test_set_audio_device_with_name(self):
"""Test set_audio_device stores both index and name."""
manager = SpeechRecognitionManager(engine="vosk")
manager.set_audio_device(5, "BRIO Webcam")
self.assertEqual(manager.get_audio_device(), 5)
self.assertEqual(manager.get_audio_device_name(), "BRIO Webcam")
def test_set_audio_device_clears_name(self):
"""Test set_audio_device(None, None) clears both."""
manager = SpeechRecognitionManager(engine="vosk", audio_device_name="Mic")
manager.set_audio_device(None, None)
self.assertIsNone(manager.get_audio_device())
self.assertIsNone(manager.get_audio_device_name())
def test_reconfigure_with_device_name(self):
"""Test reconfigure accepts audio_device_name."""
manager = SpeechRecognitionManager(engine="vosk")
manager.reconfigure(audio_device_index=2, audio_device_name="Headset")
self.assertEqual(manager.audio_device_index, 2)
self.assertEqual(manager.get_audio_device_name(), "Headset")
class TestModuleLevelFunctions(unittest.TestCase):
"""Test module-level functions in recognition_manager."""
@@ -572,6 +605,238 @@ class TestModuleLevelFunctions(unittest.TestCase):
with patch.dict(sys.modules, {"pyaudio": mock_pyaudio}):
devices = recognition_manager.get_audio_input_devices()
def test_get_audio_input_devices_skips_unreadable_device(self):
"""Test unreadable devices are skipped during enumeration."""
from vocalinux.speech_recognition import recognition_manager
mock_pyaudio_instance = MagicMock()
mock_pyaudio_instance.get_default_input_device_info.return_value = {"index": 1}
mock_pyaudio_instance.get_device_count.return_value = 2
mock_pyaudio_instance.get_device_info_by_index.side_effect = [
IOError("device disappeared"),
{"name": "USB Mic", "maxInputChannels": 1},
]
mock_pyaudio = MagicMock()
mock_pyaudio.PyAudio.return_value = mock_pyaudio_instance
with patch.dict(sys.modules, {"pyaudio": mock_pyaudio}):
devices = recognition_manager.get_audio_input_devices()
self.assertEqual(devices, [(1, "USB Mic", True)])
def test_get_audio_input_devices_skips_output_only_device(self):
"""Test output-only devices are skipped during enumeration."""
from vocalinux.speech_recognition import recognition_manager
mock_pyaudio_instance = MagicMock()
mock_pyaudio_instance.get_default_input_device_info.return_value = {"index": 1}
mock_pyaudio_instance.get_device_count.return_value = 2
mock_pyaudio_instance.get_device_info_by_index.side_effect = [
{"name": "HDMI Output", "maxInputChannels": 0},
{"name": "USB Mic", "maxInputChannels": 1},
]
mock_pyaudio = MagicMock()
mock_pyaudio.PyAudio.return_value = mock_pyaudio_instance
with patch.dict(sys.modules, {"pyaudio": mock_pyaudio}):
devices = recognition_manager.get_audio_input_devices()
self.assertEqual(devices, [(1, "USB Mic", True)])
def test_get_audio_input_devices_import_error(self):
"""Test missing PyAudio returns an empty device list."""
from vocalinux.speech_recognition import recognition_manager
with patch.dict(sys.modules, {"pyaudio": None}):
devices = recognition_manager.get_audio_input_devices()
self.assertEqual(devices, [])
def test_get_audio_input_devices_pyaudio_error(self):
"""Test PyAudio enumeration errors return an empty device list."""
from vocalinux.speech_recognition import recognition_manager
mock_pyaudio = MagicMock()
mock_pyaudio.PyAudio.side_effect = OSError("boom")
with patch.dict(sys.modules, {"pyaudio": mock_pyaudio}):
devices = recognition_manager.get_audio_input_devices()
self.assertEqual(devices, [])
def test_is_virtual_device(self):
"""Test _is_virtual_device detects known virtual device patterns."""
from vocalinux.speech_recognition.recognition_manager import _is_virtual_device
# Virtual devices
self.assertTrue(_is_virtual_device("speech-dispatcher-dummy"))
self.assertTrue(_is_virtual_device("speech-dispatcher-generic"))
self.assertTrue(_is_virtual_device("Dummy Output"))
self.assertTrue(_is_virtual_device("Null Sink"))
self.assertTrue(_is_virtual_device("Monitor of Built-in Audio"))
self.assertTrue(_is_virtual_device("alsa_output.pci-0000"))
# Real devices
self.assertFalse(_is_virtual_device("BRIO Ultra HD Webcam"))
self.assertFalse(_is_virtual_device("Built-in Microphone"))
self.assertFalse(_is_virtual_device("USB Headset"))
# Edge cases
self.assertFalse(_is_virtual_device(""))
self.assertFalse(_is_virtual_device(None))
def test_is_virtual_device_case_insensitive(self):
"""Test _is_virtual_device is case-insensitive."""
from vocalinux.speech_recognition.recognition_manager import _is_virtual_device
self.assertTrue(_is_virtual_device("SPEECH-DISPATCHER-DUMMY"))
self.assertTrue(_is_virtual_device("NULL SINK"))
self.assertTrue(_is_virtual_device("Monitor Of Something"))
def test_get_audio_input_devices_filters_virtual(self):
"""Test that get_audio_input_devices excludes virtual devices."""
from vocalinux.speech_recognition import recognition_manager
mock_pyaudio_instance = MagicMock()
mock_pyaudio_instance.get_default_input_device_info.return_value = {"index": 0}
mock_pyaudio_instance.get_device_count.return_value = 3
mock_pyaudio_instance.get_device_info_by_index.side_effect = [
{"name": "Real Mic", "maxInputChannels": 2},
{"name": "speech-dispatcher-dummy", "maxInputChannels": 2},
{"name": "USB Headset", "maxInputChannels": 1},
]
mock_pyaudio = MagicMock()
mock_pyaudio.PyAudio.return_value = mock_pyaudio_instance
with patch.dict(sys.modules, {"pyaudio": mock_pyaudio}):
devices = recognition_manager.get_audio_input_devices()
# Should only have 2 devices (virtual one filtered out)
device_names = [d[1] for d in devices]
self.assertIn("Real Mic", device_names)
self.assertIn("USB Headset", device_names)
self.assertNotIn("speech-dispatcher-dummy", device_names)
@patch("vocalinux.speech_recognition.recognition_manager._resolve_valid_input_device")
def test_resolve_device_by_name_found(self, mock_valid):
"""Test _resolve_device_by_name finds a matching device and delegates validation."""
from vocalinux.speech_recognition.recognition_manager import _resolve_device_by_name
mock_valid.return_value = 1
mock_audio = MagicMock()
mock_audio.get_device_count.return_value = 3
mock_audio.get_device_info_by_index.side_effect = [
{"name": "Built-in Mic", "maxInputChannels": 1},
{"name": "BRIO Webcam", "maxInputChannels": 2},
{"name": "speech-dispatcher-dummy", "maxInputChannels": 2},
]
result = _resolve_device_by_name(mock_audio, "BRIO Webcam")
self.assertEqual(result, 1)
mock_valid.assert_called_once_with(mock_audio, 1)
@patch("vocalinux.speech_recognition.recognition_manager._resolve_valid_input_device")
def test_resolve_device_by_name_not_found_with_fallback(self, mock_valid):
"""Test _resolve_device_by_name falls back to index when name not found."""
from vocalinux.speech_recognition.recognition_manager import _resolve_device_by_name
mock_valid.return_value = 0
mock_audio = MagicMock()
mock_audio.get_device_count.return_value = 2
mock_audio.get_device_info_by_index.return_value = {
"name": "Mic",
"maxInputChannels": 1,
}
result = _resolve_device_by_name(mock_audio, "Missing Device", fallback_index=0)
self.assertEqual(result, 0)
mock_valid.assert_called_once_with(mock_audio, 0)
@patch("vocalinux.speech_recognition.recognition_manager._resolve_valid_input_device")
def test_resolve_device_by_name_skips_unreadable_device(self, mock_valid):
"""Test _resolve_device_by_name skips devices that cannot be read."""
from vocalinux.speech_recognition.recognition_manager import _resolve_device_by_name
mock_valid.return_value = 1
mock_audio = MagicMock()
mock_audio.get_device_count.return_value = 2
mock_audio.get_device_info_by_index.side_effect = [
OSError("device disappeared"),
{"name": "USB Mic", "maxInputChannels": 1},
]
result = _resolve_device_by_name(mock_audio, "USB Mic")
self.assertEqual(result, 1)
mock_valid.assert_called_once_with(mock_audio, 1)
@patch("vocalinux.speech_recognition.recognition_manager._resolve_valid_input_device")
def test_resolve_device_by_name_ignores_non_dict_device_info(self, mock_valid):
"""Test _resolve_device_by_name ignores malformed device info."""
from vocalinux.speech_recognition.recognition_manager import _resolve_device_by_name
mock_valid.return_value = 0
mock_audio = MagicMock()
mock_audio.get_device_count.return_value = 1
mock_audio.get_device_info_by_index.return_value = None
result = _resolve_device_by_name(mock_audio, "USB Mic", fallback_index=0)
self.assertEqual(result, 0)
mock_valid.assert_called_once_with(mock_audio, 0)
@patch("vocalinux.speech_recognition.recognition_manager._resolve_valid_input_device")
def test_resolve_device_by_name_not_found_no_fallback(self, mock_valid):
"""Test _resolve_device_by_name returns None when name not found and no fallback."""
from vocalinux.speech_recognition.recognition_manager import _resolve_device_by_name
mock_valid.return_value = None
mock_audio = MagicMock()
mock_audio.get_device_count.return_value = 1
mock_audio.get_device_info_by_index.return_value = {
"name": "Mic",
"maxInputChannels": 1,
}
result = _resolve_device_by_name(mock_audio, "Missing Device")
self.assertIsNone(result)
@patch("vocalinux.speech_recognition.recognition_manager._resolve_valid_input_device")
def test_resolve_device_by_name_empty_name_with_fallback(self, mock_valid):
"""Test _resolve_device_by_name with empty name uses fallback index."""
from vocalinux.speech_recognition.recognition_manager import _resolve_device_by_name
mock_valid.return_value = 0
mock_audio = MagicMock()
result = _resolve_device_by_name(mock_audio, None, fallback_index=0)
self.assertEqual(result, 0)
mock_valid.assert_called_once_with(mock_audio, 0)
@patch("vocalinux.speech_recognition.recognition_manager._resolve_valid_input_device")
def test_resolve_device_by_name_empty_name_no_fallback(self, mock_valid):
"""Test _resolve_device_by_name with empty name and no fallback returns None."""
from vocalinux.speech_recognition.recognition_manager import _resolve_device_by_name
mock_valid.return_value = None
mock_audio = MagicMock()
result = _resolve_device_by_name(mock_audio, None)
self.assertIsNone(result)
@patch("vocalinux.speech_recognition.recognition_manager._resolve_valid_input_device")
def test_resolve_device_by_name_get_count_error(self, mock_valid):
"""Test _resolve_device_by_name handles get_device_count errors."""
from vocalinux.speech_recognition.recognition_manager import _resolve_device_by_name
mock_valid.return_value = 0
mock_audio = MagicMock()
mock_audio.get_device_count.side_effect = IOError("boom")
result = _resolve_device_by_name(mock_audio, "Some Device", fallback_index=0)
self.assertEqual(result, 0)
mock_valid.assert_called_once_with(mock_audio, 0)
def test_show_notification(self):
"""Test _show_notification helper function."""
from vocalinux.speech_recognition.recognition_manager import _show_notification
tests/test_recognition_manager_downloads.py
+63
View File
@@ -238,6 +238,69 @@ class TestAudioReconnection:
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")