/
/
/
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 dashboards.mass.players.get_player.return_value = castplayer # type: ignore[attr-defined]
137 url = "https://mass.example.com?path=%2Fparty"
138 dashboards.mass.dashboard.resolve_dashboard_url = AsyncMock(return_value=url) # type: ignore[method-assign]
139
140 with patch(
141 "music_assistant.providers.chromecast.dashboard.send_show_dashboard"
142 ) as mock_send_show_dashboard:
143 await dashboards._on_show(str(device_uuid), DashboardType.PARTY, None)
144
145 mock_send_show_dashboard.assert_called_once_with(connected_chromecast, url)
146
147
148async def test_on_show_keeps_the_player_from_releasing_the_device() -> None:
149 """A release still pending from a stop on the player would close the dashboard."""
150 dashboards = _make_dashboards()
151 device_uuid = uuid4()
152 connected_chromecast = MagicMock()
153 connected_chromecast.socket_client.is_connected = True
154 castplayer = MagicMock(spec=ChromecastPlayer)
155 castplayer.cc = connected_chromecast
156 dashboards.mass.players.get_player.return_value = castplayer # type: ignore[attr-defined]
157 dashboards.mass.dashboard.resolve_dashboard_url = AsyncMock(return_value="https://mass.test") # type: ignore[method-assign]
158
159 with patch("music_assistant.providers.chromecast.dashboard.send_show_dashboard"):
160 await dashboards._on_show(str(device_uuid), DashboardType.PARTY, None)
161
162 castplayer.cancel_pending_app_quit.assert_called_once()
163
164
165async def test_on_show_reuses_cached_on_demand_connection() -> None:
166 """A cached on-demand connection (no MA player) is reused instead of reconnecting."""
167 dashboards = _make_dashboards()
168 device_uuid = uuid4()
169 connected_chromecast = MagicMock()
170 connected_chromecast.socket_client.is_connected = True
171 dashboards._dashboard_connections[str(device_uuid)] = connected_chromecast
172 url = "https://mass.example.com?path=%2Fparty"
173 dashboards.mass.dashboard.resolve_dashboard_url = AsyncMock(return_value=url) # type: ignore[method-assign]
174
175 with patch(
176 "music_assistant.providers.chromecast.dashboard.send_show_dashboard"
177 ) as mock_send_show_dashboard:
178 await dashboards._on_show(str(device_uuid), DashboardType.PARTY, None)
179
180 mock_send_show_dashboard.assert_called_once_with(connected_chromecast, url)
181
182
183async def test_on_show_raises_for_unknown_device() -> None:
184 """An unknown device_id (not in the browser's discovered devices) is rejected."""
185 dashboards = _make_dashboards()
186 dashboards.provider.browser = MagicMock(devices={})
187 dashboards.mass.dashboard.resolve_dashboard_url = AsyncMock( # type: ignore[method-assign]
188 return_value="https://mass.example.com?path=%2Fparty"
189 )
190
191 with pytest.raises(PlayerUnavailableError):
192 await dashboards._on_show(str(uuid4()), DashboardType.PARTY, None)
193
194
195async def test_on_show_wraps_timeout_error() -> None:
196 """A TimeoutError from the helper surfaces as PlayerUnavailableError, keeping its reason."""
197 dashboards = _make_dashboards()
198 device_uuid = uuid4()
199 connected_chromecast = MagicMock()
200 connected_chromecast.socket_client.is_connected = True
201 connected_chromecast.name = "Living Room TV"
202 dashboards._dashboard_connections[str(device_uuid)] = connected_chromecast
203 dashboards.mass.dashboard.resolve_dashboard_url = AsyncMock( # type: ignore[method-assign]
204 return_value="https://mass.example.com?path=%2Fparty"
205 )
206
207 with (
208 patch(
209 "music_assistant.providers.chromecast.dashboard.send_show_dashboard",
210 side_effect=TimeoutError("Launching app on Living Room TV failed"),
211 ),
212 pytest.raises(PlayerUnavailableError) as exc_info,
213 ):
214 await dashboards._on_show(str(device_uuid), DashboardType.PARTY, None)
215
216 assert str(exc_info.value) == "Launching app on Living Room TV failed"
217 assert exc_info.value.translation_key == "show_dashboard_failed"
218 assert exc_info.value.translation_owner == "provider.chromecast"
219
220
221### on_hide: no-op / eviction behavior
222
223
224async def test_on_hide_noop_when_nothing_connected() -> None:
225 """Hiding on a device with no registered player and no cached connection is a no-op."""
226 dashboards = _make_dashboards()
227 dashboards.provider.browser = MagicMock(devices={})
228
229 with patch(
230 "music_assistant.providers.chromecast.dashboard.send_hide_dashboard"
231 ) as mock_send_hide_dashboard:
232 await dashboards._on_hide(str(uuid4()))
233
234 mock_send_hide_dashboard.assert_not_called()
235
236
237async def test_on_hide_evicts_on_demand_connection() -> None:
238 """Hiding disconnects and drops an on-demand connection cached for the device."""
239 dashboards = _make_dashboards()
240 device_uuid = uuid4()
241 connected_chromecast = MagicMock()
242 connected_chromecast.socket_client.is_connected = True
243 dashboards._dashboard_connections[str(device_uuid)] = connected_chromecast
244
245 with patch("music_assistant.providers.chromecast.dashboard.send_hide_dashboard"):
246 await dashboards._on_hide(str(device_uuid))
247
248 connected_chromecast.disconnect.assert_called_once_with(10)
249 assert str(device_uuid) not in dashboards._dashboard_connections
250
251
252async def test_on_hide_keeps_registered_player_connection() -> None:
253 """Hiding on a live MA player's own connection must not disconnect it."""
254 dashboards = _make_dashboards()
255 device_uuid = uuid4()
256 connected_chromecast = MagicMock()
257 connected_chromecast.socket_client.is_connected = True
258 castplayer = MagicMock(spec=ChromecastPlayer)
259 castplayer.cc = connected_chromecast
260 dashboards.mass.players.get_player.return_value = castplayer # type: ignore[attr-defined]
261
262 with patch("music_assistant.providers.chromecast.dashboard.send_hide_dashboard"):
263 await dashboards._on_hide(str(device_uuid))
264
265 connected_chromecast.disconnect.assert_not_called()
266
267
268async def test_on_hide_with_stale_cache_entry_spares_player_connection() -> None:
269 """A stale on-demand entry alongside a live player must not disconnect the player's cc."""
270 dashboards = _make_dashboards()
271 device_uuid = uuid4()
272 player_cc = MagicMock()
273 player_cc.socket_client.is_connected = True
274 castplayer = MagicMock(spec=ChromecastPlayer)
275 castplayer.cc = player_cc
276 dashboards.mass.players.get_player.return_value = castplayer # type: ignore[attr-defined]
277 stale = MagicMock()
278 dashboards._dashboard_connections[str(device_uuid)] = stale
279
280 with patch("music_assistant.providers.chromecast.dashboard.send_hide_dashboard"):
281 await dashboards._on_hide(str(device_uuid))
282
283 player_cc.disconnect.assert_not_called()
284 stale.disconnect.assert_called_once_with(10)
285 assert str(device_uuid) not in dashboards._dashboard_connections
286
287
288async def test_on_hide_logs_debug_when_nothing_was_showing() -> None:
289 """A False result from send_hide_dashboard is logged at debug, not raised."""
290 dashboards = _make_dashboards()
291 device_uuid = uuid4()
292 connected_chromecast = MagicMock()
293 connected_chromecast.socket_client.is_connected = True
294 dashboards._dashboard_connections[str(device_uuid)] = connected_chromecast
295
296 with patch(
297 "music_assistant.providers.chromecast.dashboard.send_hide_dashboard",
298 return_value=False,
299 ):
300 await dashboards._on_hide(str(device_uuid))
301
302 dashboards.logger.debug.assert_called_once() # type: ignore[attr-defined]
303
304
305### unload cleanup
306
307
308async def test_unload_unregisters_all_and_disconnects_via_executor() -> None:
309 """Unload unregisters every dashboard endpoint and disconnects cached connections."""
310 dashboards = _make_dashboards()
311 unregister_callback = MagicMock()
312 dashboards.mass.dashboard.register_dashboard_handler.return_value = unregister_callback # type: ignore[attr-defined]
313 dashboards.register(uuid4(), _cast_info())
314 chromecast = MagicMock()
315 dashboards._dashboard_connections["on-demand-device"] = chromecast
316
317 await dashboards.unload()
318
319 unregister_callback.assert_called_once()
320 chromecast.disconnect.assert_called_once_with(10)
321 assert dashboards._dashboard_connections == {}
322
323
324async def test_unload_disconnects_without_executor_when_closing() -> None:
325 """During shutdown, disconnects happen synchronously instead of via the executor."""
326 dashboards = _make_dashboards()
327 dashboards.mass.closing = True # type: ignore[misc]
328 chromecast = MagicMock()
329 dashboards._dashboard_connections["on-demand-device"] = chromecast
330
331 await dashboards.unload()
332
333 chromecast.disconnect.assert_called_once_with(0)
334
335
336async def test_register_after_unload_is_noop() -> None:
337 """A late discovery callback after unload must not resurrect a stale registration."""
338 dashboards = _make_dashboards()
339
340 await dashboards.unload()
341 dashboards.register(uuid4(), _cast_info())
342
343 dashboards.mass.dashboard.register_dashboard_handler.assert_not_called() # type: ignore[attr-defined]
344
345
346### Provider wiring: discovery lifecycle hooks
347
348
349def test_discovery_registers_device_with_dashboards() -> None:
350 """Discovering an already-registered player's Cast device still (re-)registers it."""
351 provider = _make_provider()
352 uuid = uuid4()
353 disc_info = _cast_info()
354 disc_info.uuid = uuid
355 provider.browser = MagicMock(devices={uuid: disc_info})
356 castplayer = MagicMock(spec=ChromecastPlayer)
357 castplayer.available = True
358 castplayer.cast_info = MagicMock(is_audio_group=False)
359 castplayer.cc = MagicMock() # discovery fast-path reads castplayer.cc.socket_client
360 provider.mass.players.get_player.return_value = castplayer # type: ignore[attr-defined]
361
362 provider._on_chromecast_discovered(uuid, "service")
363
364 provider.dashboards.register.assert_called_once_with(uuid, disc_info) # type: ignore[attr-defined]
365
366
367def test_discovery_registers_device_even_when_player_disabled() -> None:
368 """A device disabled as a player is still registered as a dashboard endpoint."""
369 provider = _make_provider()
370 uuid = uuid4()
371 disc_info = _cast_info()
372 disc_info.uuid = uuid
373 provider.browser = MagicMock(devices={uuid: disc_info})
374 provider.mass.config.get_raw_player_config_value.return_value = False # type: ignore[attr-defined]
375
376 provider._on_chromecast_discovered(uuid, "service")
377
378 provider.dashboards.register.assert_called_once_with(uuid, disc_info) # type: ignore[attr-defined]
379
380
381def test_removed_callback_unregisters_dashboard() -> None:
382 """A Cast device removed from discovery is unregistered as a dashboard endpoint."""
383 provider = _make_provider()
384 uuid = uuid4()
385
386 provider._on_chromecast_removed(uuid, "service", _cast_info())
387
388 provider.dashboards.unregister.assert_called_once_with(uuid) # type: ignore[attr-defined]
389