test(02-05): add failing tests for BeatDetector — AUD-03/AUD-04
- Structural tests: import, constants, initial state, thread safety - TDD RED: 14 tests all fail (module not yet created) - Tests verify call_soon_threadsafe bridge pattern (D-05) - Tests verify beat_count/bpm thread-safe properties via lock
This commit is contained in:
Claude
297
tests/test_beat_detector.py
Normal file
297
tests/test_beat_detector.py
Normal file
@@ -0,0 +1,297 @@
|
||||
"""
|
||||
Tests for BeatDetector — AUD-03 (system audio capture) and AUD-04 (beat detection).
|
||||
TDD RED phase: these tests define expected behavior before implementation.
|
||||
|
||||
Threading safety rules (D-04, D-05):
|
||||
- _audio_callback MUST NOT call asyncio APIs directly
|
||||
- Beat events MUST travel through call_soon_threadsafe
|
||||
"""
|
||||
import asyncio
|
||||
import inspect
|
||||
import threading
|
||||
import time
|
||||
import pytest
|
||||
|
||||
|
||||
# ─── Structural tests (no audio device required) ────────────────────────────
|
||||
|
||||
def test_beat_detector_importable():
|
||||
"""BeatDetector must be importable from led_sync.audio."""
|
||||
from led_sync.audio import BeatDetector
|
||||
assert BeatDetector is not None
|
||||
|
||||
|
||||
def test_beat_detector_importable_from_module():
|
||||
"""BeatDetector must be importable from led_sync.audio.beat_detector."""
|
||||
from led_sync.audio.beat_detector import BeatDetector
|
||||
assert BeatDetector is not None
|
||||
|
||||
|
||||
def test_constants_present():
|
||||
"""Module constants must match latency specification."""
|
||||
from led_sync.audio.beat_detector import SAMPLE_RATE, HOP_SIZE, BUF_SIZE
|
||||
assert SAMPLE_RATE == 44100
|
||||
assert HOP_SIZE == 512
|
||||
assert BUF_SIZE == 1024
|
||||
|
||||
|
||||
def test_latency_spec():
|
||||
"""HOP_SIZE / SAMPLE_RATE must yield < 12ms latency (target ~11ms)."""
|
||||
from led_sync.audio.beat_detector import SAMPLE_RATE, HOP_SIZE
|
||||
latency_ms = HOP_SIZE / SAMPLE_RATE * 1000
|
||||
assert latency_ms < 12.0, f"Latency {latency_ms:.2f}ms exceeds 11ms spec"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_initial_state():
|
||||
"""beat_count starts at 0, bpm starts at 0.0."""
|
||||
from led_sync.audio.beat_detector import BeatDetector
|
||||
|
||||
loop = asyncio.get_running_loop()
|
||||
|
||||
async def on_beat(beat_time: float):
|
||||
pass
|
||||
|
||||
detector = BeatDetector(loop=loop, on_beat=on_beat)
|
||||
assert detector.beat_count == 0
|
||||
assert detector.bpm == 0.0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_stop_before_start_is_noop():
|
||||
"""Calling stop() before start() must not raise any exception."""
|
||||
from led_sync.audio.beat_detector import BeatDetector
|
||||
|
||||
loop = asyncio.get_running_loop()
|
||||
|
||||
async def on_beat(beat_time: float):
|
||||
pass
|
||||
|
||||
detector = BeatDetector(loop=loop, on_beat=on_beat)
|
||||
# Must not raise
|
||||
detector.stop()
|
||||
detector.stop() # second call also noop
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_double_stop_is_noop():
|
||||
"""Calling stop() twice must not raise even if stream was opened."""
|
||||
from led_sync.audio.beat_detector import BeatDetector
|
||||
|
||||
loop = asyncio.get_running_loop()
|
||||
|
||||
async def on_beat(beat_time: float):
|
||||
pass
|
||||
|
||||
detector = BeatDetector(loop=loop, on_beat=on_beat)
|
||||
detector.stop()
|
||||
# Shouldn't raise
|
||||
detector.stop()
|
||||
|
||||
|
||||
def test_beat_count_property_is_thread_safe():
|
||||
"""beat_count property must be accessible from multiple threads (uses lock)."""
|
||||
from led_sync.audio.beat_detector import BeatDetector
|
||||
|
||||
loop = asyncio.new_event_loop()
|
||||
|
||||
async def on_beat(beat_time: float):
|
||||
pass
|
||||
|
||||
detector = BeatDetector(loop=loop, on_beat=on_beat)
|
||||
|
||||
errors = []
|
||||
results = []
|
||||
|
||||
def read_count():
|
||||
try:
|
||||
results.append(detector.beat_count)
|
||||
except Exception as e:
|
||||
errors.append(e)
|
||||
|
||||
threads = [threading.Thread(target=read_count) for _ in range(10)]
|
||||
for t in threads:
|
||||
t.start()
|
||||
for t in threads:
|
||||
t.join()
|
||||
|
||||
assert not errors, f"Thread safety errors: {errors}"
|
||||
assert all(r == 0 for r in results)
|
||||
loop.close()
|
||||
|
||||
|
||||
def test_bpm_property_is_thread_safe():
|
||||
"""bpm property must be accessible from multiple threads (uses lock)."""
|
||||
from led_sync.audio.beat_detector import BeatDetector
|
||||
|
||||
loop = asyncio.new_event_loop()
|
||||
|
||||
async def on_beat(beat_time: float):
|
||||
pass
|
||||
|
||||
detector = BeatDetector(loop=loop, on_beat=on_beat)
|
||||
|
||||
errors = []
|
||||
results = []
|
||||
|
||||
def read_bpm():
|
||||
try:
|
||||
results.append(detector.bpm)
|
||||
except Exception as e:
|
||||
errors.append(e)
|
||||
|
||||
threads = [threading.Thread(target=read_bpm) for _ in range(10)]
|
||||
for t in threads:
|
||||
t.start()
|
||||
for t in threads:
|
||||
t.join()
|
||||
|
||||
assert not errors, f"Thread safety errors: {errors}"
|
||||
assert all(r == 0.0 for r in results)
|
||||
loop.close()
|
||||
|
||||
|
||||
# ─── Audio callback thread-safety check ─────────────────────────────────────
|
||||
|
||||
def test_audio_callback_uses_call_soon_threadsafe():
|
||||
"""
|
||||
_audio_callback source code must use call_soon_threadsafe (not asyncio APIs directly).
|
||||
This is a static analysis check enforcing D-05.
|
||||
"""
|
||||
import inspect
|
||||
from led_sync.audio import beat_detector as mod
|
||||
|
||||
source = inspect.getsource(mod)
|
||||
# Must contain the thread-safe bridge
|
||||
assert "call_soon_threadsafe" in source, (
|
||||
"_audio_callback must use call_soon_threadsafe to bridge audio thread -> asyncio"
|
||||
)
|
||||
# Must NOT call asyncio.create_task or asyncio.ensure_future directly in callback
|
||||
# (these would be wrong — asyncio is not thread-safe to call directly)
|
||||
# We check by ensuring there's no bare asyncio. call in the source
|
||||
# (call_soon_threadsafe is the correct pattern)
|
||||
lines = source.splitlines()
|
||||
callback_lines = []
|
||||
in_callback = False
|
||||
brace_depth = 0
|
||||
for line in lines:
|
||||
if "def _audio_callback" in line:
|
||||
in_callback = True
|
||||
if in_callback:
|
||||
callback_lines.append(line)
|
||||
# Stop at next non-indented def (crude but sufficient)
|
||||
if len(callback_lines) > 1 and line.strip().startswith("def "):
|
||||
break
|
||||
|
||||
callback_text = "\n".join(callback_lines)
|
||||
# Must not directly call asyncio.run() or asyncio.get_event_loop() in callback
|
||||
assert "asyncio.run(" not in callback_text
|
||||
assert "asyncio.get_event_loop()" not in callback_text
|
||||
|
||||
|
||||
# ─── Simulated beat injection (no real audio device) ────────────────────────
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_beat_callback_called_via_asyncio_bridge():
|
||||
"""
|
||||
Simulate what the audio thread does: call_soon_threadsafe with on_beat coroutine.
|
||||
Verify beat_count increments and on_beat is called in asyncio context.
|
||||
"""
|
||||
from led_sync.audio.beat_detector import BeatDetector
|
||||
|
||||
loop = asyncio.get_running_loop()
|
||||
received_beats = []
|
||||
received_threads = []
|
||||
|
||||
async def on_beat(beat_time: float):
|
||||
received_beats.append(beat_time)
|
||||
received_threads.append(threading.current_thread())
|
||||
|
||||
detector = BeatDetector(loop=loop, on_beat=on_beat)
|
||||
|
||||
# Simulate what _audio_callback does internally when a beat is detected
|
||||
# We manually trigger the bridge to test it works correctly
|
||||
with detector._lock:
|
||||
detector._beat_count += 1
|
||||
detector._bpm = 120.0
|
||||
|
||||
loop.call_soon_threadsafe(
|
||||
loop.create_task,
|
||||
on_beat(1.0),
|
||||
)
|
||||
|
||||
# Give asyncio a chance to process
|
||||
await asyncio.sleep(0.05)
|
||||
|
||||
assert len(received_beats) == 1
|
||||
assert received_beats[0] == 1.0
|
||||
# on_beat should run in the asyncio thread (main thread in test)
|
||||
assert received_threads[0] == threading.main_thread()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_beat_count_increments_correctly():
|
||||
"""beat_count reflects simulated beats correctly."""
|
||||
from led_sync.audio.beat_detector import BeatDetector
|
||||
|
||||
loop = asyncio.get_running_loop()
|
||||
|
||||
async def on_beat(beat_time: float):
|
||||
pass
|
||||
|
||||
detector = BeatDetector(loop=loop, on_beat=on_beat)
|
||||
assert detector.beat_count == 0
|
||||
|
||||
# Simulate 3 beats
|
||||
for i in range(3):
|
||||
with detector._lock:
|
||||
detector._beat_count += 1
|
||||
|
||||
assert detector.beat_count == 3
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_bpm_updated():
|
||||
"""bpm reflects simulated BPM correctly."""
|
||||
from led_sync.audio.beat_detector import BeatDetector
|
||||
|
||||
loop = asyncio.get_running_loop()
|
||||
|
||||
async def on_beat(beat_time: float):
|
||||
pass
|
||||
|
||||
detector = BeatDetector(loop=loop, on_beat=on_beat)
|
||||
assert detector.bpm == 0.0
|
||||
|
||||
with detector._lock:
|
||||
detector._bpm = 128.5
|
||||
|
||||
assert detector.bpm == 128.5
|
||||
|
||||
|
||||
# ─── ImportError handling ────────────────────────────────────────────────────
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_start_raises_runtime_error_on_missing_deps(monkeypatch):
|
||||
"""start() must raise RuntimeError if sounddevice or aubio cannot be imported."""
|
||||
import builtins
|
||||
from led_sync.audio.beat_detector import BeatDetector
|
||||
|
||||
loop = asyncio.get_running_loop()
|
||||
|
||||
async def on_beat(beat_time: float):
|
||||
pass
|
||||
|
||||
detector = BeatDetector(loop=loop, on_beat=on_beat)
|
||||
|
||||
original_import = builtins.__import__
|
||||
|
||||
def mock_import(name, *args, **kwargs):
|
||||
if name == "sounddevice":
|
||||
raise ImportError("mocked missing sounddevice")
|
||||
return original_import(name, *args, **kwargs)
|
||||
|
||||
monkeypatch.setattr(builtins, "__import__", mock_import)
|
||||
|
||||
with pytest.raises(RuntimeError, match="sounddevice"):
|
||||
detector.start()
|
||||
Reference in New Issue
Block a user