music-assistant-server

1"""Tests for the sendspin per-track beat-schedule anchor offset."""
2
3from __future__ import annotations
4
5from types import SimpleNamespace
6from typing import TYPE_CHECKING, cast
7
8from music_assistant.providers.sendspin.player import SendspinPlayer
9
10if TYPE_CHECKING:
11    from music_assistant_models.queue_item import QueueItem
12
13    from music_assistant.controllers.player_queues.state import PlayerQueueData
14
15
16def _queue_item(queue_item_id: str, seek: int = 0) -> QueueItem:
17    return cast(
18        "QueueItem",
19        SimpleNamespace(
20            queue_item_id=queue_item_id,
21            streamdetails=SimpleNamespace(seek_position=seek),
22        ),
23    )
24
25
26def _log_entry(queue_item_id: str, seconds_streamed: float | None) -> SimpleNamespace:
27    return SimpleNamespace(queue_item_id=queue_item_id, seconds_streamed=seconds_streamed)
28
29
30def test_flow_offset_sums_prior_tracks() -> None:
31    """A later track anchors past the summed stream-time of all prior flow tracks."""
32    data = cast(
33        "PlayerQueueData",
34        SimpleNamespace(
35            flow_mode_stream_log=[
36                _log_entry("t1", 100.0),
37                _log_entry("t2", 200.5),
38                _log_entry("t3", None),  # current track, still streaming
39            ]
40        ),
41    )
42    offset_us = SendspinPlayer._flow_track_offset_us(data, _queue_item("t3"))
43    assert offset_us == int((100.0 + 200.5) * 1_000_000)
44