music-assistant-server

1"""
2Tests for the notifications a plugin gets when a queue drops or hands over its AudioSource.
3
4Clearing the queue is the one moment the owning plugin has no other signal to go on: the
5stream serving the source may have been torn down long before (a paused source ends its
6stream), so without this notification the plugin never learns that MA is done with it. A
7transfer of a paused source is the same blind spot - it moves to another queue without a
8stream request, so the plugin has to be told which queue holds it now.
9"""
10
11from __future__ import annotations
12
13from typing import Any
14from unittest.mock import AsyncMock, MagicMock, Mock
15
16import pytest
17from music_assistant_models.enums import PlaybackState, QueueOption
18from music_assistant_models.errors import MediaNotFoundError
19from music_assistant_models.media_items import AudioSource, ProviderMapping, Track
20from music_assistant_models.player_queue import PlayerQueue
21from music_assistant_models.queue_item import QueueItem
22
23from music_assistant.controllers.player_queues import PlayerQueuesController
24from music_assistant.controllers.player_queues.state import PlayerQueueData
25from music_assistant.models.plugin import PluginProvider
26
27
28def _audio_source() -> AudioSource:
29    """Build the live AudioSource a plugin exposes."""
30    return AudioSource(
31        item_id="main",
32        provider="spotify_connect--test",
33        name="Spotify Connect",
34        provider_mappings={
35            ProviderMapping(
36                item_id="main",
37                provider_domain="spotify_connect",
38                provider_instance="spotify_connect--test",
39            )
40        },
41    )
42
43
44def _track() -> Track:
45    """Build a plain track, i.e. a current item with no plugin behind it."""
46    return Track(
47        item_id="t1",
48        provider="test",
49        name="Track t1",
50        provider_mappings={
51            ProviderMapping(item_id="t1", provider_domain="test", provider_instance="test")
52        },
53    )
54
55
56def _controller(current_item: QueueItem | None) -> Any:
57    """
58    Build a bare controller holding a queue "q1" playing the given item, plus an empty "q2".
59
60    Tasks created along the way are closed instead of scheduled, so the notification can be
61    asserted on the plugin mock without a running loop.
62
63    :param current_item: The item the queue has as its current one, None for an empty queue.
64    """
65    ctrl = PlayerQueuesController.__new__(PlayerQueuesController)
66    ctrl.logger = Mock()
67    ctrl.mass = MagicMock()
68    ctrl.mass.create_task.side_effect = lambda coroutine: coroutine.close()
69    ctrl.signal_update = Mock()  # type: ignore[method-assign]
70    ctrl._managed_pool = Mock()
71    ctrl._smart_shuffle = Mock()
72    queue = PlayerQueue(queue_id="q1", active=True, display_name="Q1", available=True, items=0)
73    queue.current_item = current_item
74    target = PlayerQueue(queue_id="q2", active=True, display_name="Q2", available=True, items=0)
75    ctrl._queue_data = {"q1": PlayerQueueData(queue=queue), "q2": PlayerQueueData(queue=target)}
76    if current_item is not None:
77        ctrl._queue_data["q1"].items = [current_item]
78        queue.items = 1
79    return ctrl
80
81
82def _plugin_provider(ctrl: Any) -> Any:
83    """Make the controller resolve the AudioSource's provider to a plugin provider."""
84    provider = MagicMock(spec=PluginProvider)
85    ctrl.mass.get_provider.return_value = provider
86    return provider
87
88
89def test_clearing_the_queue_notifies_the_plugin() -> None:
90    """The plugin owning the current AudioSource is told the queue dropped it."""
91    ctrl = _controller(QueueItem.from_media_item("q1", _audio_source()))
92    provider = _plugin_provider(ctrl)
93
94    ctrl.clear("q1")
95
96    provider.on_source_removed.assert_called_once_with("main", "q1")
97
98
99def test_clearing_a_queue_of_tracks_notifies_nothing() -> None:
100    """A regular track has no plugin behind it to notify."""
101    ctrl = _controller(QueueItem.from_media_item("q1", _track()))
102    provider = _plugin_provider(ctrl)
103
104    ctrl.clear("q1")
105
106    provider.on_source_removed.assert_not_called()
107
108
109def test_clearing_an_empty_queue_notifies_nothing() -> None:
110    """A queue with nothing playing has no source to release."""
111    ctrl = _controller(None)
112    provider = _plugin_provider(ctrl)
113
114    ctrl.clear("q1")
115
116    provider.on_source_removed.assert_not_called()
117
118
119def test_replacing_or_transferring_the_queue_notifies_nothing() -> None:
120    """
121    Only the user clearing their queue releases the source, not the internal clear.
122
123    Playing the same live source again and transferring the queue to another player both drop
124    the queue's items on their way to re-selecting that very source. Releasing it there would
125    end the upstream session and cost the user the position they were paused at. A replace
126    that brings different media is a gap this leaves open, not a case it decides.
127    """
128    ctrl = _controller(QueueItem.from_media_item("q1", _audio_source()))
129    provider = _plugin_provider(ctrl)
130
131    ctrl._clear("q1", skip_stop=True)
132
133    provider.on_source_removed.assert_not_called()
134
135
136def test_clearing_a_queue_whose_plugin_is_gone_still_clears() -> None:
137    """An unloaded plugin must not keep the user from clearing their queue."""
138    ctrl = _controller(QueueItem.from_media_item("q1", _audio_source()))
139    ctrl.mass.get_provider.return_value = None
140
141    ctrl.clear("q1")
142
143    assert ctrl._queue_data["q1"].items == []
144
145
146def _ready_for_transfer(ctrl: Any, target_current_item: QueueItem | None = None) -> None:
147    """
148    Stub out the playback side of transfer_queue, leaving the handover itself real.
149
150    :param ctrl: The controller to prepare.
151    :param target_current_item: What the target queue "q2" is playing when the transfer lands.
152    """
153    ctrl.stop = AsyncMock()
154    ctrl.load = AsyncMock()
155
156    def _apply_items(queue_id: str, items: list[QueueItem]) -> None:
157        ctrl._queue_data[queue_id].items = items
158
159    # only the assignment the real one does: the queues have to end up holding what they were
160    # handed, or a check on what survived the transfer reads the items it was supposed to replace
161    ctrl.update_items = Mock(side_effect=_apply_items)
162    ctrl._queue_data["q1"].queue.state = PlaybackState.PAUSED
163    if target_current_item is not None:
164        ctrl._queue_data["q2"].queue.current_item = target_current_item
165        ctrl._queue_data["q2"].items = [target_current_item]
166    target_player = MagicMock()
167    target_player.state.active_group = None
168    target_player.state.synced_to = None
169    ctrl.mass.players.get_player.return_value = target_player
170
171
172async def test_transferring_a_paused_source_notifies_the_plugin() -> None:
173    """
174    The plugin is told which queue holds its source now.
175
176    A paused source is moved without a stream request, so nothing else would tell the plugin
177    it changed hands - and every later callback would arrive for the queue it left behind.
178    """
179    ctrl = _controller(QueueItem.from_media_item("q1", _audio_source()))
180    provider = _plugin_provider(ctrl)
181    _ready_for_transfer(ctrl)
182
183    await ctrl.transfer_queue("q1", "q2")
184
185    provider.on_source_transferred.assert_awaited_once_with("main", "q1", "q2")
186
187
188async def test_transferring_a_queue_of_tracks_notifies_nothing() -> None:
189    """A queue without a live source has no plugin to hand anything over to."""
190    ctrl = _controller(QueueItem.from_media_item("q1", _track()))
191    provider = _plugin_provider(ctrl)
192    _ready_for_transfer(ctrl)
193
194    await ctrl.transfer_queue("q1", "q2")
195
196    provider.on_source_transferred.assert_not_awaited()
197
198
199async def test_a_plugin_raising_does_not_break_the_transfer() -> None:
200    """A buggy plugin must not strand the queue halfway through a transfer."""
201    ctrl = _controller(QueueItem.from_media_item("q1", _audio_source()))
202    provider = _plugin_provider(ctrl)
203    provider.on_source_transferred.side_effect = RuntimeError("plugin is broken")
204    _ready_for_transfer(ctrl)
205
206    await ctrl.transfer_queue("q1", "q2")
207
208    # reaching this at all means the raise was contained; the queue still changed hands
209    provider.on_source_transferred.assert_awaited_once()
210    assert ctrl._queue_data["q2"].queue.current_item is not None
211
212
213def _play_media_replacing_with(ctrl: Any, *items: QueueItem) -> None:
214    """Stand in for the resolve/enqueue path, leaving the queue holding the given items."""
215    ctrl._check_player_permission = Mock()
216
217    async def _handle(*_args: Any, **_kwargs: Any) -> None:
218        ctrl._queue_data["q1"].items = list(items)
219
220    ctrl._handle_play_media = _handle
221
222
223async def test_starting_other_media_releases_the_source_it_replaced() -> None:
224    """
225    Media that pushes a live source out of the queue releases it.
226
227    Album, artist and playlist all default to "replace", so this - not the clear button - is how
228    a paused live source usually leaves the queue.
229    """
230    ctrl = _controller(QueueItem.from_media_item("q1", _audio_source()))
231    provider = _plugin_provider(ctrl)
232    _play_media_replacing_with(ctrl, QueueItem.from_media_item("q1", _track()))
233
234    await ctrl.play_media("q1", "test://album/al1", QueueOption.REPLACE)
235
236    provider.on_source_removed.assert_called_once_with("main", "q1")
237
238
239async def test_starting_the_same_source_again_keeps_it() -> None:
240    """
241    Playing the paused source again must not release it.
242
243    A replace drops the queue's items before loading the new ones, so the source briefly leaves
244    the queue on its way back in. Releasing it there would end the upstream session and restart
245    playback from the top instead of resuming where the user paused.
246    """
247    ctrl = _controller(QueueItem.from_media_item("q1", _audio_source()))
248    provider = _plugin_provider(ctrl)
249    # the source comes back as a freshly built queue item, so identity is by media, not item id
250    _play_media_replacing_with(ctrl, QueueItem.from_media_item("q1", _audio_source()))
251
252    await ctrl.play_media("q1", "spotify_connect--test://audio_source/main", QueueOption.REPLACE)
253
254    provider.on_source_removed.assert_not_called()
255
256
257async def test_media_played_alongside_the_source_keeps_it() -> None:
258    """An option that leaves the source in the queue is not the user done with it."""
259    source_item = QueueItem.from_media_item("q1", _audio_source())
260    ctrl = _controller(source_item)
261    provider = _plugin_provider(ctrl)
262    _play_media_replacing_with(ctrl, source_item, QueueItem.from_media_item("q1", _track()))
263
264    await ctrl.play_media("q1", "test://album/al1", QueueOption.PLAY)
265
266    provider.on_source_removed.assert_not_called()
267
268
269async def test_media_that_fails_to_load_still_releases_the_emptied_queue() -> None:
270    """
271    A replace gives up the queue's items before it knows the new media is playable.
272
273    Nothing plays afterwards and the source is just as gone, so it has to be released even
274    though starting the media failed.
275    """
276    ctrl = _controller(QueueItem.from_media_item("q1", _audio_source()))
277    provider = _plugin_provider(ctrl)
278    _play_media_replacing_with(ctrl)
279
280    async def _fail(*_args: Any, **_kwargs: Any) -> None:
281        ctrl._queue_data["q1"].items = []
282        raise MediaNotFoundError("No playable items found")
283
284    ctrl._handle_play_media = _fail
285
286    with pytest.raises(MediaNotFoundError):
287        await ctrl.play_media("q1", "test://album/gone", QueueOption.REPLACE)
288
289    provider.on_source_removed.assert_called_once_with("main", "q1")
290
291
292async def test_transferring_onto_a_paused_source_releases_it() -> None:
293    """
294    A queue moved onto a player that was paused on a live source releases that source.
295
296    The transfer overwrites what the target was playing, so its source leaves the queue just as
297    surely as a clear would take it - and nothing else would tell its plugin.
298    """
299    ctrl = _controller(QueueItem.from_media_item("q1", _track()))
300    provider = _plugin_provider(ctrl)
301    _ready_for_transfer(ctrl, target_current_item=QueueItem.from_media_item("q2", _audio_source()))
302
303    await ctrl.transfer_queue("q1", "q2")
304
305    provider.on_source_removed.assert_called_once_with("main", "q2")
306
307
308async def test_a_source_left_behind_by_a_play_is_still_released_later() -> None:
309    """
310    A source kept in the queue alongside other media is released when it does leave.
311
312    Starting a track with "play" leaves the live source in the queue behind it, so nothing is
313    released then. Clearing the queue afterwards is what finally removes it - and by then it is
314    no longer the item the queue is playing.
315    """
316    source_item = QueueItem.from_media_item("q1", _audio_source())
317    ctrl = _controller(source_item)
318    provider = _plugin_provider(ctrl)
319    # as a "play" leaves it: the source still queued, a track playing in front of it
320    track_item = QueueItem.from_media_item("q1", _track())
321    ctrl._queue_data["q1"].items = [source_item, track_item]
322    ctrl._queue_data["q1"].queue.current_item = track_item
323
324    ctrl.clear("q1")
325
326    provider.on_source_removed.assert_called_once_with("main", "q1")
327
328
329async def test_a_source_left_behind_by_a_play_is_released_when_replaced() -> None:
330    """The same source, no longer playing, is released when new media takes the queue over."""
331    source_item = QueueItem.from_media_item("q1", _audio_source())
332    ctrl = _controller(source_item)
333    provider = _plugin_provider(ctrl)
334    track_item = QueueItem.from_media_item("q1", _track())
335    ctrl._queue_data["q1"].items = [source_item, track_item]
336    ctrl._queue_data["q1"].queue.current_item = track_item
337    _play_media_replacing_with(ctrl, QueueItem.from_media_item("q1", _track()))
338
339    await ctrl.play_media("q1", "test://album/al1", QueueOption.REPLACE)
340
341    provider.on_source_removed.assert_called_once_with("main", "q1")
342
343
344async def test_transferring_hands_over_a_source_that_was_not_playing() -> None:
345    """
346    Every live source the queue holds moves with it, not just the one that was playing.
347
348    A source left in the queue by a "play" travels to the target like any other item, so its
349    plugin has to follow it there - otherwise a later clear on the target releases nothing.
350    """
351    source_item = QueueItem.from_media_item("q1", _audio_source())
352    track_item = QueueItem.from_media_item("q1", _track())
353    ctrl = _controller(track_item)
354    provider = _plugin_provider(ctrl)
355    ctrl._queue_data["q1"].items = [source_item, track_item]
356    _ready_for_transfer(ctrl)
357
358    await ctrl.transfer_queue("q1", "q2")
359
360    provider.on_source_transferred.assert_awaited_once_with("main", "q1", "q2")
361