/
/
/
1"""
2Tests for how ChromecastPlayer releases a Cast device on stop.
3
4Quitting the receiver app ends the Cast session, and the device plays its 'cast
5connected' chime whenever a new one is started. Stopping must therefore leave the
6session in place for a while, so a follow-up command (an announcement stops playback
7before it plays) does not have to start a new one.
8"""
9
10from __future__ import annotations
11
12from functools import partial
13from typing import cast
14from unittest.mock import MagicMock
15
16import pytest
17from music_assistant_models.enums import PlayerType
18
19from music_assistant.providers.chromecast.constants import (
20 APP_MEDIA_RECEIVER,
21 APP_QUIT_DELAY,
22 MASS_APP_ID,
23 SENDSPIN_CAST_APP_ID,
24)
25from music_assistant.providers.chromecast.player import ChromecastPlayer
26
27
28async def _stop(fake: MagicMock) -> None:
29 await ChromecastPlayer.stop(cast("ChromecastPlayer", fake))
30
31
32async def _quit_app_when_unused(fake: MagicMock) -> None:
33 await ChromecastPlayer._quit_app_when_unused(cast("ChromecastPlayer", fake))
34
35
36def _fake_cast(
37 *,
38 player_type: PlayerType = PlayerType.PLAYER,
39 running_app_id: str | None = MASS_APP_ID,
40 player_state: str = "PLAYING",
41 media_session_id: int | None = 1,
42) -> MagicMock:
43 """
44 Build a MagicMock Cast player.
45
46 :param player_type: Type of the player.
47 :param running_app_id: App id the receiver is running, if any.
48 :param player_state: Media player state the receiver reports.
49 :param media_session_id: Media session the receiver reports, if any.
50 """
51 fake = MagicMock()
52 fake.type = player_type
53 fake.available = True
54 fake.cc.app_id = running_app_id
55 fake.app_quit_sent = False
56 fake._app_quit_task_id = "cast_quit_app_test"
57 fake.cancel_pending_app_quit = partial(
58 ChromecastPlayer.cancel_pending_app_quit, cast("ChromecastPlayer", fake)
59 )
60 fake._schedule_app_release = partial(
61 ChromecastPlayer._schedule_app_release, cast("ChromecastPlayer", fake)
62 )
63 fake._quit_app = partial(ChromecastPlayer._quit_app, cast("ChromecastPlayer", fake))
64 status = fake.cc.media_controller.status
65 status.player_state = player_state
66 status.player_is_playing = player_state in ("PLAYING", "BUFFERING")
67 status.player_is_paused = player_state == "PAUSED"
68 status.media_session_id = media_session_id
69 fake._flow_stream_underrun.return_value = False
70 return fake
71
72
73async def test_stop_keeps_the_cast_session_alive() -> None:
74 """Stopping a player must not end the Cast session right away."""
75 fake = _fake_cast()
76
77 await _stop(fake)
78
79 fake.cc.media_controller.stop.assert_called_once()
80 fake.cc.quit_app.assert_not_called()
81
82
83async def test_stop_schedules_the_device_release() -> None:
84 """The device is released a little later, so a follow-up command can reuse it."""
85 fake = _fake_cast()
86
87 await _stop(fake)
88
89 fake.mass.call_later.assert_called_once_with(
90 APP_QUIT_DELAY, fake._quit_app_when_unused, task_id=fake._app_quit_task_id
91 )
92
93
94@pytest.mark.parametrize("app_id", [SENDSPIN_CAST_APP_ID, "CC32E753", None])
95async def test_stop_ends_a_foreign_cast_session_right_away(app_id: str | None) -> None:
96 """A session of another app is not ours to keep around."""
97 fake = _fake_cast(running_app_id=app_id)
98
99 await _stop(fake)
100
101 fake.cc.quit_app.assert_called_once()
102 fake.mass.call_later.assert_not_called()
103 assert fake.app_quit_sent is True
104
105
106async def test_stop_on_a_group_leaves_the_app_alone() -> None:
107 """A cast group is released by its power command, not by stop."""
108 fake = _fake_cast(player_type=PlayerType.GROUP)
109
110 await _stop(fake)
111
112 fake.cc.media_controller.stop.assert_called_once()
113 fake.cc.quit_app.assert_not_called()
114 fake.mass.call_later.assert_not_called()
115
116
117async def test_stop_without_a_media_session_is_not_sent() -> None:
118 """Nothing was ever loaded, so there is no playback to stop."""
119 fake = _fake_cast(media_session_id=None)
120
121 await _stop(fake)
122
123 fake.cc.media_controller.stop.assert_not_called()
124 fake.cc.quit_app.assert_not_called()
125 fake.mass.call_later.assert_called_once()
126
127
128async def test_launching_an_app_cancels_a_pending_release() -> None:
129 """The device is claimed again within the grace period, so it must not be released."""
130 fake = _fake_cast(running_app_id=MASS_APP_ID)
131
132 await ChromecastPlayer._launch_app(cast("ChromecastPlayer", fake))
133
134 # a timer that already fired is re-created as a task under the same id, so both
135 # have to be cancelled to reliably catch the release
136 fake.mass.cancel_timer.assert_called_once_with(fake._app_quit_task_id)
137 fake.mass.cancel_task.assert_called_once_with(fake._app_quit_task_id)
138
139
140@pytest.mark.parametrize("app_id", [MASS_APP_ID, APP_MEDIA_RECEIVER])
141async def test_unused_receiver_app_is_quit(app_id: str) -> None:
142 """Nothing came along within the grace period, so the device is released."""
143 fake = _fake_cast(running_app_id=app_id, player_state="IDLE")
144
145 await _quit_app_when_unused(fake)
146
147 fake.cc.quit_app.assert_called_once()
148
149
150async def test_unreachable_device_is_not_quit() -> None:
151 """The device dropped off within the grace period, so there is nothing to send to."""
152 fake = _fake_cast(player_state="IDLE")
153 fake.available = False
154
155 await _quit_app_when_unused(fake)
156
157 fake.cc.quit_app.assert_not_called()
158
159
160async def test_app_of_another_sender_is_not_quit() -> None:
161 """Another app took over the device, so it is no longer ours to release."""
162 fake = _fake_cast(running_app_id=SENDSPIN_CAST_APP_ID, player_state="IDLE")
163
164 await _quit_app_when_unused(fake)
165
166 fake.cc.quit_app.assert_not_called()
167
168
169@pytest.mark.parametrize("player_state", ["PLAYING", "BUFFERING", "PAUSED"])
170async def test_receiver_app_in_use_is_not_quit(player_state: str) -> None:
171 """The receiver app got used again (such as by a dashboard), so it is left running."""
172 fake = _fake_cast(player_state=player_state)
173
174 await _quit_app_when_unused(fake)
175
176 fake.cc.quit_app.assert_not_called()
177
178
179async def test_a_device_that_ran_dry_at_the_end_of_the_flow_stream_is_quit() -> None:
180 """A device stuck buffering at flow EOF has no audio coming, so it is not in use."""
181 fake = _fake_cast(player_state="BUFFERING")
182 fake._flow_stream_underrun.return_value = True
183
184 await _quit_app_when_unused(fake)
185
186 fake.cc.quit_app.assert_called_once()
187
188
189async def test_a_sent_quit_is_recorded() -> None:
190 """A quit that is on the wire cannot be recalled, so the player has to remember it."""
191 fake = _fake_cast(player_state="IDLE")
192
193 await _quit_app_when_unused(fake)
194
195 assert fake.app_quit_sent is True
196
197
198@pytest.mark.parametrize(
199 ("available", "running_app_id", "player_state"),
200 [
201 (False, MASS_APP_ID, "IDLE"),
202 (True, SENDSPIN_CAST_APP_ID, "IDLE"),
203 (True, MASS_APP_ID, "PLAYING"),
204 (True, MASS_APP_ID, "PAUSED"),
205 ],
206 ids=["unavailable", "foreign_app", "playing", "paused"],
207)
208async def test_a_skipped_quit_is_not_recorded(
209 available: bool, running_app_id: str, player_state: str
210) -> None:
211 """Nothing was sent, so the Cast session is still fine to load into."""
212 fake = _fake_cast(running_app_id=running_app_id, player_state=player_state)
213 fake.available = available
214
215 await _quit_app_when_unused(fake)
216
217 assert fake.app_quit_sent is False
218
219
220async def test_powering_off_a_group_records_the_release() -> None:
221 """A receiver that never answers the quit keeps reporting the app for the full timeout."""
222 fake = _fake_cast(player_type=PlayerType.GROUP)
223
224 await ChromecastPlayer.power(cast("ChromecastPlayer", fake), False)
225
226 fake.cc.quit_app.assert_called_once()
227 assert fake.app_quit_sent is True
228