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