"""Tests for AudioRecorder."""
import pytest
import numpy as np
from unittest.mock import MagicMock
class TestAudioRecorder:
"""Tests for AudioRecorder."""
def test_init_defaults(self):
"""AudioRecorder initializes with default values."""
from src.localvoicemode.audio.recorder import AudioRecorder
recorder = AudioRecorder()
assert recorder.sample_rate == 16000
assert recorder.recording is False
assert recorder._vad_factory is None
def test_init_with_custom_sample_rate(self):
"""AudioRecorder respects custom sample rate."""
from src.localvoicemode.audio.recorder import AudioRecorder
recorder = AudioRecorder(sample_rate=48000)
assert recorder.sample_rate == 48000
def test_init_with_vad_factory(self):
"""AudioRecorder accepts vad_factory for dependency injection."""
from src.localvoicemode.audio.recorder import AudioRecorder
mock_vad = MagicMock()
factory = lambda: mock_vad
recorder = AudioRecorder(vad_factory=factory)
assert recorder._vad_factory is factory
def test_level_callback(self):
"""AudioRecorder calls level callback on update."""
from src.localvoicemode.audio.recorder import AudioRecorder
levels_received = []
def callback(level, history):
levels_received.append(level)
recorder = AudioRecorder()
recorder.set_level_callback(callback)
recorder._update_level(0.5)
assert len(levels_received) == 1
assert levels_received[0] > 0
def test_update_level_normalizes(self):
"""AudioRecorder._update_level normalizes RMS values."""
from src.localvoicemode.audio.recorder import AudioRecorder
recorder = AudioRecorder()
# Low RMS
recorder._update_level(0.0)
assert recorder.current_level == 0.0
# Mid RMS
recorder._update_level(0.15)
assert recorder.current_level == 0.5 # 0.15 / 0.3 = 0.5
# High RMS (capped at 1.0)
recorder._update_level(0.6)
assert recorder.current_level == 1.0
def test_level_history_shifts(self):
"""AudioRecorder._update_level shifts history correctly."""
from src.localvoicemode.audio.recorder import AudioRecorder
recorder = AudioRecorder()
initial_history = recorder.level_history.copy()
recorder._update_level(0.3)
# History should have shifted left and added new value at end
assert len(recorder.level_history) == len(initial_history)
assert recorder.level_history[-1] == 1.0 # 0.3 / 0.3 = 1.0
def test_recording_state(self):
"""AudioRecorder tracks recording state correctly."""
from src.localvoicemode.audio.recorder import AudioRecorder
recorder = AudioRecorder()
assert recorder.recording is False
# After start
recorder.recording = True
assert recorder.recording is True
def test_audio_buffer_initialized_empty(self):
"""AudioRecorder starts with empty audio buffer."""
from src.localvoicemode.audio.recorder import AudioRecorder
recorder = AudioRecorder()
assert recorder.audio_buffer == []
class TestAudioRecorderVAD:
"""Tests for AudioRecorder VAD recording."""
def test_record_vad_uses_injected_factory(self, mock_sounddevice):
"""AudioRecorder._record_vad_smart_turn uses injected vad_factory."""
from src.localvoicemode.audio.recorder import AudioRecorder
# Create mock VAD that immediately signals turn complete
mock_silero = MagicMock()
mock_silero.predict.return_value = 0.6 # Above threshold = speech detected
mock_smart_turn = MagicMock()
mock_smart_turn.silero = mock_silero
mock_smart_turn.predict_endpoint.return_value = {
"prediction": 1, # Turn complete
"probability": 0.9,
}
factory_called = []
def factory():
factory_called.append(True)
return mock_smart_turn
recorder = AudioRecorder(vad_factory=factory)
# We can't fully test record_vad without mocking sounddevice.InputStream
# but we can verify the factory is stored
assert recorder._vad_factory is factory
def test_record_vad_returns_array(self, mock_sounddevice, mocker):
"""AudioRecorder.record_vad returns numpy array."""
from src.localvoicemode.audio.recorder import AudioRecorder
# Create mock VAD
mock_silero = MagicMock()
mock_silero.predict.side_effect = [
0.2, 0.2, 0.2, # No speech
0.8, 0.8, # Speech
0.2, 0.2, 0.2, 0.2, 0.2, 0.2, 0.2, 0.2, 0.2, 0.2,
0.2, 0.2, 0.2, 0.2, 0.2, 0.2, 0.2, 0.2, 0.2, 0.2, # Silence
]
mock_smart_turn = MagicMock()
mock_smart_turn.silero = mock_silero
mock_smart_turn.predict_endpoint.return_value = {
"prediction": 1,
"probability": 0.9,
}
recorder = AudioRecorder(vad_factory=lambda: mock_smart_turn)
# Mock InputStream
mock_stream = MagicMock()
mock_stream.read.return_value = (np.zeros(512, dtype=np.float32), False)
mock_stream.__enter__ = MagicMock(return_value=mock_stream)
mock_stream.__exit__ = MagicMock(return_value=False)
mock_stream.stop = MagicMock()
mock_stream.start = MagicMock()
mocker.patch("sounddevice.InputStream", return_value=mock_stream)
result = recorder.record_vad(max_duration=0.5)
assert isinstance(result, np.ndarray)
class TestAudioRecorderPTT:
"""Tests for AudioRecorder push-to-talk mode."""
def test_start_recording_sets_state(self, mock_sounddevice, mocker):
"""AudioRecorder.start_recording sets recording state."""
from src.localvoicemode.audio.recorder import AudioRecorder
# Mock InputStream
mock_stream = MagicMock()
mock_stream.start = MagicMock()
mocker.patch("sounddevice.InputStream", return_value=mock_stream)
recorder = AudioRecorder()
assert recorder.recording is False
recorder.start_recording()
assert recorder.recording is True
assert recorder.audio_buffer == []
def test_stop_recording_returns_audio(self, mock_sounddevice, mocker):
"""AudioRecorder.stop_recording returns captured audio."""
from src.localvoicemode.audio.recorder import AudioRecorder
# Mock InputStream
mock_stream = MagicMock()
mock_stream.start = MagicMock()
mock_stream.stop = MagicMock()
mock_stream.close = MagicMock()
mocker.patch("sounddevice.InputStream", return_value=mock_stream)
recorder = AudioRecorder()
recorder.start_recording()
# Simulate some audio being captured
recorder.audio_buffer = [
np.ones(1600, dtype=np.float32),
np.ones(1600, dtype=np.float32),
]
result = recorder.stop_recording()
assert isinstance(result, np.ndarray)
assert len(result) == 3200 # 1600 + 1600
assert recorder.recording is False
def test_stop_recording_empty_buffer(self, mock_sounddevice, mocker):
"""AudioRecorder.stop_recording handles empty buffer."""
from src.localvoicemode.audio.recorder import AudioRecorder
# Mock InputStream
mock_stream = MagicMock()
mock_stream.start = MagicMock()
mock_stream.stop = MagicMock()
mock_stream.close = MagicMock()
mocker.patch("sounddevice.InputStream", return_value=mock_stream)
recorder = AudioRecorder()
recorder.start_recording()
# Don't add anything to buffer
result = recorder.stop_recording()
assert isinstance(result, np.ndarray)
assert len(result) == 0
