/
/
/
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_task_id = "cast_quit_app_test"
56 fake.cancel_pending_app_quit = partial(
57 ChromecastPlayer.cancel_pending_app_quit, cast("ChromecastPlayer", fake)
58 )
59 fake._schedule_app_release = partial(
60 ChromecastPlayer._schedule_app_release, cast("ChromecastPlayer", fake)
61 )
62 status = fake.cc.media_controller.status
63 status.player_state = player_state
64 status.player_is_playing = player_state in ("PLAYING", "BUFFERING")
65 status.player_is_paused = player_state == "PAUSED"
66 status.media_session_id = media_session_id
67 fake._flow_stream_underrun.return_value = False
68 return fake
69
70
71async def test_stop_keeps_the_cast_session_alive() -> None:
72 """Stopping a player must not end the Cast session right away."""
73 fake = _fake_cast()
74
75 await _stop(fake)
76
77 fake.cc.media_controller.stop.assert_called_once()
78 fake.cc.quit_app.assert_not_called()
79
80
81async def test_stop_schedules_the_device_release() -> None:
82 """The device is released a little later, so a follow-up command can reuse it."""
83 fake = _fake_cast()
84
85 await _stop(fake)
86
87 fake.mass.call_later.assert_called_once_with(
88 APP_QUIT_DELAY, fake._quit_app_when_unused, task_id=fake._app_quit_task_id
89 )
90
91
92@pytest.mark.parametrize("app_id", [SENDSPIN_CAST_APP_ID, "CC32E753", None])
93async def test_stop_ends_a_foreign_cast_session_right_away(app_id: str | None) -> None:
94 """A session of another app is not ours to keep around."""
95 fake = _fake_cast(running_app_id=app_id)
96
97 await _stop(fake)
98
99 fake.cc.quit_app.assert_called_once()
100 fake.mass.call_later.assert_not_called()
101
102
103async def test_stop_on_a_group_leaves_the_app_alone() -> None:
104 """A cast group is released by its power command, not by stop."""
105 fake = _fake_cast(player_type=PlayerType.GROUP)
106
107 await _stop(fake)
108
109 fake.cc.media_controller.stop.assert_called_once()
110 fake.cc.quit_app.assert_not_called()
111 fake.mass.call_later.assert_not_called()
112
113
114async def test_stop_without_a_media_session_is_not_sent() -> None:
115 """Nothing was ever loaded, so there is no playback to stop."""
116 fake = _fake_cast(media_session_id=None)
117
118 await _stop(fake)
119
120 fake.cc.media_controller.stop.assert_not_called()
121 fake.cc.quit_app.assert_not_called()
122 fake.mass.call_later.assert_called_once()
123
124
125async def test_launching_an_app_cancels_a_pending_release() -> None:
126 """The device is claimed again within the grace period, so it must not be released."""
127 fake = _fake_cast(running_app_id=MASS_APP_ID)
128
129 await ChromecastPlayer._launch_app(cast("ChromecastPlayer", fake))
130
131 fake.mass.cancel_timer.assert_called_once_with(fake._app_quit_task_id)
132
133
134@pytest.mark.parametrize("app_id", [MASS_APP_ID, APP_MEDIA_RECEIVER])
135async def test_unused_receiver_app_is_quit(app_id: str) -> None:
136 """Nothing came along within the grace period, so the device is released."""
137 fake = _fake_cast(running_app_id=app_id, player_state="IDLE")
138
139 await _quit_app_when_unused(fake)
140
141 fake.cc.quit_app.assert_called_once()
142
143
144async def test_unreachable_device_is_not_quit() -> None:
145 """The device dropped off within the grace period, so there is nothing to send to."""
146 fake = _fake_cast(player_state="IDLE")
147 fake.available = False
148
149 await _quit_app_when_unused(fake)
150
151 fake.cc.quit_app.assert_not_called()
152
153
154async def test_app_of_another_sender_is_not_quit() -> None:
155 """Another app took over the device, so it is no longer ours to release."""
156 fake = _fake_cast(running_app_id=SENDSPIN_CAST_APP_ID, player_state="IDLE")
157
158 await _quit_app_when_unused(fake)
159
160 fake.cc.quit_app.assert_not_called()
161
162
163@pytest.mark.parametrize("player_state", ["PLAYING", "BUFFERING", "PAUSED"])
164async def test_receiver_app_in_use_is_not_quit(player_state: str) -> None:
165 """The receiver app got used again (such as by a dashboard), so it is left running."""
166 fake = _fake_cast(player_state=player_state)
167
168 await _quit_app_when_unused(fake)
169
170 fake.cc.quit_app.assert_not_called()
171
172
173async def test_a_device_that_ran_dry_at_the_end_of_the_flow_stream_is_quit() -> None:
174 """A device stuck buffering at flow EOF has no audio coming, so it is not in use."""
175 fake = _fake_cast(player_state="BUFFERING")
176 fake._flow_stream_underrun.return_value = True
177
178 await _quit_app_when_unused(fake)
179
180 fake.cc.quit_app.assert_called_once()
181