/
/
/
1"""Tests for the Fully Kiosk provider's dashboard registration and display control."""
2
3from __future__ import annotations
4
5from typing import cast
6from unittest.mock import AsyncMock, MagicMock, patch
7
8import pytest
9from fullykiosk.exceptions import FullyKioskError
10from music_assistant_models.enums import DashboardType
11from music_assistant_models.errors import PlayerCommandFailed, PlayerUnavailableError
12
13from music_assistant.constants import CONF_SSL_FINGERPRINT, CONF_USE_SSL, CONF_VERIFY_SSL
14from music_assistant.providers.fully_kiosk.dashboard import FullyKioskDashboards
15from music_assistant.providers.fully_kiosk.player import FullyKioskPlayer
16from music_assistant.providers.fully_kiosk.provider import FullyKioskProvider
17from tests.common import MockProvider, create_mock_config
18
19PLAYER_ID = "fully_kiosk_10.0.0.5_2323"
20
21
22def _make_dashboards(player: FullyKioskPlayer | None = None) -> FullyKioskDashboards:
23 """Build a FullyKioskDashboards instance whose mocked players controller returns player."""
24 provider = MagicMock()
25 provider.domain = "fully_kiosk"
26 provider.translation_owner = "provider.fully_kiosk"
27 provider.mass.players.get_player.return_value = player
28 provider.mass.dashboard.register_dashboard_handler.return_value = MagicMock()
29 return FullyKioskDashboards(provider)
30
31
32def _make_player(
33 *,
34 player_id: str = PLAYER_ID,
35 display_name: str = "Living Room Kiosk",
36 connected: bool = True,
37) -> FullyKioskPlayer:
38 """Build a real FullyKioskPlayer, optionally with a connected client mock."""
39 provider = MockProvider("fully_kiosk", instance_id="fully_kiosk--1")
40 provider.mass.config.get_base_player_config.return_value = create_mock_config(display_name)
41 player = FullyKioskPlayer(provider=provider, player_id=player_id, host="10.0.0.5") # type: ignore[arg-type]
42 player._attr_name = display_name
43 if connected:
44 client = MagicMock()
45 client.screenOn = AsyncMock()
46 client.stopScreensaver = AsyncMock()
47 client.toForeground = AsyncMock()
48 client.loadUrl = AsyncMock()
49 client.loadStartUrl = AsyncMock()
50 player.fully_kiosk = client
51 return player
52
53
54def _make_connectable_player(*, player_id: str = PLAYER_ID) -> tuple[FullyKioskPlayer, MagicMock]:
55 """Build a real player with a mocked provider.dashboards, ready to drive _connect()/poll()."""
56 provider = MockProvider("fully_kiosk", instance_id="fully_kiosk--1")
57 provider.dashboards = MagicMock()
58 provider.mass.config.get_base_player_config.return_value = create_mock_config("Kiosk")
59 player = FullyKioskPlayer(provider=provider, player_id=player_id, host="10.0.0.5") # type: ignore[arg-type]
60 player.get_setup_value = MagicMock(return_value="secret") # type: ignore[method-assign]
61 ssl_config = {CONF_USE_SSL: False, CONF_VERIFY_SSL: True, CONF_SSL_FINGERPRINT: ""}
62 player.config.get_value = MagicMock(side_effect=ssl_config.get) # type: ignore[method-assign]
63 return player, provider.dashboards
64
65
66def _make_client(*, fails: bool = False) -> MagicMock:
67 """Build a FullyKiosk client stub for patching into player._connect()."""
68 client = MagicMock()
69 client.getDeviceInfo = AsyncMock(
70 side_effect=FullyKioskError("Error", "boom") if fails else None
71 )
72 client.deviceInfo = {}
73 return client
74
75
76def _client_of(player: FullyKioskPlayer) -> MagicMock:
77 """Return a connected player's client mock, narrowing away the FullyKiosk | None type."""
78 assert player.fully_kiosk is not None
79 return cast("MagicMock", player.fully_kiosk)
80
81
82### Player connect/poll wiring
83
84
85async def test_connect_registers_dashboard_endpoint() -> None:
86 """A successful connect registers the player as a dashboard endpoint."""
87 player, dashboards = _make_connectable_player()
88 client = _make_client()
89 with patch("music_assistant.providers.fully_kiosk.player.FullyKiosk", return_value=client):
90 await player._connect()
91
92 assert player.available is True
93 dashboards.register.assert_called_once_with(player)
94
95
96async def test_connect_registers_dashboard_endpoint_with_synced_name() -> None:
97 """A successful connect registers under the device's synced name, not the placeholder."""
98 player, dashboards = _make_connectable_player()
99 client = _make_client()
100 client.deviceInfo = {"deviceName": "Kitchen Tablet"}
101 captured_names: list[str] = []
102 dashboards.register.side_effect = lambda registered_player: captured_names.append(
103 registered_player.display_name
104 )
105 # a real config change (e.g. the password being set) clears the cached display_name
106 # via set_config() before on_config_updated()/_connect() runs; mirror that here
107 player.set_config(player.config)
108 with patch("music_assistant.providers.fully_kiosk.player.FullyKiosk", return_value=client):
109 await player._connect()
110
111 assert captured_names == ["Kitchen Tablet"]
112
113
114async def test_connect_fingerprint_without_ssl_unregisters_dashboard_endpoint() -> None:
115 """A config change to a fingerprint without SSL drops any existing dashboard endpoint."""
116 player, dashboards = _make_connectable_player()
117 client = _make_client()
118 with patch("music_assistant.providers.fully_kiosk.player.FullyKiosk", return_value=client):
119 await player._connect()
120 dashboards.reset_mock()
121
122 ssl_config = {CONF_USE_SSL: False, CONF_VERIFY_SSL: True, CONF_SSL_FINGERPRINT: "aa:bb"}
123 player.config.get_value = MagicMock(side_effect=ssl_config.get) # type: ignore[method-assign]
124
125 await player._connect()
126
127 assert player.available is False
128 dashboards.unregister.assert_called_once_with(PLAYER_ID)
129
130
131async def test_poll_failure_unregisters_then_reconnect_reregisters() -> None:
132 """A poll failure drops the dashboard endpoint; a subsequent reconnect re-registers it."""
133 player, dashboards = _make_connectable_player()
134 client = _make_client()
135 with patch("music_assistant.providers.fully_kiosk.player.FullyKiosk", return_value=client):
136 await player._connect()
137 dashboards.reset_mock()
138
139 client.getDeviceInfo.side_effect = FullyKioskError("Error", "boom")
140 await player.poll()
141 unavailable = player.available
142 assert unavailable is False
143 dashboards.unregister.assert_called_once_with(PLAYER_ID)
144
145 client.getDeviceInfo.side_effect = None
146 await player.poll() # fully_kiosk is None, so poll() reconnects
147
148 reconnected = player.available
149 assert reconnected is True
150 dashboards.register.assert_called_once_with(player)
151
152
153async def test_connect_failure_unregisters_dashboard_endpoint() -> None:
154 """A failed (re)connect drops any existing dashboard endpoint registration."""
155 player, dashboards = _make_connectable_player()
156 client = _make_client(fails=True)
157 with patch("music_assistant.providers.fully_kiosk.player.FullyKiosk", return_value=client):
158 await player._connect()
159
160 assert player.available is False
161 dashboards.unregister.assert_called_once_with(PLAYER_ID)
162
163
164async def test_password_cleared_unregisters_dashboard_endpoint() -> None:
165 """Clearing the password drops the dashboard endpoint without a network call."""
166 player, dashboards = _make_connectable_player()
167 client = _make_client()
168 with patch("music_assistant.providers.fully_kiosk.player.FullyKiosk", return_value=client):
169 await player._connect()
170 dashboards.reset_mock()
171
172 player.get_setup_value = MagicMock(return_value=None) # type: ignore[method-assign]
173 await player.on_config_updated()
174
175 assert player.available is False
176 dashboards.unregister.assert_called_once_with(PLAYER_ID)
177
178
179async def test_on_unload_unregisters_dashboard_endpoint() -> None:
180 """Unloading the player drops its dashboard endpoint registration."""
181 player, dashboards = _make_connectable_player()
182
183 await player.on_unload()
184
185 dashboards.unregister.assert_called_once_with(PLAYER_ID)
186
187
188### Provider unload
189
190
191async def test_provider_unload_awaits_dashboards_unload() -> None:
192 """Provider unload tears down every dashboard endpoint it registered."""
193 provider = FullyKioskProvider.__new__(FullyKioskProvider)
194 provider.dashboards = AsyncMock()
195
196 await provider.unload()
197
198 provider.dashboards.unload.assert_awaited_once()
199
200
201### Registration lifecycle
202
203
204def test_register_creates_dashboard_endpoint_with_all_types() -> None:
205 """A connected player registers a dashboard endpoint supporting all three types."""
206 player = _make_player()
207 dashboards = _make_dashboards(player)
208
209 dashboards.register(player)
210
211 handler = dashboards.mass.dashboard.register_dashboard_handler
212 handler.assert_called_once() # type: ignore[attr-defined]
213 device = handler.call_args[0][0] # type: ignore[attr-defined]
214 assert device.dashboard_id == PLAYER_ID
215 assert device.name == "Living Room Kiosk"
216 assert device.supported_types == {
217 DashboardType.NOW_PLAYING,
218 DashboardType.PARTY,
219 DashboardType.MUSIC_QUIZ,
220 }
221 assert device.provider_domain_hint == "fully_kiosk"
222 assert PLAYER_ID in dashboards._unregister_callbacks
223
224
225def test_register_is_noop_after_unload() -> None:
226 """A late register call after unload must not resurrect an endpoint."""
227 player = _make_player()
228 dashboards = _make_dashboards(player)
229 dashboards._unloaded = True
230
231 dashboards.register(player)
232
233 dashboards.mass.dashboard.register_dashboard_handler.assert_not_called() # type: ignore[attr-defined]
234
235
236@pytest.mark.parametrize("other_player", [False, True])
237def test_register_is_noop_for_removed_player(other_player: bool) -> None:
238 """A register call for a player the players controller no longer knows about is a no-op."""
239 player = _make_player()
240 current: FullyKioskPlayer | None = _make_player() if other_player else None
241 dashboards = _make_dashboards(current)
242
243 dashboards.register(player)
244
245 dashboards.mass.dashboard.register_dashboard_handler.assert_not_called() # type: ignore[attr-defined]
246
247
248def test_unregister_calls_callback_and_reregister_creates_new_one() -> None:
249 """Connection loss unregisters the endpoint; a subsequent reconnect re-registers it."""
250 player = _make_player()
251 dashboards = _make_dashboards(player)
252 unregister = MagicMock()
253 dashboards.mass.dashboard.register_dashboard_handler.return_value = unregister # type: ignore[attr-defined]
254 dashboards.register(player)
255 assert PLAYER_ID in dashboards._unregister_callbacks
256
257 dashboards.unregister(PLAYER_ID)
258
259 unregister.assert_called_once()
260 assert PLAYER_ID not in dashboards._unregister_callbacks
261
262 # reconnect re-registers a fresh endpoint
263 dashboards.register(player)
264 assert PLAYER_ID in dashboards._unregister_callbacks
265 assert dashboards.mass.dashboard.register_dashboard_handler.call_count == 2 # type: ignore[attr-defined]
266
267
268def test_unregister_unknown_player_is_noop() -> None:
269 """Unregistering a player id that was never registered does nothing."""
270 dashboards = _make_dashboards()
271
272 dashboards.unregister(PLAYER_ID) # must not raise
273
274
275async def test_unload_unregisters_all_endpoints() -> None:
276 """Unload unregisters every endpoint and blocks further registration."""
277 player = _make_player()
278 dashboards = _make_dashboards(player)
279 unregister = MagicMock()
280 dashboards.mass.dashboard.register_dashboard_handler.return_value = unregister # type: ignore[attr-defined]
281 dashboards.register(player)
282
283 await dashboards.unload()
284
285 unregister.assert_called_once()
286 assert dashboards._unloaded
287 assert not dashboards._unregister_callbacks
288
289 dashboards.register(player)
290 dashboards.mass.dashboard.register_dashboard_handler.assert_called_once() # type: ignore[attr-defined]
291
292
293### on_show
294
295
296async def test_on_show_wakes_and_loads_url_in_order() -> None:
297 """on_show resolves the local url and drives the device in the documented order."""
298 dashboards = _make_dashboards()
299 player = _make_player()
300 dashboards.mass.players.get_player.return_value = player # type: ignore[attr-defined]
301 target = "http://192.168.1.10:8095/?dashboard=ABC123&path=%2Fparty"
302 dashboards.mass.dashboard.resolve_dashboard_url = AsyncMock(return_value=target) # type: ignore[method-assign]
303 client = _client_of(player)
304 order = MagicMock()
305 order.attach_mock(client.screenOn, "screen_on")
306 order.attach_mock(client.stopScreensaver, "stop_screensaver")
307 order.attach_mock(client.toForeground, "to_foreground")
308 order.attach_mock(client.loadUrl, "load_url")
309
310 await dashboards._on_show(PLAYER_ID, DashboardType.PARTY, None)
311
312 dashboards.mass.dashboard.resolve_dashboard_url.assert_awaited_once_with(
313 DashboardType.PARTY, None, prefer_local=True
314 )
315 assert [call[0] for call in order.mock_calls] == [
316 "screen_on",
317 "stop_screensaver",
318 "to_foreground",
319 "load_url",
320 ]
321 client.loadUrl.assert_awaited_once_with(target)
322
323
324async def test_on_show_missing_player_raises_unavailable() -> None:
325 """on_show for a player that vanished raises PlayerUnavailableError."""
326 dashboards = _make_dashboards()
327 dashboards.mass.players.get_player.return_value = None # type: ignore[attr-defined]
328
329 with pytest.raises(PlayerUnavailableError):
330 await dashboards._on_show(PLAYER_ID, DashboardType.PARTY, None)
331
332
333async def test_on_show_disconnected_player_raises_unavailable() -> None:
334 """on_show for a player without an active client connection raises PlayerUnavailableError."""
335 dashboards = _make_dashboards()
336 player = _make_player(connected=False)
337 dashboards.mass.players.get_player.return_value = player # type: ignore[attr-defined]
338
339 with pytest.raises(PlayerUnavailableError):
340 await dashboards._on_show(PLAYER_ID, DashboardType.PARTY, None)
341
342
343async def test_on_show_device_error_raises_translated_unavailable_without_leaking_url() -> None:
344 """A device-side failure surfaces as a translated error that never echoes the url."""
345 dashboards = _make_dashboards()
346 player = _make_player(display_name="Living Room Kiosk")
347 dashboards.mass.players.get_player.return_value = player # type: ignore[attr-defined]
348 target = "http://192.168.1.10:8095/?dashboard=SECRETCODE&path=%2Fparty"
349 dashboards.mass.dashboard.resolve_dashboard_url = AsyncMock(return_value=target) # type: ignore[method-assign]
350 _client_of(player).toForeground.side_effect = FullyKioskError("Error", "boom")
351
352 with pytest.raises(PlayerUnavailableError) as exc_info:
353 await dashboards._on_show(PLAYER_ID, DashboardType.PARTY, None)
354
355 assert exc_info.value.translation_key == "show_dashboard_failed"
356 assert exc_info.value.translation_owner == "provider.fully_kiosk"
357 assert exc_info.value.translation_args == ["Living Room Kiosk"]
358 assert "SECRETCODE" not in str(exc_info.value)
359 assert target not in str(exc_info.value)
360 assert exc_info.value.__cause__ is None
361
362
363### on_hide
364
365
366async def test_on_hide_loads_start_url() -> None:
367 """on_hide returns the kiosk to its configured start page."""
368 dashboards = _make_dashboards()
369 player = _make_player()
370 dashboards.mass.players.get_player.return_value = player # type: ignore[attr-defined]
371
372 await dashboards._on_hide(PLAYER_ID)
373
374 _client_of(player).loadStartUrl.assert_awaited_once()
375
376
377async def test_on_hide_disconnected_player_is_noop() -> None:
378 """on_hide for an already-disconnected player is a graceful no-op."""
379 dashboards = _make_dashboards()
380 player = _make_player(connected=False)
381 dashboards.mass.players.get_player.return_value = player # type: ignore[attr-defined]
382
383 await dashboards._on_hide(PLAYER_ID) # must not raise
384
385
386async def test_on_hide_missing_player_is_noop() -> None:
387 """on_hide for a player that vanished is a graceful no-op."""
388 dashboards = _make_dashboards()
389 dashboards.mass.players.get_player.return_value = None # type: ignore[attr-defined]
390
391 await dashboards._on_hide(PLAYER_ID) # must not raise
392
393
394async def test_on_hide_device_error_raises_command_failed() -> None:
395 """A device-side failure while hiding surfaces as PlayerCommandFailed."""
396 dashboards = _make_dashboards()
397 player = _make_player()
398 dashboards.mass.players.get_player.return_value = player # type: ignore[attr-defined]
399 _client_of(player).loadStartUrl.side_effect = FullyKioskError("Error", "boom")
400
401 with pytest.raises(PlayerCommandFailed) as exc_info:
402 await dashboards._on_hide(PLAYER_ID)
403
404 assert exc_info.value.__cause__ is None
405