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