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