3ad5e91890
- 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
298 lines
8.9 KiB
Python
298 lines
8.9 KiB
Python
"""
|
|
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()
|