music-assistant-server

1"""Tests for sendspin controller seek / seek-relative event handling."""
2
3from __future__ import annotations
4
5from types import SimpleNamespace
6from typing import cast
7from unittest.mock import AsyncMock
8
9from aiosendspin.server.roles import ControllerSeekEvent, ControllerSeekRelativeEvent
10
11from music_assistant.providers.sendspin.player import SendspinPlayer
12
13
14def _player(
15    *, duration: float | None, elapsed: float = 0.0, has_item: bool = True
16) -> tuple[SendspinPlayer, AsyncMock]:
17    current_item = SimpleNamespace(duration=duration) if has_item else None
18    queue = SimpleNamespace(
19        queue_id="q1",
20        current_item=current_item,
21        corrected_elapsed_time=elapsed,
22    )
23    seek = AsyncMock()
24    player_queues = SimpleNamespace(get_active_queue=lambda _pid: queue, seek=seek)
25    fake = SimpleNamespace(player_id="p1", mass=SimpleNamespace(player_queues=player_queues))
26    return cast("SendspinPlayer", fake), seek
27
28
29async def test_absolute_seek_converts_ms_to_seconds() -> None:
30    """An absolute seek forwards the position to the queue in seconds."""
31    player, seek = _player(duration=100.0)
32    await SendspinPlayer._handle_controller_event(player, ControllerSeekEvent(position_ms=5400))
33    seek.assert_awaited_once_with("q1", 5)
34
35
36async def test_absolute_seek_clamps_to_duration() -> None:
37    """An absolute seek past the end clamps to the track duration."""
38    player, seek = _player(duration=100.0)
39    await SendspinPlayer._handle_controller_event(player, ControllerSeekEvent(position_ms=120_000))
40    seek.assert_awaited_once_with("q1", 100)
41
42
43async def test_absolute_seek_ignored_without_current_item() -> None:
44    """An absolute seek is dropped when no item is loaded."""
45    player, seek = _player(duration=None, has_item=False)
46    await SendspinPlayer._handle_controller_event(player, ControllerSeekEvent(position_ms=5400))
47    seek.assert_not_awaited()
48
49
50async def test_absolute_seek_ignored_when_duration_unknown() -> None:
51    """An absolute seek is dropped when the current item has no duration."""
52    player, seek = _player(duration=None)
53    await SendspinPlayer._handle_controller_event(player, ControllerSeekEvent(position_ms=5400))
54    seek.assert_not_awaited()
55
56
57async def test_relative_seek_clamps_to_duration() -> None:
58    """A relative seek past the end clamps to the track duration."""
59    player, seek = _player(duration=100.0, elapsed=95.0)
60    await SendspinPlayer._handle_controller_event(
61        player, ControllerSeekRelativeEvent(offset_ms=30_000)
62    )
63    seek.assert_awaited_once_with("q1", 100)
64
65
66async def test_relative_seek_clamps_to_zero() -> None:
67    """A relative seek before the start clamps to zero."""
68    player, seek = _player(duration=100.0, elapsed=5.0)
69    await SendspinPlayer._handle_controller_event(
70        player, ControllerSeekRelativeEvent(offset_ms=-30_000)
71    )
72    seek.assert_awaited_once_with("q1", 0)
73
74
75async def test_relative_seek_ignored_when_duration_unknown() -> None:
76    """A relative seek is dropped when the current item has no duration."""
77    player, seek = _player(duration=None, elapsed=10.0)
78    await SendspinPlayer._handle_controller_event(
79        player, ControllerSeekRelativeEvent(offset_ms=30_000)
80    )
81    seek.assert_not_awaited()
82