/
/
/
1"""Tests for Chromecast dashboard session-liveness detection."""
2
3from __future__ import annotations
4
5from unittest.mock import AsyncMock, MagicMock, patch
6from uuid import uuid4
7
8from music_assistant_models.enums import DashboardType
9from pychromecast.socket_client import CONNECTION_STATUS_CONNECTED, CONNECTION_STATUS_LOST
10
11from music_assistant.providers.chromecast.constants import MASS_APP_ID
12from music_assistant.providers.chromecast.dashboard import (
13 DASHBOARD_CONNECTION_LOSS_GRACE,
14 ChromecastDashboards,
15 _ActiveDashboardCast,
16)
17
18
19def _make_dashboards() -> ChromecastDashboards:
20 """Build a ChromecastDashboards instance with a fully mocked owning provider."""
21 provider = MagicMock()
22 provider.translation_owner = "provider.chromecast"
23 provider.domain = "chromecast"
24 provider.mass.closing = False
25 # run_in_executor just calls the function inline, synchronously
26 provider.mass.loop.run_in_executor = AsyncMock(
27 side_effect=lambda _executor, func, *args: func(*args)
28 )
29 provider.mass.loop.call_soon_threadsafe = MagicMock(side_effect=lambda func, *args: func(*args))
30 provider.mass.loop.call_later = MagicMock()
31 provider.mass.players.get_player.return_value = None
32 return ChromecastDashboards(provider)
33
34
35def _chromecast(app_id: str = MASS_APP_ID, session_id: str = "session-1") -> MagicMock:
36 """Build a connected Chromecast mock, already showing the MA receiver app."""
37 chromecast = MagicMock()
38 chromecast.socket_client.is_connected = True
39 chromecast.status = MagicMock(app_id=app_id, session_id=session_id)
40 return chromecast
41
42
43async def _show(dashboards: ChromecastDashboards, device_id: str, chromecast: MagicMock) -> None:
44 """Drive a successful _on_show for device_id against an already-connected chromecast."""
45 dashboards.mass.dashboard.resolve_dashboard_url = AsyncMock( # type: ignore[method-assign]
46 return_value="https://x"
47 )
48 dashboards._dashboard_connections[device_id] = chromecast
49 with patch("music_assistant.providers.chromecast.dashboard.send_show_dashboard"):
50 await dashboards._on_show(device_id, DashboardType.PARTY, None)
51
52
53### Cast status: receiver app identity changes
54
55
56async def test_cast_status_different_app_ends_active_session() -> None:
57 """A cast status reporting a different app than ours ends the dashboard session."""
58 dashboards = _make_dashboards()
59 device_id = str(uuid4())
60 chromecast = _chromecast()
61 await _show(dashboards, device_id, chromecast)
62 listener = chromecast.register_status_listener.call_args[0][0]
63
64 other_status = MagicMock(app_id="OTHERAPP", session_id="other-session", display_name="YouTube")
65 listener.new_cast_status(other_status)
66
67 dashboards.mass.dashboard.end_session.assert_called_once_with( # type: ignore[attr-defined]
68 f"chromecast_{device_id}", "the receiver app was closed (active app: YouTube)"
69 )
70 assert device_id not in dashboards._active_casts
71
72
73async def test_cast_status_same_app_and_session_keeps_active_session() -> None:
74 """MA audio playing in the same receiver app session leaves the dashboard session intact."""
75 dashboards = _make_dashboards()
76 device_id = str(uuid4())
77 chromecast = _chromecast(session_id="session-1")
78 await _show(dashboards, device_id, chromecast)
79 listener = chromecast.register_status_listener.call_args[0][0]
80
81 same_status = MagicMock(app_id=MASS_APP_ID, session_id="session-1")
82 listener.new_cast_status(same_status)
83
84 dashboards.mass.dashboard.end_session.assert_not_called() # type: ignore[attr-defined]
85 assert device_id in dashboards._active_casts
86
87
88def test_handle_cast_status_without_active_cast_is_noop() -> None:
89 """A cast status for a device with no tracked active dashboard cast is ignored."""
90 dashboards = _make_dashboards()
91 status = MagicMock(app_id="OTHERAPP", session_id="other-session")
92
93 dashboards._handle_cast_status(str(uuid4()), status)
94
95 dashboards.mass.dashboard.end_session.assert_not_called() # type: ignore[attr-defined]
96
97
98async def test_reshow_invalidates_previous_listener() -> None:
99 """A re-show registers a fresh listener and stale callbacks from the old one are ignored."""
100 dashboards = _make_dashboards()
101 device_id = str(uuid4())
102 chromecast = _chromecast()
103
104 await _show(dashboards, device_id, chromecast)
105 await _show(dashboards, device_id, chromecast)
106
107 first_listener = chromecast.register_status_listener.call_args_list[0][0][0]
108 second_listener = chromecast.register_status_listener.call_args_list[1][0][0]
109
110 other_status = MagicMock(app_id="OTHERAPP", session_id="x", display_name="YouTube")
111 first_listener.new_cast_status(other_status)
112 dashboards.mass.dashboard.end_session.assert_not_called() # type: ignore[attr-defined]
113 assert device_id in dashboards._active_casts
114
115 second_listener.new_cast_status(other_status)
116 dashboards.mass.dashboard.end_session.assert_called_once() # type: ignore[attr-defined]
117
118
119### Connection status: loss grace timer
120
121
122def test_connection_lost_arms_timer_and_reconnect_cancels_it() -> None:
123 """A lost connection arms the grace timer; reconnecting before it fires cancels it."""
124 dashboards = _make_dashboards()
125 device_id = str(uuid4())
126 timer_handle = MagicMock()
127 dashboards.mass.loop.call_later.return_value = timer_handle # type: ignore[attr-defined]
128 dashboards._active_casts[device_id] = _ActiveDashboardCast(
129 cast_session_id="session-1", listener=MagicMock()
130 )
131
132 dashboards._handle_connection_status(device_id, MagicMock(status=CONNECTION_STATUS_LOST))
133
134 dashboards.mass.loop.call_later.assert_called_once_with( # type: ignore[attr-defined]
135 DASHBOARD_CONNECTION_LOSS_GRACE,
136 dashboards._session_lost,
137 device_id,
138 "connection to the device was lost",
139 )
140 assert dashboards._active_casts[device_id].loss_timer is timer_handle
141
142 dashboards._handle_connection_status(device_id, MagicMock(status=CONNECTION_STATUS_CONNECTED))
143
144 timer_handle.cancel.assert_called_once()
145 assert dashboards._active_casts[device_id].loss_timer is None
146
147
148def test_connection_lost_does_not_stack_a_second_timer() -> None:
149 """A second LOST status while a timer is already pending does not arm another one."""
150 dashboards = _make_dashboards()
151 device_id = str(uuid4())
152 dashboards._active_casts[device_id] = _ActiveDashboardCast(
153 cast_session_id="session-1", listener=MagicMock()
154 )
155
156 dashboards._handle_connection_status(device_id, MagicMock(status=CONNECTION_STATUS_LOST))
157 dashboards._handle_connection_status(device_id, MagicMock(status=CONNECTION_STATUS_LOST))
158
159 dashboards.mass.loop.call_later.assert_called_once() # type: ignore[attr-defined]
160
161
162def test_connection_loss_timer_callback_ends_session() -> None:
163 """The grace timer firing reports the session lost with the connection-loss reason."""
164 dashboards = _make_dashboards()
165 device_id = str(uuid4())
166 dashboards._active_casts[device_id] = _ActiveDashboardCast(
167 cast_session_id="session-1", listener=MagicMock()
168 )
169
170 dashboards._session_lost(device_id, "connection to the device was lost")
171
172 dashboards.mass.dashboard.end_session.assert_called_once_with( # type: ignore[attr-defined]
173 f"chromecast_{device_id}", "connection to the device was lost"
174 )
175 assert device_id not in dashboards._active_casts
176
177
178def test_handle_connection_status_without_active_cast_is_noop() -> None:
179 """A connection status for a device with no tracked active dashboard cast is ignored."""
180 dashboards = _make_dashboards()
181
182 dashboards._handle_connection_status(str(uuid4()), MagicMock(status=CONNECTION_STATUS_LOST))
183
184 dashboards.mass.loop.call_later.assert_not_called() # type: ignore[attr-defined]
185
186
187### Hide: no false-positive liveness report
188
189
190async def test_hide_cancels_loss_timer_without_ending_session() -> None:
191 """Hiding cancels a pending loss timer and drops the entry, without reporting it lost."""
192 dashboards = _make_dashboards()
193 device_id = str(uuid4())
194 timer_handle = MagicMock()
195 dashboards._active_casts[device_id] = _ActiveDashboardCast(
196 cast_session_id="session-1", listener=MagicMock(), loss_timer=timer_handle
197 )
198 dashboards.provider.browser = MagicMock(devices={})
199
200 with patch("music_assistant.providers.chromecast.dashboard.send_hide_dashboard"):
201 await dashboards._on_hide(device_id)
202
203 timer_handle.cancel.assert_called_once()
204 assert device_id not in dashboards._active_casts
205 dashboards.mass.dashboard.end_session.assert_not_called() # type: ignore[attr-defined]
206
207
208async def test_stale_connection_status_after_session_lost_is_ignored() -> None:
209 """A connection callback from an already-invalidated listener does not re-arm the timer."""
210 dashboards = _make_dashboards()
211 device_id = str(uuid4())
212 chromecast = _chromecast()
213 await _show(dashboards, device_id, chromecast)
214 listener = chromecast.register_status_listener.call_args[0][0]
215
216 dashboards._session_lost(device_id, "connection to the device was lost")
217 listener.new_connection_status(MagicMock(status=CONNECTION_STATUS_LOST))
218
219 dashboards.mass.loop.call_later.assert_not_called() # type: ignore[attr-defined]
220