/
/
/
1"""Unit tests for the output protocol config entries a player renders."""
2
3from __future__ import annotations
4
5import json
6from pathlib import Path
7from typing import TYPE_CHECKING
8from unittest.mock import MagicMock, patch
9
10from music_assistant_models.config_entries import ConfigEntry
11from music_assistant_models.enums import ConfigEntryType
12from music_assistant_models.player import OutputProtocol
13
14from music_assistant import constants as _constants
15from music_assistant.constants import (
16 CONF_ENABLED,
17 CONF_ENTRY_CROSSFADE_DIFFERENT_SAMPLE_RATES,
18 CONF_ENTRY_FLOW_MODE,
19 CONF_FLOW_MODE,
20 CONF_PLAYERS,
21 CONF_PREFERRED_OUTPUT_PROTOCOL,
22 CONF_PROTOCOL_EXPERIMENTAL_NOTE,
23 CONF_PROTOCOL_KEY_SPLITTER,
24)
25from music_assistant.mass import MusicAssistant
26from music_assistant.models.player import LinkedOutputProtocol
27
28if TYPE_CHECKING:
29 from music_assistant_models.config_entries import ConfigValueOption
30
31# the common strings live next to the constants module, so this path holds from anywhere
32_STRINGS_PATH = Path(_constants.__file__).resolve().parent / "strings.json"
33
34_AIRPLAY_ID = "airplay_aabbccddeeff"
35_DLNA_ID = "dlna_aabbccddeeff"
36_DLNA_PREFIX = f"{_DLNA_ID}{CONF_PROTOCOL_KEY_SPLITTER}"
37_PARENT_ID = "soundtouch_aabbccddeeff"
38
39
40def _make_output_protocol(
41 output_protocol_id: str, domain: str, priority: int, *, available: bool
42) -> OutputProtocol:
43 """Return an output protocol entry for a linked protocol player."""
44 return OutputProtocol(
45 output_protocol_id=output_protocol_id,
46 name=domain.title(),
47 protocol_domain=domain,
48 priority=priority,
49 available=available,
50 )
51
52
53def _make_protocol_player(*, available: bool, needs_setup: bool) -> MagicMock:
54 """Return a protocol player mock in the given availability/setup state."""
55 player = MagicMock()
56 player.available = available
57 player.needs_setup = needs_setup
58 player.available_for_playback = available and not needs_setup
59 return player
60
61
62def _make_provider_manifest(domain: str) -> MagicMock:
63 """Return a provider manifest mock named after its domain."""
64 manifest = MagicMock()
65 manifest.name = domain.title()
66 return manifest
67
68
69async def _preferred_output_entry(
70 mass: MusicAssistant,
71 protocols: list[OutputProtocol],
72 protocol_player: MagicMock | None = None,
73) -> ConfigEntry:
74 """Build the preferred-output-protocol entry for a player with the given outputs."""
75 player = MagicMock()
76 player.needs_setup = False
77 player.output_protocols = protocols
78 mass.players = MagicMock()
79 mass.players.get_player.return_value = protocol_player
80 # the entries name each protocol after its provider, which mass_minimal does not load
81 with patch.object(mass, "get_provider_manifest", side_effect=_make_provider_manifest):
82 entries = await mass.config._create_output_protocol_config_entries(player)
83 return next(entry for entry in entries if entry.key == CONF_PREFERRED_OUTPUT_PROTOCOL)
84
85
86async def _protocol_block_entries(
87 mass: MusicAssistant, protocol_entries: list[ConfigEntry]
88) -> dict[str, ConfigEntry]:
89 """
90 Build the config block a player renders for a single (non-native) output protocol.
91
92 :param mass: the MusicAssistant instance to build the entries with.
93 :param protocol_entries: the config entries the protocol's own player reports.
94 """
95 player = MagicMock()
96 player.needs_setup = False
97 player.output_protocols = [_make_output_protocol(_DLNA_ID, "dlna", 50, available=True)]
98 protocol_player = _make_protocol_player(available=True, needs_setup=False)
99 protocol_player.translation_owner = "dlna"
100 mass.players = MagicMock()
101 mass.players.get_player.return_value = protocol_player
102 with (
103 patch.object(mass, "get_provider_manifest", side_effect=_make_provider_manifest),
104 # the block is only built for a protocol whose provider is loaded
105 patch.object(mass, "get_provider", return_value=MagicMock()),
106 patch.object(mass.config, "_get_player_config_entries", return_value=protocol_entries),
107 ):
108 entries = await mass.config._create_output_protocol_config_entries(player)
109 return {entry.key: entry for entry in entries}
110
111
112async def _control_only_player_entries(
113 mass: MusicAssistant,
114 parent_entries: list[ConfigEntry],
115 protocol_entries: list[ConfigEntry],
116) -> dict[str, ConfigEntry]:
117 """
118 Build the full config surface of a control-only player, the way the api renders it.
119
120 :param mass: the MusicAssistant instance to build the entries with.
121 :param parent_entries: the config entries the control-only player itself reports.
122 :param protocol_entries: the config entries its linked protocol player reports.
123 """
124 player = MagicMock()
125 player.player_id = _PARENT_ID
126 player.needs_setup = False
127 # no native protocol: this player only controls, playback goes through the linked protocol
128 player.output_protocols = [_make_output_protocol(_DLNA_ID, "dlna", 50, available=True)]
129 player.linked_output_protocols = [
130 LinkedOutputProtocol(output_protocol_id=_DLNA_ID, protocol_domain="dlna", priority=50)
131 ]
132 protocol_player = _make_protocol_player(available=True, needs_setup=False)
133 protocol_player.player_id = _DLNA_ID
134 protocol_player.translation_owner = "dlna"
135 players = {_PARENT_ID: player, _DLNA_ID: protocol_player}
136 own_entries = {_PARENT_ID: parent_entries, _DLNA_ID: protocol_entries}
137 mass.players = MagicMock()
138 mass.players.get_player.side_effect = lambda player_id, *_: players.get(player_id)
139 mass.players.player_controls.return_value = []
140 with (
141 patch.object(mass, "get_provider_manifest", side_effect=_make_provider_manifest),
142 # the block is only built for a protocol whose provider is loaded
143 patch.object(mass, "get_provider", return_value=MagicMock()),
144 patch.object(
145 mass.config,
146 "_get_player_config_entries",
147 side_effect=lambda target: own_entries[target.player_id],
148 ),
149 ):
150 entries = await mass.config.get_player_config_entries(_PARENT_ID)
151 return {entry.key: entry for entry in entries}
152
153
154def _option(entry: ConfigEntry, value: str) -> ConfigValueOption:
155 """Return the entry's option for the given value."""
156 return next(option for option in entry.options if option.value == value)
157
158
159def _assert_disabled_with_reason(option: ConfigValueOption, reason: str) -> None:
160 """Assert the option is disabled for the given reason, and that the reason has a string."""
161 assert option.disabled is True
162 assert option.translation_key == reason
163 # the reason is only rendered when it is authored, so guard against drift
164 strings = json.loads(_STRINGS_PATH.read_text(encoding="utf-8"))
165 assert reason in strings["config_entries"][CONF_PREFERRED_OUTPUT_PROTOCOL]["disabled_reasons"]
166
167
168async def test_output_that_needs_setup_is_offered_disabled(mass_minimal: MusicAssistant) -> None:
169 """An output awaiting setup stays listed, disabled, and says why."""
170 entry = await _preferred_output_entry(
171 mass_minimal,
172 [
173 _make_output_protocol(_AIRPLAY_ID, "airplay", 10, available=False),
174 _make_output_protocol(_DLNA_ID, "dlna", 50, available=True),
175 ],
176 _make_protocol_player(available=True, needs_setup=True),
177 )
178 _assert_disabled_with_reason(_option(entry, _AIRPLAY_ID), "needs_setup")
179 assert _option(entry, _DLNA_ID).disabled is False
180
181
182async def test_offline_output_reports_unavailable(mass_minimal: MusicAssistant) -> None:
183 """An output whose player is gone reads as unavailable rather than needing setup."""
184 entry = await _preferred_output_entry(
185 mass_minimal,
186 [_make_output_protocol(_AIRPLAY_ID, "airplay", 10, available=False)],
187 None,
188 )
189 _assert_disabled_with_reason(_option(entry, _AIRPLAY_ID), "unavailable")
190
191
192async def test_output_turned_off_reports_turned_off(mass_minimal: MusicAssistant) -> None:
193 """An output the user turned off says so, so the enable toggle below makes sense."""
194 mass_minimal.config.set(f"{CONF_PLAYERS}/{_AIRPLAY_ID}/{CONF_ENABLED}", False)
195 entry = await _preferred_output_entry(
196 mass_minimal,
197 [_make_output_protocol(_AIRPLAY_ID, "airplay", 10, available=False)],
198 _make_protocol_player(available=True, needs_setup=True),
199 )
200 _assert_disabled_with_reason(_option(entry, _AIRPLAY_ID), "turned_off")
201
202
203async def test_default_is_never_a_disabled_option(mass_minimal: MusicAssistant) -> None:
204 """A native output that can not be used must not become the entry's default."""
205 entry = await _preferred_output_entry(
206 mass_minimal,
207 [
208 OutputProtocol(
209 output_protocol_id="native",
210 name="Sonos",
211 protocol_domain="sonos",
212 priority=0,
213 available=False,
214 is_native=True,
215 ),
216 _make_output_protocol(_DLNA_ID, "dlna", 50, available=True),
217 ],
218 )
219 assert entry.default_value == "auto"
220 assert _option(entry, entry.default_value).disabled is False
221
222
223async def test_protocol_entry_keeps_dependency_on_its_own_sibling(
224 mass_minimal: MusicAssistant,
225) -> None:
226 """An entry gated on a sibling stays gated on it after being copied into the block."""
227 entries = await _protocol_block_entries(
228 mass_minimal,
229 [
230 ConfigEntry(key="display", type=ConfigEntryType.BOOLEAN, default_value=False),
231 ConfigEntry(
232 key="visualization",
233 type=ConfigEntryType.STRING,
234 default_value="none",
235 depends_on="display",
236 depends_on_value=True,
237 ),
238 ],
239 )
240 visualization = entries[f"{_DLNA_PREFIX}visualization"]
241 assert visualization.depends_on == f"{_DLNA_PREFIX}display"
242 # an entry pointing at a key that is not in the block reads as unmet, so it would hide
243 assert visualization.depends_on in entries
244 assert visualization.depends_on_value is True
245
246
247async def test_protocol_entry_without_dependency_follows_the_protocol_toggle(
248 mass_minimal: MusicAssistant,
249) -> None:
250 """An entry with no dependency of its own is gated on the protocol's enable toggle."""
251 entries = await _protocol_block_entries(
252 mass_minimal,
253 [ConfigEntry(key="buffer_depth", type=ConfigEntryType.INTEGER, default_value=5)],
254 )
255 enabled_key = f"{_DLNA_PREFIX}{CONF_ENABLED}"
256 assert entries[f"{_DLNA_PREFIX}buffer_depth"].depends_on == enabled_key
257 assert enabled_key in entries
258
259
260async def test_unresolvable_dependency_drops_its_value_condition(
261 mass_minimal: MusicAssistant,
262) -> None:
263 """An entry whose dependency is absent falls back without carrying its condition over."""
264 entries = await _protocol_block_entries(
265 mass_minimal,
266 [
267 ConfigEntry(
268 key="flow_mode_sample_rate",
269 type=ConfigEntryType.STRING,
270 default_value="smart",
271 depends_on=CONF_FLOW_MODE,
272 depends_on_value_not=True,
273 )
274 ],
275 )
276 entry = entries[f"{_DLNA_PREFIX}flow_mode_sample_rate"]
277 assert entry.depends_on == f"{_DLNA_PREFIX}{CONF_ENABLED}"
278 # the condition was written for flow mode; against the toggle it would invert the gate
279 assert entry.depends_on_value_not is None
280
281
282async def test_crossfade_entry_still_tracks_flow_mode(mass_minimal: MusicAssistant) -> None:
283 """The shared 'only without flow mode' entries keep their meaning inside a protocol block."""
284 entries = await _protocol_block_entries(
285 mass_minimal,
286 [CONF_ENTRY_FLOW_MODE, CONF_ENTRY_CROSSFADE_DIFFERENT_SAMPLE_RATES],
287 )
288 crossfade = entries[f"{_DLNA_PREFIX}{CONF_ENTRY_CROSSFADE_DIFFERENT_SAMPLE_RATES.key}"]
289 assert crossfade.depends_on == f"{_DLNA_PREFIX}{CONF_FLOW_MODE}"
290 assert crossfade.depends_on in entries
291 assert crossfade.depends_on_value_not is True
292 # the shared constant is reused across players, so the block must not have mutated it
293 assert CONF_ENTRY_CROSSFADE_DIFFERENT_SAMPLE_RATES.depends_on == CONF_FLOW_MODE
294
295
296async def test_protocol_dependency_never_binds_to_the_players_own_setting(
297 mass_minimal: MusicAssistant,
298) -> None:
299 """A copied entry follows the protocol's own setting, not the player's same-named one."""
300 entries = await _control_only_player_entries(
301 mass_minimal,
302 [ConfigEntry(key=CONF_FLOW_MODE, type=ConfigEntryType.BOOLEAN, default_value=True)],
303 [
304 ConfigEntry(key=CONF_FLOW_MODE, type=ConfigEntryType.BOOLEAN, default_value=False),
305 ConfigEntry(
306 key="flow_mode_sample_rate",
307 type=ConfigEntryType.STRING,
308 default_value="smart",
309 depends_on=CONF_FLOW_MODE,
310 depends_on_value=True,
311 ),
312 ],
313 )
314 sample_rate = entries[f"{_DLNA_PREFIX}flow_mode_sample_rate"]
315 assert sample_rate.depends_on == f"{_DLNA_PREFIX}{CONF_FLOW_MODE}"
316 assert sample_rate.depends_on_value is True
317 # both settings are rendered side by side on this player, so a bare key would have
318 # gated the protocol's entry on the player's own flow mode instead of the protocol's
319 assert entries[f"{_DLNA_PREFIX}{CONF_FLOW_MODE}"].default_value is False
320 assert entries[CONF_FLOW_MODE].default_value is True
321
322
323async def _preferred_entry_with_default_domain(
324 mass: MusicAssistant,
325 protocols: list[OutputProtocol],
326 default_domain: str | None,
327) -> ConfigEntry:
328 """Build the preferred-output entry for a player that declares a default protocol domain."""
329 player = MagicMock()
330 player.needs_setup = False
331 player.output_protocols = protocols
332 player.default_output_protocol_domain = default_domain
333 mass.players = MagicMock()
334 mass.players.get_player.return_value = None
335 with patch.object(mass, "get_provider_manifest", side_effect=_make_provider_manifest):
336 entries = await mass.config._create_output_protocol_config_entries(player)
337 return next(entry for entry in entries if entry.key == CONF_PREFERRED_OUTPUT_PROTOCOL)
338
339
340async def test_default_domain_available_stays_auto(
341 mass_minimal: MusicAssistant,
342) -> None:
343 """A no-native player defaults to auto even when its default-domain output is available."""
344 entry = await _preferred_entry_with_default_domain(
345 mass_minimal,
346 [
347 _make_output_protocol(_DLNA_ID, "dlna", 50, available=True),
348 _make_output_protocol(_AIRPLAY_ID, "airplay", 10, available=True),
349 ],
350 default_domain="dlna",
351 )
352 # The stored default must not depend on which linked protocols happen to be available;
353 # runtime selection applies the default domain, but the persisted entry stays "auto".
354 assert entry.default_value == "auto"
355 # auto and both protocols remain selectable so the user can still override
356 assert {option.value for option in entry.options} >= {"auto", _DLNA_ID, _AIRPLAY_ID}
357
358
359async def test_default_domain_unavailable_stays_auto(mass_minimal: MusicAssistant) -> None:
360 """A no-native player defaults to auto even when its default-domain output is unavailable."""
361 entry = await _preferred_entry_with_default_domain(
362 mass_minimal,
363 [
364 _make_output_protocol(_DLNA_ID, "dlna", 50, available=False),
365 _make_output_protocol(_AIRPLAY_ID, "airplay", 10, available=True),
366 ],
367 default_domain="dlna",
368 )
369 assert entry.default_value == "auto"
370
371
372async def test_default_domain_absent_stays_auto(mass_minimal: MusicAssistant) -> None:
373 """A player whose default domain is not among its outputs falls back to auto."""
374 entry = await _preferred_entry_with_default_domain(
375 mass_minimal,
376 [_make_output_protocol(_AIRPLAY_ID, "airplay", 10, available=True)],
377 default_domain="dlna",
378 )
379 assert entry.default_value == "auto"
380
381
382async def test_no_default_domain_stays_auto(mass_minimal: MusicAssistant) -> None:
383 """Without a declared default domain, a player with no native output defaults to auto."""
384 entry = await _preferred_entry_with_default_domain(
385 mass_minimal,
386 [_make_output_protocol(_DLNA_ID, "dlna", 50, available=True)],
387 default_domain=None,
388 )
389 assert entry.default_value == "auto"
390
391
392async def test_available_native_ignores_default_domain(mass_minimal: MusicAssistant) -> None:
393 """An available native output still wins over a declared default protocol domain."""
394 native = OutputProtocol(
395 output_protocol_id="native_x",
396 name="Native",
397 protocol_domain="soundtouch",
398 is_native=True,
399 priority=1,
400 available=True,
401 )
402 entry = await _preferred_entry_with_default_domain(
403 mass_minimal,
404 [native, _make_output_protocol(_DLNA_ID, "dlna", 50, available=True)],
405 default_domain="dlna",
406 )
407 assert entry.default_value == "native"
408
409
410async def test_experimental_output_defaults_to_off_with_a_warning(
411 mass_minimal: MusicAssistant,
412) -> None:
413 """An output its provider flagged as experimental is opt-in and says why."""
414 mass_minimal.config.set(
415 f"{CONF_PLAYERS}/{_DLNA_ID}/values/{CONF_PROTOCOL_EXPERIMENTAL_NOTE}",
416 "sendspin_cast_experimental",
417 )
418 entries = await _protocol_block_entries(mass_minimal, [])
419 enabled = entries[f"{_DLNA_PREFIX}{CONF_ENABLED}"]
420 assert enabled.default_value is False
421 assert enabled.translation_key == "protocol_enable_experimental"
422 # the label is only rendered when it is authored, so guard against drift
423 strings = json.loads(_STRINGS_PATH.read_text(encoding="utf-8"))
424 assert "protocol_enable_experimental" in strings["config_entries"]
425 warning = entries[f"{_DLNA_PREFIX}sendspin_cast_experimental"]
426 assert warning.type is ConfigEntryType.ALERT
427 # the note is owned by the protocol's provider, not by the player rendering it
428 assert warning.translation_owner == "provider.dlna"
429 assert warning.translation_key == "sendspin_cast_experimental"
430 # the warning explains why the toggle is off, so it must not be hidden behind it
431 assert warning.depends_on is None
432
433
434async def test_warning_of_a_switched_off_output_still_resolves(
435 mass_minimal: MusicAssistant,
436) -> None:
437 """A note on an output whose player is gone resolves against that player's provider."""
438 mass_minimal.config.set(
439 f"{CONF_PLAYERS}/{_DLNA_ID}",
440 {"player_id": _DLNA_ID, "provider": "sendspin--abc123", "enabled": False},
441 )
442 mass_minimal.config.set(
443 f"{CONF_PLAYERS}/{_DLNA_ID}/values/{CONF_PROTOCOL_EXPERIMENTAL_NOTE}",
444 "sendspin_cast_experimental",
445 )
446 # the config parse stamps every entry with the player's own owner, so an output with
447 # no registered player to ask must fall back to the provider in its stored config
448 assert mass_minimal.config._stored_protocol_translation_owner(_DLNA_ID) == "provider.sendspin"
449
450
451async def test_note_of_an_unknown_output_has_no_owner(mass_minimal: MusicAssistant) -> None:
452 """An output without a stored config resolves to no owner rather than a wrong one."""
453 assert mass_minimal.config._stored_protocol_translation_owner("nope_1234") is None
454
455
456async def test_plain_output_keeps_its_default_and_label(mass_minimal: MusicAssistant) -> None:
457 """An output without the flag is unchanged: on by default, no warning."""
458 entries = await _protocol_block_entries(mass_minimal, [])
459 enabled = entries[f"{_DLNA_PREFIX}{CONF_ENABLED}"]
460 assert enabled.default_value is True
461 assert enabled.translation_key == "protocol_enable"
462 assert not [key for key in entries if "experimental" in key]
463