music-assistant-server

1"""
2Regression tests for control-only players losing their own config entries.
3
4Reproduces music-assistant/support#4766: a player that only handles control
5(power/volume/source) and delegates playback to a linked protocol player (e.g.
6Bose SoundTouch streaming via DLNA) has no native output protocol. The injected
7protocol entries replaced the player's own entries entirely, so everything its
8``get_config_entries()`` returned disappeared from the config UI.
9"""
10
11import logging
12from unittest.mock import AsyncMock, MagicMock
13
14from music_assistant_models.config_entries import ConfigEntry
15from music_assistant_models.enums import ConfigEntryType, PlayerFeature, PlayerType, ProviderType
16from music_assistant_models.player import OutputProtocol
17
18from music_assistant.constants import (
19    CONF_PREFERRED_OUTPUT_PROTOCOL,
20    CONF_PROTOCOL_KEY_SPLITTER,
21)
22from music_assistant.mass import MusicAssistant
23from music_assistant.models.player import DeviceInfo, Player
24
25PARENT_ID = "soundtouch_123"
26CHILD_ID = "dlna_AABBCCDDEEFF"
27OWN_ENTRY_KEY = "preset_slot"
28
29
30class _TestProvider:
31    """Minimal PlayerProvider stand-in backed by the real MusicAssistant."""
32
33    def __init__(self, mass: MusicAssistant, domain: str) -> None:
34        """Initialize the test provider."""
35        self.mass = mass
36        self.domain = domain
37        self.instance_id = domain
38        self.translation_owner = f"provider.{domain}"
39        self.name = f"{domain.title()} Provider"
40        self.available = True
41        self.logger = logging.getLogger(f"test.{domain}")
42        self.manifest = MagicMock()
43        self.manifest.domain = domain
44        self.manifest.name = self.name
45        self.manifest.type = ProviderType.PLAYER
46        # registered in mass._providers, so the stand-in must survive the provider
47        # bookkeeping that mass.stop()/unload_provider applies to every provider
48        self.type = ProviderType.PLAYER
49        self.players: list[Player] = []
50
51    async def unload(self, is_removed: bool = False) -> None:
52        """Unload the provider (nothing to clean up)."""
53
54
55class _ControlOnlyPlayer(Player):
56    """Player that only offers control and delegates playback to a protocol player."""
57
58    def __init__(self, provider: _TestProvider, player_id: str, name: str) -> None:
59        """Initialize the test player."""
60        super().__init__(provider, player_id)  # type: ignore[arg-type]
61        self._attr_name = name
62        self._attr_type = PlayerType.PLAYER
63        self._attr_available = True
64        self._attr_powered = True
65        # deliberately no PLAY_MEDIA: this player cannot play media itself,
66        # which is what makes all of its output protocols non-native
67        self._attr_supported_features = {PlayerFeature.VOLUME_SET, PlayerFeature.POWER}
68        self._attr_device_info = DeviceInfo(model="Test Model", manufacturer="Test Manufacturer")
69        self._cache.clear()
70        self.update_state(signal_event=False)
71
72    async def get_config_entries(self) -> list[ConfigEntry]:
73        """Return the player's own config entries."""
74        return [
75            ConfigEntry(
76                key=OWN_ENTRY_KEY,
77                type=ConfigEntryType.STRING,
78                label="Preset slot",
79                required=False,
80            ),
81            # collides with an entry the protocol block already emits: must be deduped
82            ConfigEntry(
83                key=CONF_PREFERRED_OUTPUT_PROTOCOL,
84                type=ConfigEntryType.STRING,
85                label="Should be ignored",
86                default_value="ignored",
87                required=False,
88            ),
89        ]
90
91    async def stop(self) -> None:
92        """Stop playback - required abstract method."""
93
94
95class _ProtocolPlayer(Player):
96    """Minimal protocol player that provides the actual playback output."""
97
98    def __init__(self, provider: _TestProvider, player_id: str, name: str) -> None:
99        """Initialize the test player."""
100        super().__init__(provider, player_id)  # type: ignore[arg-type]
101        self._attr_name = name
102        self._attr_type = PlayerType.PROTOCOL
103        self._attr_available = True
104        self._attr_powered = True
105        self._attr_supported_features = {PlayerFeature.VOLUME_SET, PlayerFeature.PLAY_MEDIA}
106        self._attr_device_info = DeviceInfo(model="Test Model", manufacturer="Test Manufacturer")
107        self._cache.clear()
108        self.update_state(signal_event=False)
109
110    async def stop(self) -> None:
111        """Stop playback - required abstract method."""
112
113
114async def _setup_control_only_player(mass: MusicAssistant) -> _ControlOnlyPlayer:
115    """Register a control-only player linked to a DLNA protocol player."""
116    soundtouch_provider = _TestProvider(mass, "soundtouch")
117    dlna_provider = _TestProvider(mass, "dlna")
118    mass._providers[soundtouch_provider.instance_id] = soundtouch_provider  # type: ignore[assignment]
119    mass._providers[dlna_provider.instance_id] = dlna_provider  # type: ignore[assignment]
120    mass._provider_manifests[soundtouch_provider.domain] = soundtouch_provider.manifest
121    mass._provider_manifests[dlna_provider.domain] = dlna_provider.manifest
122
123    parent = _ControlOnlyPlayer(soundtouch_provider, PARENT_ID, "Kitchen")
124    child = _ProtocolPlayer(dlna_provider, CHILD_ID, "Kitchen DLNA")
125
126    mass.players._players[PARENT_ID] = parent
127    mass.players._players[CHILD_ID] = child
128
129    parent.set_linked_output_protocols(
130        [
131            OutputProtocol(
132                output_protocol_id=CHILD_ID,
133                name="DLNA",
134                protocol_domain="dlna",
135                priority=50,
136            )
137        ]
138    )
139    # output_protocols is a cached property populated during __init__'s update_state
140    # (empty at that point); drop the cache so the freshly linked protocol shows up
141    parent._cache.clear()
142
143    # avoid driving the full player-manager reload machinery
144    mass.players.on_player_config_change = AsyncMock()  # type: ignore[method-assign]
145    return parent
146
147
148async def test_control_only_player_keeps_own_config_entries(mass: MusicAssistant) -> None:
149    """A player with only non-native output protocols keeps its own config entries."""
150    parent = await _setup_control_only_player(mass)
151    assert not any(protocol.is_native for protocol in parent.output_protocols)
152
153    entries = await mass.config.get_player_config_entries(PARENT_ID)
154    keys = [entry.key for entry in entries]
155
156    # the player's own entries survive alongside the injected protocol entries
157    assert OWN_ENTRY_KEY in keys
158    assert CONF_PREFERRED_OUTPUT_PROTOCOL in keys
159    assert any(CONF_PROTOCOL_KEY_SPLITTER in key for key in keys)
160
161
162async def test_control_only_player_entries_dedup_by_key(mass: MusicAssistant) -> None:
163    """Own entries that collide with an injected protocol entry are skipped."""
164    await _setup_control_only_player(mass)
165
166    entries = await mass.config.get_player_config_entries(PARENT_ID)
167    keys = [entry.key for entry in entries]
168
169    assert len(keys) == len(set(keys))
170    assert keys.count(OWN_ENTRY_KEY) == 1
171    # the protocol-owned entry wins over the player's colliding one
172    protocol_entry = next(entry for entry in entries if entry.key == CONF_PREFERRED_OUTPUT_PROTOCOL)
173    assert protocol_entry.label != "Should be ignored"
174    assert protocol_entry.options
175