/
/
/
1"""
2Tests for ChromecastPlayer receiver app launching.
3
4Regression tests for https://github.com/music-assistant/support/issues/5981, where a
5receiver refused the launch and Music Assistant carried on regardless, sending the LOAD
6to an app that was not running. Playback never started and nothing was logged.
7
8A failed launch must be reported, and the configured receiver app must not be swapped
9for the other one.
10
11A launch is also needed when the receiver still reports our app while a release of it
12is already on the wire, since the LOAD would otherwise land in a dying session.
13"""
14
15from __future__ import annotations
16
17import asyncio
18from functools import partial
19from typing import Any, cast
20from unittest.mock import MagicMock
21
22import pytest
23from music_assistant_models.errors import PlayerUnavailableError
24
25from music_assistant.providers.chromecast.constants import APP_MEDIA_RECEIVER, MASS_APP_ID
26from music_assistant.providers.chromecast.player import ChromecastPlayer
27
28
29async def _launch_app(fake: MagicMock) -> None:
30 await ChromecastPlayer._launch_app(cast("ChromecastPlayer", fake))
31
32
33def _fake_cast(
34 *,
35 running_app_id: str | None = None,
36 use_mass_app: bool = True,
37 launch_results: dict[str, bool | None],
38 refusal_reason: str | None = None,
39 app_id_after_launch: str | None = "same",
40 app_quit_sent: bool = False,
41) -> MagicMock:
42 """
43 Build a MagicMock Cast whose receiver answers launches per app id.
44
45 :param running_app_id: App id the receiver is already running, if any.
46 :param use_mass_app: Value of the ``use_mass_app`` config option.
47 :param launch_results: Per app id: True to accept, False to refuse, None to leave
48 the launch unanswered.
49 :param refusal_reason: LAUNCH_ERROR reason reported by the receiver.
50 :param app_id_after_launch: App the receiver reports running after accepting a
51 launch. "same" means the app that was asked for.
52 :param app_quit_sent: Whether a release of the receiver app is already on the wire.
53 """
54 fake = MagicMock()
55 # the launch runs in an executor and is acknowledged from that thread, so the real
56 # running loop is needed to bridge back to the waiting coroutine
57 fake.mass.loop = asyncio.get_running_loop()
58 fake.display_name = "Fake Cast"
59 fake.cc.app_id = running_app_id
60 fake.app_quit_sent = app_quit_sent
61 fake.config.get_value = MagicMock(return_value=use_mass_app)
62 fake.launch_attempts = []
63 fake.cc.socket_client.receiver_controller.launch_failure.reason = refusal_reason
64
65 def launch_app(
66 app_id: str,
67 *,
68 force_launch: bool = False, # noqa: ARG001
69 callback_function: Any = None,
70 ) -> None:
71 fake.launch_attempts.append(app_id)
72 result = launch_results.get(app_id)
73 if result is None:
74 # receiver ignores the request, so the callback is never invoked
75 return
76 if result:
77 fake.cc.app_id = app_id if app_id_after_launch == "same" else app_id_after_launch
78 callback_function(result, None)
79
80 fake.cc.socket_client.receiver_controller.launch_app = launch_app
81 fake._log_launch_failure = partial(ChromecastPlayer._log_launch_failure, fake)
82 return fake
83
84
85def _warnings(fake: MagicMock) -> str:
86 return " ".join(str(call) for call in fake.logger.warning.call_args_list)
87
88
89@pytest.fixture(autouse=True)
90def _instant_launch_timeout(monkeypatch: pytest.MonkeyPatch) -> None:
91 """Keep the unanswered-launch cases from waiting out the real timeout."""
92 monkeypatch.setattr("music_assistant.providers.chromecast.player.APP_LAUNCH_TIMEOUT", 0.01)
93
94
95async def test_refused_launch_is_not_treated_as_success() -> None:
96 """A refused launch must fail instead of letting a LOAD be sent."""
97 fake = _fake_cast(launch_results={MASS_APP_ID: False})
98
99 with pytest.raises(PlayerUnavailableError):
100 await _launch_app(fake)
101
102
103async def test_ignored_launch_fails() -> None:
104 """An unanswered launch fails once the timeout expires."""
105 fake = _fake_cast(launch_results={MASS_APP_ID: None})
106
107 with pytest.raises(PlayerUnavailableError):
108 await _launch_app(fake)
109
110
111async def test_successful_launch_returns() -> None:
112 """A confirmed launch completes without raising."""
113 fake = _fake_cast(launch_results={MASS_APP_ID: True})
114
115 await _launch_app(fake)
116
117 assert fake.launch_attempts == [MASS_APP_ID]
118
119
120async def test_configured_app_is_never_swapped() -> None:
121 """A failed launch does not silently retry the other receiver app."""
122 fake = _fake_cast(launch_results={MASS_APP_ID: False, APP_MEDIA_RECEIVER: True})
123
124 with pytest.raises(PlayerUnavailableError):
125 await _launch_app(fake)
126
127 assert fake.launch_attempts == [MASS_APP_ID]
128
129
130async def test_default_receiver_used_when_mass_app_disabled() -> None:
131 """With the option disabled the default media receiver is launched."""
132 fake = _fake_cast(use_mass_app=False, launch_results={APP_MEDIA_RECEIVER: True})
133
134 await _launch_app(fake)
135
136 assert fake.launch_attempts == [APP_MEDIA_RECEIVER]
137
138
139async def test_configured_app_already_running_is_left_alone() -> None:
140 """No LAUNCH is sent when the configured receiver app is already active."""
141 fake = _fake_cast(running_app_id=MASS_APP_ID, launch_results={MASS_APP_ID: True})
142
143 await _launch_app(fake)
144
145 assert fake.launch_attempts == []
146
147
148async def test_app_that_is_being_quit_is_relaunched() -> None:
149 """A receiver still reporting our app is no proof of a usable session after a quit."""
150 fake = _fake_cast(
151 running_app_id=MASS_APP_ID,
152 launch_results={MASS_APP_ID: True},
153 app_quit_sent=True,
154 )
155
156 await _launch_app(fake)
157
158 assert fake.launch_attempts == [MASS_APP_ID]
159
160
161async def test_confirmed_launch_clears_the_pending_quit() -> None:
162 """The new session is not the one that was quit, so it is usable again."""
163 fake = _fake_cast(
164 running_app_id=MASS_APP_ID,
165 launch_results={MASS_APP_ID: True},
166 app_quit_sent=True,
167 )
168
169 await _launch_app(fake)
170
171 assert fake.app_quit_sent is False
172
173
174@pytest.mark.parametrize(
175 ("launch_results", "app_id_after_launch"),
176 [
177 ({MASS_APP_ID: False}, "same"),
178 ({MASS_APP_ID: None}, "same"),
179 ({MASS_APP_ID: True}, None),
180 ],
181 ids=["refused", "unanswered", "not_started"],
182)
183async def test_failed_launch_keeps_the_pending_quit(
184 launch_results: dict[str, bool | None], app_id_after_launch: str | None
185) -> None:
186 """No new session came up, so the one that was quit is still the doomed one."""
187 fake = _fake_cast(
188 running_app_id=MASS_APP_ID,
189 launch_results=launch_results,
190 app_id_after_launch=app_id_after_launch,
191 app_quit_sent=True,
192 )
193
194 with pytest.raises(PlayerUnavailableError):
195 await _launch_app(fake)
196
197 assert fake.app_quit_sent is True
198
199
200async def test_other_receiver_app_is_replaced_by_the_configured_one() -> None:
201 """
202 The other compatible receiver app is not accepted as-is.
203
204 Treating both receiver apps as interchangeable made the use_mass_app setting a
205 no-op for as long as the other app was running on the device.
206 """
207 fake = _fake_cast(running_app_id=APP_MEDIA_RECEIVER, launch_results={MASS_APP_ID: True})
208
209 await _launch_app(fake)
210
211 assert fake.launch_attempts == [MASS_APP_ID]
212
213
214async def test_disabled_mass_app_is_replaced_by_the_default_receiver() -> None:
215 """With the option disabled, a running MA app is replaced by the default receiver."""
216 fake = _fake_cast(
217 running_app_id=MASS_APP_ID,
218 use_mass_app=False,
219 launch_results={APP_MEDIA_RECEIVER: True},
220 )
221
222 await _launch_app(fake)
223
224 assert fake.launch_attempts == [APP_MEDIA_RECEIVER]
225
226
227async def test_refusal_reason_is_logged() -> None:
228 """The reason the receiver gave is included in the warning."""
229 fake = _fake_cast(launch_results={MASS_APP_ID: False}, refusal_reason="NOT_FOUND")
230
231 with pytest.raises(PlayerUnavailableError):
232 await _launch_app(fake)
233
234 assert "NOT_FOUND" in _warnings(fake)
235
236
237async def test_failure_suggests_disabling_the_mass_app() -> None:
238 """A failed Music Assistant app launch suggests turning the option off."""
239 fake = _fake_cast(launch_results={MASS_APP_ID: False})
240
241 with pytest.raises(PlayerUnavailableError):
242 await _launch_app(fake)
243
244 assert "disabling" in _warnings(fake)
245
246
247async def test_failure_suggests_enabling_the_mass_app() -> None:
248 """A failed default receiver launch suggests turning the option on."""
249 fake = _fake_cast(use_mass_app=False, launch_results={APP_MEDIA_RECEIVER: False})
250
251 with pytest.raises(PlayerUnavailableError):
252 await _launch_app(fake)
253
254 assert "enabling" in _warnings(fake)
255
256
257async def test_acknowledged_launch_that_did_not_start_the_app_fails() -> None:
258 """A launch the receiver accepted but did not act on must still fail."""
259 fake = _fake_cast(launch_results={MASS_APP_ID: True}, app_id_after_launch=None)
260
261 with pytest.raises(PlayerUnavailableError):
262 await _launch_app(fake)
263