/
/
/
1"""Tests for the generic LinkPlay grouping/identity shell of the WiiM provider."""
2
3from __future__ import annotations
4
5from types import SimpleNamespace
6from typing import TYPE_CHECKING, Any, cast
7from unittest.mock import AsyncMock, MagicMock, patch
8
9import pytest
10from music_assistant_models.enums import (
11 IdentifierType,
12 PlaybackState,
13 PlayerFeature,
14 PlayerType,
15)
16from pywiim import WiiMError
17
18from music_assistant.controllers.players.protocol_linking import ProtocolLinkingMixin
19from music_assistant.models.player import LinkedOutputProtocol
20from music_assistant.providers.wiim.constants import PLAYER_ID_PREFIX
21from music_assistant.providers.wiim.grouping import NativeGroupRole
22from music_assistant.providers.wiim.helpers import (
23 is_official_manufacturer,
24 linkplay_group_compatible,
25 linkplay_slave_uuid_to_player_id,
26 linkplay_slave_uuid_to_udn,
27)
28from music_assistant.providers.wiim.linkplay_player import LinkPlayPlayer
29from music_assistant.providers.wiim.provider import WiimProvider
30
31if TYPE_CHECKING:
32 from pywiim.models import DeviceInfo as PywiimDeviceInfo
33
34# Verified Edifier MS50A identity used across the tests.
35EDIFIER_HTTP_UUID = "FF97F002783E65056579F15F"
36EDIFIER_UDN = "uuid:FF97F002-783E-6505-6579-F15FFF97F002"
37EDIFIER_PLAYER_ID = f"{PLAYER_ID_PREFIX}{EDIFIER_UDN}"
38
39# A second generic device, used for same-backend grouping tests.
40PEER_HTTP_UUID = "AA11BB22CC33DD44EE55FF66"
41PEER_UDN = "uuid:AA11BB22-CC33-DD44-EE55-FF66AA11BB22"
42PEER_PLAYER_ID = f"{PLAYER_ID_PREFIX}{PEER_UDN}"
43
44
45class TestIdentityNormalization:
46 """The 24-char HTTP UUID must map deterministically to the UPnP UDN/player id."""
47
48 def test_http_uuid_to_udn_matches_hardware(self) -> None:
49 """The Edifier HTTP UUID resolves to its verified UPnP UDN."""
50 assert linkplay_slave_uuid_to_udn(EDIFIER_HTTP_UUID) == EDIFIER_UDN
51
52 def test_http_uuid_to_player_id(self) -> None:
53 """The player id is the prefixed UDN."""
54 assert linkplay_slave_uuid_to_player_id(EDIFIER_HTTP_UUID) == EDIFIER_PLAYER_ID
55
56 def test_lowercase_input_normalizes_to_canonical_udn(self) -> None:
57 """Case does not matter; the canonical UDN is uppercase."""
58 assert linkplay_slave_uuid_to_udn(EDIFIER_HTTP_UUID.lower()) == EDIFIER_UDN
59
60 @pytest.mark.parametrize(
61 "value",
62 [
63 EDIFIER_UDN,
64 EDIFIER_UDN.removeprefix("uuid:"),
65 EDIFIER_UDN.removeprefix("uuid:").replace("-", ""),
66 EDIFIER_UDN.lower(),
67 ],
68 )
69 def test_full_udn_forms_normalize_to_canonical(self, value: str) -> None:
70 """Slaves that report a full 32-hex UDN (any form) resolve to the same UDN."""
71 assert linkplay_slave_uuid_to_udn(value) == EDIFIER_UDN
72 assert linkplay_slave_uuid_to_player_id(value) == EDIFIER_PLAYER_ID
73
74 @pytest.mark.parametrize(
75 "value", ["", "tooshort", "ZZ97F002783E65056579F15F", "1" * 30, "1" * 40]
76 )
77 def test_invalid_uuid_returns_none(self, value: str) -> None:
78 """Input that is neither a 24-hex HTTP UUID nor a 32-hex UDN is rejected."""
79 assert linkplay_slave_uuid_to_udn(value) is None
80 assert linkplay_slave_uuid_to_player_id(value) is None
81
82
83class TestManufacturerClassification:
84 """Only WiiM/Audio Pro manufacturers select the official backend."""
85
86 @pytest.mark.parametrize("manufacturer", ["Linkplay", "linkplay technology", "Audio Pro AB"])
87 def test_official_manufacturers(self, manufacturer: str) -> None:
88 """Official manufacturers are recognised (mirrors the official SDK)."""
89 assert is_official_manufacturer(manufacturer) is True
90
91 @pytest.mark.parametrize("manufacturer", ["Edifier Inc", "", None, "Arylic", "WiiM"])
92 def test_generic_manufacturers(self, manufacturer: str | None) -> None:
93 """Everything else is treated as generic LinkPlay."""
94 assert is_official_manufacturer(manufacturer) is False
95
96
97def _slaves(uuids: list[str]) -> list[dict[str, str]]:
98 """Build a pywiim get_slaves_info()-style slave list."""
99 return [{"uuid": uuid, "ip": f"10.0.0.{index + 2}"} for index, uuid in enumerate(uuids)]
100
101
102def _device_info(wmrm_version: str = "4.2", *, legacy: bool = False) -> PywiimDeviceInfo:
103 """Build a device-info stand-in; legacy=True marks it as a Wi-Fi-Direct device."""
104 return cast(
105 "PywiimDeviceInfo",
106 SimpleNamespace(wmrm_version=wmrm_version, needs_wifi_direct_multiroom=legacy),
107 )
108
109
110@pytest.fixture
111def mock_provider() -> MagicMock:
112 """Create a mock WiimProvider suitable for constructing players."""
113 provider = MagicMock()
114 provider.instance_id = "wiim_test"
115 provider.domain = "wiim"
116 provider.players = []
117 provider.native_groups = _mock_native_groups()
118 provider.mass = MagicMock()
119 provider.mass.players = MagicMock()
120 config = MagicMock()
121 config.name = None
122 config.default_name = "Edifier MS50A"
123 config.enabled = True
124 config.player_type = None
125 config.get_value = MagicMock(return_value=None)
126 provider.mass.config.get_base_player_config.return_value = config
127 # used by the core final-state calculation to resolve power/volume controls
128 provider.mass.config.get_raw_player_config_value = MagicMock(
129 side_effect=lambda _player_id, _key, default=None: default
130 )
131 return provider
132
133
134def _mock_native_groups() -> MagicMock:
135 """Create a coordinator mock that reports a standalone topology by default."""
136 groups = MagicMock()
137 groups.role_of.return_value = NativeGroupRole.STANDALONE
138 groups.members_of.return_value = []
139 groups.can_group_with.return_value = set()
140 groups.refresh_leader = AsyncMock()
141 groups.reconcile = AsyncMock()
142 groups.set_members = AsyncMock()
143 groups.schedule_reconcile = MagicMock()
144 groups.schedule_republish = MagicMock()
145 groups.unregister = MagicMock()
146 groups.is_unknown_leader_follower = MagicMock(return_value=False)
147 groups.set_self_role = MagicMock(return_value=False)
148 return groups
149
150
151@pytest.fixture
152def mock_client() -> MagicMock:
153 """Create a mock low-level WiiMClient with the GroupAPI methods the shell uses."""
154 client = MagicMock()
155 client.host = "192.168.1.50"
156 client.get_device_info_model = AsyncMock(
157 return_value=SimpleNamespace(
158 uuid=EDIFIER_HTTP_UUID, wmrm_version="4.2", firmware="Linkplay.4.6.430230"
159 )
160 )
161 client.get_slaves_info = AsyncMock(return_value=_slaves([]))
162 client.get_device_group_info = AsyncMock(return_value=SimpleNamespace(role="solo"))
163 client.capabilities = {}
164 client.join_slave = AsyncMock()
165 client.leave_group = AsyncMock()
166 return client
167
168
169@pytest.fixture
170def mock_upnp_device() -> MagicMock:
171 """Create a mock async-upnp-client UpnpDevice."""
172 device = MagicMock()
173 device.manufacturer = "Edifier Inc"
174 device.model_name = "Edifier MS50A"
175 device.friendly_name = "Edifier MS50A"
176 device.udn = EDIFIER_UDN
177 return device
178
179
180def _make_shell(
181 provider: MagicMock,
182 client: MagicMock,
183 upnp_device: MagicMock,
184 player_id: str = EDIFIER_PLAYER_ID,
185) -> LinkPlayPlayer:
186 player = LinkPlayPlayer(
187 provider=provider,
188 player_id=player_id,
189 client=client,
190 upnp_device=upnp_device,
191 description_url="http://192.168.1.50:49152/description.xml",
192 mac_address="AA:BB:CC:DD:EE:FF",
193 )
194 player.update_state = MagicMock() # type: ignore[misc,method-assign]
195 # a modern, router-based device so compatibility checks pass by default
196 player._cached_device_info = _device_info("4.2")
197 return player
198
199
200class TestShellConstruction:
201 """The shell exposes device identity and a grouping-only native capability."""
202
203 def test_is_a_native_player(
204 self, mock_provider: MagicMock, mock_client: MagicMock, mock_upnp_device: MagicMock
205 ) -> None:
206 """The shell is a native PlayerType.PLAYER so the controller can link protocols."""
207 player = _make_shell(mock_provider, mock_client, mock_upnp_device)
208 assert player._attr_type == PlayerType.PLAYER
209
210 def test_prefers_dlna_default_output(
211 self, mock_provider: MagicMock, mock_client: MagicMock, mock_upnp_device: MagicMock
212 ) -> None:
213 """The shell steers automatic output selection at its linked DLNA protocol."""
214 player = _make_shell(mock_provider, mock_client, mock_upnp_device)
215 assert player.default_output_protocol_domain == "dlna"
216
217 def test_device_info_identifiers(
218 self, mock_provider: MagicMock, mock_client: MagicMock, mock_upnp_device: MagicMock
219 ) -> None:
220 """UUID/IP/MAC identifiers are set so DLNA/AirPlay children link by identity."""
221 player = _make_shell(mock_provider, mock_client, mock_upnp_device)
222 identifiers = player.device_info.identifiers
223 assert identifiers[IdentifierType.UUID] == EDIFIER_UDN.removeprefix("uuid:")
224 assert identifiers[IdentifierType.IP_ADDRESS] == "192.168.1.50"
225 assert identifiers[IdentifierType.MAC_ADDRESS] == "AA:BB:CC:DD:EE:FF"
226
227 def test_has_no_native_playback(
228 self, mock_provider: MagicMock, mock_client: MagicMock, mock_upnp_device: MagicMock
229 ) -> None:
230 """The shell must not implement native playback: no play_media, no PLAY_MEDIA."""
231 player = _make_shell(mock_provider, mock_client, mock_upnp_device)
232 # play_media is only defined on the base Player as the default reject stub, never
233 # overridden here, so the shell exposes no native PLAY_MEDIA feature.
234 assert "play_media" not in LinkPlayPlayer.__dict__
235 assert PlayerFeature.PLAY_MEDIA not in player._attr_supported_features
236 # no eventing/state-machine internals leaked from the old backend
237 for leaked in ("_push_state", "_connect_eventing", "_refresh_state", "_dmr_device"):
238 assert not hasattr(player, leaked)
239
240
241class TestHealthGating:
242 """Native grouping is only offered while the LinkPlay HTTP API is reachable."""
243
244 def test_set_members_feature_requires_health(
245 self, mock_provider: MagicMock, mock_client: MagicMock, mock_upnp_device: MagicMock
246 ) -> None:
247 """SET_MEMBERS is withdrawn when the LinkPlay API is unreachable."""
248 player = _make_shell(mock_provider, mock_client, mock_upnp_device)
249 player._linkplay_available = True
250 assert PlayerFeature.SET_MEMBERS in player.supported_features
251 player._linkplay_available = False
252 assert PlayerFeature.SET_MEMBERS not in player.supported_features
253
254 def test_prefers_native_grouping_when_healthy(
255 self, mock_provider: MagicMock, mock_client: MagicMock, mock_upnp_device: MagicMock
256 ) -> None:
257 """A reachable shell prefers native LinkPlay grouping over a linked protocol."""
258 player = _make_shell(mock_provider, mock_client, mock_upnp_device)
259 player._linkplay_available = True
260 assert player.prefer_native_grouping is True
261 player._linkplay_available = False
262 assert player.prefer_native_grouping is False
263
264 def test_can_group_with_delegates_to_coordinator(
265 self, mock_provider: MagicMock, mock_client: MagicMock, mock_upnp_device: MagicMock
266 ) -> None:
267 """can_group_with returns exactly the coordinator's candidate set for this player."""
268 player = _make_shell(mock_provider, mock_client, mock_upnp_device)
269 mock_provider.native_groups.can_group_with.return_value = {PEER_PLAYER_ID}
270 assert player.can_group_with == {PEER_PLAYER_ID}
271 mock_provider.native_groups.can_group_with.assert_called_once_with(player)
272
273 def test_native_available_reflects_health(
274 self, mock_provider: MagicMock, mock_client: MagicMock, mock_upnp_device: MagicMock
275 ) -> None:
276 """native_available (used by the coordinator) tracks the LinkPlay API reachability."""
277 healthy = _make_shell(mock_provider, mock_client, mock_upnp_device)
278 healthy._linkplay_available = True
279 assert healthy.native_available is True
280 unhealthy = _make_shell(mock_provider, mock_client, mock_upnp_device)
281 unhealthy._linkplay_available = False
282 assert unhealthy.native_available is False
283
284 def test_native_group_compat_follows_can_group_with(
285 self, mock_provider: MagicMock, mock_client: MagicMock, mock_upnp_device: MagicMock
286 ) -> None:
287 """A peer is native-compatible exactly when the coordinator offers it as a candidate."""
288 player = _make_shell(mock_provider, mock_client, mock_upnp_device)
289 peer = MagicMock(player_id=PEER_PLAYER_ID)
290 mock_provider.native_groups.can_group_with.return_value = {PEER_PLAYER_ID}
291 assert player.is_native_group_compatible(peer) is True
292 other = MagicMock(player_id="wiim_uuid:not-a-candidate")
293 assert player.is_native_group_compatible(other) is False
294
295
296class TestPlaybackAvailability:
297 """Playback availability derives from linked protocols, not the LinkPlay API (B2)."""
298
299 def test_available_from_linked_protocol_even_when_api_unhealthy(
300 self, mock_provider: MagicMock, mock_client: MagicMock, mock_upnp_device: MagicMock
301 ) -> None:
302 """A reachable linked DLNA keeps the shell available even if the LinkPlay API is down."""
303 player = _make_shell(mock_provider, mock_client, mock_upnp_device)
304 player._linkplay_available = False
305 player.set_linked_output_protocols([LinkedOutputProtocol("dlna_x", "dlna", priority=50)])
306 protocol_player = MagicMock(available_for_playback=True)
307 mock_provider.mass.players.get_player.return_value = protocol_player
308 assert player.available is True
309
310 def test_unavailable_without_linked_protocols(
311 self, mock_provider: MagicMock, mock_client: MagicMock, mock_upnp_device: MagicMock
312 ) -> None:
313 """With no linked protocol players there is nothing to play through."""
314 player = _make_shell(mock_provider, mock_client, mock_upnp_device)
315 assert player.available is False
316
317 def test_backing_ids_derive_from_linked_protocols(
318 self, mock_provider: MagicMock, mock_client: MagicMock, mock_upnp_device: MagicMock
319 ) -> None:
320 """The backing protocol ids come from the linked output protocols."""
321 player = _make_shell(mock_provider, mock_client, mock_upnp_device)
322 player.set_linked_output_protocols(
323 [
324 LinkedOutputProtocol("dlna_x", "dlna", priority=50),
325 LinkedOutputProtocol("ap_x", "airplay", priority=10),
326 ]
327 )
328 assert player._backing_protocol_player_ids() == ["dlna_x", "ap_x"]
329
330
331class TestGrouping:
332 """Grouping is delegated to the shared native coordinator across both backends."""
333
334 async def test_set_members_delegates_to_coordinator(
335 self, mock_provider: MagicMock, mock_client: MagicMock, mock_upnp_device: MagicMock
336 ) -> None:
337 """set_members forwards the add/remove batch to the coordinator unchanged."""
338 player = _make_shell(mock_provider, mock_client, mock_upnp_device)
339 await player.set_members(
340 player_ids_to_add=[PEER_PLAYER_ID], player_ids_to_remove=["wiim_uuid:gone"]
341 )
342 mock_provider.native_groups.set_members.assert_awaited_once_with(
343 player, [PEER_PLAYER_ID], ["wiim_uuid:gone"]
344 )
345
346 def test_api_unreachable_gates_only_native_grouping(
347 self, mock_provider: MagicMock, mock_client: MagicMock, mock_upnp_device: MagicMock
348 ) -> None:
349 """An unreachable LinkPlay API withdraws only native grouping, not the broad lock."""
350 player = _make_shell(mock_provider, mock_client, mock_upnp_device)
351 player._linkplay_available = False
352 # the broad lock stays off, so core may still group this device via a linked protocol
353 assert player.grouping_locked is False
354 # but native grouping is gated: the raw SET_MEMBERS is withdrawn ...
355 assert PlayerFeature.SET_MEMBERS not in player.supported_features
356 # ... and the coordinator offers no native peers for an unreachable device
357 assert player.can_group_with == set()
358
359 def test_unknown_leader_follower_locks_grouping(
360 self, mock_provider: MagicMock, mock_client: MagicMock, mock_upnp_device: MagicMock
361 ) -> None:
362 """A reachable shell that follows an undiscovered group withdraws ALL grouping."""
363 player = _make_shell(mock_provider, mock_client, mock_upnp_device)
364 player._linkplay_available = True
365 mock_provider.native_groups.is_unknown_leader_follower.return_value = True
366 # the broad lock holds, so even a linked-protocol group is withdrawn in the final state
367 assert player.grouping_locked is True
368
369 def test_api_outage_locks_a_native_group_member(
370 self, mock_provider: MagicMock, mock_client: MagicMock, mock_upnp_device: MagicMock
371 ) -> None:
372 """A device already in a native group whose API drops is broadly locked (can't leave it)."""
373 player = _make_shell(mock_provider, mock_client, mock_upnp_device)
374 player._linkplay_available = False
375 mock_provider.native_groups.role_of.return_value = NativeGroupRole.LEADER
376 # a protocol regroup would leave it in both groups, since the native ungroup can't run
377 assert player.grouping_locked is True
378
379 def test_healthy_native_group_member_not_broadly_locked(
380 self, mock_provider: MagicMock, mock_client: MagicMock, mock_upnp_device: MagicMock
381 ) -> None:
382 """A reachable native group member can still be regrouped (core natively ungroups first)."""
383 player = _make_shell(mock_provider, mock_client, mock_upnp_device)
384 player._linkplay_available = True
385 mock_provider.native_groups.role_of.return_value = NativeGroupRole.FOLLOWER
386 assert player.grouping_locked is False
387
388 def test_grouping_rebuild_lock_serializes_with_address_change(
389 self, mock_provider: MagicMock, mock_client: MagicMock, mock_upnp_device: MagicMock
390 ) -> None:
391 """The lock the coordinator holds during a command is the shell's address-rebuild lock."""
392 player = _make_shell(mock_provider, mock_client, mock_upnp_device)
393 assert player.grouping_rebuild_lock is player._rebuild_lock
394
395 def test_native_follower_suppresses_playback(
396 self, mock_provider: MagicMock, mock_client: MagicMock, mock_upnp_device: MagicMock
397 ) -> None:
398 """A native follower reports idle and no media instead of its delegated state."""
399 player = _make_shell(mock_provider, mock_client, mock_upnp_device)
400 mock_provider.native_groups.role_of.return_value = NativeGroupRole.FOLLOWER
401 assert player.playback_state == PlaybackState.IDLE
402 assert player.current_media is None
403 assert player.active_source is None
404
405
406class TestTopology:
407 """The coordinator publishes resolved membership onto the shell."""
408
409 def test_on_native_group_update_publishes_leader_members(
410 self, mock_provider: MagicMock, mock_client: MagicMock, mock_upnp_device: MagicMock
411 ) -> None:
412 """A leader publishes exactly the members the coordinator resolved for it."""
413 player = _make_shell(mock_provider, mock_client, mock_upnp_device)
414 mock_provider.native_groups.members_of.return_value = [EDIFIER_PLAYER_ID, PEER_PLAYER_ID]
415 player.on_native_group_update()
416 assert player._attr_group_members == [EDIFIER_PLAYER_ID, PEER_PLAYER_ID]
417
418 def test_on_native_group_update_clears_members_for_follower(
419 self, mock_provider: MagicMock, mock_client: MagicMock, mock_upnp_device: MagicMock
420 ) -> None:
421 """A standalone/follower shell publishes no members of its own."""
422 player = _make_shell(mock_provider, mock_client, mock_upnp_device)
423 player._attr_group_members = ["stale"]
424 mock_provider.native_groups.members_of.return_value = []
425 player.on_native_group_update()
426 assert player._attr_group_members == []
427
428 async def test_poll_pushes_topology_to_coordinator(
429 self, mock_provider: MagicMock, mock_client: MagicMock, mock_upnp_device: MagicMock
430 ) -> None:
431 """A poll refreshes reachability and forces a coordinator topology read."""
432 player = _make_shell(mock_provider, mock_client, mock_upnp_device)
433 await player.poll()
434 mock_client.get_device_info_model.assert_awaited()
435 mock_provider.native_groups.refresh_leader.assert_awaited_with(player, force=True)
436
437 def test_becoming_follower_clears_active_output_protocol(
438 self, mock_provider: MagicMock, mock_client: MagicMock, mock_upnp_device: MagicMock
439 ) -> None:
440 """A shell that was playing through DLNA drops that output when it becomes a follower."""
441 player = _make_shell(mock_provider, mock_client, mock_upnp_device)
442 player.set_active_output_protocol("dlna_x")
443 mock_provider.native_groups.role_of.return_value = NativeGroupRole.FOLLOWER
444
445 player.on_native_group_update()
446
447 assert player.active_output_protocol is None
448 assert player.playback_state == PlaybackState.IDLE
449 assert player.current_media is None
450
451 def test_leaving_follower_does_not_resurrect_protocol(
452 self, mock_provider: MagicMock, mock_client: MagicMock, mock_upnp_device: MagicMock
453 ) -> None:
454 """The dropped output stays cleared after leaving; normal playback reselects it."""
455 player = _make_shell(mock_provider, mock_client, mock_upnp_device)
456 player.set_active_output_protocol("dlna_x")
457 mock_provider.native_groups.role_of.return_value = NativeGroupRole.FOLLOWER
458 player.on_native_group_update()
459
460 mock_provider.native_groups.role_of.return_value = NativeGroupRole.STANDALONE
461 player.on_native_group_update()
462
463 assert player.active_output_protocol is None
464
465 def test_standalone_update_keeps_active_output_protocol(
466 self, mock_provider: MagicMock, mock_client: MagicMock, mock_upnp_device: MagicMock
467 ) -> None:
468 """A non-follower topology update never touches the active output (no churn)."""
469 player = _make_shell(mock_provider, mock_client, mock_upnp_device)
470 player.set_active_output_protocol("dlna_x")
471
472 player.on_native_group_update() # role stays standalone
473
474 assert player.active_output_protocol == "dlna_x"
475
476
477class TestGroupCompatibility:
478 """Only compatible, modern router-based generic LinkPlay devices may be grouped."""
479
480 def test_same_major_generation_compatible(self) -> None:
481 """4.2 and 4.3 (same WMRM major, router-based) can be grouped."""
482 first = _device_info("4.2")
483 second = _device_info("4.3")
484 assert linkplay_group_compatible(first, second) is True
485
486 def test_wifi_direct_rejected(self) -> None:
487 """A legacy Wi-Fi-Direct device is never grouped."""
488 first = _device_info("4.2")
489 second = _device_info("2.0", legacy=True)
490 assert linkplay_group_compatible(first, second) is False
491
492 def test_different_major_generation_rejected(self) -> None:
493 """Different WMRM major generations are not grouped."""
494 first = _device_info("4.2")
495 second = _device_info("3.0")
496 assert linkplay_group_compatible(first, second) is False
497
498 def test_unknown_device_info_rejected(self) -> None:
499 """An unknown (missing) device info is treated as incompatible."""
500 known = _device_info("4.2")
501 assert linkplay_group_compatible(None, known) is False
502
503 def test_unknown_generation_rejected(self) -> None:
504 """A device whose WMRM generation cannot be determined is not grouped."""
505 known = _device_info("4.2")
506 unknown = cast(
507 "PywiimDeviceInfo",
508 SimpleNamespace(wmrm_version=None, needs_wifi_direct_multiroom=False),
509 )
510 assert linkplay_group_compatible(known, unknown) is False
511
512
513class TestRefreshResilience:
514 """A poll refreshes reachability and pushes topology without blocking on a blip."""
515
516 async def test_reachable_refresh_marks_healthy(
517 self, mock_provider: MagicMock, mock_client: MagicMock, mock_upnp_device: MagicMock
518 ) -> None:
519 """A successful device-info read keeps the shell reachable and caches the info."""
520 player = _make_shell(mock_provider, mock_client, mock_upnp_device)
521 player._linkplay_available = False
522 await player.poll()
523 assert player._linkplay_available is True
524 assert player._cached_device_info is not None
525
526 async def test_unreachable_api_marks_unhealthy(
527 self, mock_provider: MagicMock, mock_client: MagicMock, mock_upnp_device: MagicMock
528 ) -> None:
529 """A failed device-info read marks the shell unhealthy, withdrawing NATIVE grouping."""
530 player = _make_shell(mock_provider, mock_client, mock_upnp_device)
531 player._linkplay_available = True
532 mock_client.get_device_info_model = AsyncMock(side_effect=WiiMError("down"))
533 await player.poll()
534 assert player._linkplay_available is False
535 # only native grouping is gated; the broad lock stays off so a linked protocol can group
536 assert PlayerFeature.SET_MEMBERS not in player.supported_features
537 assert player.grouping_locked is False
538
539 async def test_setup_without_primed_info_does_full_refresh(
540 self, mock_provider: MagicMock, mock_client: MagicMock, mock_upnp_device: MagicMock
541 ) -> None:
542 """A shell built without a primed device info probes the API during setup."""
543 player = LinkPlayPlayer(
544 provider=mock_provider,
545 player_id=EDIFIER_PLAYER_ID,
546 client=mock_client,
547 upnp_device=mock_upnp_device,
548 description_url="http://192.168.1.50:49152/description.xml",
549 )
550 player.update_state = MagicMock() # type: ignore[misc,method-assign]
551 mock_provider.players = [player]
552 assert player._linkplay_available is False
553 await player.setup()
554 mock_client.get_device_info_model.assert_awaited_once()
555 assert player._linkplay_available is True
556
557
558class TestAddressChange:
559 """A moved device rebuilds its low-level client without touching the MA player."""
560
561 async def test_successful_rebuild_swaps_client(
562 self, mock_provider: MagicMock, mock_client: MagicMock, mock_upnp_device: MagicMock
563 ) -> None:
564 """A reachable new address swaps in a fresh client and updates the IP identifier."""
565 player = _make_shell(mock_provider, mock_client, mock_upnp_device)
566 new_client = MagicMock(host="192.168.1.99")
567 new_client.get_device_info_model = AsyncMock(return_value=SimpleNamespace(uuid=""))
568 new_client.get_slaves_info = AsyncMock(return_value=_slaves([]))
569 mock_provider.players = [player]
570 with patch(
571 "music_assistant.providers.wiim.linkplay_player.WiiMClient", return_value=new_client
572 ):
573 await player.async_handle_address_change(
574 "192.168.1.99", mock_upnp_device, "http://192.168.1.99:49152/description.xml"
575 )
576 assert player._client is new_client
577 assert player.device_info.identifiers[IdentifierType.IP_ADDRESS] == "192.168.1.99"
578
579 async def test_failed_rebuild_keeps_old_client(
580 self, mock_provider: MagicMock, mock_client: MagicMock, mock_upnp_device: MagicMock
581 ) -> None:
582 """If the new address is unreachable, the existing client is preserved."""
583 player = _make_shell(mock_provider, mock_client, mock_upnp_device)
584 new_client = MagicMock(host="192.168.1.99")
585 new_client.get_device_info_model = AsyncMock(side_effect=WiiMError("unreachable"))
586 with patch(
587 "music_assistant.providers.wiim.linkplay_player.WiiMClient", return_value=new_client
588 ):
589 await player.async_handle_address_change(
590 "192.168.1.99", mock_upnp_device, "http://192.168.1.99:49152/description.xml"
591 )
592 assert player._client is mock_client
593
594
595class TestProviderRouting:
596 """Discovery classifies a device and routes it to the correct backend."""
597
598 async def test_generic_device_registers_shell(
599 self, mock_provider: MagicMock, mock_client: MagicMock, mock_upnp_device: MagicMock
600 ) -> None:
601 """A reachable generic LinkPlay device is registered as a shell."""
602 mock_provider.players = []
603 registered: list[Any] = []
604 mock_provider.mass.players.register_or_update = AsyncMock(
605 side_effect=lambda p: registered.append(p)
606 )
607 with patch("music_assistant.providers.wiim.provider.WiiMClient", return_value=mock_client):
608 await WiimProvider.try_add_linkplay_player(
609 mock_provider,
610 EDIFIER_PLAYER_ID,
611 "192.168.1.50",
612 mock_upnp_device,
613 "http://192.168.1.50:49152/description.xml",
614 "AA:BB:CC:DD:EE:FF",
615 )
616 assert registered
617 assert isinstance(registered[0], LinkPlayPlayer)
618 # discovery does a single authoritative device-info probe that primes the shell:
619 # setup must not repeat it, and the shell is registered already reachable.
620 mock_client.get_device_info_model.assert_awaited_once()
621 assert registered[0]._linkplay_available is True
622 # the live topology is read only after the player is registered (a read taken during
623 # setup, before registration, would be discarded by the coordinator).
624 mock_provider.native_groups.refresh_leader.assert_awaited_with(registered[0], force=True)
625
626 async def test_unreachable_device_not_registered(
627 self, mock_provider: MagicMock, mock_upnp_device: MagicMock
628 ) -> None:
629 """A device that does not answer the LinkPlay API is not registered."""
630 mock_provider.mass.players.register_or_update = AsyncMock()
631 bad_client = MagicMock()
632 bad_client.get_device_info_model = AsyncMock(side_effect=WiiMError("no api"))
633 with patch("music_assistant.providers.wiim.provider.WiiMClient", return_value=bad_client):
634 await WiimProvider.try_add_linkplay_player(
635 mock_provider,
636 EDIFIER_PLAYER_ID,
637 "192.168.1.50",
638 mock_upnp_device,
639 "http://192.168.1.50:49152/description.xml",
640 )
641 mock_provider.mass.players.register_or_update.assert_not_called()
642
643
644class TestDefaultProtocolSelection:
645 """The default_output_protocol_domain steers automatic output selection."""
646
647 def _controller(self, protocol_players: dict[str, Any], preferred: Any = None) -> MagicMock:
648 controller = MagicMock()
649 controller.get_player.side_effect = protocol_players.get
650 controller.mass.config.get_raw_player_config_value.return_value = preferred
651 controller._is_protocol_grouped.return_value = False
652 controller.logger = MagicMock()
653 return controller
654
655 def _shell_player(self, links: list[LinkedOutputProtocol]) -> MagicMock:
656 player = MagicMock()
657 player.default_output_protocol_domain = "dlna"
658 player.supported_features = set() # no native PLAY_MEDIA
659 player.linked_output_protocols = links
660 player.get_linked_protocol.side_effect = lambda pid: next(
661 (link for link in links if link.output_protocol_id == pid), None
662 )
663 return player
664
665 def test_prefers_default_domain_when_available(self) -> None:
666 """With DLNA available, the DLNA output is chosen over the higher-priority AirPlay."""
667 links = [
668 LinkedOutputProtocol("dlna_x", "dlna", priority=50),
669 LinkedOutputProtocol("ap_x", "airplay", priority=10),
670 ]
671 players = {
672 "dlna_x": MagicMock(available_for_playback=True),
673 "ap_x": MagicMock(available_for_playback=True),
674 }
675 controller = self._controller(players)
676 player = self._shell_player(links)
677 target, _ = ProtocolLinkingMixin._select_best_output_protocol(controller, player)
678 assert target is players["dlna_x"]
679
680 def test_falls_back_to_priority_when_default_absent(self) -> None:
681 """Without an available DLNA, selection falls back to priority (AirPlay)."""
682 links = [
683 LinkedOutputProtocol("dlna_x", "dlna", priority=50),
684 LinkedOutputProtocol("ap_x", "airplay", priority=10),
685 ]
686 players = {
687 "dlna_x": MagicMock(available_for_playback=False),
688 "ap_x": MagicMock(available_for_playback=True),
689 }
690 controller = self._controller(players)
691 player = self._shell_player(links)
692 target, _ = ProtocolLinkingMixin._select_best_output_protocol(controller, player)
693 assert target is players["ap_x"]
694
695 def test_explicit_user_preference_wins(self) -> None:
696 """An explicit stored user preference overrides the default domain."""
697 links = [
698 LinkedOutputProtocol("dlna_x", "dlna", priority=50),
699 LinkedOutputProtocol("ap_x", "airplay", priority=10),
700 ]
701 players = {
702 "dlna_x": MagicMock(available_for_playback=True),
703 "ap_x": MagicMock(available_for_playback=True),
704 }
705 controller = self._controller(players, preferred="ap_x")
706 player = self._shell_player(links)
707 target, _ = ProtocolLinkingMixin._select_best_output_protocol(controller, player)
708 assert target is players["ap_x"]
709
710 def test_explicit_auto_resolves_to_default_domain(self) -> None:
711 """Selecting Auto resolves to the default domain, just like an unset value."""
712 links = [
713 LinkedOutputProtocol("dlna_x", "dlna", priority=50),
714 LinkedOutputProtocol("ap_x", "airplay", priority=10),
715 ]
716 players = {
717 "dlna_x": MagicMock(available_for_playback=True),
718 "ap_x": MagicMock(available_for_playback=True),
719 }
720 controller = self._controller(players, preferred="auto")
721 player = self._shell_player(links)
722 target, _ = ProtocolLinkingMixin._select_best_output_protocol(controller, player)
723 assert target is players["dlna_x"]
724
725
726class TestFinalGroupingState:
727 """The FINAL state gates native grouping by API health but keeps linked-protocol grouping."""
728
729 @staticmethod
730 def _final_supported_features(player: LinkPlayPlayer) -> set[PlayerFeature]:
731 return cast(
732 "set[PlayerFeature]",
733 player._Player__final_supported_features, # type: ignore[attr-defined]
734 )
735
736 def _link_grouping_protocol(self, player: LinkPlayPlayer, mock_provider: MagicMock) -> None:
737 """Link a DLNA protocol player that itself supports SET_MEMBERS."""
738 player.set_linked_output_protocols([LinkedOutputProtocol("dlna_x", "dlna", priority=50)])
739 protocol_player = MagicMock()
740 protocol_player.available = True
741 protocol_player.available_for_playback = True
742 protocol_player.supported_features = {PlayerFeature.SET_MEMBERS, PlayerFeature.PAUSE}
743 mock_provider.mass.players.get_player.return_value = protocol_player
744
745 def test_protocol_grouping_survives_api_outage(
746 self, mock_provider: MagicMock, mock_client: MagicMock, mock_upnp_device: MagicMock
747 ) -> None:
748 """With the LinkPlay API down, a linked protocol still supplies SET_MEMBERS to the final state."""
749 player = _make_shell(mock_provider, mock_client, mock_upnp_device)
750 player._linkplay_available = False
751 self._link_grouping_protocol(player, mock_provider)
752
753 # native raw is gated, but the broad lock is off, so the linked protocol re-adds it
754 assert PlayerFeature.SET_MEMBERS not in player.supported_features
755 assert player.grouping_locked is False
756 assert PlayerFeature.SET_MEMBERS in self._final_supported_features(player)
757
758 def test_unknown_leader_follower_withdraws_even_protocol_grouping(
759 self, mock_provider: MagicMock, mock_client: MagicMock, mock_upnp_device: MagicMock
760 ) -> None:
761 """The broad lock withdraws grouping in the final state even a linked protocol would add."""
762 player = _make_shell(mock_provider, mock_client, mock_upnp_device)
763 player._linkplay_available = True
764 mock_provider.native_groups.is_unknown_leader_follower.return_value = True
765 self._link_grouping_protocol(player, mock_provider)
766
767 assert player.grouping_locked is True
768 assert PlayerFeature.SET_MEMBERS not in self._final_supported_features(player)
769
770 def test_healthy_native_keeps_set_members(
771 self, mock_provider: MagicMock, mock_client: MagicMock, mock_upnp_device: MagicMock
772 ) -> None:
773 """A reachable, standalone shell keeps native SET_MEMBERS and no broad lock."""
774 player = _make_shell(mock_provider, mock_client, mock_upnp_device)
775 player._linkplay_available = True
776 assert PlayerFeature.SET_MEMBERS in player.supported_features
777 assert player.grouping_locked is False
778
779
780class TestAvailabilityRepublish:
781 """A native-availability flip re-publishes peers so their candidate sets don't go stale."""
782
783 async def test_health_flip_republishes_peers(
784 self, mock_provider: MagicMock, mock_client: MagicMock, mock_upnp_device: MagicMock
785 ) -> None:
786 """Losing the LinkPlay API re-publishes every native peer."""
787 player = _make_shell(mock_provider, mock_client, mock_upnp_device)
788 player._linkplay_available = True
789 mock_client.get_device_info_model = AsyncMock(side_effect=WiiMError("down"))
790
791 await player._refresh_reachability()
792
793 mock_provider.native_groups.schedule_republish.assert_called()
794
795 async def test_no_republish_when_health_unchanged(
796 self, mock_provider: MagicMock, mock_client: MagicMock, mock_upnp_device: MagicMock
797 ) -> None:
798 """A refresh that does not change availability does not churn peers."""
799 player = _make_shell(mock_provider, mock_client, mock_upnp_device)
800 player._linkplay_available = True
801
802 await player._refresh_reachability() # stays reachable
803
804 mock_provider.native_groups.schedule_republish.assert_not_called()
805
806 async def test_address_change_recovery_republishes_peers(
807 self, mock_provider: MagicMock, mock_client: MagicMock, mock_upnp_device: MagicMock
808 ) -> None:
809 """Recovering an unavailable shell via an address change re-publishes its peers."""
810 player = _make_shell(mock_provider, mock_client, mock_upnp_device)
811 player._linkplay_available = False
812 new_client = MagicMock(host="192.168.1.99")
813 new_client.get_device_info_model = AsyncMock(return_value=SimpleNamespace(uuid=""))
814 new_client.get_slaves_info = AsyncMock(return_value=_slaves([]))
815 with patch(
816 "music_assistant.providers.wiim.linkplay_player.WiiMClient", return_value=new_client
817 ):
818 await player.async_handle_address_change(
819 "192.168.1.99", mock_upnp_device, "http://192.168.1.99:49152/description.xml"
820 )
821
822 mock_provider.native_groups.schedule_republish.assert_called()
823