/
/
/
1"""
2Regression tests for removing a player config with linked protocol players.
3
4Reproduces the case where a player is first disabled and then removed: disabling
5cascades to the linked protocol players, so none of them is registered anymore when
6the removal comes in. The leftover (disabled) protocol config is not shown anywhere
7and keeps the device from ever registering again.
8
9Also covers the mirrored case where the player being removed is not registered (e.g.
10its provider was unloaded) while one of its protocol players still is: that protocol
11player must be detached from the removed player instead of keeping a dead parent link.
12"""
13
14from collections.abc import Callable, Generator
15from types import SimpleNamespace
16
17import pytest
18from music_assistant_models.enums import PlaybackState, PlayerType
19
20from music_assistant.constants import CONF_PLAYER_DSP, CONF_PLAYERS, CONF_PROTOCOL_PARENT_ID
21from music_assistant.mass import MusicAssistant
22
23PARENT_ID = "up_esp32"
24PROTOCOL_ID = "spb_esp32"
25
26
27class StubProtocolPlayer:
28 """Minimal stand-in for a registered protocol player."""
29
30 def __init__(self, parent_id: str | None) -> None:
31 """Initialize the stub with the given (live) protocol parent."""
32 self.player_id = PROTOCOL_ID
33 # a registered player is looked up and polled by the running server,
34 # so carry the state fields those paths read
35 self.state = SimpleNamespace(
36 type=PlayerType.PROTOCOL,
37 playback_state=PlaybackState.IDLE,
38 available=True,
39 enabled=True,
40 )
41 self.needs_poll = False
42 self.protocol_parent_id = parent_id
43 self.refreshed = False
44
45 def set_protocol_parent_id(self, parent_id: str | None) -> None:
46 """Set the live protocol parent."""
47 self.protocol_parent_id = parent_id
48
49 def refresh_state(self) -> None:
50 """Record that the state was refreshed."""
51 self.refreshed = True
52
53
54@pytest.fixture(name="register_protocol_player")
55def register_protocol_player_fixture(
56 mass: MusicAssistant,
57) -> Generator[Callable[[str | None], StubProtocolPlayer]]:
58 """Register a stub protocol player on the player controller for the test."""
59
60 def _register(parent_id: str | None) -> StubProtocolPlayer:
61 protocol_player = StubProtocolPlayer(parent_id)
62 mass.players._players[PROTOCOL_ID] = protocol_player # type: ignore[assignment]
63 return protocol_player
64
65 yield _register
66 mass.players._players.pop(PROTOCOL_ID, None)
67
68
69def _pop_scheduled_evaluation(mass: MusicAssistant) -> bool:
70 """Return True if a protocol evaluation is pending for the protocol player."""
71 if handle := mass.players._pending_protocol_evaluations.pop(PROTOCOL_ID, None):
72 handle.cancel()
73 return True
74 return False
75
76
77def _store_configs(mass: MusicAssistant, enabled: bool) -> None:
78 """Store a universal player config with a single linked protocol player."""
79 mass.config.set(
80 f"{CONF_PLAYERS}/{PARENT_ID}",
81 {
82 "player_id": PARENT_ID,
83 "provider": "universal_player",
84 "player_type": "player",
85 "enabled": enabled,
86 "values": {"linked_protocol_ids": [PROTOCOL_ID]},
87 },
88 )
89 mass.config.set(
90 f"{CONF_PLAYERS}/{PROTOCOL_ID}",
91 {
92 "player_id": PROTOCOL_ID,
93 "provider": "sendspin",
94 "player_type": "protocol",
95 "enabled": enabled,
96 "values": {CONF_PROTOCOL_PARENT_ID: PARENT_ID},
97 },
98 )
99 mass.config.set(f"{CONF_PLAYER_DSP}/{PROTOCOL_ID}", {"enabled": True})
100
101
102async def test_remove_wipes_unregistered_protocol_configs(mass: MusicAssistant) -> None:
103 """Removing a disabled player also wipes the config of its linked protocol player."""
104 _store_configs(mass, enabled=False)
105
106 await mass.config.remove_player_config(PARENT_ID)
107
108 assert mass.config.get(f"{CONF_PLAYERS}/{PARENT_ID}") is None
109 assert mass.config.get(f"{CONF_PLAYERS}/{PROTOCOL_ID}") is None
110 assert mass.config.get(f"{CONF_PLAYER_DSP}/{PROTOCOL_ID}") is None
111
112
113async def test_remove_wipes_protocol_configs_with_a_half_broken_link(
114 mass: MusicAssistant,
115) -> None:
116 """A protocol player is wiped by its own parent reference, not by the parent's list."""
117 _store_configs(mass, enabled=False)
118 mass.config.set(f"{CONF_PLAYERS}/{PARENT_ID}/values/linked_protocol_ids", [])
119
120 await mass.config.remove_player_config(PARENT_ID)
121
122 assert mass.config.get(f"{CONF_PLAYERS}/{PROTOCOL_ID}") is None
123
124
125async def test_remove_keeps_reparented_protocol_configs(mass: MusicAssistant) -> None:
126 """A protocol player that already moved to another parent keeps its config."""
127 _store_configs(mass, enabled=True)
128 mass.config.set(f"{CONF_PLAYERS}/{PROTOCOL_ID}/values/{CONF_PROTOCOL_PARENT_ID}", "cast_1")
129
130 mass.players.delete_player_config(PARENT_ID)
131
132 assert mass.config.get(f"{CONF_PLAYERS}/{PARENT_ID}") is None
133 assert mass.config.get(f"{CONF_PLAYERS}/{PROTOCOL_ID}") is not None
134
135
136async def test_remove_keeps_registered_protocol_configs(
137 mass: MusicAssistant,
138 register_protocol_player: Callable[[str | None], StubProtocolPlayer],
139) -> None:
140 """A protocol player that is still registered keeps its config to be re-parented."""
141 _store_configs(mass, enabled=True)
142 register_protocol_player(PARENT_ID)
143
144 mass.players.delete_player_config(PARENT_ID)
145
146 assert mass.config.get(f"{CONF_PLAYERS}/{PARENT_ID}") is None
147 assert mass.config.get(f"{CONF_PLAYERS}/{PROTOCOL_ID}") is not None
148 assert mass.config.get(f"{CONF_PLAYER_DSP}/{PROTOCOL_ID}") is not None
149 _pop_scheduled_evaluation(mass)
150
151
152async def test_remove_detaches_registered_protocol_player(
153 mass: MusicAssistant,
154 register_protocol_player: Callable[[str | None], StubProtocolPlayer],
155) -> None:
156 """Removing an unregistered parent detaches its still registered protocol player."""
157 _store_configs(mass, enabled=True)
158 protocol_player = register_protocol_player(PARENT_ID)
159
160 await mass.config.remove_player_config(PARENT_ID)
161
162 assert mass.config.get(f"{CONF_PLAYERS}/{PARENT_ID}") is None
163 assert mass.config.get(f"{CONF_PLAYERS}/{PROTOCOL_ID}") is not None
164 assert protocol_player.protocol_parent_id is None
165 assert protocol_player.refreshed
166 assert mass.config.get(f"{CONF_PLAYERS}/{PROTOCOL_ID}/values/{CONF_PROTOCOL_PARENT_ID}") is None
167 assert _pop_scheduled_evaluation(mass)
168
169
170async def test_remove_detaches_protocol_player_waiting_for_its_parent(
171 mass: MusicAssistant,
172 register_protocol_player: Callable[[str | None], StubProtocolPlayer],
173) -> None:
174 """A protocol player that only has the parent link in its config is detached too."""
175 _store_configs(mass, enabled=True)
176 protocol_player = register_protocol_player(None)
177
178 await mass.config.remove_player_config(PARENT_ID)
179
180 assert mass.config.get(f"{CONF_PLAYERS}/{PROTOCOL_ID}/values/{CONF_PROTOCOL_PARENT_ID}") is None
181 assert protocol_player.protocol_parent_id is None
182 assert _pop_scheduled_evaluation(mass)
183
184
185async def test_remove_leaves_unrelated_protocol_player_alone(
186 mass: MusicAssistant,
187 register_protocol_player: Callable[[str | None], StubProtocolPlayer],
188) -> None:
189 """A protocol player of another parent keeps its link when a player is removed."""
190 _store_configs(mass, enabled=True)
191 protocol_player = register_protocol_player("cast_1")
192
193 mass.players.delete_player_config(PARENT_ID)
194
195 assert protocol_player.protocol_parent_id == "cast_1"
196 assert not _pop_scheduled_evaluation(mass)
197