96 lines
2.8 KiB
Python
96 lines
2.8 KiB
Python
"""Tests for AudioPipeline."""
|
|
|
|
from tuimble.audio import FRAME_SIZE, SAMPLE_RATE, AudioPipeline
|
|
|
|
|
|
def test_default_construction():
|
|
ap = AudioPipeline()
|
|
assert ap._sample_rate == SAMPLE_RATE
|
|
assert ap._frame_size == FRAME_SIZE
|
|
assert ap._input_device is None
|
|
assert ap._output_device is None
|
|
assert ap.capturing is False
|
|
|
|
|
|
def test_custom_construction():
|
|
ap = AudioPipeline(sample_rate=24000, frame_size=480,
|
|
input_device=1, output_device=2)
|
|
assert ap._sample_rate == 24000
|
|
assert ap._frame_size == 480
|
|
assert ap._input_device == 1
|
|
assert ap._output_device == 2
|
|
|
|
|
|
def test_capturing_toggle():
|
|
ap = AudioPipeline()
|
|
assert ap.capturing is False
|
|
ap.capturing = True
|
|
assert ap.capturing is True
|
|
ap.capturing = False
|
|
assert ap.capturing is False
|
|
|
|
|
|
def test_get_capture_frame_empty():
|
|
ap = AudioPipeline()
|
|
assert ap.get_capture_frame() is None
|
|
|
|
|
|
def test_get_capture_frame_returns_queued():
|
|
ap = AudioPipeline()
|
|
ap._capture_queue.put(b"\x01\x02\x03")
|
|
assert ap.get_capture_frame() == b"\x01\x02\x03"
|
|
assert ap.get_capture_frame() is None
|
|
|
|
|
|
def test_queue_playback_and_callback():
|
|
"""Verify playback callback writes PCM directly to output buffer."""
|
|
ap = AudioPipeline()
|
|
frame_bytes = FRAME_SIZE * 2 # 16-bit mono = 2 bytes per sample
|
|
pcm = bytes(range(256)) * (frame_bytes // 256) + bytes(range(frame_bytes % 256))
|
|
assert len(pcm) == frame_bytes
|
|
|
|
ap.queue_playback(pcm)
|
|
|
|
outdata = bytearray(frame_bytes)
|
|
ap._playback_callback(outdata, FRAME_SIZE, None, None)
|
|
assert bytes(outdata) == pcm
|
|
|
|
|
|
def test_playback_callback_silence_on_empty():
|
|
"""Empty queue produces silence."""
|
|
ap = AudioPipeline()
|
|
frame_bytes = FRAME_SIZE * 2
|
|
outdata = bytearray(b"\xff" * frame_bytes)
|
|
ap._playback_callback(outdata, FRAME_SIZE, None, None)
|
|
assert outdata == bytearray(frame_bytes) # all zeros
|
|
|
|
|
|
def test_playback_callback_short_pcm_pads_silence():
|
|
"""PCM shorter than output buffer gets zero-padded."""
|
|
ap = AudioPipeline()
|
|
frame_bytes = FRAME_SIZE * 2
|
|
short_pcm = b"\x42" * 100
|
|
ap.queue_playback(short_pcm)
|
|
|
|
outdata = bytearray(frame_bytes)
|
|
ap._playback_callback(outdata, FRAME_SIZE, None, None)
|
|
assert outdata[:100] == bytearray(b"\x42" * 100)
|
|
assert outdata[100:] == bytearray(frame_bytes - 100)
|
|
|
|
|
|
def test_queue_playback_overflow_drops():
|
|
"""Full queue drops new data silently."""
|
|
ap = AudioPipeline(frame_size=FRAME_SIZE)
|
|
# Fill the queue
|
|
for i in range(ap._playback_queue.maxsize):
|
|
ap.queue_playback(b"\x00")
|
|
# This should not raise
|
|
ap.queue_playback(b"\xff")
|
|
assert ap._playback_queue.qsize() == ap._playback_queue.maxsize
|
|
|
|
|
|
def test_stop_without_start():
|
|
"""Stop on unstarted pipeline should not raise."""
|
|
ap = AudioPipeline()
|
|
ap.stop()
|