fix(audio): never record from another microphone when one is configured
CI / security (pull_request) Successful in 1m10s
CI / lint (pull_request) Successful in 23s
CI / ai-review (pull_request) Successful in 25s

A configured device that was virtual, output-only, renamed or unplugged
was replaced by the hardware default input, with only a log warning. The
capture source now refuses an explicit selection that cannot be used as
configured: it raises AudioDeviceUnavailableError, skips reconnection
(retrying cannot help), and reports it through the desktop notification
path. System Default (unset, or -1) keeps its implicit default.

Behaviour change: an unplugged or renamed configured microphone now fails
visibly instead of recording from the default input. Pulse source
selection is unchanged.

Fixes #42
This commit is contained in:
2026-11-06 23:58:15 +05:30
parent 378a4f2da3
commit 71b97fe633
10 changed files with 392 additions and 114 deletions
+4 -1
View File
@@ -44,7 +44,10 @@ The clipboard copy behind these pastes is temporary: your previous clipboard com
Open **Settings → Audio → Input Device** to select a specific microphone or choose
**System Default**. A valid explicit microphone selection uses that device and
bypasses the System Default preference below.
bypasses the System Default preference below. If the selected microphone is
unplugged, renamed, or is a virtual device, dictation does not switch to another
input: Voculinux shows an "Audio Input Unavailable" notification and records
nothing. Choose a microphone again, or System Default, in Settings.
With **System Default**, Voculinux leaves valid default or virtual inputs under
the audio system's control rather than pinning a hardware device.
+90 -26
View File
@@ -17,7 +17,7 @@ import subprocess
import time
from typing import TYPE_CHECKING, Any, Callable, Optional, cast
from .errors import CaptureStalledError
from .errors import AudioDeviceUnavailableError, CaptureStalledError
from .pipewire import (
PIPEWIRE_INDEX_BASE,
_test_pipewire_input,
@@ -427,7 +427,7 @@ def _resolve_valid_input_device(audio: Any, preferred_index: Optional[int] = Non
device_name = "unknown"
logger.warning(
"Configured audio device [%s] (%s) is not a safe input device. "
"Falling back to a valid input device.",
"Choosing another input device.",
preferred_index,
device_name,
)
@@ -438,6 +438,86 @@ def _resolve_valid_input_device(audio: Any, preferred_index: Optional[int] = Non
return input_device_indices[0]
def _selects_system_default(device_index: Optional[int], device_name: Optional[str]) -> bool:
"""Whether a stored selection means System Default.
Settings stores System Default as -1 and clears both fields; an unnamed
index of -1 (PortAudio's own default sentinel) is the same choice.
"""
return not device_name and (device_index is None or device_index == -1)
def _configured_label(device_name: Optional[str], device_index: Optional[int]) -> str:
"""Name a configured selection in user-facing messages."""
label = device_name or f"index {device_index}"
return f"Configured audio device '{label}'"
def _require_usable_configured_input(
audio: Any, device_name: Optional[str], device_index: Optional[int]
) -> None:
"""Refuse an explicit microphone that cannot be used exactly as configured.
The device must be found by name (or by index when no name was stored) and
must be a real input. Matching follows :func:`_resolve_device_by_name`, so
a device accepted here is the one it returns. Returns quietly when device
enumeration itself fails, leaving the stored index to PortAudio as before.
Raises:
AudioDeviceUnavailableError: the device is missing, renamed, virtual or
output-only. Nothing is opened, so no other input is recorded.
"""
label = _configured_label(device_name, device_index)
try:
device_count = int(audio.get_device_count())
except (IOError, OSError, TypeError, ValueError, AttributeError):
return
match: Optional[tuple[int, Any]] = None
for index in range(max(device_count, 0)):
try:
info = audio.get_device_info_by_index(index)
except (IOError, OSError, TypeError, ValueError, AttributeError):
continue
if device_name:
if isinstance(info, dict) and info.get("name", "") == device_name:
match = (index, info)
break
elif index == device_index:
match = (index, info)
break
if match is None:
raise AudioDeviceUnavailableError(
f"{label} is not connected or was renamed; nothing was recorded from another input."
)
_, info = match
if not isinstance(info, dict):
return # Cannot classify it; keep the same leniency as _resolve_valid_input_device.
name = str(info.get("name", ""))
channels = info.get("maxInputChannels", 0)
has_input = isinstance(channels, (int, float)) and channels > 0
if _is_virtual_device(name) or not has_input:
raise AudioDeviceUnavailableError(
f"{label} is not a usable microphone (virtual or output-only); "
"nothing was recorded from another input."
)
def _resolve_configured_input(
audio: Any, device_name: Optional[str], device_index: Optional[int]
) -> int:
"""Resolve an explicit microphone selection, or raise. Never substitutes a device."""
_require_usable_configured_input(audio, device_name, device_index)
resolved = _resolve_device_by_name(audio, device_name, device_index)
if resolved is None:
raise AudioDeviceUnavailableError(
f"{_configured_label(device_name, device_index)} "
"could not be opened; nothing was recorded from another input."
)
return resolved
def _resolve_system_default(audio: Any) -> Optional[int]:
"""Prefer ALSA's sound-server PCM when PortAudio substituted hardware.
@@ -971,25 +1051,16 @@ class PortAudioCaptureSource:
# PulseAudio source opens the default device with PULSE_SOURCE set.
pulse_selected, pulse_source = _resolve_pulse_selection(self.device_index, self.device_name)
pulse_name = pulse_source.name if pulse_source is not None else None
use_system_default = pulse_selected or (
self.device_index is None and self.device_name is None
use_system_default = pulse_selected or _selects_system_default(
self.device_index, self.device_name
)
if use_system_default:
resolved_device_index = _resolve_system_default(audio)
else:
resolved_device_index = _resolve_device_by_name(
# An explicit selection is used exactly or refused (issue #42).
resolved_device_index = _resolve_configured_input(
audio, self.device_name, self.device_index
)
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 = _resolve_system_default(audio)
use_system_default = True
# Log available devices for debugging (skip virtual devices)
logger.debug("Available audio input devices:")
@@ -1117,23 +1188,16 @@ class PortAudioCaptureSource:
self.device_index, self.device_name
)
pulse_name = pulse_source.name if pulse_source is not None else None
use_system_default = pulse_selected or (
self.device_index is None and self.device_name is None
use_system_default = pulse_selected or _selects_system_default(
self.device_index, self.device_name
)
if use_system_default:
resolved_device_index = _resolve_system_default(audio_instance)
else:
resolved_device_index = _resolve_device_by_name(
# An explicit selection is used exactly or refused (issue #42).
resolved_device_index = _resolve_configured_input(
audio_instance, self.device_name, self.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 = _resolve_system_default(audio_instance)
# Negotiate the format and open the capture stream in ONE
# PortAudio open — Bluetooth SCO devices abort with heap
+9
View File
@@ -3,3 +3,12 @@
class CaptureStalledError(IOError):
"""The capture device stopped delivering audio frames."""
class AudioDeviceUnavailableError(IOError):
"""The configured capture device cannot be used as configured.
Raised instead of recording from another input: a configured device that
is missing, renamed, virtual or output-only must fail visibly (issue #42).
The message is user-facing and names the configured device.
"""
+8 -3
View File
@@ -28,7 +28,7 @@ from dataclasses import dataclass
from typing import Any, Callable, Dict, List, Mapping, Optional, Tuple
from ..utils.host_process import host_env
from .errors import CaptureStalledError
from .errors import AudioDeviceUnavailableError, CaptureStalledError
logger = logging.getLogger(__name__)
@@ -368,7 +368,8 @@ class PipeWireCaptureSource:
Raises:
FileNotFoundError: when ``pw-record`` is not installed.
IOError: when the stored selection no longer resolves to a source.
AudioDeviceUnavailableError: when the stored selection no longer
resolves to a source (an IOError subclass).
"""
if self.stream is not None:
return
@@ -378,7 +379,11 @@ class PipeWireCaptureSource:
if source is None:
# Deliberately no PortAudio fallback: silently switching to a
# microphone would capture the wrong audio.
raise IOError("PipeWire source is no longer available")
label = self.device_name or f"index {self.device_index}"
raise AudioDeviceUnavailableError(
f"System audio source '{label}' is no longer available; "
"nothing was recorded from another input."
)
self.source = source
self.stream = self._popen(
[
@@ -30,6 +30,7 @@ if TYPE_CHECKING:
from ..app_identity import APP_NAME, RESOURCE_DIR_NAME, user_agent
from ..audio.capture import ( # noqa: F401 (re-exported for compatibility)
_STICKY_LOCK_MIN_MEAN_SQUARE,
AudioDeviceUnavailableError,
CaptureStalledError,
PortAudioCaptureSource,
_alsa_handler,
@@ -112,6 +113,9 @@ class _CaptureSession:
self.stopping = False
self.stall_recovery_attempted = False
self.stall_error: Optional[OSError] = None
# A configured device that cannot be used (issue #42), kept for the
# failure notice when the read error that ended the session is not it.
self.unavailable_error: Optional[AudioDeviceUnavailableError] = None
self.device_name = "System Default"
self.segments: queue.Queue = queue.Queue(maxsize=32)
# A dispatched callback can outlive the bounded recognition join.
@@ -3955,6 +3959,10 @@ class SpeechRecognitionManager:
def recover(error: OSError) -> bool:
if not recording():
return False
if isinstance(error, AudioDeviceUnavailableError):
# A configured device that cannot be used does not recover by retrying.
session.unavailable_error = session.unavailable_error or error
return False
if session.stall_error is None:
session.device_name = self._capture_device_name(source)
if isinstance(error, CaptureStalledError):
@@ -3965,6 +3973,8 @@ class SpeechRecognitionManager:
session.stall_recovery_attempted = True
recovered = self._attempt_audio_reconnection(audio, should_continue=recording)
validation_error = getattr(self._capture_source, "last_error", None)
if isinstance(validation_error, AudioDeviceUnavailableError):
session.unavailable_error = session.unavailable_error or validation_error
if isinstance(validation_error, CaptureStalledError):
session.stall_error = session.stall_error or validation_error
session.stall_recovery_attempted = True
@@ -4324,7 +4334,17 @@ class SpeechRecognitionManager:
play_error_sound()
if isinstance(error, CaptureStalledError):
session.stall_error = session.stall_error or error
if session.stall_error is not None:
unavailable = session.unavailable_error
if unavailable is None and isinstance(error, AudioDeviceUnavailableError):
unavailable = error
if unavailable is not None:
logger.error("Audio input unavailable: %s", unavailable)
_show_notification(
"Audio Input Unavailable",
f"{unavailable} Choose a microphone in Settings, then try dictation again.",
"dialog-warning",
)
elif session.stall_error is not None:
logger.error("Microphone '%s' stalled: %s", session.device_name, session.stall_error)
_show_notification(
"Microphone Stalled",
+76
View File
@@ -385,3 +385,79 @@ def test_successful_reopen_updates_identity_for_a_later_stall(
for stream in streams:
stream.close.assert_called_once()
audio.terminate.assert_called_once()
def test_unusable_configured_device_is_reported_without_retrying(
manager: rm.SpeechRecognitionManager,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""A configured virtual mic fails visibly; it is not reopened or replaced (#42)."""
source = capture.PortAudioCaptureSource(device_index=0, device_name="pipewire")
devices = [
{"index": 0, "name": "pipewire", "maxInputChannels": 32},
{"index": 1, "name": "Built-in Mic", "maxInputChannels": 2},
]
audio = Mock()
audio.get_device_count.return_value = 2
audio.get_device_info_by_index.side_effect = lambda i: devices[i]
audio.get_default_input_device_info.return_value = {"index": 1, "name": "Built-in Mic"}
monkeypatch.setitem(sys.modules, "pyaudio", Mock(PyAudio=lambda: audio, paInt16=8))
monkeypatch.setattr(manager, "_new_capture_source", lambda: source)
opener = Mock(return_value=(1, 16000, Mock()))
monkeypatch.setattr(capture, "_open_capture_stream", opener)
reconnect = Mock(return_value=False)
monkeypatch.setattr(manager, "_attempt_audio_reconnection", reconnect)
manager.should_record = True
manager.state = RecognitionState.LISTENING
manager._record_audio()
opener.assert_not_called()
audio.open.assert_not_called()
reconnect.assert_not_called()
title, message = rm._show_notification.call_args.args[:2]
assert title == "Audio Input Unavailable"
assert "'pipewire'" in message
assert "Choose a microphone in Settings" in message
rm.play_error_sound.assert_called()
assert manager.state == RecognitionState.IDLE
def test_configured_device_lost_mid_dictation_is_reported(
manager: rm.SpeechRecognitionManager,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""A read error followed by a refused reconnect still reports the cause (#42)."""
source = capture.PortAudioCaptureSource(device_index=0, device_name="USB Microphone")
devices = {
0: {"index": 0, "name": "USB Microphone", "maxInputChannels": 1},
1: {"index": 1, "name": "Built-in Mic", "maxInputChannels": 2},
}
audio = Mock()
audio.get_device_count.return_value = 2
audio.get_device_info_by_index.side_effect = lambda i: devices[i]
audio.get_default_input_device_info.return_value = {"index": 1, "name": "Built-in Mic"}
monkeypatch.setitem(sys.modules, "pyaudio", Mock(PyAudio=lambda: audio, paInt16=8))
monkeypatch.setattr(manager, "_new_capture_source", lambda: source)
opener = Mock(return_value=(1, 16000, Mock()))
monkeypatch.setattr(capture, "_open_capture_stream", opener)
def read(**_kwargs: object) -> bytes:
# The microphone is renamed while dictating: the reconnect must refuse it.
devices[0] = {"index": 0, "name": "Renamed USB", "maxInputChannels": 1}
raise OSError("device lost")
monkeypatch.setattr(source, "read_chunk", read)
# Reconnection rebuilds the source from the manager's configured device.
manager.audio_device_index = 0
manager.audio_device_name = "USB Microphone"
manager.should_record = True
manager.state = RecognitionState.LISTENING
manager._record_audio()
opener.assert_called_once()
title, message = rm._show_notification.call_args.args[:2]
assert title == "Audio Input Unavailable"
assert "'USB Microphone'" in message
assert manager.state == RecognitionState.IDLE
+9
View File
@@ -11,6 +11,7 @@ import unittest
from typing import Any, Callable, Dict, List, Optional, Tuple
from unittest.mock import MagicMock, patch
from vocalinux.audio.errors import AudioDeviceUnavailableError
from vocalinux.audio.pipewire import (
PIPEWIRE_INDEX_BASE,
SYSTEM_AUDIO_PREFIX,
@@ -398,6 +399,14 @@ class TestPipeWireCaptureSource(unittest.TestCase):
with self.assertRaises(IOError):
capture.open()
def test_open_reports_a_gone_source_as_unavailable(self) -> None:
# No PortAudio fallback: the typed error lets the manager tell the user.
with patch("vocalinux.audio.pipewire.resolve_pipewire_source", return_value=None):
capture = PipeWireCaptureSource(device_index=PIPEWIRE_INDEX_BASE, which=lambda n: "x")
with self.assertRaises(AudioDeviceUnavailableError) as ctx:
capture.open()
assert "no longer available" in str(ctx.exception)
def test_read_chunk_returns_exact_frame_bytes(self) -> None:
payload = b"\x01\x02" * 4096
process = _FakeProcess(payload)
+13 -6
View File
@@ -389,13 +389,13 @@ class TestPulseSourceEnv:
assert leaked is None
def _pyaudio_with_one_mic() -> Tuple[MagicMock, MagicMock]:
def _pyaudio_with_one_mic(name: str = "default") -> Tuple[MagicMock, MagicMock]:
mock_audio = MagicMock()
mock_audio.get_device_count.return_value = 1
mock_audio.get_default_input_device_info.return_value = {"index": 0, "name": "default"}
mock_audio.get_default_input_device_info.return_value = {"index": 0, "name": name}
mock_audio.get_device_info_by_index.return_value = {
"index": 0,
"name": "default",
"name": name,
"maxInputChannels": 32,
}
return mock_audio, MagicMock(PyAudio=MagicMock(return_value=mock_audio))
@@ -434,9 +434,13 @@ class TestCaptureIntegration:
assert devices == [(PULSE_INDEX_BASE, "Yeti", False)]
def _open_recording_env(
self, device_index: int, device_name: str, sources: List[PulseSource]
self,
device_index: int,
device_name: str,
sources: List[PulseSource],
mic_name: str = "default",
) -> Tuple[Dict[str, object], Optional[str]]:
mock_audio, mock_pyaudio = _pyaudio_with_one_mic()
mock_audio, mock_pyaudio = _pyaudio_with_one_mic(mic_name)
seen: Dict[str, object] = {}
def fake_open(audio: Any, index: Optional[int]) -> Tuple[int, int, MagicMock]:
@@ -467,7 +471,10 @@ class TestCaptureIntegration:
assert leaked is None
def test_portaudio_selection_is_untouched(self) -> None:
seen, _leaked = self._open_recording_env(0, "default", self.sources)
# A real configured microphone; an explicit virtual "default" is now refused (#42).
seen, _leaked = self._open_recording_env(
0, "USB Microphone", self.sources, mic_name="USB Microphone"
)
assert seen["env"] is None
def test_audio_test_uses_the_chosen_source(self) -> None:
+120 -21
View File
@@ -160,8 +160,8 @@ class TestAudioDeviceDetection(unittest.TestCase):
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."""
def test_record_audio_refuses_system_default_for_unusable_configured_device(self):
"""An unusable configured mic must not reopen the system default (issue #42)."""
manager = _make_manager(
engine="whisper_cpp",
audio_device_index=0,
@@ -172,14 +172,6 @@ class TestAudioDeviceDetection(unittest.TestCase):
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.get_read_available.return_value = 1024
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 = {
@@ -196,17 +188,9 @@ class TestAudioDeviceDetection(unittest.TestCase):
with (
patch.dict("sys.modules", {"pyaudio": mock_pyaudio}),
patch(
"vocalinux.audio.capture._resolve_device_by_name",
return_value=None,
),
patch(
"vocalinux.audio.capture._resolve_valid_input_device",
return_value=None,
),
patch(
"vocalinux.audio.capture._open_capture_stream",
return_value=(1, 16000, mock_stream),
return_value=(1, 16000, MagicMock()),
) as mock_open,
patch("vocalinux.speech_recognition.recognition_manager.play_error_sound"),
):
@@ -214,8 +198,9 @@ class TestAudioDeviceDetection(unittest.TestCase):
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()
mock_open.assert_not_called()
mock_audio.open.assert_not_called()
assert manager._buffered_capture_failed
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."""
@@ -946,6 +931,120 @@ class TestDownmixToMono(unittest.TestCase):
assert sticky3 == 2
class TestExplicitMicrophoneSelection(unittest.TestCase):
"""A configured microphone is recorded exactly, or not at all (issue #42)."""
def _audio(self, devices):
audio = MagicMock()
audio.get_device_count.return_value = len(devices)
audio.get_device_info_by_index.side_effect = lambda i: devices[i]
audio.get_default_input_device_info.return_value = {"index": 2, "name": "Built-in Mic"}
return audio
def _pyaudio_module(self):
module = MagicMock(paInt16=8)
module.PyAudio.return_value = MagicMock()
return module
def _devices(self):
return [
{"index": 0, "name": "pipewire", "maxInputChannels": 32},
{"index": 1, "name": "USB Microphone", "maxInputChannels": 1},
{"index": 2, "name": "Built-in Mic", "maxInputChannels": 2},
]
def test_virtual_configured_device_is_refused_not_replaced(self):
"""A configured virtual device must not open the hardware default instead."""
from vocalinux.audio.capture import PortAudioCaptureSource
from vocalinux.audio.errors import AudioDeviceUnavailableError
audio = self._audio(self._devices())
source = PortAudioCaptureSource(device_index=0, device_name="pipewire")
with (
patch.dict("sys.modules", {"pyaudio": self._pyaudio_module()}),
patch(
"vocalinux.audio.capture._open_capture_stream",
return_value=(1, 16000, MagicMock()),
) as opener,
):
with self.assertRaises(AudioDeviceUnavailableError) as ctx:
source.open(audio)
opener.assert_not_called()
audio.open.assert_not_called()
assert "'pipewire'" in str(ctx.exception)
def test_configured_device_that_disappeared_is_not_switched(self):
"""A missing configured mic must not reuse the stored index for another input."""
from vocalinux.audio.capture import PortAudioCaptureSource
from vocalinux.audio.errors import AudioDeviceUnavailableError
devices = self._devices()
devices[1] = {"index": 1, "name": "Built-in Mic", "maxInputChannels": 2}
devices[2] = {"index": 2, "name": "Other Mic", "maxInputChannels": 1}
audio = self._audio(devices)
source = PortAudioCaptureSource(device_index=1, device_name="USB Microphone")
with (
patch.dict("sys.modules", {"pyaudio": self._pyaudio_module()}),
patch(
"vocalinux.audio.capture._open_capture_stream",
return_value=(1, 16000, MagicMock()),
) as opener,
):
with self.assertRaises(AudioDeviceUnavailableError):
source.open(audio)
opener.assert_not_called()
def test_configured_usable_device_opens_by_name(self):
"""A configured microphone that is present still opens at its index."""
from vocalinux.audio.capture import PortAudioCaptureSource
audio = self._audio(self._devices())
source = PortAudioCaptureSource(device_index=1, device_name="USB Microphone")
with (
patch.dict("sys.modules", {"pyaudio": self._pyaudio_module()}),
patch(
"vocalinux.audio.capture._open_capture_stream",
return_value=(1, 16000, MagicMock()),
) as opener,
):
source.open(audio)
opener.assert_called_once_with(audio, 1)
def test_system_default_sentinel_still_opens_the_default_device(self):
"""A stored -1 is PortAudio's System Default, not a device to look up."""
from vocalinux.audio.capture import PortAudioCaptureSource
audio = self._audio(self._devices())
source = PortAudioCaptureSource(device_index=-1, device_name=None)
with (
patch.dict("sys.modules", {"pyaudio": self._pyaudio_module()}),
patch(
"vocalinux.audio.capture._open_capture_stream",
return_value=(1, 16000, MagicMock()),
) as opener,
):
source.open(audio)
opener.assert_called_once_with(audio, None)
def test_reopen_refuses_a_virtual_configured_device(self):
"""Reconnection reports the refusal through last_error and opens nothing."""
from vocalinux.audio.capture import PortAudioCaptureSource
from vocalinux.audio.errors import AudioDeviceUnavailableError
audio = self._audio(self._devices())
source = PortAudioCaptureSource(device_index=0, device_name="pipewire")
with (
patch.dict("sys.modules", {"pyaudio": self._pyaudio_module()}),
patch(
"vocalinux.audio.capture._open_capture_stream",
return_value=(1, 16000, MagicMock()),
) as opener,
):
assert source.reopen(audio) is False
opener.assert_not_called()
assert isinstance(source.last_error, AudioDeviceUnavailableError)
class TestRecordAudioNegotiationFallback(unittest.TestCase):
"""Cover _record_audio fallback when stream negotiation returns None."""
+42 -56
View File
@@ -1295,76 +1295,62 @@ 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."""
def test_attempt_audio_reconnection_refuses_missing_configured_device(self):
"""A configured mic that is gone must fail visibly, not reopen the default (#42)."""
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.get_read_available.return_value = 1024
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
mock_audio_instance.get_device_count.return_value = 1
mock_audio_instance.get_device_info_by_index.return_value = {
"index": 0,
"name": "Built-in Mic",
"maxInputChannels": 2,
}
mock_audio_instance.get_default_input_device_info.return_value = {
"index": 0,
"name": "Built-in Mic",
}
with (
patch.dict("sys.modules", {"pyaudio": mock_pyaudio_mod}),
patch("time.sleep"),
patch(
"vocalinux.audio.capture._resolve_device_by_name",
return_value=None,
) as mock_resolve_name,
patch(
"vocalinux.audio.capture._resolve_valid_input_device",
return_value=1,
) as mock_resolve_default,
patch(
"vocalinux.audio.capture._get_supported_channels",
return_value=1,
),
patch(
"vocalinux.audio.capture._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):
"""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()
mock_stream.get_read_available.return_value = 1024
with (
patch.dict("sys.modules", {"pyaudio": mock_pyaudio_mod}),
patch("time.sleep"),
patch(
"vocalinux.audio.capture._resolve_device_by_name",
return_value=None,
),
patch(
"vocalinux.audio.capture._resolve_valid_input_device",
return_value=None,
),
patch(
"vocalinux.audio.capture._open_capture_stream",
return_value=(1, 16000, mock_stream),
return_value=(1, 16000, MagicMock()),
) as mock_open,
):
result = manager._attempt_audio_reconnection(mock_audio_instance)
assert result is True
mock_open.assert_called_once_with(mock_audio_instance, None)
assert result is False
mock_open.assert_not_called()
mock_audio_instance.open.assert_not_called()
def test_attempt_audio_reconnection_refuses_when_no_safe_input_exists(self):
"""With no usable input enumerated, reconnection refuses rather than using the 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_audio_instance.get_device_count.return_value = 1
mock_audio_instance.get_device_info_by_index.return_value = {
"index": 0,
"name": "pipewire",
"maxInputChannels": 32,
}
with (
patch.dict("sys.modules", {"pyaudio": mock_pyaudio_mod}),
patch("time.sleep"),
patch(
"vocalinux.audio.capture._open_capture_stream",
return_value=(1, 16000, MagicMock()),
) as mock_open,
):
result = manager._attempt_audio_reconnection(mock_audio_instance)
assert result is False
mock_open.assert_not_called()
def test_attempt_audio_reconnection_max_attempts(self):
"""Test reconnection stops after max attempts."""