/
/
/
1"""Tests for the party plugin."""
2
3from __future__ import annotations
4
5import asyncio
6from typing import cast
7from unittest.mock import AsyncMock, MagicMock, patch
8
9import pytest
10from music_assistant_models.auth import Scope
11from music_assistant_models.config_entries import ProviderConfig
12from music_assistant_models.enums import ConfigEntryType, MediaType, PlaybackState, ProviderType
13from music_assistant_models.errors import ActionUnavailable, InvalidDataError
14
15from music_assistant.controllers.config import ConfigController
16from music_assistant.helpers.shared_playback import SharedPlaybackMode
17from music_assistant.providers.party import (
18 CONF_ENABLE_ADD_QUEUE,
19 CONF_ENABLE_BOOST,
20 CONF_ENABLE_GUEST_ACCESS,
21 CONF_PARTY_DURATION,
22 CONF_PARTY_MODE,
23 CONF_PREVENT_DUPLICATE_TRACKS,
24 PartyPlugin,
25)
26
27
28def _create_party_plugin() -> PartyPlugin:
29 """Create a minimally configured party plugin for unit tests."""
30 plugin = PartyPlugin.__new__(PartyPlugin)
31 plugin.mass = MagicMock()
32 plugin.mass.music = MagicMock()
33 plugin.mass.player_queues = MagicMock()
34 plugin.logger = MagicMock()
35 plugin.config = MagicMock()
36 plugin._queue_lock = asyncio.Lock()
37 plugin._session = None
38 plugin._session_lock = asyncio.Lock()
39 plugin.get_party_player = AsyncMock(return_value="party_queue") # type: ignore[method-assign]
40 config_values = {
41 CONF_ENABLE_GUEST_ACCESS: True,
42 CONF_ENABLE_BOOST: True,
43 CONF_ENABLE_ADD_QUEUE: True,
44 CONF_PREVENT_DUPLICATE_TRACKS: True,
45 CONF_PARTY_DURATION: 8,
46 }
47 plugin.config.get_value.side_effect = config_values.__getitem__
48 return plugin
49
50
51@pytest.mark.asyncio
52async def test_add_to_queue_rechecks_duplicates_during_priority_insert() -> None:
53 """Reject a duplicate that appears after the initial queue lookup."""
54 plugin = _create_party_plugin()
55 player_queues = cast("MagicMock", plugin.mass.player_queues)
56 music = cast("MagicMock", plugin.mass.music)
57 uri = "spotify://track/123"
58
59 queue = MagicMock()
60 queue.state = PlaybackState.PLAYING
61 queue.current_index = 0
62 queue.index_in_buffer = 0
63 player_queues.get.return_value = queue
64 player_queues.items.return_value = []
65 player_queues.load = AsyncMock()
66
67 async def mutate_queue_during_resolve(_uri: str) -> MagicMock:
68 media_item = MagicMock()
69 media_item.media_type = MediaType.TRACK
70 player_queues.items.return_value = [MagicMock(uri=uri, extra_attributes={})]
71 return media_item
72
73 music.get_item_by_uri = AsyncMock(side_effect=mutate_queue_during_resolve)
74 queue_item = MagicMock()
75 queue_item.extra_attributes = {}
76
77 with (
78 patch("music_assistant.providers.party.build_queue_item", return_value=queue_item),
79 pytest.raises(InvalidDataError, match="already in the queue"),
80 ):
81 await plugin.add_to_queue(uri)
82
83 player_queues.load.assert_not_awaited()
84
85
86def _create_session_test_plugin(mode: str) -> PartyPlugin:
87 """Create a party plugin with a real get_party_player for session tests."""
88 plugin = PartyPlugin.__new__(PartyPlugin)
89 plugin.mass = MagicMock()
90 plugin.logger = MagicMock()
91 plugin.config = MagicMock()
92 plugin._session = None
93 plugin._session_lock = asyncio.Lock()
94 config_values = {
95 CONF_ENABLE_GUEST_ACCESS: True,
96 CONF_PARTY_MODE: mode,
97 }
98 plugin.config.get_value.side_effect = config_values.__getitem__
99 return plugin
100
101
102@pytest.mark.asyncio
103async def test_get_party_player_remote_mode_returns_session_queue() -> None:
104 """In remote mode the party player is the session's virtual player queue."""
105 plugin = _create_session_test_plugin(SharedPlaybackMode.REMOTE.value)
106 session = MagicMock()
107 session.queue_id = "sendspin_virtual_party"
108 plugin._get_session = AsyncMock(return_value=session) # type: ignore[method-assign]
109
110 assert await plugin.get_party_player() == "sendspin_virtual_party"
111
112
113@pytest.mark.asyncio
114async def test_get_party_player_remote_mode_no_session() -> None:
115 """In remote mode without a session (sendspin missing) no queue is returned."""
116 plugin = _create_session_test_plugin(SharedPlaybackMode.REMOTE.value)
117 plugin._get_session = AsyncMock(return_value=None) # type: ignore[method-assign]
118
119 assert await plugin.get_party_player() is None
120
121
122@pytest.mark.asyncio
123async def test_listen_in_without_session_raises() -> None:
124 """Listen-in is rejected when no session is available (venue auto mode)."""
125 plugin = _create_session_test_plugin(SharedPlaybackMode.VENUE.value)
126 plugin._get_or_create_session_locked = AsyncMock(return_value=None) # type: ignore[method-assign]
127
128 with (
129 pytest.raises(InvalidDataError, match="not available"),
130 ):
131 await plugin.listen_in("web_player_1")
132
133
134@pytest.mark.asyncio
135async def test_listen_in_attaches_guest_player() -> None:
136 """Listen-in attaches the guest's web player to the session."""
137 plugin = _create_session_test_plugin(SharedPlaybackMode.REMOTE.value)
138 session = MagicMock()
139 session.queue_id = "sendspin_virtual_party"
140
141 async def _assert_locked(_web_player_id: str) -> None:
142 assert plugin._session_lock.locked()
143
144 session.add_guest_listener = AsyncMock(side_effect=_assert_locked)
145 plugin._get_or_create_session_locked = AsyncMock(return_value=session) # type: ignore[method-assign]
146
147 result = await plugin.listen_in("web_player_1")
148
149 assert result == {"success": True, "queue_id": "sendspin_virtual_party"}
150 session.add_guest_listener.assert_awaited_once_with("web_player_1")
151
152
153@pytest.mark.parametrize("mode", [SharedPlaybackMode.VENUE.value, SharedPlaybackMode.REMOTE.value])
154@pytest.mark.asyncio
155async def test_get_party_config_exposes_mode(mode: str) -> None:
156 """get_party_config surfaces the configured playback mode to the guest frontend."""
157 plugin = _create_party_plugin()
158 cast("MagicMock", plugin.config.get_value).side_effect = {CONF_PARTY_MODE: mode}.get
159
160 config = await plugin.get_party_config()
161
162 assert config.mode == mode
163
164
165@pytest.mark.asyncio
166async def test_guest_readable_commands_use_guest_scope() -> None:
167 """party/url and party/config stay on a guest-readable scope, never a host-only one."""
168 plugin = _create_party_plugin()
169 plugin._unregister_handles = []
170
171 await plugin.loaded_in_mass()
172
173 scopes = {
174 call.args[0]: call.kwargs["required_scope"]
175 for call in cast("MagicMock", plugin.mass.register_api_command).call_args_list
176 }
177 assert scopes["party/url"] == Scope.PROVIDERS_READ
178 assert scopes["party/config"] == Scope.PROVIDERS_READ
179 assert scopes["party/listen_in"] == Scope.PLAYERS_CONTROL
180 assert scopes["party/stop_listen_in"] == Scope.PLAYERS_CONTROL
181 assert scopes["party/can_listen_in"] == Scope.PLAYERS_CONTROL
182
183
184@pytest.mark.parametrize("expiry", [8, 48])
185@pytest.mark.asyncio
186async def test_get_party_url_passes_configured_expiry(expiry: int) -> None:
187 """get_party_url passes the configured expiry through to the join code helper."""
188 plugin = _create_party_plugin()
189 cast("MagicMock", plugin.config.get_value).side_effect = {
190 CONF_ENABLE_GUEST_ACCESS: True,
191 CONF_PARTY_DURATION: expiry,
192 }.get
193
194 guest_user = MagicMock()
195 with (
196 patch(
197 "music_assistant.providers.party.guest_access.get_or_create_guest_user",
198 AsyncMock(return_value=guest_user),
199 ),
200 patch(
201 "music_assistant.providers.party.guest_access.get_or_create_join_code",
202 AsyncMock(return_value="abc123"),
203 ) as mock_get_code,
204 patch(
205 "music_assistant.providers.party.guest_access.build_join_url",
206 return_value="http://example/?join=abc123",
207 ),
208 ):
209 url = await plugin.get_party_url()
210
211 assert url == "http://example/?join=abc123"
212 mock_get_code.assert_awaited_once()
213 assert mock_get_code.call_args.kwargs["expires_in_hours"] == expiry
214
215
216def _create_config_entries_plugin(*, guest_access_enabled: bool) -> PartyPlugin:
217 """Create a party plugin for exercising get_config_entries/handle_config_action."""
218 plugin = PartyPlugin.__new__(PartyPlugin)
219 plugin.mass = MagicMock()
220 plugin.mass.players.all_players.return_value = []
221 plugin.config = MagicMock()
222 # an empty (real) dict, so get_config_value falls through to config.get_value below
223 # instead of taking the "typed entry present" branch a MagicMock would fake
224 plugin.config.values = {}
225 plugin.config.get_value.side_effect = {CONF_ENABLE_GUEST_ACCESS: guest_access_enabled}.get
226 return plugin
227
228
229@pytest.mark.parametrize("guest_access_enabled", [True, False])
230@pytest.mark.asyncio
231async def test_get_config_entries_guest_access_is_a_visible_toggle(
232 guest_access_enabled: bool,
233) -> None:
234 """Guest access is a plain visible boolean; the two removed action buttons are gone."""
235 plugin = _create_config_entries_plugin(guest_access_enabled=guest_access_enabled)
236
237 entries = await plugin.get_config_entries()
238 by_key = {entry.key: entry for entry in entries}
239
240 toggle = by_key[CONF_ENABLE_GUEST_ACCESS]
241 assert toggle.type == ConfigEntryType.BOOLEAN
242 assert toggle.hidden is False
243 assert toggle.immediate_apply is True
244 assert "action_enable_guest_access" not in by_key
245 assert "action_disable_guest_access" not in by_key
246 # the two notes still toggle their visibility from the live guest_access_enabled value
247 assert by_key["guest_disabled_note"].hidden is guest_access_enabled
248 assert by_key["guest_enabled_note"].hidden is (not guest_access_enabled)
249
250
251@pytest.mark.asyncio
252async def test_handle_config_action_rejects_former_guest_access_actions() -> None:
253 """With the buttons removed, their old action ids fall through to the base rejection."""
254 plugin = _create_config_entries_plugin(guest_access_enabled=False)
255
256 with pytest.raises(ActionUnavailable):
257 await plugin.handle_config_action("action_enable_guest_access")
258
259
260def _create_unload_plugin(*, guest_access_enabled: bool) -> PartyPlugin:
261 """Create a party plugin wired for a bare unload() call, with no session and no handles."""
262 plugin = PartyPlugin.__new__(PartyPlugin)
263 plugin.mass = MagicMock()
264 plugin.mass.config.get_raw_provider_config_value.return_value = guest_access_enabled
265 plugin.logger = MagicMock()
266 plugin.config = MagicMock()
267 plugin.config.instance_id = "party--test"
268 plugin._unregister_handles = []
269 plugin._session = None
270 plugin._session_lock = asyncio.Lock()
271 plugin._revoke_guest_tokens = AsyncMock() # type: ignore[method-assign]
272 return plugin
273
274
275@pytest.mark.asyncio
276async def test_unload_revokes_guest_tokens_when_guest_access_is_off() -> None:
277 """Switching guest access off makes the ensuing unload revoke the guest tokens."""
278 plugin = _create_unload_plugin(guest_access_enabled=False)
279
280 await plugin.unload()
281
282 cast("AsyncMock", plugin._revoke_guest_tokens).assert_awaited_once()
283 # the live stored value is what decides, not the config snapshot taken at init, and an
284 # absent key must read as disabled (matching the config entry's default_value)
285 cast("MagicMock", plugin.mass.config.get_raw_provider_config_value).assert_called_once_with(
286 "party--test", CONF_ENABLE_GUEST_ACCESS, default=False
287 )
288
289
290@pytest.mark.asyncio
291async def test_unload_keeps_guest_tokens_while_guest_access_is_on() -> None:
292 """A plain reload with guest access still on leaves the guest tokens intact."""
293 plugin = _create_unload_plugin(guest_access_enabled=True)
294
295 await plugin.unload()
296
297 cast("AsyncMock", plugin._revoke_guest_tokens).assert_not_awaited()
298
299
300@pytest.mark.asyncio
301async def test_unload_revokes_guest_tokens_on_removal() -> None:
302 """Removing the plugin revokes the guest tokens even with guest access still on."""
303 plugin = _create_unload_plugin(guest_access_enabled=True)
304
305 await plugin.unload(is_removed=True)
306
307 cast("AsyncMock", plugin._revoke_guest_tokens).assert_awaited_once()
308
309
310async def _create_stored_config_controller(*, guest_access_enabled: bool) -> ConfigController:
311 """Store the party config the way a real save does and return a controller holding it."""
312 entries = await _create_config_entries_plugin(
313 guest_access_enabled=guest_access_enabled
314 ).get_config_entries()
315 config = ProviderConfig.parse(
316 entries,
317 {"type": ProviderType.PLUGIN, "domain": "party", "instance_id": "party--test"},
318 )
319 config.update({CONF_ENABLE_GUEST_ACCESS: guest_access_enabled})
320
321 controller = ConfigController.__new__(ConfigController)
322 controller._data = {"providers": {"party--test": config.to_raw()}}
323 controller.initialized = True
324 return controller
325
326
327@pytest.mark.parametrize("guest_access_enabled", [True, False])
328@pytest.mark.asyncio
329async def test_stored_guest_access_value_survives_a_save(guest_access_enabled: bool) -> None:
330 """
331 Only non-default values are persisted, so a disabled toggle is stored as an absent key.
332
333 unload() must still read that absent key back as disabled.
334 """
335 controller = await _create_stored_config_controller(guest_access_enabled=guest_access_enabled)
336
337 stored_values = controller._data["providers"]["party--test"]["values"]
338 assert (CONF_ENABLE_GUEST_ACCESS in stored_values) is guest_access_enabled
339 assert (
340 controller.get_raw_provider_config_value(
341 "party--test", CONF_ENABLE_GUEST_ACCESS, default=False
342 )
343 is guest_access_enabled
344 )
345
346
347@pytest.mark.asyncio
348async def test_unload_revokes_guest_tokens_after_a_real_save_of_guest_access_off() -> None:
349 """Switching guest access off and reloading revokes the tokens against real stored config."""
350 plugin = _create_unload_plugin(guest_access_enabled=True)
351 # the real stored config replaces the stub, so the absent key is what drives the outcome
352 plugin.mass.config = await _create_stored_config_controller(guest_access_enabled=False)
353
354 await plugin.unload()
355
356 cast("AsyncMock", plugin._revoke_guest_tokens).assert_awaited_once()
357