music-assistant-server

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