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