/
/
/
1"""Tests for keeping the player's upcoming track in step with the queue."""
2
3from __future__ import annotations
4
5from typing import Any
6from unittest.mock import MagicMock, Mock
7
8from music_assistant_models.enums import MediaType, PlaybackState, RepeatMode
9from music_assistant_models.media_items import ItemMapping, ProviderMapping, Track
10from music_assistant_models.player_queue import PlayerQueue
11from music_assistant_models.queue_item import QueueItem
12from music_assistant_models.unique_list import UniqueList
13
14from music_assistant.controllers.player_queues import PlayerQueuesController
15from music_assistant.controllers.player_queues.state import PlayerQueueData
16
17TRACKS = ["t0", "t1", "t2", "t3", "t4"]
18
19
20def _track(item_id: str) -> Track:
21 """Build a playable Track on the 'test' provider."""
22 return Track(
23 item_id=item_id,
24 provider="test",
25 name=f"Track {item_id}",
26 duration=60,
27 artists=UniqueList(
28 [ItemMapping(item_id="a", provider="test", name="A", media_type=MediaType.ARTIST)]
29 ),
30 provider_mappings={
31 ProviderMapping(item_id=item_id, provider_domain="test", provider_instance="test")
32 },
33 )
34
35
36def _controller(
37 *,
38 current_index: int = 0,
39 index_in_buffer: int | None = 1,
40 state: PlaybackState = PlaybackState.PLAYING,
41 enqueued_offset: int = 1,
42) -> Any:
43 """
44 Build a controller whose player has been handed the track at `enqueued_offset`.
45
46 :param current_index: The index the player is playing.
47 :param index_in_buffer: The index the streams engine has read ahead to.
48 :param state: The queue's playback state.
49 :param enqueued_offset: Offset from current_index of the track the player already holds.
50 """
51 ctrl = PlayerQueuesController.__new__(PlayerQueuesController)
52 ctrl.logger = Mock()
53 ctrl.mass = MagicMock()
54 ctrl.signal_update = Mock() # type: ignore[method-assign]
55 ctrl._enqueue_next_item = Mock() # type: ignore[method-assign]
56 ctrl._smart_shuffle = Mock()
57 ctrl._smart_shuffle.is_enabled = Mock(return_value=False)
58 ctrl.mass.streams.is_smart_fades_active = Mock(return_value=False)
59 queue = PlayerQueue(queue_id="q1", active=True, display_name="Q1", available=True, items=0)
60 queue_data = PlayerQueueData(queue=queue)
61 ctrl._queue_data = {"q1": queue_data}
62 items = [QueueItem.from_media_item("q1", _track(item_id)) for item_id in TRACKS]
63 queue_data.items = items
64 queue_data.next_item_id_enqueued = items[current_index + enqueued_offset].queue_item_id
65 queue.items = len(items)
66 queue.state = state
67 queue.current_index = current_index
68 queue.current_item = items[current_index]
69 queue.index_in_buffer = index_in_buffer
70 queue.repeat_mode = RepeatMode.OFF
71 return ctrl
72
73
74def _enqueued(ctrl: Any) -> str | None:
75 """Return the provider item id handed to the player, if any."""
76 if not ctrl._enqueue_next_item.called:
77 return None
78 item = ctrl._enqueue_next_item.call_args.args[1]
79 return str(item.media_item.item_id)
80
81
82async def test_a_changed_upcoming_track_reaches_the_player_while_a_track_is_buffered() -> None:
83 """A track buffered ahead does not stop the player being told the upcoming track changed."""
84 ctrl = _controller(current_index=0, index_in_buffer=1)
85 items = ctrl._queue_data["q1"].items
86 reordered = [items[0], items[4], items[1], items[2], items[3]]
87
88 ctrl.update_items("q1", reordered)
89
90 assert _enqueued(ctrl) == "t4"
91
92
93async def test_an_unchanged_upcoming_track_is_not_handed_over_again() -> None:
94 """A queue change that leaves the upcoming track alone sends nothing to the player."""
95 ctrl = _controller(current_index=0, index_in_buffer=0)
96 items = ctrl._queue_data["q1"].items
97 reordered = [items[0], items[1], items[4], items[2], items[3]]
98
99 ctrl.update_items("q1", reordered)
100
101 assert _enqueued(ctrl) is None
102
103
104async def test_repeat_one_hands_the_playing_track_back_to_the_player() -> None:
105 """Switching to repeat-one makes the playing track the upcoming one on the player too."""
106 ctrl = _controller(current_index=0, index_in_buffer=1)
107
108 await ctrl.set_repeat("q1", RepeatMode.ONE)
109
110 assert _enqueued(ctrl) == "t0"
111
112
113async def test_crossfade_hands_the_same_track_over_again() -> None:
114 """Crossfade changes how the upcoming track is streamed, so it is handed over again."""
115 ctrl = _controller(current_index=0, index_in_buffer=1)
116
117 ctrl.set_crossfade("q1", crossfade_enabled=True)
118
119 assert _enqueued(ctrl) == "t1"
120
121
122async def test_nothing_is_handed_over_while_a_track_is_starting() -> None:
123 """A starting track moves the two positions one after the other, so the gap is skipped."""
124 ctrl = _controller(current_index=0, index_in_buffer=3)
125 ctrl._queue_data["q1"].transitioning = True
126 items = ctrl._queue_data["q1"].items
127
128 ctrl.update_items("q1", [items[0], items[4], items[1], items[2], items[3]])
129
130 assert _enqueued(ctrl) is None
131
132
133async def test_nothing_is_handed_over_while_the_queue_holds_no_position() -> None:
134 """A queue mid-replace has no committed position, so no upcoming track is picked for it."""
135 ctrl = _controller(current_index=0, index_in_buffer=None)
136 items = ctrl._queue_data["q1"].items
137
138 ctrl.update_items("q1", [items[0], items[4], items[1], items[2], items[3]])
139
140 assert _enqueued(ctrl) is None
141
142
143async def test_nothing_is_handed_over_while_the_queue_is_not_playing() -> None:
144 """A paused or idle queue is not handed an upcoming track."""
145 ctrl = _controller(current_index=0, index_in_buffer=1, state=PlaybackState.PAUSED)
146 items = ctrl._queue_data["q1"].items
147
148 ctrl.update_items("q1", [items[0], items[4], items[1], items[2], items[3]])
149
150 assert _enqueued(ctrl) is None
151