From 6d73f5bfe6aec371847c129d476271fba0bd4998 Mon Sep 17 00:00:00 2001 From: CarolinePascal Date: Sat, 9 Aug 2025 01:09:17 +0200 Subject: [PATCH] test(Microphone): removing unittest.TestCase class architecture to add tests parametrization on multiprocessing/multithreading use --- tests/microphones/test_portaudio.py | 734 ++++++++++++++++------------ 1 file changed, 409 insertions(+), 325 deletions(-) diff --git a/tests/microphones/test_portaudio.py b/tests/microphones/test_portaudio.py index efb52e24d..b8c8a65b4 100644 --- a/tests/microphones/test_portaudio.py +++ b/tests/microphones/test_portaudio.py @@ -13,11 +13,11 @@ # limitations under the License. import os -import unittest from copy import deepcopy from pathlib import Path import numpy as np +import pytest from soundfile import read from lerobot.microphones.portaudio.configuration_portaudio import PortAudioMicrophoneConfig @@ -43,386 +43,470 @@ LEROBOT_USE_REAL_PORTAUDIO_MICROPHONE_TESTS = ( ) -class TestPortAudioMicrophoneConfiguration(unittest.TestCase): - """Test the PortAudioMicrophone configuration and initialization.""" - - def test_config_creation(self): - """Test creating a valid configuration.""" - config = PortAudioMicrophoneConfig(microphone_index=0, sample_rate=48000, channels=[1, 2]) - self.assertEqual(config.microphone_index, 0) - self.assertEqual(config.sample_rate, 48000) - self.assertEqual(config.channels, [1, 2]) - - def test_config_creation_missing_microphone_index(self): - """Test creating a configuration with missing microphone index.""" - with self.assertRaises(TypeError): - PortAudioMicrophoneConfig(sample_rate=48000, channels=[1, 2]) - - def test_config_creation_missing_sample_rate(self): - """Test creating a configuration with missing sample rate.""" - config = PortAudioMicrophoneConfig(microphone_index=0, channels=[1, 2]) - self.assertIsNone(config.sample_rate) - - def test_config_creation_missing_channels(self): - """Test creating a configuration with missing channels.""" - config = PortAudioMicrophoneConfig(microphone_index=0, sample_rate=48000) - self.assertIsNone(config.channels) +@pytest.fixture +def test_sdk(): + """Fixture to provide either real or fake SDK based on environment variable.""" + if LEROBOT_USE_REAL_PORTAUDIO_MICROPHONE_TESTS: + return SounddeviceSDKAdapter() + else: + return FakeSounddeviceSDKAdapter() -class TestPortAudioMicrophoneDeviceValidation(unittest.TestCase): - """Test device validation and configuration.""" +# Configuration Tests - def setUp(self): - if LEROBOT_USE_REAL_PORTAUDIO_MICROPHONE_TESTS: - self.test_sdk = SounddeviceSDKAdapter() - else: - self.test_sdk = FakeSounddeviceSDKAdapter() - def _create_config( - device: int | str | None = None, kind: str | None = None - ) -> PortAudioMicrophoneConfig: - device_info = self.test_sdk.query_devices(device, kind) - config = PortAudioMicrophoneConfig( - microphone_index=device_info["index"], - sample_rate=device_info["default_samplerate"], - channels=np.arange(device_info["max_input_channels"]) + 1, - ) - return config +def test_config_creation(): + """Test creating a valid configuration.""" + config = PortAudioMicrophoneConfig(microphone_index=0, sample_rate=48000, channels=[1, 2]) + assert config.microphone_index == 0 + assert config.sample_rate == 48000 + assert config.channels == [1, 2] - self._create_config = _create_config - self.default_config = self._create_config(kind="input") +def test_config_creation_missing_microphone_index(): + """Test creating a configuration with missing microphone index.""" + with pytest.raises(TypeError): + PortAudioMicrophoneConfig(sample_rate=48000, channels=[1, 2]) - def test_find_microphones(self): - microphones = PortAudioMicrophone.find_microphones(sounddevice_sdk=self.test_sdk) - for microphone in microphones: - self.assertIsInstance(microphone["index"], int) - self.assertIsInstance(microphone["name"], str) - self.assertIsInstance(microphone["sample_rate"], int) - self.assertIsInstance(microphone["channels"], np.ndarray) - self.assertGreater(len(microphone["channels"]), 0) +def test_config_creation_missing_sample_rate(): + """Test creating a configuration with missing sample rate.""" + config = PortAudioMicrophoneConfig(microphone_index=0, channels=[1, 2]) + assert config.sample_rate is None - def test_init_defaults(self): - microphone = PortAudioMicrophone(self.default_config, sounddevice_sdk=self.test_sdk) - device_info = self.test_sdk.query_devices(kind="input") - self.assertIsNotNone(microphone) - self.assertEqual(microphone.microphone_index, device_info["index"]) - self.assertEqual(microphone.sample_rate, device_info["default_samplerate"]) - np.testing.assert_array_equal(microphone.channels, np.arange(device_info["max_input_channels"]) + 1) - self.assertFalse(microphone.is_connected) - self.assertFalse(microphone.is_recording) +def test_config_creation_missing_channels(): + """Test creating a configuration with missing channels.""" + config = PortAudioMicrophoneConfig(microphone_index=0, sample_rate=48000) + assert config.channels is None - def test_connect_success(self): - microphone = PortAudioMicrophone(self.default_config, sounddevice_sdk=self.test_sdk) + +@pytest.fixture +def default_config(test_sdk): + """Fixture to provide a default configuration for input devices.""" + device_info = test_sdk.query_devices(kind="input") + return PortAudioMicrophoneConfig( + microphone_index=device_info["index"], + sample_rate=device_info["default_samplerate"], + channels=np.arange(device_info["max_input_channels"]) + 1, + ) + + +# Microphone Tests + + +def test_find_microphones(test_sdk): + """Test finding microphones.""" + microphones = PortAudioMicrophone.find_microphones(sounddevice_sdk=test_sdk) + + for microphone in microphones: + assert isinstance(microphone["index"], int) + assert isinstance(microphone["name"], str) + assert isinstance(microphone["sample_rate"], int) + assert isinstance(microphone["channels"], np.ndarray) + assert len(microphone["channels"]) > 0 + + +def test_init_defaults(default_config, test_sdk): + """Test microphone initialization with defaults.""" + microphone = PortAudioMicrophone(default_config, sounddevice_sdk=test_sdk) + + device_info = test_sdk.query_devices(kind="input") + assert microphone is not None + assert microphone.microphone_index == device_info["index"] + assert microphone.sample_rate == device_info["default_samplerate"] + np.testing.assert_array_equal(microphone.channels, np.arange(device_info["max_input_channels"]) + 1) + assert not microphone.is_connected + assert not microphone.is_recording + + +def test_connect_success(default_config, test_sdk): + """Test successful connection.""" + microphone = PortAudioMicrophone(default_config, sounddevice_sdk=test_sdk) + microphone.connect() + + assert microphone.is_connected + assert not microphone.is_recording + assert not microphone.is_writing + + +def test_connect_empty_config(default_config, test_sdk): + """Test connection with empty config values.""" + config = deepcopy(default_config) + config.sample_rate = None + config.channels = None + microphone = PortAudioMicrophone(config, sounddevice_sdk=test_sdk) + microphone.connect() + + device_info = test_sdk.query_devices(kind="input") + assert microphone.sample_rate == device_info["default_samplerate"] + np.testing.assert_array_equal(microphone.channels, np.arange(device_info["max_input_channels"]) + 1) + + +def test_connect_already_connected(default_config, test_sdk): + """Test connecting when already connected.""" + microphone = PortAudioMicrophone(default_config, sounddevice_sdk=test_sdk) + microphone.connect() + + with pytest.raises(DeviceAlreadyConnectedError): microphone.connect() - self.assertTrue(microphone.is_connected) - self.assertFalse(microphone.is_recording) - self.assertFalse(microphone.is_writing) - def test_connect_empty_config(self): - config = deepcopy(self.default_config) - config.sample_rate = None - config.channels = None - microphone = PortAudioMicrophone(config, sounddevice_sdk=self.test_sdk) +def test_connect_invalid_device(test_sdk): + """Test connecting with invalid device (output device).""" + device_info = test_sdk.query_devices(kind="output") + config = PortAudioMicrophoneConfig( + microphone_index=device_info["index"], + sample_rate=device_info["default_samplerate"], + channels=np.arange(device_info["max_input_channels"]) + 1, + ) + microphone = PortAudioMicrophone(config, sounddevice_sdk=test_sdk) + + with pytest.raises(RuntimeError): microphone.connect() - device_info = self.test_sdk.query_devices(kind="input") - self.assertEqual(microphone.sample_rate, device_info["default_samplerate"]) - np.testing.assert_array_equal(microphone.channels, np.arange(device_info["max_input_channels"]) + 1) - def test_connect_already_connected(self): - microphone = PortAudioMicrophone(self.default_config, sounddevice_sdk=self.test_sdk) +def test_connect_invalid_index(default_config, test_sdk): + """Test connecting with invalid device index.""" + config = deepcopy(default_config) + config.microphone_index = -1 + microphone = PortAudioMicrophone(config, sounddevice_sdk=test_sdk) + + with pytest.raises(RuntimeError): microphone.connect() - with self.assertRaises(DeviceAlreadyConnectedError): - microphone.connect() - def test_connect_invalid_device(self): - config = self._create_config(kind="output") - microphone = PortAudioMicrophone(config, sounddevice_sdk=self.test_sdk) +def test_connect_invalid_sample_rate(default_config, test_sdk): + """Test connecting with invalid sample rate.""" + config = deepcopy(default_config) + config.sample_rate = -1 + microphone = PortAudioMicrophone(config, sounddevice_sdk=test_sdk) - with self.assertRaises(RuntimeError): - microphone.connect() - - def test_connect_invalid_index(self): - config = deepcopy(self.default_config) - config.microphone_index = -1 - microphone = PortAudioMicrophone(config, sounddevice_sdk=self.test_sdk) - - with self.assertRaises(RuntimeError): - microphone.connect() - - def test_connect_invalid_sample_rate(self): - config = deepcopy(self.default_config) - config.sample_rate = -1 - microphone = PortAudioMicrophone(config, sounddevice_sdk=self.test_sdk) - - with self.assertRaises(RuntimeError): - microphone.connect() - - def test_connect_float_sample_rate(self): - config = deepcopy(self.default_config) - config.sample_rate = int(config.sample_rate) - 0.5 - microphone = PortAudioMicrophone(config, sounddevice_sdk=self.test_sdk) + with pytest.raises(RuntimeError): microphone.connect() - self.assertIsInstance(microphone.sample_rate, int) - self.assertEqual(microphone.sample_rate, int(config.sample_rate)) - def test_connect_lower_sample_rate(self): - config = deepcopy(self.default_config) - config.sample_rate = 1000 # Lowest possible sample rate - microphone = PortAudioMicrophone(config, sounddevice_sdk=self.test_sdk) +def test_connect_float_sample_rate(default_config, test_sdk): + """Test connecting with float sample rate.""" + config = deepcopy(default_config) + config.sample_rate = int(config.sample_rate) - 0.5 + microphone = PortAudioMicrophone(config, sounddevice_sdk=test_sdk) + microphone.connect() + assert isinstance(microphone.sample_rate, int) + assert microphone.sample_rate == int(config.sample_rate) + + +def test_connect_lower_sample_rate(default_config, test_sdk): + """Test connecting with lower sample rate.""" + config = deepcopy(default_config) + config.sample_rate = 1000 # Lowest possible sample rate + microphone = PortAudioMicrophone(config, sounddevice_sdk=test_sdk) + + microphone.connect() + assert microphone.sample_rate == 1000 + + +def test_connect_invalid_channels(default_config, test_sdk): + """Test connecting with invalid channels.""" + config = deepcopy(default_config) + config.channels = np.append(default_config.channels, -1) + microphone = PortAudioMicrophone(config, sounddevice_sdk=test_sdk) + + with pytest.raises(RuntimeError): microphone.connect() - self.assertEqual(microphone.sample_rate, 1000) - def test_connect_invalid_channels(self): - config = deepcopy(self.default_config) - config.channels = np.append(self.default_config.channels, -1) - microphone = PortAudioMicrophone(config, sounddevice_sdk=self.test_sdk) - with self.assertRaises(RuntimeError): - microphone.connect() +def test_disconnect_success(default_config, test_sdk): + """Test successful disconnection.""" + microphone = PortAudioMicrophone(default_config, sounddevice_sdk=test_sdk) + microphone.connect() + microphone.disconnect() - def test_disconnect_success(self): - microphone = PortAudioMicrophone(self.default_config, sounddevice_sdk=self.test_sdk) - microphone.connect() + assert not microphone.is_connected + assert not microphone.is_recording + assert not microphone.is_writing + + +def test_disconnect_not_connected(default_config, test_sdk): + """Test disconnecting when not connected.""" + microphone = PortAudioMicrophone(default_config, sounddevice_sdk=test_sdk) + + with pytest.raises(DeviceNotConnectedError): microphone.disconnect() - self.assertFalse(microphone.is_connected) - self.assertFalse(microphone.is_recording) - self.assertFalse(microphone.is_writing) - def test_disconnect_not_connected(self): - microphone = PortAudioMicrophone(self.default_config, sounddevice_sdk=self.test_sdk) +@pytest.mark.parametrize("multiprocessing", [True, False]) +def test_start_recording_success(default_config, test_sdk, multiprocessing): + """Test successful recording start.""" + microphone = PortAudioMicrophone(default_config, sounddevice_sdk=test_sdk) + microphone.connect() + microphone.start_recording(multiprocessing=multiprocessing) - with self.assertRaises(DeviceNotConnectedError): - microphone.disconnect() + assert microphone.is_recording + assert microphone.is_connected + assert not microphone.is_writing - def test_start_recording_success(self): - microphone = PortAudioMicrophone(self.default_config, sounddevice_sdk=self.test_sdk) - microphone.connect() - microphone.start_recording() - self.assertTrue(microphone.is_recording) - self.assertTrue(microphone.is_connected) - self.assertFalse(microphone.is_writing) +@pytest.mark.parametrize("multiprocessing", [True, False]) +def test_recording_not_connected(default_config, test_sdk, multiprocessing): + """Test starting recording when not connected.""" + microphone = PortAudioMicrophone(default_config, sounddevice_sdk=test_sdk) - def test_recoring_not_connected(self): - microphone = PortAudioMicrophone(self.default_config, sounddevice_sdk=self.test_sdk) + with pytest.raises(DeviceNotConnectedError): + microphone.start_recording(multiprocessing=multiprocessing) - with self.assertRaises(DeviceNotConnectedError): - microphone.start_recording() - def test_start_recording_already_recording(self): - microphone = PortAudioMicrophone(self.default_config, sounddevice_sdk=self.test_sdk) - microphone.connect() - microphone.start_recording() +@pytest.mark.parametrize("multiprocessing", [True, False]) +def test_start_recording_already_recording(default_config, test_sdk, multiprocessing): + """Test starting recording when already recording.""" + microphone = PortAudioMicrophone(default_config, sounddevice_sdk=test_sdk) + microphone.connect() + microphone.start_recording(multiprocessing=multiprocessing) - with self.assertRaises(DeviceAlreadyRecordingError): - microphone.start_recording() + with pytest.raises(DeviceAlreadyRecordingError): + microphone.start_recording(multiprocessing=multiprocessing) - def test_start_writing_success(self): - microphone = PortAudioMicrophone(self.default_config, sounddevice_sdk=self.test_sdk) - microphone.connect() - microphone.start_recording(output_file="test.wav") - self.assertTrue(microphone.is_recording) - self.assertTrue(microphone.is_connected) - self.assertTrue(microphone.is_writing) - self.assertTrue(Path("test.wav").exists()) +@pytest.mark.parametrize("multiprocessing", [True, False]) +def test_start_writing_success(tmp_path, default_config, test_sdk, multiprocessing): + """Test successful writing start.""" + microphone = PortAudioMicrophone(default_config, sounddevice_sdk=test_sdk) + microphone.connect() + microphone.start_recording(output_file=tmp_path / "test.wav", multiprocessing=multiprocessing) - def test_stop_recording_success(self): - microphone = PortAudioMicrophone(self.default_config, sounddevice_sdk=self.test_sdk) - microphone.connect() - microphone.start_recording() - busy_wait(RECORDING_DURATION) - microphone.stop_recording() + assert microphone.is_recording + assert microphone.is_connected + assert microphone.is_writing + assert (tmp_path / "test.wav").exists() - self.assertFalse(microphone.is_recording) - self.assertTrue(microphone.is_connected) - self.assertFalse(microphone.is_writing) - def test_stop_writing_success(self): - microphone = PortAudioMicrophone(self.default_config, sounddevice_sdk=self.test_sdk) - microphone.connect() - microphone.start_recording(output_file="test.wav") - busy_wait(RECORDING_DURATION) - microphone.stop_recording() +@pytest.mark.parametrize("multiprocessing", [True, False]) +def test_start_writing_file_already_exists_no_overwrite(tmp_path, default_config, test_sdk, multiprocessing): + """Test writing with file that already exists.""" + (tmp_path / "test.wav").touch() + microphone = PortAudioMicrophone(default_config, sounddevice_sdk=test_sdk) + microphone.connect() - self.assertFalse(microphone.is_recording) - self.assertTrue(microphone.is_connected) - self.assertFalse(microphone.is_writing) - self.assertTrue(Path("test.wav").exists()) - - def test_stop_recording_not_connected(self): - microphone = PortAudioMicrophone(self.default_config, sounddevice_sdk=self.test_sdk) - - with self.assertRaises(DeviceNotConnectedError): - microphone.stop_recording() - - def test_stop_recording_not_recording(self): - microphone = PortAudioMicrophone(self.default_config, sounddevice_sdk=self.test_sdk) - microphone.connect() - - with self.assertRaises(DeviceNotRecordingError): - microphone.stop_recording() - - def test_disconnect_while_recording(self): - microphone = PortAudioMicrophone(self.default_config, sounddevice_sdk=self.test_sdk) - microphone.connect() - microphone.start_recording() - busy_wait(RECORDING_DURATION) - microphone.disconnect() - - self.assertFalse(microphone.is_connected) - self.assertFalse(microphone.is_recording) - self.assertFalse(microphone.is_writing) - - def test_disconnect_while_writing(self): - microphone = PortAudioMicrophone(self.default_config, sounddevice_sdk=self.test_sdk) - microphone.connect() - microphone.start_recording(output_file="test.wav") - busy_wait(RECORDING_DURATION) - microphone.disconnect() - - self.assertFalse(microphone.is_connected) - self.assertFalse(microphone.is_recording) - self.assertFalse(microphone.is_writing) - self.assertTrue(Path("test.wav").exists()) - - def test_read_success(self): - microphone = PortAudioMicrophone(self.default_config, sounddevice_sdk=self.test_sdk) - microphone.connect() - microphone.start_recording() - - busy_wait(RECORDING_DURATION) - - data = microphone.read() - - device_info = self.test_sdk.query_devices(kind="input") - self.assertIsNotNone(data) - self.assertEqual(data.shape[1], len(self.default_config.channels)) - self.assertAlmostEqual( - data.shape[0], - RECORDING_DURATION * self.default_config.sample_rate, - delta=2 * self.default_config.sample_rate * device_info["default_low_input_latency"], + with pytest.raises(FileExistsError): + microphone.start_recording( + output_file=tmp_path / "test.wav", multiprocessing=multiprocessing, overwrite=False ) - def test_writing_success(self): - microphone = PortAudioMicrophone(self.default_config, sounddevice_sdk=self.test_sdk) - microphone.connect() - microphone.start_recording(output_file="test.wav") + (tmp_path / "test.wav").unlink() - busy_wait(RECORDING_DURATION) +@pytest.mark.parametrize("multiprocessing", [True, False]) +def test_stop_recording_success(default_config, test_sdk, multiprocessing): + """Test successful recording stop.""" + microphone = PortAudioMicrophone(default_config, sounddevice_sdk=test_sdk) + microphone.connect() + microphone.start_recording(multiprocessing=multiprocessing) + busy_wait(RECORDING_DURATION) + microphone.stop_recording() + + assert not microphone.is_recording + assert microphone.is_connected + assert not microphone.is_writing + + +@pytest.mark.parametrize("multiprocessing", [True, False]) +def test_stop_writing_success(tmp_path, default_config, test_sdk, multiprocessing): + """Test successful writing stop.""" + microphone = PortAudioMicrophone(default_config, sounddevice_sdk=test_sdk) + microphone.connect() + microphone.start_recording(output_file=tmp_path / "test.wav", multiprocessing=multiprocessing) + busy_wait(RECORDING_DURATION) + microphone.stop_recording() + + assert not microphone.is_recording + assert microphone.is_connected + assert not microphone.is_writing + assert (tmp_path / "test.wav").exists() + + +def test_stop_recording_not_connected(default_config, test_sdk): + """Test stopping recording when not connected.""" + microphone = PortAudioMicrophone(default_config, sounddevice_sdk=test_sdk) + + with pytest.raises(DeviceNotConnectedError): microphone.stop_recording() - data, samplerate = read("test.wav") - device_info = self.test_sdk.query_devices(kind="input") - self.assertEqual(samplerate, self.default_config.sample_rate) - self.assertEqual(data.shape[1], len(self.default_config.channels)) - self.assertAlmostEqual( - data.shape[0], - RECORDING_DURATION * self.default_config.sample_rate, - delta=2 * self.default_config.sample_rate * device_info["default_low_input_latency"], - ) +def test_stop_recording_not_recording(default_config, test_sdk): + """Test stopping recording when not recording.""" + microphone = PortAudioMicrophone(default_config, sounddevice_sdk=test_sdk) + microphone.connect() - def test_read_while_writing(self): - microphone = PortAudioMicrophone(self.default_config, sounddevice_sdk=self.test_sdk) - microphone.connect() - microphone.start_recording(output_file="test.wav") - - busy_wait(RECORDING_DURATION) - - read_data = microphone.read() + with pytest.raises(DeviceNotRecordingError): microphone.stop_recording() - writing_data, _ = read("test.wav") - device_info = self.test_sdk.query_devices(kind="input") - self.assertAlmostEqual( - writing_data.shape[0], - RECORDING_DURATION * self.default_config.sample_rate, - delta=2 * self.default_config.sample_rate * device_info["default_low_input_latency"], - ) - self.assertAlmostEqual( - read_data.shape[0], - RECORDING_DURATION * self.default_config.sample_rate, - delta=2 * self.default_config.sample_rate * device_info["default_low_input_latency"], - ) +@pytest.mark.parametrize("multiprocessing", [True, False]) +def test_disconnect_while_recording(default_config, test_sdk, multiprocessing): + """Test disconnecting while recording.""" + microphone = PortAudioMicrophone(default_config, sounddevice_sdk=test_sdk) + microphone.connect() + microphone.start_recording(multiprocessing=multiprocessing) + busy_wait(RECORDING_DURATION) + microphone.disconnect() - def test_async_start_recording(self): - microphones = { - "microphone_1": PortAudioMicrophone(self.default_config, sounddevice_sdk=self.test_sdk), - "microphone_2": PortAudioMicrophone(self.default_config, sounddevice_sdk=self.test_sdk), - } - for microphone in microphones.values(): - microphone.connect() - - async_microphones_start_recording(microphones) - - for microphone in microphones.values(): - self.assertTrue(microphone.is_recording) - self.assertTrue(microphone.is_connected) - self.assertFalse(microphone.is_writing) - - def test_async_start_writing(self): - microphones = { - "microphone_1": PortAudioMicrophone(self.default_config, sounddevice_sdk=self.test_sdk), - "microphone_2": PortAudioMicrophone(self.default_config, sounddevice_sdk=self.test_sdk), - } - for microphone in microphones.values(): - microphone.connect() - - async_microphones_start_recording(microphones, output_files=["test_1.wav", "test_2.wav"]) - - for microphone in microphones.values(): - self.assertTrue(microphone.is_recording) - self.assertTrue(microphone.is_connected) - self.assertTrue(microphone.is_writing) - self.assertTrue(Path("test_1.wav").exists()) - self.assertTrue(Path("test_2.wav").exists()) - - def test_async_stop_recording(self): - microphones = { - "microphone_1": PortAudioMicrophone(self.default_config, sounddevice_sdk=self.test_sdk), - "microphone_2": PortAudioMicrophone(self.default_config, sounddevice_sdk=self.test_sdk), - } - for microphone in microphones.values(): - microphone.connect() - - async_microphones_start_recording(microphones) - async_microphones_stop_recording(microphones) - - for microphone in microphones.values(): - self.assertFalse(microphone.is_recording) - self.assertTrue(microphone.is_connected) - self.assertFalse(microphone.is_writing) - - def test_async_stop_writing(self): - microphones = { - "microphone_1": PortAudioMicrophone(self.default_config, sounddevice_sdk=self.test_sdk), - "microphone_2": PortAudioMicrophone(self.default_config, sounddevice_sdk=self.test_sdk), - } - for microphone in microphones.values(): - microphone.connect() - - async_microphones_start_recording(microphones, output_files=["test_1.wav", "test_2.wav"]) - async_microphones_stop_recording(microphones) - - for microphone in microphones.values(): - self.assertFalse(microphone.is_recording) - self.assertTrue(microphone.is_connected) - self.assertFalse(microphone.is_writing) - self.assertTrue(Path("test_1.wav").exists()) - self.assertTrue(Path("test_2.wav").exists()) + assert not microphone.is_connected + assert not microphone.is_recording + assert not microphone.is_writing -if __name__ == "__main__": - unittest.main(argv=["first-arg-is-ignored"], exit=False) +@pytest.mark.parametrize("multiprocessing", [True, False]) +def test_disconnect_while_writing(tmp_path, default_config, test_sdk, multiprocessing): + """Test disconnecting while writing.""" + microphone = PortAudioMicrophone(default_config, sounddevice_sdk=test_sdk) + microphone.connect() + microphone.start_recording(output_file=tmp_path / "test.wav", multiprocessing=multiprocessing) + busy_wait(RECORDING_DURATION) + microphone.disconnect() + + assert not microphone.is_connected + assert not microphone.is_recording + assert not microphone.is_writing + assert Path("test.wav").exists() + + +@pytest.mark.parametrize("multiprocessing", [True, False]) +def test_read_success(default_config, test_sdk, multiprocessing): + """Test successful reading of audio data.""" + microphone = PortAudioMicrophone(default_config, sounddevice_sdk=test_sdk) + microphone.connect() + microphone.start_recording(multiprocessing=multiprocessing) + + busy_wait(RECORDING_DURATION) + + data = microphone.read() + + device_info = test_sdk.query_devices(kind="input") + assert data is not None + assert data.shape[1] == len(default_config.channels) + assert ( + abs(data.shape[0] - RECORDING_DURATION * default_config.sample_rate) + <= 2 * default_config.sample_rate * device_info["default_low_input_latency"] + ) + + +@pytest.mark.parametrize("multiprocessing", [True, False]) +def test_writing_success(tmp_path, default_config, test_sdk, multiprocessing): + """Test successful writing to file.""" + microphone = PortAudioMicrophone(default_config, sounddevice_sdk=test_sdk) + microphone.connect() + microphone.start_recording(output_file=tmp_path / "test.wav", multiprocessing=multiprocessing) + + busy_wait(RECORDING_DURATION) + + microphone.stop_recording() + + data, samplerate = read(tmp_path / "test.wav") + + device_info = test_sdk.query_devices(kind="input") + assert samplerate == default_config.sample_rate + assert data.shape[1] == len(default_config.channels) + assert ( + abs(data.shape[0] - RECORDING_DURATION * default_config.sample_rate) + <= 2 * default_config.sample_rate * device_info["default_low_input_latency"] + ) + + +@pytest.mark.parametrize("multiprocessing", [True, False]) +def test_read_while_writing(tmp_path, default_config, test_sdk, multiprocessing): + """Test reading while writing.""" + microphone = PortAudioMicrophone(default_config, sounddevice_sdk=test_sdk) + microphone.connect() + microphone.start_recording(output_file=tmp_path / "test.wav", multiprocessing=multiprocessing) + + busy_wait(RECORDING_DURATION) + + read_data = microphone.read() + microphone.stop_recording() + + writing_data, _ = read(tmp_path / "test.wav") + + device_info = test_sdk.query_devices(kind="input") + assert ( + abs(writing_data.shape[0] - RECORDING_DURATION * default_config.sample_rate) + <= 2 * default_config.sample_rate * device_info["default_low_input_latency"] + ) + assert ( + abs(read_data.shape[0] - RECORDING_DURATION * default_config.sample_rate) + <= 2 * default_config.sample_rate * device_info["default_low_input_latency"] + ) + + +def test_async_start_recording(default_config, test_sdk): + """Test async recording start.""" + microphones = { + "microphone_1": PortAudioMicrophone(default_config, sounddevice_sdk=test_sdk), + "microphone_2": PortAudioMicrophone(default_config, sounddevice_sdk=test_sdk), + } + for microphone in microphones.values(): + microphone.connect() + + async_microphones_start_recording(microphones) + + for microphone in microphones.values(): + assert microphone.is_recording + assert microphone.is_connected + assert not microphone.is_writing + + +def test_async_start_writing(default_config, test_sdk): + """Test async writing start.""" + microphones = { + "microphone_1": PortAudioMicrophone(default_config, sounddevice_sdk=test_sdk), + "microphone_2": PortAudioMicrophone(default_config, sounddevice_sdk=test_sdk), + } + for microphone in microphones.values(): + microphone.connect() + + async_microphones_start_recording(microphones, output_files=["test_1.wav", "test_2.wav"]) + + for microphone in microphones.values(): + assert microphone.is_recording + assert microphone.is_connected + assert microphone.is_writing + assert Path("test_1.wav").exists() + assert Path("test_2.wav").exists() + + +def test_async_stop_recording(default_config, test_sdk): + """Test async recording stop.""" + microphones = { + "microphone_1": PortAudioMicrophone(default_config, sounddevice_sdk=test_sdk), + "microphone_2": PortAudioMicrophone(default_config, sounddevice_sdk=test_sdk), + } + for microphone in microphones.values(): + microphone.connect() + + async_microphones_start_recording(microphones) + async_microphones_stop_recording(microphones) + + for microphone in microphones.values(): + assert not microphone.is_recording + assert microphone.is_connected + assert not microphone.is_writing + + +def test_async_stop_writing(default_config, test_sdk): + """Test async writing stop.""" + microphones = { + "microphone_1": PortAudioMicrophone(default_config, sounddevice_sdk=test_sdk), + "microphone_2": PortAudioMicrophone(default_config, sounddevice_sdk=test_sdk), + } + for microphone in microphones.values(): + microphone.connect() + + async_microphones_start_recording(microphones, output_files=["test_1.wav", "test_2.wav"]) + async_microphones_stop_recording(microphones) + + for microphone in microphones.values(): + assert not microphone.is_recording + assert microphone.is_connected + assert not microphone.is_writing + assert Path("test_1.wav").exists() + assert Path("test_2.wav").exists()