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