/
/
/
1"""
2Tests for what a play request naming a live audio source does to the queue.
3
4A live source is not queue content: it plays on the player while the queue keeps
5its own items, so selecting it is the real operation and a play request naming one
6is forwarded there. Replaces the queue-release tests, which covered a source
7leaving a queue it can no longer be in.
8"""
9
10from typing import Any
11from unittest.mock import AsyncMock, MagicMock
12
13import pytest
14from music_assistant_models.enums import QueueOption
15from music_assistant_models.errors import InvalidCommand
16from music_assistant_models.media_items import AudioSource, Track
17from music_assistant_models.media_items.provider_mapping import ProviderMapping
18from music_assistant_models.player_queue import PlayerQueue
19from music_assistant_models.queue_item import QueueItem
20from music_assistant_models.unique_list import UniqueList
21
22from music_assistant.controllers.player_queues import PlayerQueuesController
23from music_assistant.controllers.player_queues.state import PlayerQueueData
24
25QUEUE_ID = "q1"
26SOURCE_URI = "spotify_connect--test://audio_source/main"
27
28
29def _track() -> Track:
30 return Track(
31 item_id="t1",
32 provider="test",
33 name="A track the user queued",
34 artists=UniqueList(),
35 provider_mappings={
36 ProviderMapping(item_id="t1", provider_domain="test", provider_instance="test")
37 },
38 )
39
40
41def _audio_source() -> AudioSource:
42 return AudioSource(
43 item_id="main",
44 provider="spotify_connect--test",
45 name="Spotify Connect",
46 provider_mappings={
47 ProviderMapping(
48 item_id="main",
49 provider_domain="spotify_connect",
50 provider_instance="spotify_connect--test",
51 )
52 },
53 )
54
55
56def _controller(*items: QueueItem) -> Any:
57 """Build a bare controller holding a queue with the given items."""
58 ctrl = PlayerQueuesController.__new__(PlayerQueuesController)
59 ctrl.mass = MagicMock()
60 ctrl.logger = MagicMock()
61 ctrl.mass.players.select_source = AsyncMock()
62 queue = PlayerQueue(
63 queue_id=QUEUE_ID,
64 active=True,
65 display_name="Q1",
66 available=True,
67 items=len(items),
68 )
69 ctrl._queue_data = {QUEUE_ID: PlayerQueueData(queue=queue, items=list(items))}
70 ctrl._check_player_permission = MagicMock() # type: ignore[method-assign]
71 ctrl.get = MagicMock(return_value=queue) # type: ignore[method-assign]
72 ctrl._handle_play_media = AsyncMock() # type: ignore[method-assign]
73 return ctrl
74
75
76async def test_a_source_uri_is_selected_and_the_queue_is_left_alone() -> None:
77 """
78 The source is selected on the player and nothing is enqueued.
79
80 That is the whole point of the source living on the player: the queue keeps the
81 items the user had, so they are still there to resume once the source is done.
82 """
83 existing = QueueItem.from_media_item(QUEUE_ID, _track())
84 ctrl = _controller(existing)
85
86 await ctrl.play_media(QUEUE_ID, SOURCE_URI, QueueOption.REPLACE)
87
88 ctrl.mass.players.select_source.assert_awaited_once_with(QUEUE_ID, SOURCE_URI)
89 ctrl._handle_play_media.assert_not_awaited()
90 assert ctrl._queue_data[QUEUE_ID].items == [existing]
91
92
93async def test_a_source_media_item_is_selected_too() -> None:
94 """A play request carrying the media item rather than its uri behaves the same."""
95 ctrl = _controller()
96
97 await ctrl.play_media(QUEUE_ID, _audio_source(), QueueOption.PLAY)
98
99 ctrl.mass.players.select_source.assert_awaited_once_with(QUEUE_ID, SOURCE_URI)
100 ctrl._handle_play_media.assert_not_awaited()
101
102
103async def test_ordinary_media_still_goes_to_the_queue() -> None:
104 """Anything that is not a live source is enqueued as before."""
105 ctrl = _controller()
106
107 await ctrl.play_media(QUEUE_ID, _track(), QueueOption.PLAY)
108
109 ctrl.mass.players.select_source.assert_not_awaited()
110 ctrl._handle_play_media.assert_awaited_once()
111
112
113async def test_a_source_named_among_other_media_is_refused() -> None:
114 """
115 A live source cannot be combined with other media, so the request is refused.
116
117 It is selected on a player rather than queued, so there is nothing to line it up
118 behind or alongside. Letting the batch through would put it back in the queue,
119 which is the one thing this whole shape exists to stop.
120 """
121 ctrl = _controller()
122
123 with pytest.raises(InvalidCommand, match="plays on its own"):
124 await ctrl.play_media(QUEUE_ID, [SOURCE_URI, "library://track/1"], QueueOption.PLAY)
125
126 ctrl.mass.players.select_source.assert_not_awaited()
127 ctrl._handle_play_media.assert_not_awaited()
128
129
130async def test_a_source_media_item_among_other_media_is_refused_too() -> None:
131 """The same holds when the batch carries the media item rather than its uri."""
132 ctrl = _controller()
133
134 with pytest.raises(InvalidCommand, match="plays on its own"):
135 await ctrl.play_media(QUEUE_ID, [_audio_source(), _track()], QueueOption.PLAY)
136
137 ctrl._handle_play_media.assert_not_awaited()
138
139
140async def test_two_sources_at_once_are_refused() -> None:
141 """A player plays one source at a time, so two in a request is not a choice to make."""
142 ctrl = _controller()
143
144 with pytest.raises(InvalidCommand, match="plays on its own"):
145 await ctrl.play_media(QUEUE_ID, [_audio_source(), _audio_source()], QueueOption.PLAY)
146
147 ctrl.mass.players.select_source.assert_not_awaited()
148