From a6ba5239af9b01036cb897a3f3ebe494faba27c1 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 3 Apr 2026 14:50:04 +0200 Subject: [PATCH] test(02-04): add failing tests for ChoreographyScheduler MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 8 tests covering: dispatch timing, payload format, pause/resume, seek, stop, on_event callback, position tracking - All tests fail (RED) — scheduler module not yet implemented --- tests/test_scheduler.py | 183 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 183 insertions(+) create mode 100644 tests/test_scheduler.py diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py new file mode 100644 index 0000000..700bb74 --- /dev/null +++ b/tests/test_scheduler.py @@ -0,0 +1,183 @@ +""" +Tests for ChoreographyScheduler — TDD RED phase. + +Verifies: + - 3 events fire at correct timestamps (within 20ms drift per D-08) + - Pause suppresses dispatch; resume continues + - Seek skips events before seek position + - Payload format: {zone, animation, params} — no "v":1 (transport injects that) + - stop() resets state cleanly +""" +import asyncio +import time + +import pytest +import pytest_asyncio + +from led_sync.models import ChoreoEvent +from led_sync.scheduler import ChoreographyScheduler + + +class MockTransport: + """Minimal stand-in for ESP32Transport.""" + + connected = True + last_status = None + + def __init__(self): + self.sent: list[dict] = [] + self.sent_times: list[float] = [] + + def send_command(self, cmd: dict) -> None: + self.sent.append(cmd) + self.sent_times.append(time.monotonic()) + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture +def transport(): + return MockTransport() + + +@pytest.fixture +def scheduler(transport): + return ChoreographyScheduler(transport) + + +def make_events() -> list[ChoreoEvent]: + return [ + ChoreoEvent(timestamp=0.0, zone="wand", animation="chase", params={"speed": 0.5}), + ChoreoEvent(timestamp=0.05, zone="schrank", animation="pulse", params={}), + ChoreoEvent(timestamp=0.1, zone="all", animation="rainbow", params={}), + ] + + +def make_two_events() -> list[ChoreoEvent]: + return [ + ChoreoEvent(timestamp=0.0, zone="wand", animation="breathe", params={}), + ChoreoEvent(timestamp=0.15, zone="wand", animation="sparkle", params={}), + ] + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_three_events_dispatched(scheduler, transport): + """All 3 events dispatch, in order, within 20ms drift each.""" + events = make_events() + t0 = time.monotonic() + scheduler.play(events) + await asyncio.sleep(0.25) + + assert len(transport.sent) == 3, f"Expected 3 commands, got {len(transport.sent)}" + for i, (cmd, fired_at) in enumerate(zip(transport.sent, transport.sent_times)): + drift = abs((fired_at - t0) - events[i].timestamp) + assert drift < 0.020, f"Event {i}: drift {drift * 1000:.1f}ms > 20ms" + + +@pytest.mark.asyncio +async def test_payload_format(scheduler, transport): + """Command dict contains zone, animation, params — no 'v':1 (transport adds that).""" + events = [ChoreoEvent(timestamp=0.0, zone="wand", animation="chase", params={"speed": 0.5})] + scheduler.play(events) + await asyncio.sleep(0.05) + + assert len(transport.sent) == 1 + cmd = transport.sent[0] + assert cmd["zone"] == "wand" + assert cmd["animation"] == "chase" + assert cmd["params"] == {"speed": 0.5} + assert "v" not in cmd, "Transport injects v:1, scheduler must not include it" + + +@pytest.mark.asyncio +async def test_pause_suppresses_dispatch(scheduler, transport): + """No events fire while paused.""" + events = make_two_events() + scheduler.play(events) + await asyncio.sleep(0.05) + scheduler.pause() + count_at_pause = len(transport.sent) + await asyncio.sleep(0.25) # second event would have fired by now if not paused + assert len(transport.sent) == count_at_pause, "Events fired while paused!" + + +@pytest.mark.asyncio +async def test_resume_fires_remaining_events(scheduler, transport): + """After resume, remaining events fire; no double-firing.""" + events = make_two_events() + scheduler.play(events) + await asyncio.sleep(0.05) + scheduler.pause() + await asyncio.sleep(0.1) + scheduler.resume() + await asyncio.sleep(0.3) + assert len(transport.sent) == 2, f"Expected 2 total, got {len(transport.sent)}" + + +@pytest.mark.asyncio +async def test_seek_skips_past_events(scheduler, transport): + """seek(0.1) skips first event (ts=0.0), fires only events >= 0.1.""" + events = make_two_events() # ts=0.0 and ts=0.15 + scheduler.play(events, seek_seconds=0.1) + await asyncio.sleep(0.25) + assert len(transport.sent) == 1, f"Seek should skip first event; got {len(transport.sent)}" + assert transport.sent[0]["animation"] == "sparkle" + + +@pytest.mark.asyncio +async def test_stop_resets_state(scheduler, transport): + """stop() cancels running task, resets state.""" + events = make_events() + scheduler.play(events) + assert scheduler.is_running + scheduler.stop() + await asyncio.sleep(0.01) + assert not scheduler.is_running + assert scheduler.start_time is None + assert scheduler.pause_offset == 0.0 + assert scheduler.paused_at is None + + +@pytest.mark.asyncio +async def test_on_event_callback(transport): + """on_event callback is called once per dispatched event.""" + fired: list[ChoreoEvent] = [] + scheduler = ChoreographyScheduler(transport, on_event=fired.append) + events = make_events() + scheduler.play(events) + await asyncio.sleep(0.25) + assert len(fired) == 3 + assert fired[0].animation == "chase" + + +@pytest.mark.asyncio +async def test_current_position_during_play(scheduler): + """current_position advances monotonically during playback.""" + events = make_events() + scheduler.play(events) + await asyncio.sleep(0.05) + pos = scheduler.current_position + assert 0.04 <= pos <= 0.12, f"current_position={pos:.3f} out of expected range" + scheduler.stop() + + +@pytest.mark.asyncio +async def test_current_position_frozen_when_paused(scheduler, transport): + """current_position is frozen while paused.""" + events = make_two_events() + scheduler.play(events) + await asyncio.sleep(0.05) + scheduler.pause() + pos1 = scheduler.current_position + await asyncio.sleep(0.1) + pos2 = scheduler.current_position + assert abs(pos1 - pos2) < 0.002, f"Position changed during pause: {pos1} -> {pos2}" + scheduler.stop()