led-sync-studio/tests/test_scheduler.py

"""
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()