/
/
/
1"""Tests for the chromecast provider's dashboard registration and casting transport."""
2
3from __future__ import annotations
4
5import threading
6from unittest.mock import AsyncMock, MagicMock, patch
7from uuid import uuid4
8
9import pytest
10from music_assistant_models.dashboard import DashboardDevice
11from music_assistant_models.enums import DashboardType
12from music_assistant_models.errors import PlayerUnavailableError
13from pychromecast.const import CAST_TYPE_CHROMECAST
14
15from music_assistant.providers.chromecast.dashboard import ChromecastDashboards
16from music_assistant.providers.chromecast.player import ChromecastPlayer
17from music_assistant.providers.chromecast.provider import ChromecastProvider
18
19
20def _make_dashboards() -> ChromecastDashboards:
21 """Build a ChromecastDashboards instance with a fully mocked owning provider."""
22 provider = MagicMock()
23 provider.translation_owner = "provider.chromecast"
24 provider.domain = "chromecast"
25 provider.mass.closing = False
26 # run_in_executor just calls the function inline, synchronously
27 provider.mass.loop.run_in_executor = AsyncMock(
28 side_effect=lambda _executor, func, *args: func(*args)
29 )
30 provider.mass.players.get_player.return_value = None
31 return ChromecastDashboards(provider)
32
33
34def _cast_info(cast_type: str = CAST_TYPE_CHROMECAST, name: str = "Living Room TV") -> MagicMock:
35 info = MagicMock()
36 info.cast_type = cast_type
37 info.friendly_name = name
38 return info
39
40
41def _make_provider() -> ChromecastProvider:
42 """Build a ChromecastProvider without running its (network-touching) __init__."""
43 provider = ChromecastProvider.__new__(ChromecastProvider)
44 provider.mass = MagicMock()
45 provider.mass.closing = False
46 provider.mass.loop.call_soon_threadsafe = MagicMock(side_effect=lambda func, *args: func(*args))
47 provider.mass.players.get_player.return_value = None
48 provider.mass.config.get_raw_player_config_value.return_value = True
49 provider.logger = MagicMock()
50 provider._discover_lock = threading.Lock()
51 provider._pending_discoveries = set()
52 provider.dashboards = MagicMock()
53 return provider
54
55
56### Registration on discovery
57
58
59def test_register_ignores_non_video_capable_device() -> None:
60 """Audio speakers/groups are not registered as dashboard endpoints."""
61 dashboards = _make_dashboards()
62
63 dashboards.register(uuid4(), _cast_info(cast_type="audio"))
64
65 dashboards.mass.dashboard.register_dashboard_handler.assert_not_called() # type: ignore[attr-defined]
66
67
68def test_register_video_capable_device() -> None:
69 """A video-capable Cast device registers as a dashboard endpoint."""
70 dashboards = _make_dashboards()
71 uuid = uuid4()
72
73 dashboards.register(uuid, _cast_info(name="Living Room TV"))
74
75 dashboards.mass.dashboard.register_dashboard_handler.assert_called_once() # type: ignore[attr-defined]
76 device = dashboards.mass.dashboard.register_dashboard_handler.call_args[0][0] # type: ignore[attr-defined]
77 assert device == DashboardDevice(
78 dashboard_id=f"chromecast_{uuid}",
79 name="Living Room TV",
80 supported_types={
81 DashboardType.PARTY,
82 DashboardType.NOW_PLAYING,
83 DashboardType.MUSIC_QUIZ,
84 },
85 provider_domain_hint="chromecast",
86 )
87
88
89### Unregistration on removal
90
91
92def test_unregister_calls_stored_unregister_callback() -> None:
93 """Removing a discovered device unregisters it from the dashboard controller."""
94 dashboards = _make_dashboards()
95 uuid = uuid4()
96 unregister_callback = MagicMock()
97 dashboards.mass.dashboard.register_dashboard_handler.return_value = unregister_callback # type: ignore[attr-defined]
98 dashboards.register(uuid, _cast_info())
99
100 dashboards.unregister(uuid)
101
102 unregister_callback.assert_called_once()
103
104
105def test_unregister_disconnects_cached_connection() -> None:
106 """Removing a discovered device also drops its cached on-demand connection."""
107 dashboards = _make_dashboards()
108 uuid = uuid4()
109 chromecast = MagicMock()
110 dashboards._dashboard_connections[str(uuid)] = chromecast
111
112 dashboards.unregister(uuid)
113
114 assert not dashboards._dashboard_connections
115 chromecast.disconnect.assert_called_once_with(0)
116
117
118def test_unregister_unknown_device_is_noop() -> None:
119 """Unregistering a device that was never registered does not raise."""
120 dashboards = _make_dashboards()
121
122 dashboards.unregister(uuid4())
123
124
125### on_show: connection reuse and error wrapping
126
127
128async def test_on_show_reuses_connected_player_cc() -> None:
129 """A connected registered player's own Chromecast connection is reused."""
130 dashboards = _make_dashboards()
131 device_uuid = uuid4()
132 connected_chromecast = MagicMock()
133 connected_chromecast.socket_client.is_connected = True
134 castplayer = MagicMock(spec=ChromecastPlayer)
135 castplayer.cc = connected_chromecast
136 castplayer.app_quit_sent = False
137 dashboards.mass.players.get_player.return_value = castplayer # type: ignore[attr-defined]
138 url = "https://mass.example.com?path=%2Fparty"
139 dashboards.mass.dashboard.resolve_dashboard_url = AsyncMock(return_value=url) # type: ignore[method-assign]
140
141 with patch(
142 "music_assistant.providers.chromecast.dashboard.send_show_dashboard"
143 ) as mock_send_show_dashboard:
144 await dashboards._on_show(str(device_uuid), DashboardType.PARTY, None)
145
146 mock_send_show_dashboard.assert_called_once_with(connected_chromecast, url, force_launch=False)
147
148
149async def test_on_show_keeps_the_player_from_releasing_the_device() -> None:
150 """A release still pending from a stop on the player would close the dashboard."""
151 dashboards = _make_dashboards()
152 device_uuid = uuid4()
153 connected_chromecast = MagicMock()
154 connected_chromecast.socket_client.is_connected = True
155 castplayer = MagicMock(spec=ChromecastPlayer)
156 castplayer.cc = connected_chromecast
157 castplayer.app_quit_sent = False
158 dashboards.mass.players.get_player.return_value = castplayer # type: ignore[attr-defined]
159 dashboards.mass.dashboard.resolve_dashboard_url = AsyncMock(return_value="https://mass.test") # type: ignore[method-assign]
160
161 with patch("music_assistant.providers.chromecast.dashboard.send_show_dashboard"):
162 await dashboards._on_show(str(device_uuid), DashboardType.PARTY, None)
163
164 castplayer.cancel_pending_app_quit.assert_called_once()
165
166
167@pytest.mark.parametrize("app_quit_sent", [True, False])
168async def test_on_show_forces_a_launch_when_a_release_is_on_the_wire(
169 app_quit_sent: bool,
170) -> None:
171 """A release that can no longer be cancelled closes the app, so a new session is needed."""
172 dashboards = _make_dashboards()
173 device_uuid = uuid4()
174 connected_chromecast = MagicMock()
175 connected_chromecast.socket_client.is_connected = True
176 castplayer = MagicMock(spec=ChromecastPlayer)
177 castplayer.cc = connected_chromecast
178 castplayer.app_quit_sent = app_quit_sent
179 dashboards.mass.players.get_player.return_value = castplayer # type: ignore[attr-defined]
180 url = "https://mass.example.com?path=%2Fparty"
181 dashboards.mass.dashboard.resolve_dashboard_url = AsyncMock(return_value=url) # type: ignore[method-assign]
182
183 with patch(
184 "music_assistant.providers.chromecast.dashboard.send_show_dashboard"
185 ) as mock_send_show_dashboard:
186 await dashboards._on_show(str(device_uuid), DashboardType.PARTY, None)
187
188 mock_send_show_dashboard.assert_called_once_with(
189 connected_chromecast, url, force_launch=app_quit_sent
190 )
191 # the dashboard just established a session, so the player can trust it again
192 assert castplayer.app_quit_sent is False
193
194
195async def test_on_show_leaves_a_release_sent_while_it_was_launching() -> None:
196 """Only the release this launch replaced is resolved, not one that arrived meanwhile."""
197 dashboards = _make_dashboards()
198 device_uuid = uuid4()
199 connected_chromecast = MagicMock()
200 connected_chromecast.socket_client.is_connected = True
201 castplayer = MagicMock(spec=ChromecastPlayer)
202 castplayer.cc = connected_chromecast
203 castplayer.app_quit_sent = False
204 dashboards.mass.players.get_player.return_value = castplayer # type: ignore[attr-defined]
205 dashboards.mass.dashboard.resolve_dashboard_url = AsyncMock(return_value="https://mass.test") # type: ignore[method-assign]
206
207 def _release_the_device(*_args: object, **_kwargs: object) -> None:
208 castplayer.app_quit_sent = True
209
210 with patch(
211 "music_assistant.providers.chromecast.dashboard.send_show_dashboard",
212 side_effect=_release_the_device,
213 ):
214 await dashboards._on_show(str(device_uuid), DashboardType.PARTY, None)
215
216 assert castplayer.app_quit_sent is True
217
218
219async def test_on_show_reuses_cached_on_demand_connection() -> None:
220 """A cached on-demand connection (no MA player) is reused instead of reconnecting."""
221 dashboards = _make_dashboards()
222 device_uuid = uuid4()
223 connected_chromecast = MagicMock()
224 connected_chromecast.socket_client.is_connected = True
225 dashboards._dashboard_connections[str(device_uuid)] = connected_chromecast
226 url = "https://mass.example.com?path=%2Fparty"
227 dashboards.mass.dashboard.resolve_dashboard_url = AsyncMock(return_value=url) # type: ignore[method-assign]
228
229 with patch(
230 "music_assistant.providers.chromecast.dashboard.send_show_dashboard"
231 ) as mock_send_show_dashboard:
232 await dashboards._on_show(str(device_uuid), DashboardType.PARTY, None)
233
234 mock_send_show_dashboard.assert_called_once_with(connected_chromecast, url, force_launch=False)
235
236
237async def test_on_show_raises_for_unknown_device() -> None:
238 """An unknown device_id (not in the browser's discovered devices) is rejected."""
239 dashboards = _make_dashboards()
240 dashboards.provider.browser = MagicMock(devices={})
241 dashboards.mass.dashboard.resolve_dashboard_url = AsyncMock( # type: ignore[method-assign]
242 return_value="https://mass.example.com?path=%2Fparty"
243 )
244
245 with pytest.raises(PlayerUnavailableError):
246 await dashboards._on_show(str(uuid4()), DashboardType.PARTY, None)
247
248
249async def test_on_show_wraps_timeout_error() -> None:
250 """A TimeoutError from the helper surfaces as PlayerUnavailableError, keeping its reason."""
251 dashboards = _make_dashboards()
252 device_uuid = uuid4()
253 connected_chromecast = MagicMock()
254 connected_chromecast.socket_client.is_connected = True
255 connected_chromecast.name = "Living Room TV"
256 dashboards._dashboard_connections[str(device_uuid)] = connected_chromecast
257 dashboards.mass.dashboard.resolve_dashboard_url = AsyncMock( # type: ignore[method-assign]
258 return_value="https://mass.example.com?path=%2Fparty"
259 )
260
261 with (
262 patch(
263 "music_assistant.providers.chromecast.dashboard.send_show_dashboard",
264 side_effect=TimeoutError("Launching app on Living Room TV failed"),
265 ),
266 pytest.raises(PlayerUnavailableError) as exc_info,
267 ):
268 await dashboards._on_show(str(device_uuid), DashboardType.PARTY, None)
269
270 assert str(exc_info.value) == "Launching app on Living Room TV failed"
271 assert exc_info.value.translation_key == "show_dashboard_failed"
272 assert exc_info.value.translation_owner == "provider.chromecast"
273
274
275### on_hide: no-op / eviction behavior
276
277
278async def test_on_hide_noop_when_nothing_connected() -> None:
279 """Hiding on a device with no registered player and no cached connection is a no-op."""
280 dashboards = _make_dashboards()
281 dashboards.provider.browser = MagicMock(devices={})
282
283 with patch(
284 "music_assistant.providers.chromecast.dashboard.send_hide_dashboard"
285 ) as mock_send_hide_dashboard:
286 await dashboards._on_hide(str(uuid4()))
287
288 mock_send_hide_dashboard.assert_not_called()
289
290
291async def test_on_hide_evicts_on_demand_connection() -> None:
292 """Hiding disconnects and drops an on-demand connection cached for the device."""
293 dashboards = _make_dashboards()
294 device_uuid = uuid4()
295 connected_chromecast = MagicMock()
296 connected_chromecast.socket_client.is_connected = True
297 dashboards._dashboard_connections[str(device_uuid)] = connected_chromecast
298
299 with patch("music_assistant.providers.chromecast.dashboard.send_hide_dashboard"):
300 await dashboards._on_hide(str(device_uuid))
301
302 connected_chromecast.disconnect.assert_called_once_with(10)
303 assert str(device_uuid) not in dashboards._dashboard_connections
304
305
306async def test_on_hide_keeps_registered_player_connection() -> None:
307 """Hiding on a live MA player's own connection must not disconnect it."""
308 dashboards = _make_dashboards()
309 device_uuid = uuid4()
310 connected_chromecast = MagicMock()
311 connected_chromecast.socket_client.is_connected = True
312 castplayer = MagicMock(spec=ChromecastPlayer)
313 castplayer.cc = connected_chromecast
314 dashboards.mass.players.get_player.return_value = castplayer # type: ignore[attr-defined]
315
316 with patch("music_assistant.providers.chromecast.dashboard.send_hide_dashboard"):
317 await dashboards._on_hide(str(device_uuid))
318
319 connected_chromecast.disconnect.assert_not_called()
320
321
322async def test_on_hide_with_stale_cache_entry_spares_player_connection() -> None:
323 """A stale on-demand entry alongside a live player must not disconnect the player's cc."""
324 dashboards = _make_dashboards()
325 device_uuid = uuid4()
326 player_cc = MagicMock()
327 player_cc.socket_client.is_connected = True
328 castplayer = MagicMock(spec=ChromecastPlayer)
329 castplayer.cc = player_cc
330 dashboards.mass.players.get_player.return_value = castplayer # type: ignore[attr-defined]
331 stale = MagicMock()
332 dashboards._dashboard_connections[str(device_uuid)] = stale
333
334 with patch("music_assistant.providers.chromecast.dashboard.send_hide_dashboard"):
335 await dashboards._on_hide(str(device_uuid))
336
337 player_cc.disconnect.assert_not_called()
338 stale.disconnect.assert_called_once_with(10)
339 assert str(device_uuid) not in dashboards._dashboard_connections
340
341
342async def test_on_hide_logs_debug_when_nothing_was_showing() -> None:
343 """A False result from send_hide_dashboard is logged at debug, not raised."""
344 dashboards = _make_dashboards()
345 device_uuid = uuid4()
346 connected_chromecast = MagicMock()
347 connected_chromecast.socket_client.is_connected = True
348 dashboards._dashboard_connections[str(device_uuid)] = connected_chromecast
349
350 with patch(
351 "music_assistant.providers.chromecast.dashboard.send_hide_dashboard",
352 return_value=False,
353 ):
354 await dashboards._on_hide(str(device_uuid))
355
356 dashboards.logger.debug.assert_called_once() # type: ignore[attr-defined]
357
358
359### unload cleanup
360
361
362async def test_unload_unregisters_all_and_disconnects_via_executor() -> None:
363 """Unload unregisters every dashboard endpoint and disconnects cached connections."""
364 dashboards = _make_dashboards()
365 unregister_callback = MagicMock()
366 dashboards.mass.dashboard.register_dashboard_handler.return_value = unregister_callback # type: ignore[attr-defined]
367 dashboards.register(uuid4(), _cast_info())
368 chromecast = MagicMock()
369 dashboards._dashboard_connections["on-demand-device"] = chromecast
370
371 await dashboards.unload()
372
373 unregister_callback.assert_called_once()
374 chromecast.disconnect.assert_called_once_with(10)
375 assert dashboards._dashboard_connections == {}
376
377
378async def test_unload_disconnects_without_executor_when_closing() -> None:
379 """During shutdown, disconnects happen synchronously instead of via the executor."""
380 dashboards = _make_dashboards()
381 dashboards.mass.closing = True # type: ignore[misc]
382 chromecast = MagicMock()
383 dashboards._dashboard_connections["on-demand-device"] = chromecast
384
385 await dashboards.unload()
386
387 chromecast.disconnect.assert_called_once_with(0)
388
389
390async def test_register_after_unload_is_noop() -> None:
391 """A late discovery callback after unload must not resurrect a stale registration."""
392 dashboards = _make_dashboards()
393
394 await dashboards.unload()
395 dashboards.register(uuid4(), _cast_info())
396
397 dashboards.mass.dashboard.register_dashboard_handler.assert_not_called() # type: ignore[attr-defined]
398
399
400### Provider wiring: discovery lifecycle hooks
401
402
403def test_discovery_registers_device_with_dashboards() -> None:
404 """Discovering an already-registered player's Cast device still (re-)registers it."""
405 provider = _make_provider()
406 uuid = uuid4()
407 disc_info = _cast_info()
408 disc_info.uuid = uuid
409 provider.browser = MagicMock(devices={uuid: disc_info})
410 castplayer = MagicMock(spec=ChromecastPlayer)
411 castplayer.available = True
412 castplayer.cast_info = MagicMock(is_audio_group=False)
413 castplayer.cc = MagicMock() # discovery fast-path reads castplayer.cc.socket_client
414 provider.mass.players.get_player.return_value = castplayer # type: ignore[attr-defined]
415
416 provider._on_chromecast_discovered(uuid, "service")
417
418 provider.dashboards.register.assert_called_once_with(uuid, disc_info) # type: ignore[attr-defined]
419
420
421def test_discovery_registers_device_even_when_player_disabled() -> None:
422 """A device disabled as a player is still registered as a dashboard endpoint."""
423 provider = _make_provider()
424 uuid = uuid4()
425 disc_info = _cast_info()
426 disc_info.uuid = uuid
427 provider.browser = MagicMock(devices={uuid: disc_info})
428 provider.mass.config.get_raw_player_config_value.return_value = False # type: ignore[attr-defined]
429
430 provider._on_chromecast_discovered(uuid, "service")
431
432 provider.dashboards.register.assert_called_once_with(uuid, disc_info) # type: ignore[attr-defined]
433
434
435def test_removed_callback_unregisters_dashboard() -> None:
436 """A Cast device removed from discovery is unregistered as a dashboard endpoint."""
437 provider = _make_provider()
438 uuid = uuid4()
439
440 provider._on_chromecast_removed(uuid, "service", _cast_info())
441
442 provider.dashboards.unregister.assert_called_once_with(uuid) # type: ignore[attr-defined]
443