music-assistant-server

1"""
2Tests that replacing a queue's contents swaps them in one step.
3
4A replace used to empty the queue up front and tear its audio down, then resolve the new media
5over the network before loading it - so clients saw an empty queue with nothing playing for as
6long as the providers took to answer. The queue now keeps playing its current content until the
7new items are ready, and hands the audio over in a single swap.
8"""
9
10from __future__ import annotations
11
12from typing import Any, cast
13from unittest.mock import AsyncMock, MagicMock, Mock
14
15from music_assistant_models.enums import MediaType, PlaybackState, QueueOption
16from music_assistant_models.media_items import (
17    ItemMapping,
18    Playlist,
19    ProviderMapping,
20    Track,
21)
22from music_assistant_models.player_queue import PlayerQueue
23from music_assistant_models.queue_item import QueueItem
24from music_assistant_models.unique_list import UniqueList
25
26from music_assistant.controllers.player_queues import PlayerQueuesController
27from music_assistant.controllers.player_queues.state import PlayerQueueData
28
29NEW_TRACKS = ["n1", "n2", "n3"]
30PLAYING_TRACKS = ["p1", "p2", "p3"]
31
32
33def _track(item_id: str) -> Track:
34    """Build a playable Track on the 'test' provider."""
35    return Track(
36        item_id=item_id,
37        provider="test",
38        name=f"Track {item_id}",
39        duration=60,
40        artists=UniqueList(
41            [ItemMapping(item_id="a", provider="test", name="A", media_type=MediaType.ARTIST)]
42        ),
43        provider_mappings={
44            ProviderMapping(item_id=item_id, provider_domain="test", provider_instance="test")
45        },
46    )
47
48
49def _playlist(item_id: str = "pl1", *, dynamic: bool = False) -> Playlist:
50    """Build a playlist, optionally a dynamic one (an always-on smart mix)."""
51    playlist = Playlist(
52        item_id=item_id,
53        provider="test",
54        name=f"Playlist {item_id}",
55        provider_mappings={
56            ProviderMapping(item_id=item_id, provider_domain="test", provider_instance="test")
57        },
58    )
59    playlist.is_dynamic = dynamic
60    return playlist
61
62
63def _controller(**queue_kwargs: Any) -> Any:
64    """Build a bare controller whose queue "q1" is playing, with the resolver stubbed."""
65    ctrl = PlayerQueuesController.__new__(PlayerQueuesController)
66    ctrl.logger = Mock()
67    ctrl.mass = MagicMock()
68    ctrl.mass.players.get_player = Mock(return_value=Mock(extra_data={}))
69    lock_cm = MagicMock()
70    lock_cm.__aenter__ = AsyncMock(return_value=None)
71    lock_cm.__aexit__ = AsyncMock(return_value=None)
72    ctrl.mass.players.get_player_lock = Mock(return_value=lock_cm)
73    ctrl.signal_update = Mock()  # type: ignore[method-assign]
74    ctrl.on_player_update = Mock()  # type: ignore[method-assign]
75    ctrl.play_index = AsyncMock()  # type: ignore[method-assign]
76    ctrl.get_next_item = Mock(return_value=None)  # type: ignore[method-assign]
77    ctrl.get_config_value = Mock(return_value=QueueOption.REPLACE.value)  # type: ignore[method-assign]
78    ctrl._managed_pool = Mock()
79    ctrl._managed_pool.fill = AsyncMock(
80        side_effect=lambda *_a, **_kw: [_track(item_id) for item_id in NEW_TRACKS]
81    )
82    ctrl._smart_shuffle = Mock()
83    ctrl._smart_shuffle.is_enabled = Mock(return_value=False)
84    ctrl._media_resolver = Mock()
85    ctrl._media_resolver._resolve_media_items = AsyncMock(
86        side_effect=lambda *_a, **_kw: [_track(item_id) for item_id in NEW_TRACKS]
87    )
88    queue = PlayerQueue(
89        queue_id="q1", active=True, display_name="Q1", available=True, items=0, **queue_kwargs
90    )
91    ctrl._queue_data = {"q1": PlayerQueueData(queue=queue)}
92    return ctrl
93
94
95def _queue(ctrl: Any) -> PlayerQueue:
96    """Return the controller's queue."""
97    return cast("PlayerQueue", ctrl._queue_data["q1"].queue)
98
99
100def _item_ids(ctrl: Any) -> list[str]:
101    """Return the item ids of the tracks currently loaded in the queue, in play order."""
102    return [
103        item.media_item.item_id
104        for item in ctrl._queue_data["q1"].items
105        if item.media_item is not None
106    ]
107
108
109def _load_playing_queue(ctrl: Any, *, with_buffers: bool = False) -> list[QueueItem]:
110    """Fill the queue with items as if it were playing its first one."""
111    items = [QueueItem.from_media_item("q1", _track(item_id)) for item_id in PLAYING_TRACKS]
112    if with_buffers:
113        for item in items:
114            item.streamdetails = MagicMock()
115            item.streamdetails.buffer = MagicMock()
116            item.streamdetails.buffer.clear = AsyncMock()
117    ctrl._queue_data["q1"].items = items
118    queue = _queue(ctrl)
119    queue.items = len(items)
120    queue.state = PlaybackState.PLAYING
121    queue.current_index = 0
122    queue.current_item = items[0]
123    queue.index_in_buffer = 0
124    return items
125
126
127def _observe_item_counts(ctrl: Any) -> list[int]:
128    """Record the number of items every queue-items update publishes."""
129    counts: list[int] = []
130    original = PlayerQueuesController.update_items
131
132    def _spy(queue_id: str, queue_items: list[QueueItem]) -> None:
133        counts.append(len(queue_items))
134        original(ctrl, queue_id, queue_items)
135
136    ctrl.update_items = _spy
137    return counts
138
139
140async def test_replace_never_advertises_an_empty_queue() -> None:
141    """
142    The swap goes out as a single update, so clients never see the queue empty in between.
143
144    The empty state used to be published before the new media was resolved, which meant it stayed
145    on screen for as long as the providers took to answer.
146    """
147    ctrl = _controller()
148    _load_playing_queue(ctrl)
149    counts = _observe_item_counts(ctrl)
150
151    await ctrl.play_media("q1", _playlist(), QueueOption.REPLACE)
152
153    assert counts == [len(NEW_TRACKS)]
154    assert _item_ids(ctrl) == NEW_TRACKS
155
156
157async def test_replace_releases_the_audio_of_the_items_it_swapped_out() -> None:
158    """
159    The outgoing items hand their buffers back once the new item has taken over.
160
161    Nothing reaches an item after the swap, so a buffer left attached would keep its producer -
162    and the provider stream slot behind it - alive until its own inactivity timeout expires.
163    """
164    ctrl = _controller()
165    replaced = _load_playing_queue(ctrl, with_buffers=True)
166    details = [cast("Any", item.streamdetails) for item in replaced]
167    order: list[str] = []
168    for detail in details:
169        detail.buffer.clear = AsyncMock(side_effect=lambda: order.append("release"))
170    buffers = [detail.buffer for detail in details]
171    ctrl.play_index = AsyncMock(side_effect=lambda *_a, **_kw: order.append("play"))
172
173    await ctrl.play_media("q1", _playlist(), QueueOption.REPLACE)
174
175    for buffer in buffers:
176        buffer.clear.assert_awaited_once()
177    assert all(detail.buffer is None for detail in details)
178    # the source slot the outgoing item holds has to be free before the new one is started, or a
179    # provider that allows only one stream refuses the track the user just picked
180    assert order == ["release", "release", "release", "play"]
181
182
183async def test_replace_does_not_hand_the_player_a_next_item_from_the_new_list() -> None:
184    """
185    The buffered index is dropped before the swap, so no stale position picks the next track.
186
187    The player is still sitting on the old playing index while the items are exchanged; loading
188    the new ones against that index would enqueue an arbitrary track as the one to play next.
189    """
190    ctrl = _controller()
191    _load_playing_queue(ctrl)
192    # a live successor, so the branch under test can actually fire
193    ctrl.get_next_item = Mock(
194        side_effect=lambda queue_id, _cur_index: ctrl._queue_data[queue_id].items[0]
195    )
196    enqueued = Mock()
197    ctrl._enqueue_next_item = enqueued
198
199    await ctrl.play_media("q1", _playlist(), QueueOption.REPLACE)
200
201    enqueued.assert_not_called()
202    assert _queue(ctrl).index_in_buffer is None
203
204
205async def test_replace_starting_at_a_chosen_track_never_advertises_a_partial_queue() -> None:
206    """
207    Pinning the track the user picked still publishes the queue exactly once.
208
209    The pinned item used to be loaded on its own before the rest of the batch was shuffled in
210    behind it, which put a one-item queue on screen in between - and the shuffle in between can
211    hit the database, so it is a real window rather than a theoretical one.
212    """
213    ctrl = _controller(shuffle_enabled=True)
214    ctrl._smart_shuffle.is_enabled = Mock(return_value=False)
215    _load_playing_queue(ctrl)
216    counts = _observe_item_counts(ctrl)
217
218    await ctrl.play_media(
219        "q1", _playlist(), QueueOption.REPLACE, start_item="test://track/n1", shuffle=True
220    )
221
222    assert counts == [len(NEW_TRACKS)]
223    # the chosen track is the one that starts, whatever the shuffle did with the rest
224    assert _item_ids(ctrl)[0] == NEW_TRACKS[0]
225    assert sorted(_item_ids(ctrl)) == sorted(NEW_TRACKS)
226
227
228async def test_replacing_a_dynamic_queue_hides_the_rebuild_from_clients() -> None:
229    """
230    The pool is fetched over the network with the queue already emptied, so updates are held off.
231
232    Player updates would otherwise reconcile against the half-built queue and publish it as empty
233    with nothing playing - the very state this is meant to remove, and on the dynamic path the
234    fetch is the slowest part of the whole operation.
235    """
236    ctrl = _controller(is_dynamic=True, shuffle_enabled=True)
237    _load_playing_queue(ctrl)
238    observed: list[bool] = []
239
240    def _record_fill(*_args: Any, **_kwargs: Any) -> list[Track]:
241        observed.append(ctrl._queue_data["q1"].transitioning)
242        return [_track(item_id) for item_id in NEW_TRACKS]
243
244    ctrl._managed_pool.fill = AsyncMock(side_effect=_record_fill)
245
246    await ctrl.play_media("q1", _playlist("dyn1", dynamic=True), QueueOption.REPLACE)
247
248    assert observed == [True]
249    # and the flag is handed back afterwards, or the queue would stop reconciling for good
250    assert ctrl._queue_data["q1"].transitioning is False
251
252
253async def test_replacing_a_dynamic_queue_rebuilds_it_from_the_front() -> None:
254    """
255    A replace onto a smart mix takes the place of what was playing instead of stacking behind it.
256
257    The rebuild normally keeps the current and already-buffered tracks so the crossfade is not
258    disturbed, which is right for an add or a play but would leave a replace appending its pool
259    behind the old queue.
260    """
261    ctrl = _controller(is_dynamic=True, shuffle_enabled=True)
262    _load_playing_queue(ctrl)
263
264    await ctrl.play_media("q1", _playlist("dyn1", dynamic=True), QueueOption.REPLACE)
265
266    assert _queue(ctrl).is_dynamic is True
267    assert _item_ids(ctrl) == NEW_TRACKS
268    ctrl.play_index.assert_awaited_once_with("q1", 0)
269
270
271async def test_replacing_an_ended_queue_keeps_the_sources_it_just_stored() -> None:
272    """
273    A queue picked up from its end is replaced like any other, sources included.
274
275    The finished queue used to be emptied on its way through, which also dropped the sources
276    recorded for the media being started - taking autoplay's seed with them.
277    """
278    ctrl = _controller(ended=True)
279    _load_playing_queue(ctrl)
280    _queue(ctrl).ended = True
281
282    await ctrl.play_media("q1", _playlist(), QueueOption.REPLACE)
283
284    assert _queue(ctrl).ended is False
285    assert [source.item_id for source in ctrl._queue_data["q1"].source_items] == ["pl1"]
286    assert _item_ids(ctrl) == NEW_TRACKS
287