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

fix(audio): stop Bluetooth mic probing from corrupting the heap (#599)

Browse Source 
* fix(audio): prevent Bluetooth mic probe heap corruption

Open PortAudio streams only once when negotiating capture format, and
always stop_stream() before close(). Separate channel/rate probing was
aborting on Bluetooth SCO devices (e.g. FreeBuds) with malloc corruption.

Fixes #567

* refactor(audio): open capture stream once instead of probe-close-reopen

Adversarial review of the first fix found it still closed the probe
stream and immediately reopened the device — the same rapid open/close
pattern blamed for the Bluetooth SCO heap corruption, just fewer cycles.

_open_capture_stream() now returns the successfully negotiated stream
and all three callers (mic test, recording, reconnect) use it directly,
so each capture session performs exactly one PortAudio open. Also drops
the dead known_working_rate parameter.

* test(audio): cover negotiation fallback paths for codecov patch

Add tests for mic-test, recording, and reconnect fallback branches when
_open_capture_stream returns no stream, plus Bluetooth settle-on-failure
and empty-read reconnect cleanup. Brings patch diff coverage above 80%.
This commit is contained in:
Jatin K Malik
2026-07-29 02:49:55 +05:30
committed by GitHub
parent 8322306aac
commit ca24264c85
3 changed files with 550 additions and 149 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/vocalinux/speech_recognition/recognition_manager.py
+224 -148
View File
@@ -210,6 +210,57 @@ def _is_virtual_device(device_name: str) -> bool:
return any(pattern in name_lower for pattern in virtual_patterns)
def _is_bluetooth_device(device_name: Optional[str]) -> bool:
"""Return True if the device name looks like a Bluetooth headset/mic."""
if not device_name:
return False
name_lower = device_name.lower()
# Prefer explicit Bluetooth/BlueZ markers. Avoid bare "headset" which also
# matches many wired USB headsets that do not need SCO settle delays.
bluetooth_patterns = (
"bluetooth",
"bluez",
"hands-free",
"handsfree",
)
return any(pattern in name_lower for pattern in bluetooth_patterns)
def _safe_close_stream(stream) -> None:
"""Stop and close a PortAudio stream without raising.
Closing an active stream (especially Bluetooth SCO/HFP capture) without
stopping it first is a known trigger for PortAudio heap corruption and
process abort via malloc assertions.
"""
if stream is None:
return
try:
stop = getattr(stream, "stop_stream", None)
if callable(stop):
stop()
except Exception:
pass
try:
close = getattr(stream, "close", None)
if callable(close):
close()
except Exception:
pass
def _get_device_info_safe(audio, device_index: Optional[int] = None) -> dict:
"""Fetch PortAudio device info, returning {} on failure."""
try:
if device_index is not None:
info = audio.get_device_info_by_index(device_index)
else:
info = audio.get_default_input_device_info()
return info if isinstance(info, dict) else {}
except (IOError, OSError, TypeError, ValueError, AttributeError):
return {}
def get_audio_input_devices() -> list:
"""
Get a list of available audio input devices, excluding virtual devices.
@@ -357,52 +408,63 @@ def _resolve_valid_input_device(audio, preferred_index: Optional[int] = None) ->
return input_device_indices[0]
def _get_supported_channels(audio, device_index: Optional[int] = None) -> int:
def _open_capture_stream(audio, device_index: Optional[int] = None) -> tuple[int, int, object]:
"""
Detect the supported number of channels for the audio device.
Negotiate a working (channels, sample_rate) and return the opened stream.
Some audio devices (particularly professional audio interfaces and certain
onboard audio chips) only support specific channel configurations. This
function tests mono (1) and stereo (2) to find a working configuration.
Historically Vocalinux probed channels and sample rates separately, each
opening and closing PortAudio streams, then opened the real capture stream
on top — three open/close cycles in under a second. Bluetooth headset
capture (SCO/HFP / PipeWire "Bluetooth internal capture stream") is
especially sensitive to that pattern and can abort the process with malloc
heap corruption (see GitHub issue #567).
Pro-audio USB interfaces (MUPRO, Vocaster, etc.) often only support 48kHz
and will reject 16kHz probes. This function uses the device's default
sample rate first, then falls back to common rates.
This function opens PortAudio exactly once per capture session: candidate
formats are tried in order (device default rate first, mono before stereo,
stereo skipped entirely for mono-only devices) and the FIRST successfully
opened stream is returned to the caller for actual capture — never closed
and reopened.
Args:
audio: PyAudio instance
device_index: The device index to test (None for default)
device_index: The device index to open (None for default)
Returns:
int: Number of channels supported (1 or 2), defaults to 1
Tuple of (channels, sample_rate, stream). If no format works, returns
(1, 16000, None) and the caller may attempt its own fallback open.
"""
import pyaudio
FORMAT = pyaudio.paInt16
CHUNK = 1024
COMMON_RATES = [48000, 44100, 32000, 22050, 16000, 8000]
rates_to_try = []
try:
if device_index is not None:
device_info = audio.get_device_info_by_index(device_index)
else:
device_info = audio.get_default_input_device_info()
device_info = _get_device_info_safe(audio, device_index)
device_name = device_info.get("name")
rates_to_try: list[int] = []
default_rate = int(device_info.get("defaultSampleRate", 0))
if default_rate > 0:
rates_to_try.append(default_rate)
logger.debug(f"Device reports default sample rate: {default_rate}Hz")
except (IOError, OSError) as e:
logger.debug(f"Could not get device info for channel probing: {e}")
default_rate = int(device_info.get("defaultSampleRate", 0) or 0)
if default_rate > 0:
rates_to_try.append(default_rate)
logger.debug(f"Device reports default sample rate: {default_rate}Hz")
for rate in COMMON_RATES:
if rate not in rates_to_try:
rates_to_try.append(rate)
for channels in [1, 2]:
# Never probe stereo on a device that reports a single input channel
# (opening with more channels than supported is itself a known
# PortAudio/ALSA corruption trigger).
reported_channels = int(device_info.get("maxInputChannels", 0) or 0)
channel_options = [1] if reported_channels == 1 else [1, 2]
bluetooth = _is_bluetooth_device(device_name)
# Brief settle helps BlueZ/Pulse release SCO between failed open attempts.
settle_s = 0.15 if bluetooth else 0.0
for channels in channel_options:
for rate in rates_to_try:
stream = None
try:
stream_kwargs = {
"format": FORMAT,
@@ -414,20 +476,41 @@ def _get_supported_channels(audio, device_index: Optional[int] = None) -> int:
if device_index is not None:
stream_kwargs["input_device_index"] = device_index
test_stream = audio.open(**stream_kwargs)
test_stream.close()
logger.debug(f"Device supports {channels} channel(s) at {rate}Hz")
return channels
stream = audio.open(**stream_kwargs)
logger.debug(f"Opened capture stream: {channels} channel(s) at {rate}Hz")
return channels, rate, stream
except (IOError, OSError) as e:
_safe_close_stream(stream)
error_str = str(e).lower()
if "invalid number of channels" in error_str or "-9998" in error_str:
logger.debug(f"Device rejected {channels} channel(s) at {rate}Hz: {e}")
else:
logger.debug(f"Channel test failed at {rate}Hz: {e}")
continue
logger.debug(f"Capture open failed ({channels}ch @ {rate}Hz): {e}")
if settle_s:
time.sleep(settle_s)
logger.warning("Could not determine supported channel count, defaulting to 1")
return 1
logger.warning("Could not open capture stream, defaulting to 1ch/16000Hz")
return 1, 16000, None
def _get_supported_channels(audio, device_index: Optional[int] = None) -> int:
"""
Detect the supported number of channels for the audio device.
Production code should use :func:`_open_capture_stream`, which keeps the
negotiated stream open instead of closing and reopening it. This helper is
retained for standalone format queries.
Args:
audio: PyAudio instance
device_index: The device index to test (None for default)
Returns:
int: Number of channels supported (1 or 2), defaults to 1
"""
channels, _rate, stream = _open_capture_stream(audio, device_index)
_safe_close_stream(stream)
return channels
def _get_supported_sample_rate(audio, device_index: Optional[int], channels: int = 1) -> int:
@@ -438,6 +521,10 @@ def _get_supported_sample_rate(audio, device_index: Optional[int], channels: int
(e.g., 48kHz) and will fail with the default 16kHz. This function tests
common sample rates and returns the highest supported one.
Production code should use :func:`_open_capture_stream`, which negotiates
channels and rate with a single PortAudio open. This helper is retained
for standalone format queries.
Args:
audio: PyAudio instance
device_index: The device index to test
@@ -454,38 +541,21 @@ def _get_supported_sample_rate(audio, device_index: Optional[int], channels: int
# Common sample rates to try, ordered from highest to lowest quality
COMMON_RATES = [48000, 44100, 32000, 22050, 16000, 8000]
# First, try the device's default sample rate
try:
if device_index is not None:
device_info = audio.get_device_info_by_index(device_index)
else:
device_info = audio.get_default_input_device_info()
device_info = _get_device_info_safe(audio, device_index)
device_name = device_info.get("name")
bluetooth = _is_bluetooth_device(device_name)
settle_s = 0.15 if bluetooth else 0.0
default_rate = int(device_info.get("defaultSampleRate", 0))
if default_rate > 0 and default_rate in COMMON_RATES:
# Test if the default rate actually works
try:
stream_kwargs = {
"format": FORMAT,
"channels": channels,
"rate": default_rate,
"input": True,
"frames_per_buffer": CHUNK,
}
if device_index is not None:
stream_kwargs["input_device_index"] = device_index
test_stream = audio.open(**stream_kwargs)
test_stream.close()
logger.debug(f"Using device default sample rate: {default_rate}Hz")
return default_rate
except (IOError, OSError):
logger.debug(f"Device default rate {default_rate}Hz failed, trying common rates")
except (IOError, OSError) as e:
logger.debug(f"Could not get device default rate: {e}")
# Try common sample rates in order of preference
rates_to_try: list[int] = []
default_rate = int(device_info.get("defaultSampleRate", 0) or 0)
if default_rate > 0:
rates_to_try.append(default_rate)
for rate in COMMON_RATES:
if rate not in rates_to_try:
rates_to_try.append(rate)
for rate in rates_to_try:
test_stream = None
try:
stream_kwargs = {
"format": FORMAT,
@@ -498,11 +568,15 @@ def _get_supported_sample_rate(audio, device_index: Optional[int], channels: int
stream_kwargs["input_device_index"] = device_index
test_stream = audio.open(**stream_kwargs)
test_stream.close()
_safe_close_stream(test_stream)
test_stream = None
logger.debug(f"Found supported sample rate: {rate}Hz")
return rate
except (IOError, OSError):
continue
if settle_s:
time.sleep(settle_s)
finally:
_safe_close_stream(test_stream)
# Fallback to 16kHz if nothing works
logger.warning("Could not find supported sample rate, defaulting to 16000Hz")
@@ -561,31 +635,32 @@ def test_audio_input(device_index: int = None, duration: float = 1.0) -> dict:
audio.terminate()
return result
# Detect supported channel count first (some devices require stereo)
CHANNELS = _get_supported_channels(audio, device_index)
# 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)
logger.info(f"Using {CHANNELS} channel(s) for audio test")
# Detect supported sample rate for this device
RATE = _get_supported_sample_rate(audio, device_index, CHANNELS)
result["sample_rate"] = RATE
# Open stream
try:
stream_kwargs = {
"format": FORMAT,
"channels": CHANNELS,
"rate": RATE,
"input": True,
"frames_per_buffer": CHUNK,
}
if device_index is not None:
stream_kwargs["input_device_index"] = device_index
if stream is None:
# Negotiation failed; try one last plain open so the error
# message reflects the real failure.
try:
stream_kwargs = {
"format": FORMAT,
"channels": CHANNELS,
"rate": RATE,
"input": True,
"frames_per_buffer": CHUNK,
}
if device_index is not None:
stream_kwargs["input_device_index"] = device_index
stream = audio.open(**stream_kwargs)
except (IOError, OSError) as e:
result["error"] = f"Cannot open audio stream: {e}"
audio.terminate()
return result
stream = audio.open(**stream_kwargs)
except (IOError, OSError) as e:
result["error"] = f"Cannot open audio stream: {e}"
audio.terminate()
return result
# Record and analyze
all_amplitudes = []
@@ -601,8 +676,7 @@ def test_audio_input(device_index: int = None, duration: float = 1.0) -> dict:
result["error"] = f"Error reading audio: {e}"
break
stream.stop_stream()
stream.close()
_safe_close_stream(stream)
audio.terminate()
if all_amplitudes:
@@ -2460,32 +2534,14 @@ class SpeechRecognitionManager:
except (IOError, OSError):
continue
# Detect supported channel count first (some devices require stereo)
CHANNELS = _get_supported_channels(audio, resolved_device_index)
# 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, negotiated_stream = _open_capture_stream(audio, resolved_device_index)
logger.info(f"Using {CHANNELS} channel(s) for recording")
# Detect supported sample rate for the selected device
RATE = _get_supported_sample_rate(audio, resolved_device_index, CHANNELS)
self._capture_sample_rate = RATE
logger.info(f"Using sample rate: {RATE}Hz")
# Open microphone stream with optional device selection and reconnection logic
stream_kwargs = {
"format": FORMAT,
"channels": CHANNELS,
"rate": RATE,
"input": True,
"frames_per_buffer": CHUNK,
}
# Use the resolved device (skip if already system default)
try:
default_idx = audio.get_default_input_device_info().get("index")
except (IOError, OSError):
default_idx = None
if resolved_device_index != default_idx:
stream_kwargs["input_device_index"] = resolved_device_index
try:
device_info = audio.get_device_info_by_index(resolved_device_index)
logger.info(
@@ -2494,21 +2550,45 @@ class SpeechRecognitionManager:
except (IOError, OSError):
logger.warning(f"Could not get info for device index {resolved_device_index}")
try:
self._audio_stream = audio.open(**stream_kwargs)
stream = self._audio_stream
except (IOError, OSError) as e:
logger.error(f"Failed to open audio stream: {e}")
logger.error("This may indicate a problem with the audio device or permissions.")
if negotiated_stream is not None:
self._audio_stream = negotiated_stream
stream = negotiated_stream
else:
# Negotiation failed; fall back to a plain open so the
# existing reconnection/error path still applies.
stream_kwargs = {
"format": FORMAT,
"channels": CHANNELS,
"rate": RATE,
"input": True,
"frames_per_buffer": CHUNK,
}
# Attempt reconnection
if self._attempt_audio_reconnection(audio):
# Use the resolved device (skip if already system default)
try:
default_idx = audio.get_default_input_device_info().get("index")
except (IOError, OSError):
default_idx = None
if resolved_device_index != default_idx:
stream_kwargs["input_device_index"] = resolved_device_index
try:
self._audio_stream = audio.open(**stream_kwargs)
stream = self._audio_stream
else:
play_error_sound()
audio.terminate()
self._update_state(RecognitionState.ERROR)
return
except (IOError, OSError) as e:
logger.error(f"Failed to open audio stream: {e}")
logger.error(
"This may indicate a problem with the audio device or permissions."
)
# Attempt reconnection
if self._attempt_audio_reconnection(audio):
stream = self._audio_stream
else:
play_error_sound()
audio.terminate()
self._update_state(RecognitionState.ERROR)
return
logger.info("Audio recording started")
@@ -3089,33 +3169,33 @@ class SpeechRecognitionManager:
CHUNK = 1024
FORMAT = pyaudio.paInt16
# Detect supported channel count first (some devices require stereo)
CHANNELS = _get_supported_channels(audio_instance, resolved_device_index)
# 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, new_stream = _open_capture_stream(audio_instance, resolved_device_index)
logger.debug(f"Reconnecting with {CHANNELS} channel(s)")
# Detect supported sample rate for the device
RATE = _get_supported_sample_rate(audio_instance, resolved_device_index, CHANNELS)
self._capture_sample_rate = RATE
logger.debug(f"Reconnecting with sample rate: {RATE}Hz")
stream_kwargs = {
"format": FORMAT,
"channels": CHANNELS,
"rate": RATE,
"input": True,
"frames_per_buffer": CHUNK,
}
if new_stream is None:
# Negotiation failed; fall back to a plain open.
stream_kwargs = {
"format": FORMAT,
"channels": CHANNELS,
"rate": RATE,
"input": True,
"frames_per_buffer": CHUNK,
}
# Use resolved device (skip if already system default)
try:
default_idx = audio_instance.get_default_input_device_info().get("index")
except (IOError, OSError):
default_idx = None
if resolved_device_index != default_idx:
stream_kwargs["input_device_index"] = resolved_device_index
# Use resolved device (skip if already system default)
try:
default_idx = audio_instance.get_default_input_device_info().get("index")
except (IOError, OSError):
default_idx = None
if resolved_device_index != default_idx:
stream_kwargs["input_device_index"] = resolved_device_index
# Attempt to open new stream
new_stream = audio_instance.open(**stream_kwargs)
new_stream = audio_instance.open(**stream_kwargs)
# Test the stream by reading a small amount of data
test_data = new_stream.read(CHUNK, exception_on_overflow=False)
@@ -3126,11 +3206,7 @@ class SpeechRecognitionManager:
return True
else:
logger.error("Reconnected stream returned no data")
try:
new_stream.stop_stream()
new_stream.close()
except Exception:
pass
_safe_close_stream(new_stream)
return False
except (IOError, OSError) as e:
tests/test_recognition_manager_device_config.py
+282 -1
View File
@@ -21,7 +21,7 @@ import threading
import time
import unittest
from unittest import mock
from unittest.mock import MagicMock, Mock, patch
from unittest.mock import MagicMock, Mock, call, patch
import pytest
@@ -31,7 +31,11 @@ from vocalinux.speech_recognition.recognition_manager import (
_filter_non_speech,
_get_supported_channels,
_get_supported_sample_rate,
_is_bluetooth_device,
_open_capture_stream,
_safe_close_stream,
get_audio_input_devices,
test_audio_input,
)
@@ -48,6 +52,230 @@ def _make_manager(engine="whisper_cpp", **kw):
class TestAudioDeviceDetection(unittest.TestCase):
"""Test audio device enumeration functions."""
def test_is_bluetooth_device_detection(self):
"""Bluetooth headset/mic names should be recognized (Issue #567)."""
assert _is_bluetooth_device("Bluetooth internal capture stream for HUAWEI FreeBuds Pro 3")
assert _is_bluetooth_device("bluez_source.XX_XX_XX")
assert _is_bluetooth_device("Hands-Free AG Speech")
assert not _is_bluetooth_device("WH-1000XM5 Headset") # no bluetooth marker
assert not _is_bluetooth_device("USB Audio Device")
assert not _is_bluetooth_device(None)
assert not _is_bluetooth_device("")
def test_safe_close_stream_stops_before_close(self):
"""PortAudio streams must be stopped before close to avoid heap corruption."""
mock_stream = MagicMock()
_safe_close_stream(mock_stream)
mock_stream.stop_stream.assert_called_once()
mock_stream.close.assert_called_once()
# stop must happen before close
assert mock_stream.mock_calls.index(mock.call.stop_stream()) < mock_stream.mock_calls.index(
mock.call.close()
)
def test_safe_close_stream_tolerates_errors(self):
"""Cleanup must not raise even if stop/close fail."""
mock_stream = MagicMock()
mock_stream.stop_stream.side_effect = OSError("already stopped")
mock_stream.close.side_effect = OSError("already closed")
_safe_close_stream(mock_stream) # should not raise
_safe_close_stream(None) # should not raise
def test_open_capture_stream_returns_live_stream(self):
"""Negotiation must return the opened stream, never close-and-reopen."""
mock_audio = MagicMock()
mock_stream = MagicMock()
mock_audio.open.return_value = mock_stream
mock_audio.get_device_info_by_index.return_value = {
"name": "USB Mic",
"defaultSampleRate": 48000,
"maxInputChannels": 1,
}
mock_pyaudio = MagicMock(paInt16=8)
with patch.dict("sys.modules", {"pyaudio": mock_pyaudio}):
channels, rate, stream = _open_capture_stream(mock_audio, 0)
assert channels == 1
assert rate == 48000
assert stream is mock_stream
assert mock_audio.open.call_count == 1
# The negotiated stream is handed to the caller live.
mock_stream.stop_stream.assert_not_called()
mock_stream.close.assert_not_called()
def test_open_capture_stream_bluetooth_single_open(self):
"""Bluetooth SCO capture must be opened exactly once (Issue #567).
The previous channels-then-rate-then-real-open flow opened the SCO
capture stream three times in under a second, which aborts with
malloc heap corruption on devices like the HUAWEI FreeBuds Pro 3.
"""
mock_audio = MagicMock()
mock_stream = MagicMock()
mock_audio.open.return_value = mock_stream
mock_audio.get_device_info_by_index.return_value = {
"name": "Bluetooth internal capture stream for HUAWEI FreeBuds Pro 3",
"defaultSampleRate": 16000,
"maxInputChannels": 1,
}
mock_pyaudio = MagicMock(paInt16=8)
with patch.dict("sys.modules", {"pyaudio": mock_pyaudio}):
channels, rate, stream = _open_capture_stream(mock_audio, 14)
assert channels == 1
assert rate == 16000
assert stream is mock_stream
assert mock_audio.open.call_count == 1
def test_open_capture_stream_mono_device_never_probes_stereo(self):
"""Mono-only devices must never be opened with 2 channels.
Opening with more channels than the device supports is itself a
PortAudio/ALSA heap-corruption trigger.
"""
mock_audio = MagicMock()
mock_audio.open.side_effect = IOError("cannot open")
mock_audio.get_device_info_by_index.return_value = {
"name": "Bluetooth internal capture stream",
"defaultSampleRate": 16000,
"maxInputChannels": 1,
}
mock_pyaudio = MagicMock(paInt16=8)
with patch.dict("sys.modules", {"pyaudio": mock_pyaudio}):
with patch("vocalinux.speech_recognition.recognition_manager.time.sleep"):
channels, rate, stream = _open_capture_stream(mock_audio, 14)
assert stream is None
assert (channels, rate) == (1, 16000)
assert all(call.kwargs.get("channels") == 1 for call in mock_audio.open.call_args_list)
def test_open_capture_stream_failure_returns_none_stream(self):
"""Total negotiation failure returns (1, 16000, None) for caller fallback."""
mock_audio = MagicMock()
mock_audio.open.side_effect = IOError("no device")
mock_audio.get_device_info_by_index.side_effect = IOError("gone")
mock_pyaudio = MagicMock(paInt16=8)
with patch.dict("sys.modules", {"pyaudio": mock_pyaudio}):
channels, rate, stream = _open_capture_stream(mock_audio, 3)
assert (channels, rate, stream) == (1, 16000, None)
def test_test_audio_input_single_portaudio_open(self):
"""The mic test must perform exactly ONE PortAudio open (Issue #567)."""
from vocalinux.speech_recognition.recognition_manager import test_audio_input
mock_stream = MagicMock()
mock_stream.read.return_value = b"\x00\x01" * 1024
mock_audio = MagicMock()
mock_audio.open.return_value = mock_stream
mock_audio.get_device_info_by_index.return_value = {
"index": 14,
"name": "Bluetooth internal capture stream for HUAWEI FreeBuds Pro 3",
"defaultSampleRate": 16000,
"maxInputChannels": 1,
}
mock_pyaudio = MagicMock(paInt16=8)
mock_pyaudio.PyAudio.return_value = mock_audio
with patch.dict("sys.modules", {"pyaudio": mock_pyaudio}):
# Other test modules poison sys.modules["numpy"] with a MagicMock
# at import time; drop it so test_audio_input gets real numpy.
# patch.dict restores the original sys.modules afterwards.
if isinstance(sys.modules.get("numpy"), MagicMock):
del sys.modules["numpy"]
result = test_audio_input(device_index=14, duration=0.1)
assert result["success"] is True
assert mock_audio.open.call_count == 1
# The stream that was read from is the negotiated stream itself.
assert mock_stream.read.called
mock_stream.stop_stream.assert_called_once()
mock_stream.close.assert_called_once()
def test_test_audio_input_negotiation_fallback_open(self):
"""When negotiation returns no stream, mic test falls back to a plain open."""
mock_stream = MagicMock()
mock_stream.read.return_value = b"\x00\x01" * 1024
mock_audio = MagicMock()
mock_audio.open.return_value = mock_stream
mock_audio.get_device_info_by_index.return_value = {
"index": 14,
"name": "Bluetooth internal capture stream",
"defaultSampleRate": 16000,
"maxInputChannels": 1,
}
mock_pyaudio = MagicMock(paInt16=8)
mock_pyaudio.PyAudio.return_value = mock_audio
with patch.dict("sys.modules", {"pyaudio": mock_pyaudio}):
if isinstance(sys.modules.get("numpy"), MagicMock):
del sys.modules["numpy"]
with patch(
"vocalinux.speech_recognition.recognition_manager._open_capture_stream",
return_value=(1, 16000, None),
):
result = test_audio_input(device_index=14, duration=0.1)
assert result["success"] is True
mock_audio.open.assert_called_once()
def test_test_audio_input_negotiation_and_fallback_both_fail(self):
"""When negotiation and fallback open both fail, return a clear error."""
mock_audio = MagicMock()
mock_audio.open.side_effect = IOError("device busy")
mock_audio.get_device_info_by_index.return_value = {
"index": 14,
"name": "Bluetooth internal capture stream",
"defaultSampleRate": 16000,
"maxInputChannels": 1,
}
mock_pyaudio = MagicMock(paInt16=8)
mock_pyaudio.PyAudio.return_value = mock_audio
with patch.dict("sys.modules", {"pyaudio": mock_pyaudio}):
with patch(
"vocalinux.speech_recognition.recognition_manager._open_capture_stream",
return_value=(1, 16000, None),
):
result = test_audio_input(device_index=14, duration=0.1)
assert result["success"] is False
assert "Cannot open audio stream" in result["error"]
def test_get_supported_sample_rate_bluetooth_settle_on_failure(self):
"""Bluetooth devices should pause between failed sample-rate probes."""
mock_audio = MagicMock()
mock_audio.open.side_effect = IOError("busy")
mock_audio.get_device_info_by_index.return_value = {
"name": "Bluetooth internal capture stream",
"defaultSampleRate": 48000,
}
mock_pyaudio = MagicMock(paInt16=8)
with patch.dict("sys.modules", {"pyaudio": mock_pyaudio}):
with patch("vocalinux.speech_recognition.recognition_manager.time.sleep") as mock_sleep:
rate = _get_supported_sample_rate(mock_audio, 0, channels=1)
assert rate == 16000
assert mock_sleep.called
def test_open_capture_stream_logs_channel_rejection(self):
"""Invalid-channel errors during negotiation should be logged distinctly."""
mock_audio = MagicMock()
mock_audio.open.side_effect = IOError("[Errno -9998] Invalid number of channels")
mock_audio.get_device_info_by_index.return_value = {
"name": "USB Mic",
"defaultSampleRate": 48000,
"maxInputChannels": 2,
}
mock_pyaudio = MagicMock(paInt16=8)
with patch.dict("sys.modules", {"pyaudio": mock_pyaudio}):
channels, rate, stream = _open_capture_stream(mock_audio, 0)
assert (channels, rate, stream) == (1, 16000, None)
assert mock_audio.open.call_count >= 1
def test_get_supported_channels_mono_success(self):
"""Test mono channel support detection."""
mock_audio = MagicMock()
@@ -207,6 +435,59 @@ class TestAudioDeviceDetection(unittest.TestCase):
assert rate == 16000 # Default fallback
class TestRecordAudioNegotiationFallback(unittest.TestCase):
"""Cover _record_audio fallback when stream negotiation returns None."""
def test_record_audio_falls_back_when_negotiation_returns_none(self):
"""Negotiation failure must fall back to a plain PortAudio open."""
manager = _make_manager(engine="whisper_cpp")
manager.should_record = True
manager.state = RecognitionState.LISTENING
manager.silence_timeout = 0.05
manager._silero_vad = None
mock_stream = MagicMock()
def _read_once(*_args, **_kwargs):
manager.should_record = False
return b"\x00" * 2048
mock_stream.read.side_effect = _read_once
mock_audio = MagicMock()
mock_audio.get_device_count.return_value = 1
mock_audio.get_device_info_by_index.return_value = {
"index": 0,
"name": "test mic",
"maxInputChannels": 1,
}
mock_audio.get_default_input_device_info.return_value = {"index": 0}
mock_audio.open.return_value = mock_stream
mock_pyaudio = MagicMock(paInt16=8)
mock_pyaudio.PyAudio.return_value = mock_audio
with (
patch.dict("sys.modules", {"pyaudio": mock_pyaudio}),
patch(
"vocalinux.speech_recognition.recognition_manager._resolve_device_by_name",
return_value=0,
),
patch(
"vocalinux.speech_recognition.recognition_manager._open_capture_stream",
return_value=(1, 16000, None),
),
patch(
"vocalinux.speech_recognition.recognition_manager.play_error_sound",
),
):
if isinstance(sys.modules.get("numpy"), MagicMock):
del sys.modules["numpy"]
manager._record_audio()
mock_audio.open.assert_called_once()
class TestFilterNonSpeech(unittest.TestCase):
"""Test the _filter_non_speech function."""
tests/test_recognition_manager_downloads.py
+44
View File
@@ -481,6 +481,50 @@ class TestAudioReconnection:
assert second_delay > first_delay
assert second_delay == first_delay * 2
def test_attempt_audio_reconnection_negotiation_fallback(self):
"""When negotiation returns no stream, reconnect falls back to plain open."""
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}),
patch("time.sleep"),
patch(
"vocalinux.speech_recognition.recognition_manager._open_capture_stream",
return_value=(1, 16000, None),
),
):
result = manager._attempt_audio_reconnection(mock_audio_instance)
assert result is True
assert manager._audio_stream == mock_stream
mock_audio_instance.open.assert_called_once()
def test_attempt_audio_reconnection_empty_read_closes_stream(self):
"""A reconnected stream that returns no data must be closed safely."""
manager = _make_manager(engine="whisper_cpp")
mock_pyaudio_mod = MagicMock()
mock_pyaudio_mod.paInt16 = 8
mock_stream = MagicMock()
mock_stream.read.return_value = b""
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 False
mock_stream.stop_stream.assert_called_once()
mock_stream.close.assert_called_once()
class TestIBusEngineUtilities:
"""Test ibus_engine utility functions."""