/
/
/
1"""
2Tests for how ChromecastPlayer releases a Cast device once playback ends by itself.
3
4Playback that finishes on its own never gets a stop command: the queue simply runs
5out, or an announcement ends. Without a release the receiver app stays loaded, which
6on a TV or Nest Hub means Music Assistant stays on screen instead of the backdrop.
7"""
8
9from __future__ import annotations
10
11from typing import Any, cast
12from unittest.mock import MagicMock
13
14import pytest
15from music_assistant_models.enums import PlaybackState, PlayerType
16from pychromecast import IDLE_APP_ID
17
18from music_assistant.providers.chromecast.constants import (
19 APP_QUIT_DELAY,
20 MASS_APP_ID,
21)
22from music_assistant.providers.chromecast.player import ChromecastPlayer
23
24
25def _fake_player(
26 *,
27 prev_state: PlaybackState = PlaybackState.PLAYING,
28 player_type: PlayerType = PlayerType.PLAYER,
29 active_cast_group: str | None = None,
30 flow_underrun: bool = False,
31) -> Any:
32 """
33 Build a Cast player on mocked collaborators, so it handles media status for real.
34
35 :param prev_state: Playback state the player was in before the status arrives.
36 :param player_type: Type of the player.
37 :param active_cast_group: Player id of the cast group the player plays through, if any.
38 :param flow_underrun: Whether the queue's flow stream has been fully consumed.
39 """
40 # __init__ is skipped: it needs a provider, cast info and a live Chromecast connection.
41 # Typed as Any because the collaborators below are read back as the mocks they are.
42 fake = cast("Any", ChromecastPlayer.__new__(ChromecastPlayer))
43 fake.mass = MagicMock()
44 fake.logger = MagicMock()
45 fake.cc = MagicMock(app_id=MASS_APP_ID)
46 fake.update_state = MagicMock()
47 fake._flow_stream_underrun = MagicMock(return_value=flow_underrun)
48 fake._media_error_reported = False
49 fake._app_quit_task_id = "cast_quit_app_test"
50 fake.active_cast_group = active_cast_group
51 # display_name is a cached property, normally built from the config in __init__
52 fake._cache = {"display_name": "Test Cast"}
53 # the state below is exposed through read-only properties on the base Player,
54 # so it is seeded through the attributes backing them
55 fake._attr_type = player_type
56 fake._attr_powered = False
57 fake._attr_group_members = []
58 fake._attr_playback_state = prev_state
59 return fake
60
61
62def _media_status(
63 *,
64 playing: bool = False,
65 paused: bool = False,
66 buffering: bool = False,
67 content_id: str = "",
68) -> MagicMock:
69 """
70 Build a MediaStatus as the receiver reports it.
71
72 :param playing: Whether the receiver reports playback.
73 :param paused: Whether the receiver reports paused playback.
74 :param buffering: Whether the receiver reports buffering, which it counts as playing.
75 :param content_id: Content id the receiver has loaded, if any.
76 """
77 status = MagicMock()
78 status.content_id = content_id
79 status.player_state = (
80 "BUFFERING" if buffering else "PLAYING" if playing else "PAUSED" if paused else "IDLE"
81 )
82 status.player_is_playing = playing or buffering
83 status.player_is_paused = paused
84 status.player_is_idle = not playing and not paused and not buffering
85 return status
86
87
88def _assert_release_scheduled(fake: Any) -> None:
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("prev_state", [PlaybackState.PLAYING, PlaybackState.PAUSED])
95def test_playback_ending_releases_the_device(prev_state: PlaybackState) -> None:
96 """Nothing follows up on playback that ended on its own, so the device is released."""
97 fake = _fake_player(prev_state=prev_state)
98
99 fake._handle_media_status(_media_status())
100
101 assert fake._attr_playback_state == PlaybackState.IDLE
102 _assert_release_scheduled(fake)
103
104
105def test_an_announcement_on_an_idle_player_releases_the_device() -> None:
106 """
107 An announcement claims an idle device, so it must hand it back afterwards.
108
109 The player is never stopped in this flow (there was nothing to stop), which is
110 what used to leave the receiver app loaded for good.
111 """
112 fake = _fake_player(prev_state=PlaybackState.IDLE)
113
114 fake._handle_media_status(_media_status(playing=True, content_id="http://mass/announce.mp3"))
115 assert fake._attr_playback_state == PlaybackState.PLAYING
116 fake.mass.call_later.assert_not_called()
117
118 fake._handle_media_status(_media_status())
119
120 _assert_release_scheduled(fake)
121
122
123def test_a_device_that_ran_dry_at_the_end_of_the_flow_stream_releases_it() -> None:
124 """A device stuck buffering at flow EOF counts as finished, so it is released too."""
125 fake = _fake_player(flow_underrun=True)
126
127 fake._handle_media_status(_media_status(buffering=True, content_id="http://mass/flow.flac"))
128
129 assert fake._attr_playback_state == PlaybackState.IDLE
130 _assert_release_scheduled(fake)
131
132
133@pytest.mark.parametrize("app_id", [None, IDLE_APP_ID])
134def test_a_released_device_does_not_become_a_source(app_id: str | None) -> None:
135 """A released device sits on its backdrop, which is not something to select."""
136 fake = _fake_player()
137 fake.cc.app_id = app_id
138 fake._attr_source_list = []
139 # the source the released device was on, so the assertion proves it is cleared
140 fake._attr_active_source = "previous_source"
141
142 fake._handle_media_status(_media_status())
143
144 assert fake._attr_active_source is None
145 assert fake._attr_source_list == []
146
147
148def test_pausing_keeps_the_device_claimed() -> None:
149 """A paused player is meant to be resumed, so it keeps its Cast session."""
150 fake = _fake_player()
151
152 fake._handle_media_status(_media_status(paused=True, content_id="http://mass/track.flac"))
153
154 assert fake._attr_playback_state == PlaybackState.PAUSED
155 fake.mass.call_later.assert_not_called()
156
157
158def test_an_idle_player_is_not_released_again() -> None:
159 """A player that was already idle has no session of its own to release."""
160 fake = _fake_player(prev_state=PlaybackState.IDLE)
161
162 fake._handle_media_status(_media_status())
163
164 fake.mass.call_later.assert_not_called()
165
166
167@pytest.mark.parametrize(
168 ("content_id", "playing", "paused"),
169 [
170 ("https://cast.music-assistant.io/dashboard-keepalive.mp4", True, False),
171 ("https://cast.music-assistant.io/keepalive.png", False, True),
172 ],
173)
174def test_a_dashboard_keepalive_does_not_release_the_device(
175 content_id: str, playing: bool, paused: bool
176) -> None:
177 """The receiver is showing a dashboard, so the device is deliberately kept claimed."""
178 fake = _fake_player()
179
180 fake._handle_media_status(_media_status(playing=playing, paused=paused, content_id=content_id))
181
182 assert fake._attr_playback_state == PlaybackState.IDLE
183 fake.mass.call_later.assert_not_called()
184
185
186def test_a_cast_group_is_not_released_when_playback_ends() -> None:
187 """A cast group is released by its power control, not by playback ending."""
188 fake = _fake_player(player_type=PlayerType.GROUP)
189
190 fake._handle_media_status(_media_status())
191
192 assert fake._attr_playback_state == PlaybackState.IDLE
193 fake.mass.call_later.assert_not_called()
194
195
196def test_a_group_member_is_not_released_when_the_group_stops() -> None:
197 """A member follows the group's Cast session, so it has none of its own to release."""
198 fake = _fake_player(active_cast_group="group_player_1")
199 # the handler reads the group's status instead of its own, but only for a real cast player
200 group = MagicMock(spec=ChromecastPlayer)
201 group.cc = MagicMock()
202 group.cc.media_controller.status = _media_status()
203 fake.mass.players.get_player.return_value = group
204
205 fake._handle_media_status(_media_status())
206
207 assert fake._attr_playback_state == PlaybackState.IDLE
208 fake.mass.call_later.assert_not_called()
209