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