From 20375f865756e46573f5f94d185fd9c19a088d4d Mon Sep 17 00:00:00 2001 From: Jatin K Malik Date: Sun, 2 Aug 2026 19:41:34 -0700 Subject: [PATCH] fix(audio): filter unsafe virtual devices that crash capture (#629) * fix(audio): filter unsafe virtual devices that crash capture This is a defensive fix for #624 heap corruption seen when PortAudio opens PipeWire pseudo-devices such as DeepFilterNet, default, pipewire, paplay, monitor, null, dummy, or filter-chain sources. The root cause is not fully proven, but avoiding explicit capture from these unsafe virtual rows should keep dictation on real microphones while still allowing System Default through device_index=None. If this still crashes with a physical microphone selected, please confirm with the exact physical device name and audio stack details so the capture path can be narrowed further. Fixes #624 * fix(settings): sync audio device fallback with raw name restore Address Bugbot review on #629: match legacy "(default)" saved names and clear stale config/engine state when a filtered device falls back to System Default. * fix(audio): keep system default capture off pseudo device indices Stop converting System Default to an explicit PortAudio index in the mic test, and fall back to host default capture when every enumerated input is filtered as unsafe. * test(audio): expect system-default fallback on reconnect * test(audio): cover system-default fallback paths for capture --- .../speech_recognition/recognition_manager.py | 141 ++++++---- src/vocalinux/ui/settings_dialog.py | 74 ++++-- .../test_recognition_manager_device_config.py | 245 +++++++++++++++++- tests/test_recognition_manager_downloads.py | 12 +- 4 files changed, 396 insertions(+), 76 deletions(-) diff --git a/src/vocalinux/speech_recognition/recognition_manager.py b/src/vocalinux/speech_recognition/recognition_manager.py index 9defcd7..f32a184 100644 --- a/src/vocalinux/speech_recognition/recognition_manager.py +++ b/src/vocalinux/speech_recognition/recognition_manager.py @@ -217,16 +217,39 @@ def _is_virtual_device(device_name: str) -> bool: """ if not device_name: return False - name_lower = device_name.lower() + name_lower = device_name.strip().lower() + virtual_exact_names = { + "default", + "monitor", + "paplay", + "pipewire", + } + virtual_prefixes = ("default:", "pipewire:", "pipewire ") virtual_patterns = [ + ".monitor", + "_monitor", + "-monitor", + "alsa_output", # ALSA output devices exposed as monitors + "deepfilternet", + "dummy", + "filter-chain", + "filter_chain", + "monitor of", + "null sink", + "null source", + "null-sink", + "null-source", + "null_sink", + "null_source", + "paplay", "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) + return ( + name_lower in virtual_exact_names + or name_lower.startswith(virtual_prefixes) + or any(pattern in name_lower for pattern in virtual_patterns) + ) def _is_bluetooth_device(device_name: Optional[str]) -> bool: @@ -356,7 +379,7 @@ def _resolve_device_by_name( 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). + """Resolve a valid audio input device, skipping unsafe or output-only devices. Checks that the device has maxInputChannels > 0. Falls back from preferred_index → system default → first available input device. @@ -399,6 +422,11 @@ def _resolve_valid_input_device(audio, preferred_index: Optional[int] = None) -> input_device_indices.append(i) continue + device_name = info.get("name", "") + if _is_virtual_device(device_name): + logger.debug(f"Filtering virtual input device [{i}]: {device_name}") + continue + channels = info.get("maxInputChannels", 0) if isinstance(channels, (int, float)) and channels > 0: input_device_indices.append(i) @@ -415,7 +443,7 @@ def _resolve_valid_input_device(audio, preferred_index: Optional[int] = None) -> except (IOError, OSError): device_name = "unknown" logger.warning( - "Configured audio device [%s] (%s) has no input channels. " + "Configured audio device [%s] (%s) is not a safe input device. " "Falling back to a valid input device.", preferred_index, device_name, @@ -640,15 +668,17 @@ def test_audio_input(device_index: int = None, duration: float = 1.0) -> dict: audio = pyaudio.PyAudio() - # Get device info + # Get device info for display. Keep open_device_index as None for + # System Default so PortAudio opens the host default instead of an + # explicit pseudo-device index like "default" / DeepFilterNet (#624). + open_device_index = device_index try: if device_index is not None: info = audio.get_device_info_by_index(device_index) else: info = audio.get_default_input_device_info() - device_index = info.get("index") result["device_name"] = info.get("name", "Unknown") - result["device_index"] = device_index + result["device_index"] = info.get("index", device_index) except (IOError, OSError) as e: result["error"] = f"Cannot get device info: {e}" audio.terminate() @@ -657,7 +687,7 @@ def test_audio_input(device_index: int = None, duration: float = 1.0) -> dict: # Negotiate the format and open the capture stream in ONE PortAudio # open — Bluetooth SCO devices abort with heap corruption when the # stream is opened, closed, and quickly reopened (issue #567). - CHANNELS, RATE, stream = _open_capture_stream(audio, device_index) + CHANNELS, RATE, stream = _open_capture_stream(audio, open_device_index) logger.info(f"Using {CHANNELS} channel(s) for audio test") result["sample_rate"] = RATE @@ -672,8 +702,8 @@ def test_audio_input(device_index: int = None, duration: float = 1.0) -> dict: "input": True, "frames_per_buffer": CHUNK, } - if device_index is not None: - stream_kwargs["input_device_index"] = device_index + if open_device_index is not None: + stream_kwargs["input_device_index"] = open_device_index stream = audio.open(**stream_kwargs) except (IOError, OSError) as e: @@ -2528,20 +2558,24 @@ class SpeechRecognitionManager: # 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( - "Please connect a microphone and ensure it is recognized by the system." + use_system_default = self.audio_device_index is None and self.audio_device_name is None + if use_system_default: + resolved_device_index = None + else: + resolved_device_index = _resolve_device_by_name( + audio, self.audio_device_name, self.audio_device_index ) - play_error_sound() - audio.terminate() - self._update_state(RecognitionState.ERROR) - return + if resolved_device_index is None: + resolved_device_index = _resolve_valid_input_device(audio, None) + if resolved_device_index is None and not use_system_default: + # No safe enumerated mic left (e.g. only PipeWire pseudo devices). + # Fall back to PortAudio system default instead of aborting. + logger.warning( + "No safe audio input devices enumerated; " + "falling back to system default capture." + ) + resolved_device_index = None + use_system_default = True # Log available devices for debugging (skip virtual devices) logger.debug("Available audio input devices:") @@ -2565,10 +2599,14 @@ class SpeechRecognitionManager: logger.info(f"Using sample rate: {RATE}Hz") try: - device_info = audio.get_device_info_by_index(resolved_device_index) - logger.info( - f"Using audio device [{resolved_device_index}]: {device_info.get('name')}" - ) + if resolved_device_index is None: + device_info = audio.get_default_input_device_info() + logger.info(f"Using system default audio device: {device_info.get('name')}") + else: + device_info = audio.get_device_info_by_index(resolved_device_index) + logger.info( + f"Using audio device [{resolved_device_index}]: {device_info.get('name')}" + ) except (IOError, OSError): logger.warning(f"Could not get info for device index {resolved_device_index}") @@ -2591,7 +2629,7 @@ class SpeechRecognitionManager: default_idx = audio.get_default_input_device_info().get("index") except (IOError, OSError): default_idx = None - if resolved_device_index != default_idx: + if resolved_device_index is not None and resolved_device_index != default_idx: stream_kwargs["input_device_index"] = resolved_device_index try: @@ -2795,12 +2833,7 @@ class SpeechRecognitionManager: break # Clean up - if stream and hasattr(stream, "is_active") and stream.is_active(): - try: - stream.stop_stream() - stream.close() - except Exception as e: - logger.warning(f"Error closing audio stream: {e}") + _safe_close_stream(stream) if audio and hasattr(audio, "terminate"): try: @@ -3171,21 +3204,25 @@ class SpeechRecognitionManager: try: # Close existing stream if it exists if self._audio_stream: - try: - self._audio_stream.stop_stream() - self._audio_stream.close() - except Exception as e: - logger.debug(f"Error closing old audio stream: {e}") + _safe_close_stream(self._audio_stream) + self._audio_stream = None # 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 + use_system_default = self.audio_device_index is None and self.audio_device_name is None + if use_system_default: + resolved_device_index = None + else: + 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 and not use_system_default: + logger.warning( + "Reconnection: no safe input devices enumerated; " + "falling back to system default capture." + ) + resolved_device_index = None # Stream configuration CHUNK = 1024 @@ -3214,7 +3251,7 @@ class SpeechRecognitionManager: default_idx = audio_instance.get_default_input_device_info().get("index") except (IOError, OSError): default_idx = None - if resolved_device_index != default_idx: + if resolved_device_index is not None and resolved_device_index != default_idx: stream_kwargs["input_device_index"] = resolved_device_index new_stream = audio_instance.open(**stream_kwargs) diff --git a/src/vocalinux/ui/settings_dialog.py b/src/vocalinux/ui/settings_dialog.py index 9f4cae2..9a37a32 100644 --- a/src/vocalinux/ui/settings_dialog.py +++ b/src/vocalinux/ui/settings_dialog.py @@ -65,6 +65,36 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) + +def _raw_audio_device_name(device_name: Optional[str]) -> Optional[str]: + """Return the persisted device name without UI-only suffixes.""" + if device_name is None: + return None + return device_name.removesuffix(" (default)") + + +def _resolve_audio_device_selection( + devices: list, + saved_index: Optional[int], + saved_name: Optional[str], +) -> Optional[int]: + """Resolve a saved audio device to a currently listed index. + + Returns the matched device index, or None when the saved device is gone + (caller should fall back to System Default). + """ + saved_raw_name = _raw_audio_device_name(saved_name) + if saved_raw_name: + for idx, name, _is_default in devices: + if name == saved_raw_name: + return idx + if saved_index is not None: + for idx, _name, _is_default in devices: + if idx == saved_index: + return saved_index + return None + + # Define available models for each engine ENGINE_MODELS = { "vosk": [ @@ -4325,22 +4355,29 @@ For now, the engine has been reverted to VOSK.""" if saved_device is None: self.audio_device_combo.set_active_id("-1") else: - # 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") + matched_index = _resolve_audio_device_selection( + devices, saved_device, saved_device_name + ) + if matched_index is not None: + self.audio_device_combo.set_active_id(str(matched_index)) + # Migrate legacy configs that stored the UI "(default)" suffix. + saved_raw_name = _raw_audio_device_name(saved_device_name) + if saved_device_name and saved_raw_name and saved_device_name != saved_raw_name: + self.config_manager.set("audio", "device_name", saved_raw_name) + self.config_manager.set("audio", "device_index", matched_index) + self.config_manager.save_settings() + else: + logger.warning( + f"Saved audio device {saved_device} " + f"(name: {saved_device_name}) no longer available" + ) + self.audio_device_combo.set_active_id("-1") + # Keep combo, config, and engine aligned on System Default. + self.config_manager.set("audio", "device_index", None) + self.config_manager.set("audio", "device_name", None) + self.config_manager.save_settings() + if self.speech_engine is not None: + self.speech_engine.set_audio_device(None, None) logger.info(f"Found {len(devices)} audio input devices") @@ -4359,7 +4396,8 @@ For now, the engine has been reverted to VOSK.""" return device_index = int(device_id) - device_name = self.audio_device_combo.get_active_text() + device_label = self.audio_device_combo.get_active_text() + device_name = _raw_audio_device_name(device_label) if device_index == -1: self.config_manager.set("audio", "device_index", None) @@ -4376,7 +4414,7 @@ For now, the engine has been reverted to VOSK.""" 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"Selected: {device_name}") + self.audio_test_status.set_markup(f"Selected: {device_label}") def _on_test_audio_clicked(self, widget): """Handle test audio button click.""" diff --git a/tests/test_recognition_manager_device_config.py b/tests/test_recognition_manager_device_config.py index a29b50a..8e4e87b 100644 --- a/tests/test_recognition_manager_device_config.py +++ b/tests/test_recognition_manager_device_config.py @@ -32,10 +32,14 @@ from vocalinux.speech_recognition.recognition_manager import ( _get_supported_channels, _get_supported_sample_rate, _is_bluetooth_device, + _is_virtual_device, _open_capture_stream, + _resolve_valid_input_device, _safe_close_stream, get_audio_input_devices, - test_audio_input, +) +from vocalinux.speech_recognition.recognition_manager import ( + test_audio_input as _run_test_audio_input, ) @@ -52,6 +56,209 @@ def _make_manager(engine="whisper_cpp", **kw): class TestAudioDeviceDetection(unittest.TestCase): """Test audio device enumeration functions.""" + def test_is_virtual_device_filters_pipewire_pseudo_sources(self): + """PipeWire pseudo sources from issue #624 should not be opened by index.""" + unsafe_names = [ + "default", + "DeepFilterNet Source", + "paplay", + "pipewire", + "PipeWire filter-chain source", + "Monitor of Built-in Audio Analog Stereo", + "alsa_output.pci-0000_00_1f.3.analog-stereo.monitor", + "Null Sink", + "dummy", + "speech-dispatcher-dummy", + ] + + for name in unsafe_names: + assert _is_virtual_device(name), name + + assert not _is_virtual_device("USB Microphone") + assert not _is_virtual_device("Built-in Audio Analog Stereo") + assert not _is_virtual_device("Built-in Mic (PipeWire)") + assert not _is_virtual_device("") + + def test_get_audio_input_devices_filters_unsafe_virtual_sources(self): + """Device enumeration should hide unsafe virtual sources but keep real mics.""" + mock_audio = MagicMock() + mock_audio.get_device_count.return_value = 4 + mock_audio.get_default_input_device_info.return_value = {"index": 2} + devices = [ + {"index": 0, "name": "DeepFilterNet Source", "maxInputChannels": 2}, + {"index": 1, "name": "Monitor of Built-in Audio Analog Stereo", "maxInputChannels": 2}, + {"index": 2, "name": "USB Microphone", "maxInputChannels": 1}, + {"index": 3, "name": "default", "maxInputChannels": 32}, + ] + mock_audio.get_device_info_by_index.side_effect = lambda i: devices[i] + mock_pyaudio = MagicMock(PyAudio=MagicMock(return_value=mock_audio)) + + with patch.dict("sys.modules", {"pyaudio": mock_pyaudio}): + assert get_audio_input_devices() == [(2, "USB Microphone", True)] + + def test_resolve_valid_input_device_skips_stale_virtual_index(self): + """A saved virtual source index should fall back to a physical microphone.""" + mock_audio = MagicMock() + mock_audio.get_device_count.return_value = 3 + mock_audio.get_default_input_device_info.return_value = {"index": 1} + devices = [ + {"index": 0, "name": "DeepFilterNet Source", "maxInputChannels": 2}, + {"index": 1, "name": "default", "maxInputChannels": 32}, + {"index": 2, "name": "USB Microphone", "maxInputChannels": 1}, + ] + mock_audio.get_device_info_by_index.side_effect = lambda i: devices[i] + + assert _resolve_valid_input_device(mock_audio, preferred_index=0) == 2 + + def test_system_default_selection_uses_no_explicit_device_index(self): + """System Default should still open through PortAudio's default device path.""" + manager = _make_manager( + engine="whisper_cpp", audio_device_index=None, audio_device_name=None + ) + 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": "default", + "maxInputChannels": 32, + } + mock_audio.get_default_input_device_info.return_value = {"index": 0, "name": "default"} + + 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._open_capture_stream", + return_value=(1, 16000, mock_stream), + ) as mock_open, + patch( + "vocalinux.speech_recognition.recognition_manager.play_error_sound", + ), + ): + if isinstance(sys.modules.get("numpy"), MagicMock): + del sys.modules["numpy"] + manager._record_audio() + + mock_open.assert_called_once_with(mock_audio, None) + + def test_record_audio_falls_back_to_system_default_when_no_safe_device(self): + """Stale/unsafe saved devices should reopen via PortAudio system default.""" + manager = _make_manager( + engine="whisper_cpp", + audio_device_index=0, + audio_device_name="DeepFilterNet Source", + ) + 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": "DeepFilterNet Source", + "maxInputChannels": 2, + } + mock_audio.get_default_input_device_info.return_value = { + "index": 0, + "name": "default", + } + 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=None, + ), + patch( + "vocalinux.speech_recognition.recognition_manager._resolve_valid_input_device", + return_value=None, + ), + patch( + "vocalinux.speech_recognition.recognition_manager._open_capture_stream", + return_value=(1, 16000, mock_stream), + ) as mock_open, + patch("vocalinux.speech_recognition.recognition_manager.play_error_sound"), + ): + if isinstance(sys.modules.get("numpy"), MagicMock): + del sys.modules["numpy"] + manager._record_audio() + + mock_open.assert_called_once_with(mock_audio, None) + mock_audio.get_default_input_device_info.assert_called() + + def test_test_audio_input_system_default_fallback_open_omits_device_index(self): + """Fallback mic-test open for System Default must omit input_device_index.""" + 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_default_input_device_info.return_value = { + "index": 7, + "name": "default", + "defaultSampleRate": 48000, + "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, 48000, None), + ): + result = _run_test_audio_input(device_index=None, duration=0.1) + + assert result["success"] is True + kwargs = mock_audio.open.call_args.kwargs + assert "input_device_index" not in kwargs + + def test_settings_persist_raw_audio_device_name(self): + """Settings should persist raw PortAudio names, not UI-only suffixes.""" + from vocalinux.ui.settings_dialog import ( + _raw_audio_device_name, + _resolve_audio_device_selection, + ) + + assert _raw_audio_device_name("USB Microphone (default)") == "USB Microphone" + assert _raw_audio_device_name("USB Microphone") == "USB Microphone" + assert _raw_audio_device_name(None) is None + + devices = [(3, "USB Microphone", True), (5, "Webcam Mic", False)] + # Legacy configs may still store the UI-only "(default)" suffix. + assert _resolve_audio_device_selection(devices, 3, "USB Microphone (default)") == 3 + assert _resolve_audio_device_selection(devices, 99, "Webcam Mic") == 5 + # Filtered-out / missing devices resolve to System Default. + assert _resolve_audio_device_selection(devices, 1, "DeepFilterNet") is None + assert _resolve_audio_device_selection(devices, 1, None) is None + 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") @@ -192,6 +399,38 @@ class TestAudioDeviceDetection(unittest.TestCase): mock_stream.stop_stream.assert_called_once() mock_stream.close.assert_called_once() + def test_test_audio_input_system_default_keeps_null_device_index(self): + """System Default must open PortAudio without an explicit device index.""" + 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_default_input_device_info.return_value = { + "index": 7, + "name": "default", + "defaultSampleRate": 48000, + "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, 48000, mock_stream), + ) as mock_open: + result = test_audio_input(device_index=None, duration=0.1) + + assert result["success"] is True + mock_open.assert_called_once_with(mock_audio, None) + # Display metadata can still report the host default index/name. + assert result["device_index"] == 7 + assert result["device_name"] == "default" + 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() @@ -214,7 +453,7 @@ class TestAudioDeviceDetection(unittest.TestCase): "vocalinux.speech_recognition.recognition_manager._open_capture_stream", return_value=(1, 16000, None), ): - result = test_audio_input(device_index=14, duration=0.1) + result = _run_test_audio_input(device_index=14, duration=0.1) assert result["success"] is True mock_audio.open.assert_called_once() @@ -237,7 +476,7 @@ class TestAudioDeviceDetection(unittest.TestCase): "vocalinux.speech_recognition.recognition_manager._open_capture_stream", return_value=(1, 16000, None), ): - result = test_audio_input(device_index=14, duration=0.1) + result = _run_test_audio_input(device_index=14, duration=0.1) assert result["success"] is False assert "Cannot open audio stream" in result["error"] diff --git a/tests/test_recognition_manager_downloads.py b/tests/test_recognition_manager_downloads.py index 4a3e5bb..a55bf47 100644 --- a/tests/test_recognition_manager_downloads.py +++ b/tests/test_recognition_manager_downloads.py @@ -401,11 +401,13 @@ class TestAudioReconnection: 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.""" + """When no safe device is enumerated, reconnect via system default.""" manager = _make_manager(engine="whisper_cpp", audio_device_name="Missing Mic") mock_pyaudio_mod = MagicMock() + mock_pyaudio_mod.paInt16 = 8 mock_audio_instance = MagicMock() + mock_stream = MagicMock() with ( patch.dict("sys.modules", {"pyaudio": mock_pyaudio_mod}), @@ -418,11 +420,15 @@ class TestAudioReconnection: "vocalinux.speech_recognition.recognition_manager._resolve_valid_input_device", return_value=None, ), + patch( + "vocalinux.speech_recognition.recognition_manager._open_capture_stream", + return_value=(1, 16000, mock_stream), + ) as mock_open, ): result = manager._attempt_audio_reconnection(mock_audio_instance) - assert result is False - mock_audio_instance.open.assert_not_called() + assert result is True + mock_open.assert_called_once_with(mock_audio_instance, None) def test_attempt_audio_reconnection_max_attempts(self): """Test reconnection stops after max attempts."""