/
/
/
1"""Tests for how a Sendspin session decides that its media is fed at playback pace."""
2
3from __future__ import annotations
4
5from unittest.mock import MagicMock
6
7from music_assistant_models.enums import MediaType
8
9from music_assistant.providers.sendspin.playback import SendspinPlaybackSession
10
11
12def test_radio_is_live_without_looking_at_the_queue() -> None:
13 """Radio is live by media type; no queue lookup is needed."""
14 session, queues = _session_mock()
15 assert session._is_live_source(_media(MediaType.RADIO)) is True
16 queues.get_item.assert_not_called()
17
18
19def test_a_realtime_track_is_live() -> None:
20 """A track whose source hands over its audio just-in-time counts as live."""
21 session, queues = _session_mock()
22 queues.get_item.return_value = MagicMock(streamdetails=MagicMock(is_realtime=True))
23 assert session._is_live_source(_media(MediaType.TRACK)) is True
24 queues.get_item.assert_called_once_with("player1", "qi-1")
25
26
27def test_an_ordinary_track_is_buffered() -> None:
28 """A track that can be read ahead keeps the buffered send-ahead."""
29 session, queues = _session_mock()
30 queues.get_item.return_value = MagicMock(streamdetails=MagicMock(is_realtime=False))
31 assert session._is_live_source(_media(MediaType.TRACK)) is False
32
33
34def test_an_unresolved_item_is_treated_as_buffered() -> None:
35 """Without streamdetails to consult there is no reason to assume a live source."""
36 session, queues = _session_mock()
37 queues.get_item.return_value = MagicMock(streamdetails=None)
38 assert session._is_live_source(_media(MediaType.TRACK)) is False
39 queues.get_item.return_value = None
40 assert session._is_live_source(_media(MediaType.TRACK)) is False
41
42
43def test_media_without_a_queue_reference_is_treated_as_buffered() -> None:
44 """Media that is not a queue item cannot be looked up, so it stays buffered."""
45 session, _queues = _session_mock()
46 media = _media(MediaType.TRACK)
47 media.queue_item_id = None
48 assert session._is_live_source(media) is False
49
50
51def _session_mock() -> tuple[SendspinPlaybackSession, MagicMock]:
52 """Return a bare playback session plus the player_queues mock it consults."""
53 session = object.__new__(SendspinPlaybackSession)
54 session.player = MagicMock()
55 return session, session.player.mass.player_queues
56
57
58def _media(media_type: MediaType) -> MagicMock:
59 """Return PlayerMedia for a queue item of the given media type."""
60 return MagicMock(media_type=media_type, source_id="player1", queue_item_id="qi-1")
61