159 lines
5 KiB
Python
159 lines
5 KiB
Python
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
import numpy as np
|
|
from unittest.mock import patch
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
SRC = ROOT / "src"
|
|
if str(SRC) not in sys.path:
|
|
sys.path.insert(0, str(SRC))
|
|
|
|
import recorder
|
|
from recorder import RecordResult, stop_recording
|
|
|
|
|
|
class _Stream:
|
|
def __init__(self, *, stop_exc: Exception | None = None, close_exc: Exception | None = None):
|
|
self.stop_exc = stop_exc
|
|
self.close_exc = close_exc
|
|
self.stop_calls = 0
|
|
self.close_calls = 0
|
|
|
|
def stop(self) -> None:
|
|
self.stop_calls += 1
|
|
if self.stop_exc is not None:
|
|
raise self.stop_exc
|
|
|
|
def close(self) -> None:
|
|
self.close_calls += 1
|
|
if self.close_exc is not None:
|
|
raise self.close_exc
|
|
|
|
|
|
class _InputStream:
|
|
def __init__(self, **kwargs):
|
|
self.kwargs = kwargs
|
|
self.started = False
|
|
|
|
def start(self) -> None:
|
|
self.started = True
|
|
|
|
def stop(self) -> None:
|
|
return
|
|
|
|
def close(self) -> None:
|
|
return
|
|
|
|
|
|
class _FakeSoundDevice:
|
|
def __init__(self):
|
|
self.default = type("Default", (), {"device": (0, None)})()
|
|
self.created_streams = []
|
|
self._devices = [
|
|
{"name": "Built-in Output", "max_input_channels": 0},
|
|
{"name": "Laptop Mic", "max_input_channels": 1},
|
|
{"name": "USB Interface", "max_input_channels": 2},
|
|
]
|
|
|
|
def query_devices(self):
|
|
return self._devices
|
|
|
|
def InputStream(self, **kwargs):
|
|
stream = _InputStream(**kwargs)
|
|
self.created_streams.append(stream)
|
|
return stream
|
|
|
|
|
|
class RecorderTests(unittest.TestCase):
|
|
def test_stop_recording_closes_stream_when_stop_raises(self):
|
|
stream = _Stream(stop_exc=RuntimeError("stop boom"))
|
|
record = RecordResult()
|
|
|
|
with self.assertRaisesRegex(RuntimeError, "stop boom"):
|
|
stop_recording(stream, record)
|
|
|
|
self.assertEqual(stream.stop_calls, 1)
|
|
self.assertEqual(stream.close_calls, 1)
|
|
|
|
def test_stop_recording_raises_close_error_when_stop_succeeds(self):
|
|
stream = _Stream(close_exc=RuntimeError("close boom"))
|
|
record = RecordResult()
|
|
|
|
with self.assertRaisesRegex(RuntimeError, "close boom"):
|
|
stop_recording(stream, record)
|
|
|
|
self.assertEqual(stream.stop_calls, 1)
|
|
self.assertEqual(stream.close_calls, 1)
|
|
|
|
def test_stop_recording_raises_stop_error_when_both_fail(self):
|
|
stream = _Stream(
|
|
stop_exc=RuntimeError("stop boom"),
|
|
close_exc=RuntimeError("close boom"),
|
|
)
|
|
record = RecordResult()
|
|
|
|
with self.assertLogs(level="WARNING") as logs:
|
|
with self.assertRaisesRegex(RuntimeError, "stop boom"):
|
|
stop_recording(stream, record)
|
|
|
|
self.assertEqual(stream.stop_calls, 1)
|
|
self.assertEqual(stream.close_calls, 1)
|
|
self.assertTrue(
|
|
any("stream close failed after stop failure: close boom" in line for line in logs.output)
|
|
)
|
|
|
|
def test_stop_recording_with_no_stream_flattens_frames(self):
|
|
record = RecordResult(frames=[np.array([[0.2], [0.4]], dtype=np.float32)])
|
|
|
|
audio = stop_recording(None, record)
|
|
|
|
np.testing.assert_allclose(audio, np.array([0.2, 0.4], dtype=np.float32))
|
|
|
|
def test_start_recording_accepts_explicit_device_name(self):
|
|
sd = _FakeSoundDevice()
|
|
with patch("recorder._sounddevice", return_value=sd):
|
|
stream, record = recorder.start_recording("Laptop")
|
|
|
|
self.assertTrue(stream.started)
|
|
self.assertEqual(stream.kwargs["device"], 1)
|
|
self.assertEqual(record.channels, 1)
|
|
|
|
def test_start_recording_accepts_explicit_device_index(self):
|
|
sd = _FakeSoundDevice()
|
|
with patch("recorder._sounddevice", return_value=sd):
|
|
stream, _record = recorder.start_recording(2)
|
|
|
|
self.assertTrue(stream.started)
|
|
self.assertEqual(stream.kwargs["device"], 2)
|
|
|
|
def test_start_recording_invalid_explicit_name_raises(self):
|
|
sd = _FakeSoundDevice()
|
|
with patch("recorder._sounddevice", return_value=sd):
|
|
with self.assertRaisesRegex(
|
|
ValueError,
|
|
"did not match any input device; available inputs: 1:Laptop Mic, 2:USB Interface",
|
|
):
|
|
recorder.start_recording("NotADevice")
|
|
|
|
def test_start_recording_invalid_explicit_index_raises(self):
|
|
sd = _FakeSoundDevice()
|
|
with patch("recorder._sounddevice", return_value=sd):
|
|
with self.assertRaisesRegex(
|
|
ValueError,
|
|
"did not match any input device; available inputs: 1:Laptop Mic, 2:USB Interface",
|
|
):
|
|
recorder.start_recording(99)
|
|
|
|
def test_start_recording_empty_input_uses_default(self):
|
|
sd = _FakeSoundDevice()
|
|
with patch("recorder._sounddevice", return_value=sd):
|
|
stream, _record = recorder.start_recording("")
|
|
|
|
self.assertTrue(stream.started)
|
|
self.assertIsNone(stream.kwargs["device"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|