music-assistant-server

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        (DashboardType.MUSIC_QUIZ, None, "/music-quiz"),
620    ],
621)
622def test_dashboard_route_resolves_expected_path(
623    dashboard: DashboardType, player_id: str | None, expected_path: str
624) -> None:
625    """Each supported dashboard type resolves to its own frontend route."""
626    controller = _make_controller()
627
628    assert controller._dashboard_route(dashboard, player_id) == expected_path
629
630
631async def test_hide_dashboard_callback_path_invokes_on_hide() -> None:
632    """A callback registration's on_hide is awaited, and the session is dropped."""
633    controller = _make_controller()
634    device = DashboardDevice(dashboard_id="dash1", name="Living Room")
635    on_hide = AsyncMock()
636    controller._dashboards["dash1"] = _RegisteredDashboard(device=device, on_hide=on_hide)
637    controller._sessions["dash1"] = DashboardSession(
638        dashboard_id="dash1", name="Living Room", dashboard=DashboardType.PARTY
639    )
640
641    await controller.hide_dashboard("dash1")
642
643    on_hide.assert_awaited_once()
644    assert "dash1" not in controller._sessions
645    controller.mass.signal_event.assert_called_once_with(  # type: ignore[attr-defined]
646        EventType.DASHBOARD_SESSIONS_UPDATED, data=[]
647    )
648
649
650async def test_hide_dashboard_callback_error_is_swallowed() -> None:
651    """A MusicAssistantError raised by on_hide is caught and logged, not propagated."""
652    controller = _make_controller()
653    device = DashboardDevice(dashboard_id="dash1", name="Living Room")
654    on_hide = AsyncMock(side_effect=MusicAssistantError("nothing to hide"))
655    controller._dashboards["dash1"] = _RegisteredDashboard(device=device, on_hide=on_hide)
656    controller._sessions["dash1"] = DashboardSession(
657        dashboard_id="dash1", name="Living Room", dashboard=DashboardType.PARTY
658    )
659
660    await controller.hide_dashboard("dash1")  # should not raise
661
662    assert "dash1" not in controller._sessions
663
664
665async def test_hide_dashboard_api_path_emits_event() -> None:
666    """An API registration emits DASHBOARD_HIDE and drops the session."""
667    controller = _make_controller()
668    device = DashboardDevice(dashboard_id="dash1", name="Living Room")
669    controller._dashboards["dash1"] = _RegisteredDashboard(device=device)
670    controller._sessions["dash1"] = DashboardSession(
671        dashboard_id="dash1", name="Living Room", dashboard=DashboardType.PARTY
672    )
673
674    await controller.hide_dashboard("dash1")
675
676    controller.mass.signal_event.assert_any_call(  # type: ignore[attr-defined]
677        EventType.DASHBOARD_HIDE, object_id="dash1"
678    )
679    controller.mass.signal_event.assert_any_call(  # type: ignore[attr-defined]
680        EventType.DASHBOARD_SESSIONS_UPDATED, data=[]
681    )
682    assert "dash1" not in controller._sessions
683
684
685async def test_hide_dashboard_unknown_id_is_graceful_noop() -> None:
686    """Hiding an unknown dashboard_id just drops any stale session, never raises."""
687    controller = _make_controller()
688
689    await controller.hide_dashboard("unknown")  # should not raise
690
691    controller.mass.signal_event.assert_called_once_with(  # type: ignore[attr-defined]
692        EventType.DASHBOARD_SESSIONS_UPDATED, data=[]
693    )
694
695
696async def test_end_session_removes_session_and_logs_warning(
697    caplog: pytest.LogCaptureFixture,
698) -> None:
699    """Ending an active session drops it, signals sessions updated, and logs a warning."""
700    controller = _make_controller()
701    controller.logger = logging.getLogger("test.dashboard")
702    device = DashboardDevice(dashboard_id="dash1", name="Living Room")
703    controller._dashboards["dash1"] = _RegisteredDashboard(device=device)
704    controller._sessions["dash1"] = DashboardSession(
705        dashboard_id="dash1", name="Living Room", dashboard=DashboardType.PARTY
706    )
707
708    with caplog.at_level(logging.WARNING, logger="test.dashboard"):
709        controller.end_session("dash1", "the receiver app was closed")
710
711    assert "dash1" not in controller._sessions
712    controller.mass.signal_event.assert_called_once_with(  # type: ignore[attr-defined]
713        EventType.DASHBOARD_SESSIONS_UPDATED, data=[]
714    )
715    assert "Living Room" in caplog.text
716    assert "the receiver app was closed" in caplog.text
717
718
719async def test_end_session_unknown_id_is_noop() -> None:
720    """Ending a session for an id without an active session is a silent no-op."""
721    controller = _make_controller()
722
723    controller.end_session("unknown", "some reason")
724
725    controller.mass.signal_event.assert_not_called()  # type: ignore[attr-defined]
726
727
728async def test_get_dashboard_sessions_returns_stored_sessions() -> None:
729    """The sessions command returns all currently tracked sessions."""
730    controller = _make_controller()
731    session = DashboardSession(
732        dashboard_id="dash1", name="Living Room", dashboard=DashboardType.PARTY
733    )
734    controller._sessions["dash1"] = session
735
736    sessions = await controller.get_dashboard_sessions()
737
738    assert sessions == [session]
739
740
741async def test_get_url_for_dashboard_returns_resolved_url() -> None:
742    """An authorized caller gets a thin wrapper around URL resolution."""
743    controller = _make_controller()
744    controller.mass.webserver.base_url = "https://mass.example.com"  # type: ignore[misc]
745
746    with (
747        patch.object(DashboardController, "_can_resolve_url_for_caller", return_value=True),
748        patch.object(DashboardController, "_get_dashboard_code", AsyncMock(return_value="code456")),
749    ):
750        url = await controller.get_url_for_dashboard(DashboardType.PARTY)
751
752    assert _query(url) == {"dashboard": "code456", "path": "/party"}
753
754
755async def test_get_url_for_dashboard_allows_users_invite_scope() -> None:
756    """A user holding the users.invite scope may resolve any dashboard url."""
757    controller = _make_controller()
758    controller.mass.webserver.base_url = "https://mass.example.com"  # type: ignore[misc]
759    user = MagicMock()
760
761    with (
762        patch(
763            "music_assistant.controllers.dashboard.controller.get_current_user",
764            return_value=user,
765        ),
766        patch(
767            "music_assistant.controllers.dashboard.controller.has_scope",
768            return_value=True,
769        ),
770        patch.object(DashboardController, "_get_dashboard_code", AsyncMock(return_value="code456")),
771    ):
772        url = await controller.get_url_for_dashboard(DashboardType.PARTY)
773
774    assert _query(url) == {"dashboard": "code456", "path": "/party"}
775
776
777async def test_get_url_for_dashboard_allows_owner_with_matching_session() -> None:
778    """A client owning a dashboard with a matching active session may resolve its own url."""
779    controller = _make_controller()
780    controller.mass.webserver.base_url = "https://mass.example.com"  # type: ignore[misc]
781    device = DashboardDevice(dashboard_id="dash1", name="Living Room")
782    controller._dashboards["dash1"] = _RegisteredDashboard(device=device, client_id="client1")
783    controller._sessions["dash1"] = DashboardSession(
784        dashboard_id="dash1", name="Living Room", dashboard=DashboardType.PARTY
785    )
786
787    with (
788        patch(
789            "music_assistant.controllers.dashboard.controller.get_current_user",
790            return_value=None,
791        ),
792        patch(
793            "music_assistant.controllers.dashboard.controller.get_current_client_id",
794            return_value="client1",
795        ),
796        patch.object(DashboardController, "_get_dashboard_code", AsyncMock(return_value="code456")),
797    ):
798        url = await controller.get_url_for_dashboard(DashboardType.PARTY)
799
800    assert _query(url) == {"dashboard": "code456", "path": "/party"}
801
802
803async def test_get_url_for_dashboard_rejects_owner_without_session() -> None:
804    """Owning a dashboard endpoint alone, without an active matching session, isn't enough."""
805    controller = _make_controller()
806    controller.mass.webserver.base_url = "https://mass.example.com"  # type: ignore[misc]
807    device = DashboardDevice(dashboard_id="dash1", name="Living Room")
808    controller._dashboards["dash1"] = _RegisteredDashboard(device=device, client_id="client1")
809
810    with (
811        patch(
812            "music_assistant.controllers.dashboard.controller.get_current_user",
813            return_value=None,
814        ),
815        patch(
816            "music_assistant.controllers.dashboard.controller.get_current_client_id",
817            return_value="client1",
818        ),
819        pytest.raises(InsufficientPermissions),
820    ):
821        await controller.get_url_for_dashboard(DashboardType.PARTY)
822
823
824async def test_get_url_for_dashboard_now_playing_session_must_match_player() -> None:
825    """A now_playing session for a different player does not authorize this request."""
826    controller = _make_controller()
827    controller.mass.webserver.base_url = "https://mass.example.com"  # type: ignore[misc]
828    device = DashboardDevice(dashboard_id="dash1", name="Living Room")
829    controller._dashboards["dash1"] = _RegisteredDashboard(device=device, client_id="client1")
830    controller._sessions["dash1"] = DashboardSession(
831        dashboard_id="dash1",
832        name="Living Room",
833        dashboard=DashboardType.NOW_PLAYING,
834        player_id="player1",
835    )
836
837    with (
838        patch(
839            "music_assistant.controllers.dashboard.controller.get_current_user",
840            return_value=None,
841        ),
842        patch(
843            "music_assistant.controllers.dashboard.controller.get_current_client_id",
844            return_value="client1",
845        ),
846        pytest.raises(InsufficientPermissions),
847    ):
848        await controller.get_url_for_dashboard(DashboardType.NOW_PLAYING, player_id="player2")
849
850
851async def test_handle_client_disconnected_removes_owned_registrations_and_sessions() -> None:
852    """Disconnecting a client drops only that client's registrations and their sessions."""
853    controller = _make_controller()
854    device1 = DashboardDevice(dashboard_id="dash1", name="Living Room")
855    device2 = DashboardDevice(dashboard_id="dash2", name="Kitchen")
856    controller._dashboards["dash1"] = _RegisteredDashboard(device=device1, client_id="client-a")
857    controller._dashboards["dash2"] = _RegisteredDashboard(device=device2, client_id="client-b")
858    controller._sessions["dash1"] = DashboardSession(
859        dashboard_id="dash1", name="Living Room", dashboard=DashboardType.PARTY
860    )
861
862    controller.handle_client_disconnected("client-a")
863
864    assert "dash1" not in controller._dashboards
865    assert "dash1" not in controller._sessions
866    assert "dash2" in controller._dashboards
867    controller.mass.signal_event.assert_any_call(  # type: ignore[attr-defined]
868        EventType.DASHBOARDS_UPDATED, data=[device2]
869    )
870    controller.mass.signal_event.assert_any_call(  # type: ignore[attr-defined]
871        EventType.DASHBOARD_SESSIONS_UPDATED, data=[]
872    )
873
874
875async def test_handle_client_disconnected_unknown_client_is_noop() -> None:
876    """Disconnecting a client with no registrations doesn't touch other clients' state."""
877    controller = _make_controller()
878    device1 = DashboardDevice(dashboard_id="dash1", name="Living Room")
879    controller._dashboards["dash1"] = _RegisteredDashboard(device=device1, client_id="client-a")
880
881    controller.handle_client_disconnected("client-other")
882
883    assert "dash1" in controller._dashboards
884    controller.mass.signal_event.assert_not_called()  # type: ignore[attr-defined]
885
886
887async def test_handle_client_disconnected_only_signals_sessions_when_changed() -> None:
888    """Sessions-updated is only signaled when a session was actually dropped."""
889    controller = _make_controller()
890    device1 = DashboardDevice(dashboard_id="dash1", name="Living Room")
891    controller._dashboards["dash1"] = _RegisteredDashboard(device=device1, client_id="client-a")
892
893    controller.handle_client_disconnected("client-a")
894
895    controller.mass.signal_event.assert_called_once_with(  # type: ignore[attr-defined]
896        EventType.DASHBOARDS_UPDATED, data=[]
897    )
898
899
900async def test_get_dashboard_code_creates_viewer_user_and_code(mass: MusicAssistant) -> None:
901    """A dashboard code can be exchanged for a token belonging to the cast viewer user."""
902    code = await mass.dashboard._get_dashboard_code()
903
904    assert code
905    result = await mass.webserver.auth.exchange_join_code(code)
906
907    assert result["success"] is True
908    user = await mass.webserver.auth.authenticate_with_token(result["access_token"])
909    assert user is not None
910    assert user.username == DASHBOARD_VIEWER_USERNAME
911
912
913async def test_get_dashboard_code_mints_fresh_code_each_call(mass: MusicAssistant) -> None:
914    """Two consecutive mints return different codes and both can be exchanged."""
915    code1 = await mass.dashboard._get_dashboard_code()
916    code2 = await mass.dashboard._get_dashboard_code()
917
918    assert code1 != code2
919
920    result1 = await mass.webserver.auth.exchange_join_code(code1)
921    result2 = await mass.webserver.auth.exchange_join_code(code2)
922
923    assert result1["success"] is True
924    assert result2["success"] is True
925