/
/
/
1"""Tests for the DashboardController."""
2
3from __future__ import annotations
4
5import logging
6from unittest.mock import AsyncMock, MagicMock, patch
7from urllib.parse import parse_qs, urlparse
8
9import pytest
10from music_assistant_models.dashboard import DashboardDevice, DashboardSession
11from music_assistant_models.enums import DashboardType, EventType
12from music_assistant_models.errors import (
13 ActionUnavailable,
14 InsufficientPermissions,
15 InvalidCommand,
16 MusicAssistantError,
17)
18
19from music_assistant.controllers.dashboard import DashboardController
20from music_assistant.controllers.dashboard.controller import (
21 ALL_DASHBOARD_TYPES,
22 DASHBOARD_VIEWER_USERNAME,
23 _RegisteredDashboard,
24)
25from music_assistant.mass import MusicAssistant
26
27
28def _make_controller() -> DashboardController:
29 """Build a DashboardController instance without running its (network-touching) __init__."""
30 controller = DashboardController.__new__(DashboardController)
31 controller.mass = MagicMock()
32 controller.logger = MagicMock()
33 controller._dashboards = {}
34 controller._sessions = {}
35 # neither casting mechanism configured by default: individual tests opt in
36 controller.mass.webserver.base_url = ""
37 controller.mass.webserver.remote_access.is_enabled = False
38 controller.mass.webserver.remote_access.remote_id = ""
39 controller.mass.version = "0.0.0"
40 return controller
41
42
43def _query(url: str) -> dict[str, str]:
44 """Parse a URL's query string into a flat dict for easy assertions."""
45 return {key: values[0] for key, values in parse_qs(urlparse(url).query).items()}
46
47
48async def test_resolve_dashboard_url_uses_https_base_url_when_configured() -> None:
49 """An https base url is preferred and used same-origin, without a remote_id."""
50 controller = _make_controller()
51 controller.mass.webserver.base_url = "https://mass.example.com" # type: ignore[misc]
52 controller.mass.webserver.remote_access.is_enabled = True # type: ignore[misc]
53 controller.mass.webserver.remote_access.remote_id = "remote123" # type: ignore[misc]
54
55 with patch.object(
56 DashboardController, "_get_dashboard_code", AsyncMock(return_value="code456")
57 ):
58 url = await controller.resolve_dashboard_url(DashboardType.PARTY, None)
59
60 assert url.startswith("https://mass.example.com?")
61 query = _query(url)
62 assert query == {"dashboard": "code456", "path": "/party"}
63
64
65async def test_resolve_dashboard_url_uses_remote_access_when_no_https_base() -> None:
66 """Falls back to the app.music-assistant.io signaling portal, including the remote_id."""
67 controller = _make_controller()
68 controller.mass.webserver.base_url = "http://192.168.1.10:8095" # type: ignore[misc]
69 controller.mass.webserver.remote_access.is_enabled = True # type: ignore[misc]
70 controller.mass.webserver.remote_access.remote_id = "remote123" # type: ignore[misc]
71 controller.mass.version = "2.17.150"
72
73 with patch.object(
74 DashboardController, "_get_dashboard_code", AsyncMock(return_value="code456")
75 ):
76 url = await controller.resolve_dashboard_url(DashboardType.PARTY, None)
77
78 assert url.startswith("https://app.music-assistant.io/stable/?")
79 query = _query(url)
80 assert query == {"remote_id": "remote123", "dashboard": "code456", "path": "/party"}
81
82
83async def test_resolve_dashboard_url_raises_when_neither_configured() -> None:
84 """Neither an https base url nor remote access is available: casting is unavailable."""
85 controller = _make_controller()
86
87 with pytest.raises(ActionUnavailable):
88 await controller.resolve_dashboard_url(DashboardType.PARTY, None)
89
90
91async def test_resolve_dashboard_url_now_playing_requires_player_id() -> None:
92 """The now_playing dashboard cannot be shown without a player_id."""
93 controller = _make_controller()
94 controller.mass.webserver.base_url = "https://mass.example.com" # type: ignore[misc]
95
96 with pytest.raises(InvalidCommand):
97 await controller.resolve_dashboard_url(DashboardType.NOW_PLAYING, None)
98
99
100async def test_resolve_dashboard_url_now_playing_embeds_player_id_in_path() -> None:
101 """The now_playing route carries the player_id as its own query param."""
102 controller = _make_controller()
103 controller.mass.webserver.base_url = "https://mass.example.com" # type: ignore[misc]
104
105 with patch.object(
106 DashboardController, "_get_dashboard_code", AsyncMock(return_value="code456")
107 ):
108 url = await controller.resolve_dashboard_url(DashboardType.NOW_PLAYING, "player1")
109
110 query = _query(url)
111 assert query["path"] == "/now-playing?player=player1"
112
113
114async def test_resolve_dashboard_url_encodes_player_id() -> None:
115 """A player_id with reserved characters is url-encoded in the now_playing route."""
116 controller = _make_controller()
117 controller.mass.webserver.base_url = "https://mass.example.com" # type: ignore[misc]
118
119 with patch.object(
120 DashboardController, "_get_dashboard_code", AsyncMock(return_value="code456")
121 ):
122 url = await controller.resolve_dashboard_url(DashboardType.NOW_PLAYING, "a&b=c d")
123
124 query = _query(url)
125 assert query["path"] == "/now-playing?player=a%26b%3Dc+d"
126
127
128@pytest.mark.parametrize(
129 ("player_id", "expected"),
130 [
131 # dlna/wiim UDNs and MAC-based ids (bluesound, squeezelite, samsung_wam)
132 ("uuid:FF98F7F4-EEC9-70AF", "uuid:FF98F7F4-EEC9-70AF"),
133 ("20:F8:3B:09:6B:92", "20:F8:3B:09:6B:92"),
134 # alexa uses the device name the user typed, verbatim
135 ("Marvin's Echo (Kitchen)!", "Marvin's+Echo+(Kitchen)!"),
136 # every remaining safe character, so the set cannot silently shrink
137 ("a*b@c,d;e$f/g", "a*b@c,d;e$f/g"),
138 ],
139)
140async def test_resolve_dashboard_url_keeps_route_safe_chars_literal(
141 player_id: str, expected: str
142) -> None:
143 """The now_playing route leaves the characters the frontend's router keeps literal."""
144 controller = _make_controller()
145 controller.mass.webserver.base_url = "https://mass.example.com" # type: ignore[misc]
146
147 with patch.object(
148 DashboardController, "_get_dashboard_code", AsyncMock(return_value="code456")
149 ):
150 url = await controller.resolve_dashboard_url(DashboardType.NOW_PLAYING, player_id)
151
152 assert _query(url)["path"] == f"/now-playing?player={expected}"
153
154
155async def test_resolve_dashboard_url_rejects_unknown_dashboard_type() -> None:
156 """An UNKNOWN dashboard type is never a valid target to cast."""
157 controller = _make_controller()
158 controller.mass.webserver.base_url = "https://mass.example.com" # type: ignore[misc]
159
160 with pytest.raises(InvalidCommand):
161 await controller.resolve_dashboard_url(DashboardType.UNKNOWN, None)
162
163
164async def test_resolve_dashboard_url_prefer_local_uses_plain_http_base() -> None:
165 """prefer_local returns the plain local base url, bypassing the https/remote requirement."""
166 controller = _make_controller()
167 controller.mass.webserver.base_url = "http://192.168.1.10:8095" # type: ignore[misc]
168 # neither an https base nor remote access is configured: the default path would raise,
169 # but prefer_local skips that gate for native LAN apps
170 with patch.object(
171 DashboardController, "_get_dashboard_code", AsyncMock(return_value="code456")
172 ):
173 url = await controller.resolve_dashboard_url(
174 DashboardType.NOW_PLAYING, "player1", prefer_local=True
175 )
176
177 assert url.startswith("http://192.168.1.10:8095?")
178 query = _query(url)
179 assert query == {"dashboard": "code456", "path": "/now-playing?player=player1"}
180
181
182@pytest.mark.parametrize(
183 ("version", "expected_channel"),
184 [
185 ("0.0.0", "nightly"),
186 ("2.17.150.dev202501271200", "nightly"),
187 ("2.17.150b1", "beta"),
188 ("2.17.150rc1", "beta"),
189 ("2.17.150", "stable"),
190 ],
191)
192def test_frontend_channel_derivation(version: str, expected_channel: str) -> None:
193 """The app.music-assistant.io channel is derived from the server version string."""
194 controller = _make_controller()
195 controller.mass.version = version
196
197 assert controller._frontend_channel() == expected_channel
198
199
200async def test_register_dashboard_stores_registration_and_signals() -> None:
201 """Registering stores the dashboard endpoint and signals DASHBOARDS_UPDATED."""
202 controller = _make_controller()
203
204 with patch(
205 "music_assistant.controllers.dashboard.controller.get_current_client_id",
206 return_value="client1",
207 ):
208 await controller.register_dashboard(
209 "dash1", "Living Room", provider_domain_hint="chromecast"
210 )
211
212 device = controller._dashboards["dash1"].device
213 assert device == DashboardDevice(
214 dashboard_id="dash1",
215 name="Living Room",
216 supported_types=set(ALL_DASHBOARD_TYPES),
217 provider_domain_hint="chromecast",
218 )
219 controller.mass.signal_event.assert_called_once_with( # type: ignore[attr-defined]
220 EventType.DASHBOARDS_UPDATED, data=[device]
221 )
222
223
224async def test_register_dashboard_respects_explicit_supported_types() -> None:
225 """Explicit supported_types are stored as given, not defaulted to all types."""
226 controller = _make_controller()
227
228 with patch(
229 "music_assistant.controllers.dashboard.controller.get_current_client_id",
230 return_value="client1",
231 ):
232 await controller.register_dashboard(
233 "dash1", "Living Room", supported_types={DashboardType.PARTY}
234 )
235
236 assert controller._dashboards["dash1"].device.supported_types == {DashboardType.PARTY}
237
238
239async def test_register_dashboard_stores_calling_client_id() -> None:
240 """The registering websocket client id is captured on the registration."""
241 controller = _make_controller()
242
243 with patch(
244 "music_assistant.controllers.dashboard.controller.get_current_client_id",
245 return_value="client123",
246 ):
247 await controller.register_dashboard("dash1", "Living Room")
248
249 assert controller._dashboards["dash1"].client_id == "client123"
250
251
252async def test_register_dashboard_requires_websocket_client() -> None:
253 """Registering without a websocket client context is rejected."""
254 controller = _make_controller()
255
256 with (
257 patch(
258 "music_assistant.controllers.dashboard.controller.get_current_client_id",
259 return_value=None,
260 ),
261 pytest.raises(InvalidCommand),
262 ):
263 await controller.register_dashboard("dash1", "Living Room")
264
265 assert "dash1" not in controller._dashboards
266
267
268async def test_register_dashboard_rejects_empty_supported_types() -> None:
269 """An explicitly empty supported_types set is a client bug, not 'supports nothing'."""
270 controller = _make_controller()
271
272 with (
273 patch(
274 "music_assistant.controllers.dashboard.controller.get_current_client_id",
275 return_value="client1",
276 ),
277 pytest.raises(InvalidCommand),
278 ):
279 await controller.register_dashboard("dash1", "Living Room", supported_types=set())
280
281 assert "dash1" not in controller._dashboards
282
283
284async def test_register_dashboard_rejects_foreign_owner_collision() -> None:
285 """A dashboard_id already registered by another client cannot be taken over."""
286 controller = _make_controller()
287
288 with patch(
289 "music_assistant.controllers.dashboard.controller.get_current_client_id",
290 return_value="client1",
291 ):
292 await controller.register_dashboard("dash1", "Living Room")
293
294 with (
295 patch(
296 "music_assistant.controllers.dashboard.controller.get_current_client_id",
297 return_value="client2",
298 ),
299 pytest.raises(InvalidCommand),
300 ):
301 await controller.register_dashboard("dash1", "Kitchen")
302
303 assert controller._dashboards["dash1"].device.name == "Living Room"
304
305
306async def test_register_dashboard_rejects_callback_owned_collision() -> None:
307 """A dashboard_id already registered by an in-server callback cannot be taken over."""
308 controller = _make_controller()
309 device = DashboardDevice(dashboard_id="dash1", name="Living Room")
310 controller._dashboards["dash1"] = _RegisteredDashboard(
311 device=device, on_show=AsyncMock(), on_hide=AsyncMock()
312 )
313
314 with (
315 patch(
316 "music_assistant.controllers.dashboard.controller.get_current_client_id",
317 return_value="client1",
318 ),
319 pytest.raises(InvalidCommand),
320 ):
321 await controller.register_dashboard("dash1", "Kitchen")
322
323 assert controller._dashboards["dash1"].device.name == "Living Room"
324
325
326async def test_register_dashboard_replaces_existing_registration() -> None:
327 """Re-registering an existing dashboard_id replaces it (e.g. client reconnect)."""
328 controller = _make_controller()
329
330 with patch(
331 "music_assistant.controllers.dashboard.controller.get_current_client_id",
332 return_value="client1",
333 ):
334 await controller.register_dashboard("dash1", "Old Name")
335 await controller.register_dashboard(
336 "dash1", "New Name", supported_types={DashboardType.PARTY}
337 )
338
339 assert len(controller._dashboards) == 1
340 assert controller._dashboards["dash1"].device.name == "New Name"
341 assert controller._dashboards["dash1"].device.supported_types == {DashboardType.PARTY}
342
343
344async def test_register_dashboard_unchanged_reregistration_does_not_signal() -> None:
345 """Re-registering with identical device data does not spam DASHBOARDS_UPDATED."""
346 controller = _make_controller()
347
348 with patch(
349 "music_assistant.controllers.dashboard.controller.get_current_client_id",
350 return_value="client1",
351 ):
352 await controller.register_dashboard("dash1", "Living Room")
353 controller.mass.signal_event.reset_mock() # type: ignore[attr-defined]
354 await controller.register_dashboard("dash1", "Living Room")
355
356 controller.mass.signal_event.assert_not_called() # type: ignore[attr-defined]
357
358
359async def test_register_dashboard_handler_unchanged_reregistration_does_not_signal() -> None:
360 """Handler re-registration with identical device data does not spam DASHBOARDS_UPDATED."""
361 controller = _make_controller()
362 device = DashboardDevice(
363 dashboard_id="dash1", name="Living Room", supported_types={DashboardType.PARTY}
364 )
365
366 controller.register_dashboard_handler(device, AsyncMock(), AsyncMock())
367 controller.mass.signal_event.reset_mock() # type: ignore[attr-defined]
368 controller.register_dashboard_handler(device, AsyncMock(), AsyncMock())
369
370 controller.mass.signal_event.assert_not_called() # type: ignore[attr-defined]
371
372
373async def test_unregister_dashboard_removes_registration_and_session() -> None:
374 """Unregistering drops the registration and any active session, signaling both events."""
375 controller = _make_controller()
376 device = DashboardDevice(dashboard_id="dash1", name="Living Room")
377 controller._dashboards["dash1"] = _RegisteredDashboard(device=device, client_id="client1")
378 controller._sessions["dash1"] = DashboardSession(
379 dashboard_id="dash1", name="Living Room", dashboard=DashboardType.PARTY
380 )
381
382 with patch(
383 "music_assistant.controllers.dashboard.controller.get_current_client_id",
384 return_value="client1",
385 ):
386 await controller.unregister_dashboard("dash1")
387
388 assert "dash1" not in controller._dashboards
389 assert "dash1" not in controller._sessions
390 controller.mass.signal_event.assert_any_call( # type: ignore[attr-defined]
391 EventType.DASHBOARDS_UPDATED, data=[]
392 )
393 controller.mass.signal_event.assert_any_call( # type: ignore[attr-defined]
394 EventType.DASHBOARD_SESSIONS_UPDATED, data=[]
395 )
396 assert controller.mass.signal_event.call_count == 2 # type: ignore[attr-defined]
397
398
399async def test_unregister_dashboard_without_session_only_signals_dashboards() -> None:
400 """Unregistering an endpoint with no active session doesn't signal sessions updated."""
401 controller = _make_controller()
402 device = DashboardDevice(dashboard_id="dash1", name="Living Room")
403 controller._dashboards["dash1"] = _RegisteredDashboard(device=device, client_id="client1")
404
405 with patch(
406 "music_assistant.controllers.dashboard.controller.get_current_client_id",
407 return_value="client1",
408 ):
409 await controller.unregister_dashboard("dash1")
410
411 controller.mass.signal_event.assert_called_once_with( # type: ignore[attr-defined]
412 EventType.DASHBOARDS_UPDATED, data=[]
413 )
414
415
416async def test_unregister_dashboard_unknown_id_is_noop() -> None:
417 """Unregistering an unknown dashboard_id is a graceful no-op, no events signaled."""
418 controller = _make_controller()
419
420 with patch(
421 "music_assistant.controllers.dashboard.controller.get_current_client_id",
422 return_value="client1",
423 ):
424 await controller.unregister_dashboard("unknown")
425
426 controller.mass.signal_event.assert_not_called() # type: ignore[attr-defined]
427
428
429async def test_unregister_dashboard_requires_websocket_client() -> None:
430 """Unregistering without a websocket client context is rejected."""
431 controller = _make_controller()
432
433 with (
434 patch(
435 "music_assistant.controllers.dashboard.controller.get_current_client_id",
436 return_value=None,
437 ),
438 pytest.raises(InvalidCommand),
439 ):
440 await controller.unregister_dashboard("dash1")
441
442
443async def test_unregister_dashboard_rejects_foreign_owner() -> None:
444 """Unregistering a dashboard_id owned by a different client is rejected."""
445 controller = _make_controller()
446 device = DashboardDevice(dashboard_id="dash1", name="Living Room")
447 controller._dashboards["dash1"] = _RegisteredDashboard(device=device, client_id="client1")
448
449 with (
450 patch(
451 "music_assistant.controllers.dashboard.controller.get_current_client_id",
452 return_value="client2",
453 ),
454 pytest.raises(InvalidCommand),
455 ):
456 await controller.unregister_dashboard("dash1")
457
458 assert "dash1" in controller._dashboards
459
460
461async def test_unregister_dashboard_rejects_callback_owned_entry() -> None:
462 """A callback-registered (client_id=None) entry cannot be unregistered via the API."""
463 controller = _make_controller()
464 device = DashboardDevice(dashboard_id="dash1", name="Living Room")
465 controller._dashboards["dash1"] = _RegisteredDashboard(
466 device=device, on_show=AsyncMock(), on_hide=AsyncMock()
467 )
468
469 with (
470 patch(
471 "music_assistant.controllers.dashboard.controller.get_current_client_id",
472 return_value="client1",
473 ),
474 pytest.raises(InvalidCommand),
475 ):
476 await controller.unregister_dashboard("dash1")
477
478 assert "dash1" in controller._dashboards
479
480
481async def test_get_dashboards_returns_all_registered() -> None:
482 """All registered dashboard endpoints are returned when no filter is given."""
483 controller = _make_controller()
484 device1 = DashboardDevice(dashboard_id="dash1", name="Living Room")
485 device2 = DashboardDevice(
486 dashboard_id="dash2", name="Kitchen", supported_types={DashboardType.PARTY}
487 )
488 controller._dashboards["dash1"] = _RegisteredDashboard(device=device1)
489 controller._dashboards["dash2"] = _RegisteredDashboard(device=device2)
490
491 devices = await controller.get_dashboards()
492
493 assert devices == [device1, device2]
494
495
496async def test_get_dashboards_filters_by_supported_type() -> None:
497 """Only endpoints declaring the requested dashboard type are returned."""
498 controller = _make_controller()
499 device1 = DashboardDevice(
500 dashboard_id="dash1", name="Living Room", supported_types={DashboardType.PARTY}
501 )
502 device2 = DashboardDevice(
503 dashboard_id="dash2", name="Kitchen", supported_types={DashboardType.NOW_PLAYING}
504 )
505 controller._dashboards["dash1"] = _RegisteredDashboard(device=device1)
506 controller._dashboards["dash2"] = _RegisteredDashboard(device=device2)
507
508 devices = await controller.get_dashboards(dashboard=DashboardType.PARTY)
509
510 assert devices == [device1]
511
512
513async def test_show_dashboard_rejects_unknown_dashboard_id() -> None:
514 """Showing fails when the dashboard_id does not resolve to a registration."""
515 controller = _make_controller()
516
517 with pytest.raises(InvalidCommand):
518 await controller.show_dashboard("unknown", DashboardType.PARTY)
519
520
521async def test_show_dashboard_rejects_unsupported_type() -> None:
522 """Showing fails when the registered endpoint doesn't support the requested type."""
523 controller = _make_controller()
524 device = DashboardDevice(
525 dashboard_id="dash1", name="Living Room", supported_types={DashboardType.NOW_PLAYING}
526 )
527 controller._dashboards["dash1"] = _RegisteredDashboard(device=device)
528
529 with pytest.raises(InvalidCommand):
530 await controller.show_dashboard("dash1", DashboardType.PARTY)
531
532
533async def test_show_dashboard_callback_path_invokes_on_show() -> None:
534 """A callback registration delegates to on_show with the dashboard and player_id only."""
535 controller = _make_controller()
536 device = DashboardDevice(
537 dashboard_id="dash1", name="Living Room", supported_types=set(ALL_DASHBOARD_TYPES)
538 )
539 on_show = AsyncMock()
540 controller._dashboards["dash1"] = _RegisteredDashboard(
541 device=device, on_show=on_show, on_hide=AsyncMock()
542 )
543
544 await controller.show_dashboard("dash1", DashboardType.PARTY)
545
546 # the controller no longer resolves the url; the consumer does that itself
547 on_show.assert_awaited_once_with(DashboardType.PARTY, None)
548
549 session = controller._sessions["dash1"]
550 assert session == DashboardSession(
551 dashboard_id="dash1", name="Living Room", dashboard=DashboardType.PARTY
552 )
553 controller.mass.signal_event.assert_called_once_with( # type: ignore[attr-defined]
554 EventType.DASHBOARD_SESSIONS_UPDATED, data=[session]
555 )
556
557
558async def test_show_dashboard_callback_error_propagates_without_storing_session() -> None:
559 """If on_show raises, the error propagates and no session is ever stored."""
560 controller = _make_controller()
561 device = DashboardDevice(
562 dashboard_id="dash1", name="Living Room", supported_types=set(ALL_DASHBOARD_TYPES)
563 )
564 on_show = AsyncMock(side_effect=MusicAssistantError("cast failed"))
565 controller._dashboards["dash1"] = _RegisteredDashboard(device=device, on_show=on_show)
566
567 with pytest.raises(MusicAssistantError):
568 await controller.show_dashboard("dash1", DashboardType.PARTY)
569
570 assert "dash1" not in controller._sessions
571 controller.mass.signal_event.assert_not_called() # type: ignore[attr-defined]
572
573
574async def test_show_dashboard_api_path_emits_event_without_url() -> None:
575 """An API registration emits DASHBOARD_SHOW with the session but no url, and stores it."""
576 controller = _make_controller()
577 device = DashboardDevice(
578 dashboard_id="dash1", name="Living Room", supported_types=set(ALL_DASHBOARD_TYPES)
579 )
580 controller._dashboards["dash1"] = _RegisteredDashboard(device=device)
581
582 await controller.show_dashboard("dash1", DashboardType.NOW_PLAYING, player_id="player1")
583
584 session = DashboardSession(
585 dashboard_id="dash1",
586 name="Living Room",
587 dashboard=DashboardType.NOW_PLAYING,
588 player_id="player1",
589 )
590 show_call = next(
591 call
592 for call in controller.mass.signal_event.call_args_list # type: ignore[attr-defined]
593 if call.args[0] == EventType.DASHBOARD_SHOW
594 )
595 assert show_call.kwargs == {"object_id": "dash1", "data": session}
596 assert controller._sessions["dash1"] == session
597
598
599async def test_show_dashboard_api_path_now_playing_requires_player_id() -> None:
600 """now_playing without a player_id is rejected before any session/event is committed."""
601 controller = _make_controller()
602 device = DashboardDevice(
603 dashboard_id="dash1", name="Living Room", supported_types=set(ALL_DASHBOARD_TYPES)
604 )
605 controller._dashboards["dash1"] = _RegisteredDashboard(device=device)
606
607 with pytest.raises(InvalidCommand):
608 await controller.show_dashboard("dash1", DashboardType.NOW_PLAYING)
609
610 assert "dash1" not in controller._sessions
611 controller.mass.signal_event.assert_not_called() # type: ignore[attr-defined]
612
613
614@pytest.mark.parametrize(
615 ("dashboard", "player_id", "expected_path"),
616 [
617 (DashboardType.PARTY, None, "/party"),
618 (DashboardType.NOW_PLAYING, "player1", "/now-playing?player=player1"),
619 # the viewer-only kiosk view, not the host page (which needs USERS_INVITE)
620 (DashboardType.MUSIC_QUIZ, None, "/music-quiz/dashboard"),
621 ],
622)
623def test_dashboard_route_resolves_expected_path(
624 dashboard: DashboardType, player_id: str | None, expected_path: str
625) -> None:
626 """Each supported dashboard type resolves to its own frontend route."""
627 controller = _make_controller()
628
629 assert controller._dashboard_route(dashboard, player_id) == expected_path
630
631
632async def test_hide_dashboard_callback_path_invokes_on_hide() -> None:
633 """A callback registration's on_hide is awaited, and the session is dropped."""
634 controller = _make_controller()
635 device = DashboardDevice(dashboard_id="dash1", name="Living Room")
636 on_hide = AsyncMock()
637 controller._dashboards["dash1"] = _RegisteredDashboard(device=device, on_hide=on_hide)
638 controller._sessions["dash1"] = DashboardSession(
639 dashboard_id="dash1", name="Living Room", dashboard=DashboardType.PARTY
640 )
641
642 await controller.hide_dashboard("dash1")
643
644 on_hide.assert_awaited_once()
645 assert "dash1" not in controller._sessions
646 controller.mass.signal_event.assert_called_once_with( # type: ignore[attr-defined]
647 EventType.DASHBOARD_SESSIONS_UPDATED, data=[]
648 )
649
650
651async def test_hide_dashboard_callback_error_is_swallowed() -> None:
652 """A MusicAssistantError raised by on_hide is caught and logged, not propagated."""
653 controller = _make_controller()
654 device = DashboardDevice(dashboard_id="dash1", name="Living Room")
655 on_hide = AsyncMock(side_effect=MusicAssistantError("nothing to hide"))
656 controller._dashboards["dash1"] = _RegisteredDashboard(device=device, on_hide=on_hide)
657 controller._sessions["dash1"] = DashboardSession(
658 dashboard_id="dash1", name="Living Room", dashboard=DashboardType.PARTY
659 )
660
661 await controller.hide_dashboard("dash1") # should not raise
662
663 assert "dash1" not in controller._sessions
664
665
666async def test_hide_dashboard_api_path_emits_event() -> None:
667 """An API registration emits DASHBOARD_HIDE and drops the session."""
668 controller = _make_controller()
669 device = DashboardDevice(dashboard_id="dash1", name="Living Room")
670 controller._dashboards["dash1"] = _RegisteredDashboard(device=device)
671 controller._sessions["dash1"] = DashboardSession(
672 dashboard_id="dash1", name="Living Room", dashboard=DashboardType.PARTY
673 )
674
675 await controller.hide_dashboard("dash1")
676
677 controller.mass.signal_event.assert_any_call( # type: ignore[attr-defined]
678 EventType.DASHBOARD_HIDE, object_id="dash1"
679 )
680 controller.mass.signal_event.assert_any_call( # type: ignore[attr-defined]
681 EventType.DASHBOARD_SESSIONS_UPDATED, data=[]
682 )
683 assert "dash1" not in controller._sessions
684
685
686async def test_hide_dashboard_unknown_id_is_graceful_noop() -> None:
687 """Hiding an unknown dashboard_id just drops any stale session, never raises."""
688 controller = _make_controller()
689
690 await controller.hide_dashboard("unknown") # should not raise
691
692 controller.mass.signal_event.assert_called_once_with( # type: ignore[attr-defined]
693 EventType.DASHBOARD_SESSIONS_UPDATED, data=[]
694 )
695
696
697async def test_end_session_removes_session_and_logs_warning(
698 caplog: pytest.LogCaptureFixture,
699) -> None:
700 """Ending an active session drops it, signals sessions updated, and logs a warning."""
701 controller = _make_controller()
702 controller.logger = logging.getLogger("test.dashboard")
703 device = DashboardDevice(dashboard_id="dash1", name="Living Room")
704 controller._dashboards["dash1"] = _RegisteredDashboard(device=device)
705 controller._sessions["dash1"] = DashboardSession(
706 dashboard_id="dash1", name="Living Room", dashboard=DashboardType.PARTY
707 )
708
709 with caplog.at_level(logging.WARNING, logger="test.dashboard"):
710 controller.end_session("dash1", "the receiver app was closed")
711
712 assert "dash1" not in controller._sessions
713 controller.mass.signal_event.assert_called_once_with( # type: ignore[attr-defined]
714 EventType.DASHBOARD_SESSIONS_UPDATED, data=[]
715 )
716 assert "Living Room" in caplog.text
717 assert "the receiver app was closed" in caplog.text
718
719
720async def test_end_session_unknown_id_is_noop() -> None:
721 """Ending a session for an id without an active session is a silent no-op."""
722 controller = _make_controller()
723
724 controller.end_session("unknown", "some reason")
725
726 controller.mass.signal_event.assert_not_called() # type: ignore[attr-defined]
727
728
729async def test_get_dashboard_sessions_returns_stored_sessions() -> None:
730 """The sessions command returns all currently tracked sessions."""
731 controller = _make_controller()
732 session = DashboardSession(
733 dashboard_id="dash1", name="Living Room", dashboard=DashboardType.PARTY
734 )
735 controller._sessions["dash1"] = session
736
737 sessions = await controller.get_dashboard_sessions()
738
739 assert sessions == [session]
740
741
742async def test_get_url_for_dashboard_returns_resolved_url() -> None:
743 """An authorized caller gets a thin wrapper around URL resolution."""
744 controller = _make_controller()
745 controller.mass.webserver.base_url = "https://mass.example.com" # type: ignore[misc]
746
747 with (
748 patch.object(DashboardController, "_can_resolve_url_for_caller", return_value=True),
749 patch.object(DashboardController, "_get_dashboard_code", AsyncMock(return_value="code456")),
750 ):
751 url = await controller.get_url_for_dashboard(DashboardType.PARTY)
752
753 assert _query(url) == {"dashboard": "code456", "path": "/party"}
754
755
756async def test_get_url_for_dashboard_allows_users_invite_scope() -> None:
757 """A user holding the users.invite scope may resolve any dashboard url."""
758 controller = _make_controller()
759 controller.mass.webserver.base_url = "https://mass.example.com" # type: ignore[misc]
760 user = MagicMock()
761
762 with (
763 patch(
764 "music_assistant.controllers.dashboard.controller.get_current_user",
765 return_value=user,
766 ),
767 patch(
768 "music_assistant.controllers.dashboard.controller.has_scope",
769 return_value=True,
770 ),
771 patch.object(DashboardController, "_get_dashboard_code", AsyncMock(return_value="code456")),
772 ):
773 url = await controller.get_url_for_dashboard(DashboardType.PARTY)
774
775 assert _query(url) == {"dashboard": "code456", "path": "/party"}
776
777
778async def test_get_url_for_dashboard_allows_owner_with_matching_session() -> None:
779 """A client owning a dashboard with a matching active session may resolve its own url."""
780 controller = _make_controller()
781 controller.mass.webserver.base_url = "https://mass.example.com" # type: ignore[misc]
782 device = DashboardDevice(dashboard_id="dash1", name="Living Room")
783 controller._dashboards["dash1"] = _RegisteredDashboard(device=device, client_id="client1")
784 controller._sessions["dash1"] = DashboardSession(
785 dashboard_id="dash1", name="Living Room", dashboard=DashboardType.PARTY
786 )
787
788 with (
789 patch(
790 "music_assistant.controllers.dashboard.controller.get_current_user",
791 return_value=None,
792 ),
793 patch(
794 "music_assistant.controllers.dashboard.controller.get_current_client_id",
795 return_value="client1",
796 ),
797 patch.object(DashboardController, "_get_dashboard_code", AsyncMock(return_value="code456")),
798 ):
799 url = await controller.get_url_for_dashboard(DashboardType.PARTY)
800
801 assert _query(url) == {"dashboard": "code456", "path": "/party"}
802
803
804async def test_get_url_for_dashboard_rejects_owner_without_session() -> None:
805 """Owning a dashboard endpoint alone, without an active matching session, isn't enough."""
806 controller = _make_controller()
807 controller.mass.webserver.base_url = "https://mass.example.com" # type: ignore[misc]
808 device = DashboardDevice(dashboard_id="dash1", name="Living Room")
809 controller._dashboards["dash1"] = _RegisteredDashboard(device=device, client_id="client1")
810
811 with (
812 patch(
813 "music_assistant.controllers.dashboard.controller.get_current_user",
814 return_value=None,
815 ),
816 patch(
817 "music_assistant.controllers.dashboard.controller.get_current_client_id",
818 return_value="client1",
819 ),
820 pytest.raises(InsufficientPermissions),
821 ):
822 await controller.get_url_for_dashboard(DashboardType.PARTY)
823
824
825async def test_get_url_for_dashboard_now_playing_session_must_match_player() -> None:
826 """A now_playing session for a different player does not authorize this request."""
827 controller = _make_controller()
828 controller.mass.webserver.base_url = "https://mass.example.com" # type: ignore[misc]
829 device = DashboardDevice(dashboard_id="dash1", name="Living Room")
830 controller._dashboards["dash1"] = _RegisteredDashboard(device=device, client_id="client1")
831 controller._sessions["dash1"] = DashboardSession(
832 dashboard_id="dash1",
833 name="Living Room",
834 dashboard=DashboardType.NOW_PLAYING,
835 player_id="player1",
836 )
837
838 with (
839 patch(
840 "music_assistant.controllers.dashboard.controller.get_current_user",
841 return_value=None,
842 ),
843 patch(
844 "music_assistant.controllers.dashboard.controller.get_current_client_id",
845 return_value="client1",
846 ),
847 pytest.raises(InsufficientPermissions),
848 ):
849 await controller.get_url_for_dashboard(DashboardType.NOW_PLAYING, player_id="player2")
850
851
852async def test_handle_client_disconnected_removes_owned_registrations_and_sessions() -> None:
853 """Disconnecting a client drops only that client's registrations and their sessions."""
854 controller = _make_controller()
855 device1 = DashboardDevice(dashboard_id="dash1", name="Living Room")
856 device2 = DashboardDevice(dashboard_id="dash2", name="Kitchen")
857 controller._dashboards["dash1"] = _RegisteredDashboard(device=device1, client_id="client-a")
858 controller._dashboards["dash2"] = _RegisteredDashboard(device=device2, client_id="client-b")
859 controller._sessions["dash1"] = DashboardSession(
860 dashboard_id="dash1", name="Living Room", dashboard=DashboardType.PARTY
861 )
862
863 controller.handle_client_disconnected("client-a")
864
865 assert "dash1" not in controller._dashboards
866 assert "dash1" not in controller._sessions
867 assert "dash2" in controller._dashboards
868 controller.mass.signal_event.assert_any_call( # type: ignore[attr-defined]
869 EventType.DASHBOARDS_UPDATED, data=[device2]
870 )
871 controller.mass.signal_event.assert_any_call( # type: ignore[attr-defined]
872 EventType.DASHBOARD_SESSIONS_UPDATED, data=[]
873 )
874
875
876async def test_handle_client_disconnected_unknown_client_is_noop() -> None:
877 """Disconnecting a client with no registrations doesn't touch other clients' state."""
878 controller = _make_controller()
879 device1 = DashboardDevice(dashboard_id="dash1", name="Living Room")
880 controller._dashboards["dash1"] = _RegisteredDashboard(device=device1, client_id="client-a")
881
882 controller.handle_client_disconnected("client-other")
883
884 assert "dash1" in controller._dashboards
885 controller.mass.signal_event.assert_not_called() # type: ignore[attr-defined]
886
887
888async def test_handle_client_disconnected_only_signals_sessions_when_changed() -> None:
889 """Sessions-updated is only signaled when a session was actually dropped."""
890 controller = _make_controller()
891 device1 = DashboardDevice(dashboard_id="dash1", name="Living Room")
892 controller._dashboards["dash1"] = _RegisteredDashboard(device=device1, client_id="client-a")
893
894 controller.handle_client_disconnected("client-a")
895
896 controller.mass.signal_event.assert_called_once_with( # type: ignore[attr-defined]
897 EventType.DASHBOARDS_UPDATED, data=[]
898 )
899
900
901async def test_get_dashboard_code_creates_viewer_user_and_code(mass: MusicAssistant) -> None:
902 """A dashboard code can be exchanged for a token belonging to the cast viewer user."""
903 code = await mass.dashboard._get_dashboard_code()
904
905 assert code
906 result = await mass.webserver.auth.exchange_join_code(code)
907
908 assert result["success"] is True
909 user = await mass.webserver.auth.authenticate_with_token(result["access_token"])
910 assert user is not None
911 assert user.username == DASHBOARD_VIEWER_USERNAME
912
913
914async def test_get_dashboard_code_mints_fresh_code_each_call(mass: MusicAssistant) -> None:
915 """Two consecutive mints return different codes and both can be exchanged."""
916 code1 = await mass.dashboard._get_dashboard_code()
917 code2 = await mass.dashboard._get_dashboard_code()
918
919 assert code1 != code2
920
921 result1 = await mass.webserver.auth.exchange_join_code(code1)
922 result2 = await mass.webserver.auth.exchange_join_code(code2)
923
924 assert result1["success"] is True
925 assert result2["success"] is True
926