add audio pipeline tests

tests/test_audio.py

@@ -0,0 +1,95 @@
+"""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_encoded_frame_empty():
+    ap = AudioPipeline()
+    assert ap.get_encoded_frame() is None
+
+
+def test_get_encoded_frame_returns_queued():
+    ap = AudioPipeline()
+    ap._capture_queue.put(b"\x01\x02\x03")
+    assert ap.get_encoded_frame() == b"\x01\x02\x03"
+    assert ap.get_encoded_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()