music-assistant-server

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