/
/
/
1"""Tests for moving items around a playing queue."""
2
3from __future__ import annotations
4
5from typing import Any, cast
6from unittest.mock import MagicMock, Mock
7
8import pytest
9from music_assistant_models.enums import MediaType, PlaybackState
10from music_assistant_models.media_items import ItemMapping, ProviderMapping, Track
11from music_assistant_models.player_queue import PlayerQueue
12from music_assistant_models.queue_item import QueueItem
13from music_assistant_models.unique_list import UniqueList
14
15from music_assistant.controllers.player_queues import PlayerQueuesController
16from music_assistant.controllers.player_queues.state import PlayerQueueData
17
18TRACKS = ["t0", "t1", "t2", "t3", "t4"]
19
20
21def _track(item_id: str) -> Track:
22 """Build a playable Track on the 'test' provider."""
23 return Track(
24 item_id=item_id,
25 provider="test",
26 name=f"Track {item_id}",
27 duration=60,
28 artists=UniqueList(
29 [ItemMapping(item_id="a", provider="test", name="A", media_type=MediaType.ARTIST)]
30 ),
31 provider_mappings={
32 ProviderMapping(item_id=item_id, provider_domain="test", provider_instance="test")
33 },
34 )
35
36
37def _controller(
38 *,
39 state: PlaybackState = PlaybackState.PLAYING,
40 current_index: int = 0,
41 index_in_buffer: int | None = 0,
42) -> Any:
43 """Build a bare controller holding queue "q1" loaded with TRACKS at the given position."""
44 ctrl = PlayerQueuesController.__new__(PlayerQueuesController)
45 ctrl.logger = Mock()
46 ctrl.mass = MagicMock()
47 ctrl.signal_update = Mock() # type: ignore[method-assign]
48 ctrl.get_next_item = Mock(return_value=None) # type: ignore[method-assign]
49 ctrl._enqueue_next_item = Mock() # type: ignore[method-assign]
50 queue = PlayerQueue(queue_id="q1", active=True, display_name="Q1", available=True, items=0)
51 ctrl._queue_data = {"q1": PlayerQueueData(queue=queue)}
52 items = [QueueItem.from_media_item("q1", _track(item_id)) for item_id in TRACKS]
53 ctrl._queue_data["q1"].items = items
54 queue.items = len(items)
55 queue.state = state
56 queue.current_index = current_index
57 queue.current_item = items[current_index]
58 queue.index_in_buffer = index_in_buffer
59 return ctrl
60
61
62def _item_id_at(ctrl: Any, index: int) -> str:
63 """Return the queue_item_id of the item sitting at the given index."""
64 return cast("str", ctrl._queue_data["q1"].items[index].queue_item_id)
65
66
67def _order(ctrl: Any) -> list[str]:
68 """Return the provider item ids of the queue's items, in play order."""
69 return [
70 item.media_item.item_id
71 for item in ctrl._queue_data["q1"].items
72 if item.media_item is not None
73 ]
74
75
76async def test_play_next_lands_behind_the_buffered_track() -> None:
77 """With a track buffered ahead, "play next" queues behind it rather than into its slot."""
78 ctrl = _controller(current_index=0, index_in_buffer=1)
79 buffered_item_id = _item_id_at(ctrl, 1)
80
81 ctrl.move_item("q1", _item_id_at(ctrl, 4), pos_shift=0)
82
83 assert _order(ctrl) == ["t0", "t1", "t4", "t2", "t3"]
84 assert _item_id_at(ctrl, 1) == buffered_item_id
85
86
87async def test_play_next_lands_behind_the_playing_track_when_nothing_is_buffered_ahead() -> None:
88 """Without a track buffered ahead, "play next" is the slot right after the playing one."""
89 ctrl = _controller(current_index=0, index_in_buffer=0)
90
91 ctrl.move_item("q1", _item_id_at(ctrl, 4), pos_shift=0)
92
93 assert _order(ctrl) == ["t0", "t4", "t1", "t2", "t3"]
94
95
96async def test_play_next_on_the_item_already_next_keeps_the_order() -> None:
97 """Asking for the item that is already first in line is a no-op."""
98 ctrl = _controller(current_index=0, index_in_buffer=1)
99
100 ctrl.move_item("q1", _item_id_at(ctrl, 2), pos_shift=0)
101
102 assert _order(ctrl) == TRACKS
103
104
105async def test_play_next_on_a_paused_queue_respects_the_buffered_track() -> None:
106 """A paused player still holds the track it was handed, so the move goes behind it."""
107 ctrl = _controller(state=PlaybackState.PAUSED, current_index=1, index_in_buffer=2)
108
109 ctrl.move_item("q1", _item_id_at(ctrl, 4), pos_shift=0)
110
111 assert _order(ctrl) == ["t0", "t1", "t2", "t4", "t3"]
112
113
114async def test_play_next_on_an_idle_queue_puts_the_item_first() -> None:
115 """On a queue that is not playing, the moved item takes the position that plays next."""
116 ctrl = _controller(state=PlaybackState.IDLE, current_index=0, index_in_buffer=None)
117
118 ctrl.move_item("q1", _item_id_at(ctrl, 3), pos_shift=0)
119
120 assert _order(ctrl) == ["t3", "t0", "t1", "t2", "t4"]
121
122
123async def test_play_next_clears_a_track_buffered_two_ahead() -> None:
124 """A buffered index further than one ahead still decides where the move lands."""
125 ctrl = _controller(current_index=0, index_in_buffer=2)
126
127 ctrl.move_item("q1", _item_id_at(ctrl, 4), pos_shift=0)
128
129 assert _order(ctrl) == ["t0", "t1", "t2", "t4", "t3"]
130
131
132async def test_moving_is_refused_while_repeat_wraps_the_queue() -> None:
133 """A queue whose buffered track wrapped back to the front refuses moves ahead of it."""
134 ctrl = _controller(current_index=4, index_in_buffer=0)
135
136 with pytest.raises(IndexError):
137 ctrl.move_item("q1", _item_id_at(ctrl, 2), pos_shift=0)
138
139 assert _order(ctrl) == TRACKS
140 assert _item_id_at(ctrl, 4) == ctrl._queue_data["q1"].queue.current_item.queue_item_id
141
142
143async def test_moving_a_buffered_item_is_refused() -> None:
144 """An item at or before the buffered one cannot be moved."""
145 ctrl = _controller(current_index=0, index_in_buffer=1)
146
147 with pytest.raises(IndexError):
148 ctrl.move_item("q1", _item_id_at(ctrl, 1), pos_shift=1)
149
150
151async def test_moving_to_the_end_is_refused_while_repeat_wraps_the_queue() -> None:
152 """Moving to the end shifts the playing track too, so it is refused on a wrapped queue."""
153 ctrl = _controller(current_index=4, index_in_buffer=0)
154
155 with pytest.raises(IndexError):
156 ctrl.move_item_end("q1", _item_id_at(ctrl, 2))
157
158 assert _order(ctrl) == TRACKS
159
160
161async def test_deleting_is_ignored_while_repeat_wraps_the_queue() -> None:
162 """Deleting ahead of the playing track shifts it, so it is ignored on a wrapped queue."""
163 ctrl = _controller(current_index=4, index_in_buffer=0)
164
165 ctrl.delete_item("q1", 2)
166
167 assert _order(ctrl) == TRACKS
168
169
170async def test_a_relative_move_cannot_land_on_a_track_the_player_owns() -> None:
171 """A move towards the front stops at the boundary instead of landing inside it."""
172 ctrl = _controller(current_index=0, index_in_buffer=2)
173
174 ctrl.move_item("q1", _item_id_at(ctrl, 4), pos_shift=-3)
175
176 assert _order(ctrl) == TRACKS
177
178
179async def test_relative_move_is_unaffected_by_the_buffered_index() -> None:
180 """A relative move still shifts the item by the requested number of positions."""
181 ctrl = _controller(current_index=0, index_in_buffer=1)
182
183 ctrl.move_item("q1", _item_id_at(ctrl, 2), pos_shift=1)
184
185 assert _order(ctrl) == ["t0", "t1", "t3", "t2", "t4"]
186