hms-mediaengine/tests/unit/test_playback.py

"""Unit-Tests Playback-Transport (PLAN.md §12.2, §12.5, §16.6)."""

from __future__ import annotations

import pytest
from hms_domain import LoopMode, TransportState
from hms_media import PlaybackController, PlaybackEventKind, PreloadSlot


def _controller(**overrides) -> PlaybackController:
    defaults = dict(
        source_id="src-1",
        in_point=0.0,
        out_point=1.0,
        loop_mode=LoopMode.LOOP,
        speed=1.0,
    )
    defaults.update(overrides)
    return PlaybackController(**defaults)


# ---------- Konstruktion und Invarianten ----------


def test_rejects_invalid_in_out_points() -> None:
    with pytest.raises(ValueError, match="in_point"):
        _controller(in_point=0.9, out_point=0.1)
    with pytest.raises(ValueError, match="in_point"):
        _controller(in_point=0.5, out_point=0.5)


def test_initial_state_stopped_at_in_point() -> None:
    c = _controller(in_point=0.2)
    assert c.state is TransportState.STOPPED
    assert c.position == pytest.approx(0.2)


# ---------- Transportbefehle (§12.5) ----------


def test_play_pause_resume_cycle() -> None:
    c = _controller()
    c.play()
    c.advance(0.1)
    pos_after_play = c.position
    c.pause()
    assert c.state is TransportState.PAUSED
    events = c.advance(1.0)  # pausiert: keine Bewegung
    assert events == []
    assert c.position == pytest.approx(pos_after_play)
    c.play()
    c.advance(0.1)
    assert c.position > pos_after_play


def test_stop_resets_position_and_direction() -> None:
    c = _controller(in_point=0.1)
    c.play()
    c.advance(0.3)
    c.stop()
    assert c.state is TransportState.STOPPED
    assert c.position == pytest.approx(0.1)
    assert c.direction == 1


def test_retrigger_restarts_from_in_point() -> None:
    c = _controller()
    c.play()
    c.advance(0.4)
    c.retrigger()  # §16.5: flankenbasierter Neustart
    assert c.position == pytest.approx(0.0)
    assert c.state is TransportState.PLAYING


def test_seek_bounds_enforced() -> None:
    c = _controller(in_point=0.2, out_point=0.8)
    c.seek(0.5)
    assert c.position == pytest.approx(0.5)
    with pytest.raises(ValueError, match="In/Out"):
        c.seek(0.9)
    with pytest.raises(ValueError, match="In/Out"):
        c.seek(0.1)


def test_speed_zero_rejected_pause_is_the_way() -> None:
    c = _controller()
    with pytest.raises(ValueError, match="speed 0"):
        c.set_speed(0.0)


def test_negative_speed_plays_backward() -> None:
    """§16.6: negative Geschwindigkeit zugelassen; Rückwärts am In-Point
    wrappt im Loop-Modus korrekt ans Ende des Fensters (§12.5)."""
    c = _controller()
    c.play()
    c.set_speed(-1.0)
    events = c.advance(0.05)
    assert c.position == pytest.approx(0.95)  # Wrap nach hinten: out - 0.05
    assert events and events[0].kind is PlaybackEventKind.LOOP_WRAP
    # danach läuft es weiter rückwärts Richtung In-Point
    c.advance(0.05)
    assert c.position == pytest.approx(0.9)


# ---------- Loop-Verhalten ----------


def test_loop_wraps_with_event() -> None:
    c = _controller(out_point=1.0)
    c.play()
    events = c.advance(1.25)  # über das Ende hinaus
    assert c.position == pytest.approx(0.25)  # Wrap: Überschuss am Anfang
    assert events and events[0].kind is PlaybackEventKind.LOOP_WRAP
    assert c.state is TransportState.PLAYING  # Loop läuft weiter


def test_once_stops_at_out_point_with_end_event() -> None:
    c = _controller(loop_mode=LoopMode.ONCE)
    c.play()
    events = c.advance(0.5)
    assert events == []  # noch nicht am Ende
    events = c.advance(0.6)  # über das Ende hinaus
    assert c.state is TransportState.STOPPED  # §12.5: Ende-Ereignis
    assert c.position == pytest.approx(1.0)
    assert events and events[0].kind is PlaybackEventKind.END_OF_MEDIA
    # weiteres Advancement bleibt gestoppt
    assert c.advance(0.1) == []


def test_ping_pong_reverses_direction() -> None:
    c = _controller(loop_mode=LoopMode.PING_PONG)
    c.play()
    events = c.advance(1.2)  # Ende erreicht → Richtungsumkehr
    assert c.direction == -1
    assert c.position == pytest.approx(0.8)  # 0.2 zurückgeprallt
    assert events and events[0].kind is PlaybackEventKind.DIRECTION_CHANGE
    events = c.advance(1.6)  # zurück zum Anfang → wieder vorwärts
    assert c.direction == 1
    assert events and events[0].kind is PlaybackEventKind.DIRECTION_CHANGE
    assert c.state is TransportState.PLAYING  # Ping-Pong läuft endlos


def test_in_out_window_respected() -> None:
    """Loop beachtet In/Out-Fenster, nicht nur 0..1 (§12.5)."""
    c = _controller(in_point=0.2, out_point=0.8)
    c.play()
    c.advance(0.7)  # 0.2 + 0.7 = 0.9 > out 0.8
    assert c.position == pytest.approx(0.3)  # Wrap innerhalb des Fensters


def test_advance_rejects_negative_dt() -> None:
    c = _controller()
    c.play()
    with pytest.raises(ValueError, match="negativ"):
        c.advance(-0.1)


def test_events_carry_source_and_monotonic_time() -> None:
    c = _controller()
    c.play()
    events = c.advance(1.5, now_ns=123456789)
    assert events[0].source_id == "src-1"
    assert events[0].monotonic_ns == 123456789


# ---------- PreloadSlot: atomarer Clipwechsel (§12.2, §16.5) ----------


def test_preload_commit_atomic_flow() -> None:
    slot = PreloadSlot()
    assert slot.commit() is None  # nichts geladen → kein Wechsel
    slot.preload("asset-a")
    assert slot.pending == "asset-a"
    assert slot.commit() is None  # noch nicht bereit → alter Clip bleibt
    slot.mark_ready("asset-a")
    assert slot.commit() == "asset-a"  # atomar umgeschaltet
    assert slot.pending is None
    assert slot.commit() is None  # Slot ist wieder leer


def test_preload_cancel_discards() -> None:
    slot = PreloadSlot()
    slot.preload("asset-a")
    slot.mark_ready("asset-a")
    slot.cancel()
    assert slot.pending is None
    assert slot.commit() is None  # abgebrochen wird nie committed


def test_preload_rejects_foreign_ready() -> None:
    slot = PreloadSlot()
    slot.preload("asset-a")
    with pytest.raises(ValueError, match="anderes Asset"):
        slot.mark_ready("asset-b")


def test_preload_new_preload_resets_ready() -> None:
    slot = PreloadSlot()
    slot.preload("asset-a")
    slot.mark_ready("asset-a")
    slot.preload("asset-b")  # umgeladen: a wird verworfen
    assert slot.ready is False
    assert slot.commit() is None