Spaces:
Sleeping
Sleeping
File size: 4,830 Bytes
c1e08a0 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 |
from unittest.mock import Mock, patch
import numpy as np
import pyaudio
import pytest
from improvisation_lab.infrastructure.audio import DirectAudioProcessor
class TestMicInput:
@pytest.fixture
def init_module(self):
self.mic_input = DirectAudioProcessor(sample_rate=44100)
@pytest.mark.usefixtures("init_module")
@patch("pyaudio.PyAudio")
def test_start_recording(self, mock_pyaudio):
"""Test recording start functionality."""
self.mic_input.start_recording()
assert self.mic_input.is_recording
# Verify that the PyAudio settings are correct
mock_pyaudio.return_value.open.assert_called_once_with(
format=pyaudio.paFloat32,
channels=1,
rate=44100,
input=True,
stream_callback=self.mic_input._audio_callback,
)
@pytest.mark.usefixtures("init_module")
def test_start_recording_when_already_recording(self):
"""Test that starting recording when already recording raises RuntimeError."""
self.mic_input.is_recording = True
with pytest.raises(RuntimeError) as exc_info:
self.mic_input.start_recording()
assert str(exc_info.value) == "Recording is already in progress"
@pytest.mark.usefixtures("init_module")
def test_audio_callback(self):
"""Test that the audio callback is called with the correct data."""
# Create sample audio data that matches the buffer size
buffer_duration = 0.2
sample_rate = 44100
buffer_size = int(sample_rate * buffer_duration)
test_data = np.array([0.1] * buffer_size, dtype=np.float32)
test_bytes = test_data.tobytes()
# Create a mock callback
mock_callback = Mock()
self.mic_input._callback = mock_callback
# Call the audio callback
result = self.mic_input._audio_callback(test_bytes, len(test_data), {}, 0)
# Verify the callback was called with the correct data
# First element of call_args is the first argument of the callback function
np.testing.assert_array_almost_equal(mock_callback.call_args[0][0], test_data)
# pyaudio.paContinue is an integer constant representing the stream status
# 0: continue, 1: complete, 2: error
assert result == (test_bytes, pyaudio.paContinue)
@pytest.mark.usefixtures("init_module")
@patch("pyaudio.PyAudio")
def test_stop_recording(self, mock_pyaudio):
"""Test recording stop functionality."""
# First start recording to set up the stream
self.mic_input.start_recording()
# Now test stopping
self.mic_input.stop_recording()
# Verify recording state
assert not self.mic_input.is_recording
assert self.mic_input._stream is None
assert self.mic_input.audio is None
# Verify that stream methods were called
mock_stream = mock_pyaudio.return_value.open.return_value
mock_stream.stop_stream.assert_called_once()
mock_stream.close.assert_called_once()
mock_pyaudio.return_value.terminate.assert_called_once()
@pytest.mark.usefixtures("init_module")
def test_stop_recording_not_recording(self):
"""Test that stopping when not recording raises an error."""
with pytest.raises(RuntimeError, match="Recording is not in progress"):
self.mic_input.stop_recording()
@pytest.mark.usefixtures("init_module")
def test_append_to_buffer(self):
"""Test appending data to the buffer."""
initial_data = np.array([0.1, 0.2], dtype=np.float32)
new_data = np.array([0.3, 0.4], dtype=np.float32)
expected_data = np.array([0.1, 0.2, 0.3, 0.4], dtype=np.float32)
self.mic_input._buffer = initial_data
self.mic_input._append_to_buffer(new_data)
np.testing.assert_array_almost_equal(self.mic_input._buffer, expected_data)
@pytest.mark.usefixtures("init_module")
def test_process_buffer(self):
"""Test processing buffer when it reaches the desired size."""
# Setup buffer with more data than buffer_size
buffer_size = self.mic_input._buffer_size
test_data = np.array([0.1] * (buffer_size + 2), dtype=np.float32)
self.mic_input._buffer = test_data
# Setup mock callback
mock_callback = Mock()
self.mic_input._callback = mock_callback
# Process buffer
self.mic_input._process_buffer()
# Verify callback was called with correct data
np.testing.assert_array_almost_equal(
mock_callback.call_args[0][0], test_data[:buffer_size]
)
# Verify remaining data in buffer
np.testing.assert_array_almost_equal(
self.mic_input._buffer, test_data[buffer_size:]
)
|