music-assistant-server

1"""Tests for the native multiroom topology coordinator shared by both backends."""
2
3from __future__ import annotations
4
5import asyncio
6from unittest.mock import AsyncMock, MagicMock
7
8import pytest
9from music_assistant_models.errors import PlayerCommandFailed
10from pywiim import WiiMError
11from wiim.exceptions import WiimException
12
13from music_assistant.providers.wiim.constants import BACKEND_GENERIC, BACKEND_OFFICIAL
14from music_assistant.providers.wiim.grouping import NativeGroupCoordinator, NativeGroupRole
15
16# Two verified LinkPlay identities: the leader and a follower whose 24-char HTTP
17# slave uuid maps to the follower's UDN-derived player id.
18LEADER_ID = "wiim_uuid:11111111-2222-3333-4444-555555555555"
19LEADER_MASTER_UUID = "11111111-2222-3333-4444-555555555555"
20OTHER_MASTER_UUID = "99999999-8888-7777-6666-555555555555"
21FOLLOWER_HTTP_UUID = "A1B2C3D4E5F6A7B8C9D0E1F2"
22FOLLOWER_ID = "wiim_uuid:A1B2C3D4-E5F6-A7B8-C9D0-E1F2A1B2C3D4"
23
24
25def _ginfo(role: str, master: str | None = LEADER_MASTER_UUID) -> MagicMock:
26    """Build a fake DeviceGroupInfo for a follower/solo device (role + master uuid)."""
27    return MagicMock(role=role, master_uuid=master, slave_uuids=[])
28
29
30def _leader_info(slave_uuids: list[str]) -> MagicMock:
31    """Build a fake DeviceGroupInfo for a master device with the given slave uuids."""
32    return MagicMock(role="master", master_uuid=None, slave_uuids=slave_uuids)
33
34
35def _leader_client(slaves: list[str]) -> MagicMock:
36    """
37    Build a leader command client whose live group info reflects a mutable slave list.
38
39    Tests mutate ``slaves`` (directly or from a grouping command's side effect) so the
40    leader-side verification observes the membership the command produced on the device.
41    ``get_slaves_info`` derives a stable per-uuid ip and ``kick_slave`` removes the slave
42    at that ip, mirroring how a follower is removed from the leader.
43    """
44
45    def _ip_for(index: int) -> str:
46        return f"192.168.1.{50 + index}"
47
48    async def _info() -> MagicMock:
49        return MagicMock(
50            role="master" if slaves else "solo", slave_uuids=list(slaves), master_uuid=None
51        )
52
53    async def _slaves_info() -> list[dict[str, str]]:
54        return [{"uuid": uuid, "ip": _ip_for(i)} for i, uuid in enumerate(slaves)]
55
56    async def _kick(slave_ip: str) -> None:
57        for i, uuid in enumerate(list(slaves)):
58            if _ip_for(i) == slave_ip:
59                slaves.remove(uuid)
60                break
61
62    client = MagicMock()
63    client.get_device_group_info = AsyncMock(side_effect=_info)
64    client.get_slaves_info = AsyncMock(side_effect=_slaves_info)
65    client.get_device_info_model = AsyncMock(
66        return_value=MagicMock(needs_wifi_direct_multiroom=False, wmrm_version="4.2")
67    )
68    client.capabilities = {"probed": True}
69    client.create_group = AsyncMock()
70    client.kick_slave = AsyncMock(side_effect=_kick)
71    return client
72
73
74def _member_client() -> MagicMock:
75    """Build a follower command-client stub for a mixed-group join target."""
76    client = MagicMock()
77    client.capabilities = {"probed": True}
78    client.get_device_info_model = AsyncMock(
79        return_value=MagicMock(needs_wifi_direct_multiroom=False, wmrm_version="4.2")
80    )
81    # a join reads the member's own live group first (it must lead nothing); default solo
82    client.get_device_group_info = AsyncMock(return_value=MagicMock(role="solo"))
83    client.get_slaves_info = AsyncMock(return_value=[])
84    return client
85
86
87def _make_player(
88    player_id: str,
89    backend: str,
90    *,
91    available: bool = True,
92    ip: str | None = "192.168.1.10",
93    command_client: MagicMock | None = None,
94    live_role: str = "solo",
95    live_master: str | None = None,
96) -> MagicMock:
97    """Build a fake native player exposing exactly the coordinator's needed surface."""
98    player = MagicMock()
99    player.player_id = player_id
100    player.linkplay_backend = backend
101    player.native_available = available
102    player.native_ip = ip
103    player.native_device_udn = player_id.removeprefix("wiim_")
104    if command_client is None:
105        command_client = MagicMock()
106        command_client.capabilities = {"probed": True}
107        command_client.get_device_group_info = AsyncMock(
108            return_value=MagicMock(role=live_role, master_uuid=live_master, slave_uuids=[])
109        )
110        command_client.get_slaves_info = AsyncMock(return_value=[])
111        command_client.get_device_info_model = AsyncMock(
112            return_value=MagicMock(needs_wifi_direct_multiroom=False, wmrm_version="4.2")
113        )
114    player.make_command_client = MagicMock(return_value=command_client)
115    player.store_command_capabilities = MagicMock()
116    player.on_native_group_update = MagicMock()
117    # a modern, router-based generation so two generic peers are offerable to each other
118    player.cached_device_info = MagicMock(needs_wifi_direct_multiroom=False, wmrm_version="4.2")
119    return player
120
121
122@pytest.fixture(autouse=True)
123def _fast_membership_verify(monkeypatch: pytest.MonkeyPatch) -> None:
124    """Collapse the grouping verification retry to a single immediate check in tests."""
125    monkeypatch.setattr("music_assistant.providers.wiim.grouping.VERIFY_MAX_WAIT", 0.0)
126    monkeypatch.setattr("music_assistant.providers.wiim.grouping.VERIFY_POLL_INTERVAL", 0.0)
127
128
129def _make_coordinator(*players: MagicMock) -> tuple[NativeGroupCoordinator, MagicMock]:
130    """Create a coordinator over a provider whose player list is the given players."""
131    provider = MagicMock()
132    provider.players = list(players)
133    provider.mass = MagicMock()
134    provider.logger = MagicMock()
135    provider.wiim_controller = MagicMock(
136        async_join_group=AsyncMock(), async_ungroup_device=AsyncMock()
137    )
138    by_id = {player.player_id: player for player in players}
139    provider.mass.players.get_player.side_effect = by_id.get
140    return NativeGroupCoordinator(provider), provider
141
142
143class TestTopologyResolution:
144    """Reconcile resolves raw slave uuids against registered players, both directions."""
145
146    async def test_generic_leader_resolves_generic_follower(self) -> None:
147        """A generic leader's slave uuid resolves to its registered generic follower."""
148        leader = _make_player(LEADER_ID, BACKEND_GENERIC)
149        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC)
150        coordinator, _ = _make_coordinator(leader, follower)
151
152        coordinator.set_leader_slaves(LEADER_ID, [FOLLOWER_HTTP_UUID])
153        await coordinator.reconcile()
154
155        assert coordinator.role_of(LEADER_ID) == NativeGroupRole.LEADER
156        assert coordinator.role_of(FOLLOWER_ID) == NativeGroupRole.FOLLOWER
157        assert coordinator.members_of(LEADER_ID) == [LEADER_ID, FOLLOWER_ID]
158        assert coordinator.leader_of(FOLLOWER_ID) == LEADER_ID
159
160    async def test_official_leader_resolves_generic_follower(self) -> None:
161        """An official leader lists a generic follower (mixed group), leader first."""
162        leader = _make_player(LEADER_ID, BACKEND_OFFICIAL)
163        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC)
164        coordinator, _ = _make_coordinator(leader, follower)
165
166        coordinator.set_leader_slaves(LEADER_ID, [FOLLOWER_HTTP_UUID])
167        await coordinator.reconcile()
168
169        assert coordinator.members_of(LEADER_ID) == [LEADER_ID, FOLLOWER_ID]
170        assert coordinator.role_of(FOLLOWER_ID) == NativeGroupRole.FOLLOWER
171
172    async def test_unknown_slave_manages_no_members_but_leads(self) -> None:
173        """A slave that resolves to no registered player yields no managed member but a leader role."""
174        leader = _make_player(LEADER_ID, BACKEND_GENERIC)
175        coordinator, _ = _make_coordinator(leader)
176
177        coordinator.set_leader_slaves(LEADER_ID, ["FFFFFFFFFFFFFFFFFFFFFFFF"])
178        await coordinator.reconcile()
179
180        # it manages no MA-resolvable members, but it is NOT standalone: it leads a native
181        # group MA cannot see, so it must not be offered for a second (protocol) group.
182        assert coordinator.members_of(LEADER_ID) == []
183        assert coordinator.role_of(LEADER_ID) == NativeGroupRole.LEADER
184
185    async def test_follower_leadership_is_ignored(self) -> None:
186        """A device that is another leader's follower cannot also be a leader."""
187        leader = _make_player(LEADER_ID, BACKEND_GENERIC)
188        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC)
189        coordinator, _ = _make_coordinator(leader, follower)
190
191        # follower still has a stale self-leadership entry while also being a slave
192        coordinator.set_leader_slaves(LEADER_ID, [FOLLOWER_HTTP_UUID])
193        coordinator.set_leader_slaves(FOLLOWER_ID, ["FFFFFFFFFFFFFFFFFFFFFFFF"])
194        await coordinator.reconcile()
195
196        assert coordinator.role_of(FOLLOWER_ID) == NativeGroupRole.FOLLOWER
197        assert coordinator.members_of(FOLLOWER_ID) == []
198
199    async def test_moved_follower_belongs_to_one_group(self) -> None:
200        """
201        A moved follower shows in only one group.
202
203        While it still lingers in its old leader's not-yet-refreshed cached list,
204        the most recently refreshed leader wins so it never appears in two groups.
205        """
206        old_leader = _make_player(LEADER_ID, BACKEND_GENERIC)
207        new_leader = _make_player("wiim_uuid:99999999-8888-7777-6666-555555555555", BACKEND_GENERIC)
208        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC)
209        coordinator, _ = _make_coordinator(old_leader, new_leader, follower)
210
211        # old leader read the follower first; the new leader read it more recently
212        coordinator.set_leader_slaves(old_leader.player_id, [FOLLOWER_HTTP_UUID])
213        coordinator.set_leader_slaves(new_leader.player_id, [FOLLOWER_HTTP_UUID])
214        coordinator._refreshed_at[old_leader.player_id] = 100.0
215        coordinator._refreshed_at[new_leader.player_id] = 200.0
216        await coordinator.reconcile()
217
218        assert coordinator.leader_of(FOLLOWER_ID) == new_leader.player_id
219        assert coordinator.members_of(new_leader.player_id) == [new_leader.player_id, FOLLOWER_ID]
220        assert coordinator.members_of(old_leader.player_id) == []
221        assert coordinator.role_of(old_leader.player_id) == NativeGroupRole.STANDALONE
222
223
224class TestReconcileNotifications:
225    """Reconcile pushes state only to players whose role or membership changed."""
226
227    async def test_only_changed_players_notified(self) -> None:
228        """A newly formed group notifies the leader and follower exactly once."""
229        leader = _make_player(LEADER_ID, BACKEND_GENERIC)
230        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC)
231        coordinator, _ = _make_coordinator(leader, follower)
232
233        coordinator.set_leader_slaves(LEADER_ID, [FOLLOWER_HTTP_UUID])
234        await coordinator.reconcile()
235
236        leader.on_native_group_update.assert_called_once()
237        follower.on_native_group_update.assert_called_once()
238
239    async def test_moved_follower_is_notified_even_when_role_unchanged(self) -> None:
240        """A follower moving between leaders (still FOLLOWER) is still notified."""
241        other_leader_id = "wiim_uuid:22222222-3333-4444-5555-666666666666"
242        leader_a = _make_player(LEADER_ID, BACKEND_GENERIC)
243        leader_b = _make_player(other_leader_id, BACKEND_GENERIC)
244        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC)
245        coordinator, _ = _make_coordinator(leader_a, leader_b, follower)
246        coordinator.set_leader_slaves(LEADER_ID, [FOLLOWER_HTTP_UUID])
247        coordinator._refreshed_at[LEADER_ID] = 100.0
248        await coordinator.reconcile()
249        follower.on_native_group_update.reset_mock()
250
251        # move to leader B: role stays FOLLOWER and members stays [], only the leader changes
252        coordinator.set_leader_slaves(LEADER_ID, [])
253        coordinator.set_leader_slaves(other_leader_id, [FOLLOWER_HTTP_UUID])
254        coordinator._refreshed_at[other_leader_id] = 200.0
255        await coordinator.reconcile()
256
257        assert coordinator.role_of(FOLLOWER_ID) == NativeGroupRole.FOLLOWER
258        assert coordinator.leader_of(FOLLOWER_ID) == other_leader_id
259        follower.on_native_group_update.assert_called_once()
260
261    async def test_stale_reverse_mapping_removed_atomically(self) -> None:
262        """When a leader drops a follower, that follower's reverse mapping is cleared."""
263        leader = _make_player(LEADER_ID, BACKEND_GENERIC)
264        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC)
265        coordinator, _ = _make_coordinator(leader, follower)
266        coordinator.set_leader_slaves(LEADER_ID, [FOLLOWER_HTTP_UUID])
267        await coordinator.reconcile()
268
269        # the leader is now alone again
270        coordinator.set_leader_slaves(LEADER_ID, [])
271        await coordinator.reconcile()
272
273        assert coordinator.leader_of(FOLLOWER_ID) is None
274        assert coordinator.role_of(FOLLOWER_ID) == NativeGroupRole.STANDALONE
275        assert coordinator.role_of(LEADER_ID) == NativeGroupRole.STANDALONE
276
277    async def test_leaders_notified_before_followers(self) -> None:
278        """A follower reads its leader's cached members, so the leader publishes first."""
279        order: list[str] = []
280        leader = _make_player(LEADER_ID, BACKEND_GENERIC)
281        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC)
282        leader.on_native_group_update = MagicMock(side_effect=lambda: order.append("leader"))
283        follower.on_native_group_update = MagicMock(side_effect=lambda: order.append("follower"))
284        # register the follower first so the natural iteration order would notify it first
285        coordinator, _ = _make_coordinator(follower, leader)
286
287        coordinator.set_leader_slaves(LEADER_ID, [FOLLOWER_HTTP_UUID])
288        await coordinator.reconcile()
289
290        assert order == ["leader", "follower"]
291
292    async def test_dissolve_notifies_old_leader_before_follower(self) -> None:
293        """On a dissolve the old leader clears its members before the follower rescans."""
294        order: list[str] = []
295        leader = _make_player(LEADER_ID, BACKEND_GENERIC)
296        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC)
297        # register the follower first so the natural iteration order would notify it first
298        coordinator, _ = _make_coordinator(follower, leader)
299        coordinator.set_leader_slaves(LEADER_ID, [FOLLOWER_HTTP_UUID])
300        await coordinator.reconcile()
301        leader.on_native_group_update = MagicMock(side_effect=lambda: order.append("leader"))
302        follower.on_native_group_update = MagicMock(side_effect=lambda: order.append("follower"))
303
304        # dissolve: the leader drops the follower, both become standalone
305        coordinator.set_leader_slaves(LEADER_ID, [])
306        await coordinator.reconcile()
307
308        assert order.index("leader") < order.index("follower")
309
310    async def test_move_notifies_both_leaders_before_follower(self) -> None:
311        """Moving a follower publishes both the old and new leader before the follower."""
312        other_leader_id = "wiim_uuid:22222222-3333-4444-5555-666666666666"
313        order: list[str] = []
314        leader_a = _make_player(LEADER_ID, BACKEND_GENERIC)
315        leader_b = _make_player(other_leader_id, BACKEND_GENERIC)
316        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC)
317        # register the follower first so the natural order would notify it first
318        coordinator, _ = _make_coordinator(follower, leader_a, leader_b)
319        coordinator.set_leader_slaves(LEADER_ID, [FOLLOWER_HTTP_UUID])
320        coordinator._refreshed_at[LEADER_ID] = 100.0
321        await coordinator.reconcile()
322        leader_a.on_native_group_update = MagicMock(side_effect=lambda: order.append("a"))
323        leader_b.on_native_group_update = MagicMock(side_effect=lambda: order.append("b"))
324        follower.on_native_group_update = MagicMock(side_effect=lambda: order.append("follower"))
325
326        # move the follower from leader A to leader B
327        coordinator.set_leader_slaves(LEADER_ID, [])
328        coordinator.set_leader_slaves(other_leader_id, [FOLLOWER_HTTP_UUID])
329        coordinator._refreshed_at[other_leader_id] = 200.0
330        await coordinator.reconcile()
331
332        assert order.index("follower") == len(order) - 1  # follower is notified last
333        assert "a" in order  # both leaders published before it
334        assert "b" in order
335
336    async def test_unknown_follower_transition_notifies_all_players(self) -> None:
337        """An unknown-leader-follower change flips every peer's can_group_with, so all re-publish."""
338        me = _make_player(LEADER_ID, BACKEND_GENERIC)
339        peer = _make_player("wiim_uuid:peer", BACKEND_OFFICIAL)
340        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC)
341        coordinator, _ = _make_coordinator(me, peer, follower)
342
343        # the device becomes an unknown-leader follower: every peer loses it as a candidate
344        coordinator.set_self_role(FOLLOWER_ID, True)
345        await coordinator.reconcile()
346        me.on_native_group_update.assert_called_once()
347        peer.on_native_group_update.assert_called_once()
348
349        for player in (me, peer, follower):
350            player.on_native_group_update.reset_mock()
351
352        # it returns to standalone: every peer regains it as a candidate
353        coordinator.set_self_role(FOLLOWER_ID, False)
354        await coordinator.reconcile()
355        me.on_native_group_update.assert_called_once()
356        peer.on_native_group_update.assert_called_once()
357        follower.on_native_group_update.assert_called_once()
358
359
360class TestSelfHealing:
361    """Topology heals as players register and as the slow TTL elapses."""
362
363    async def test_discovery_order_miss_heals_on_registration(self) -> None:
364        """A follower discovered after its leader is placed on the next reconcile."""
365        leader = _make_player(LEADER_ID, BACKEND_GENERIC)
366        coordinator, provider = _make_coordinator(leader)
367        coordinator.set_leader_slaves(LEADER_ID, [FOLLOWER_HTTP_UUID])
368        await coordinator.reconcile()
369        assert coordinator.members_of(LEADER_ID) == []  # follower not registered yet
370
371        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC)
372        provider.players.append(follower)
373        await coordinator.reconcile()
374
375        assert coordinator.members_of(LEADER_ID) == [LEADER_ID, FOLLOWER_ID]
376
377    async def test_ttl_gates_official_refresh(self) -> None:
378        """An official leader re-reads its live group state only once per TTL unless forced."""
379        client = MagicMock()
380        client.get_device_group_info = AsyncMock(return_value=_ginfo("solo", None))
381        client.get_slaves_info = AsyncMock(return_value=[])
382        leader = _make_player(LEADER_ID, BACKEND_OFFICIAL, command_client=client)
383        coordinator, _ = _make_coordinator(leader)
384
385        await coordinator.refresh_leader(leader, force=True)
386        await coordinator.refresh_leader(leader)  # within TTL -> skipped
387        assert client.get_device_group_info.await_count == 1
388
389        await coordinator.refresh_leader(leader, force=True)  # forced -> reads again
390        assert client.get_device_group_info.await_count == 2
391
392    async def test_failed_refresh_keeps_previous_members(self) -> None:
393        """A temporarily unreachable leader keeps the members it last resolved."""
394        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC)
395        client = MagicMock()
396        client.get_device_group_info = AsyncMock(
397            side_effect=[_leader_info([FOLLOWER_HTTP_UUID]), WiiMError("offline")]
398        )
399        client.get_slaves_info = AsyncMock(
400            return_value=[{"uuid": FOLLOWER_HTTP_UUID, "ip": "192.168.1.50"}]
401        )
402        leader = _make_player(LEADER_ID, BACKEND_OFFICIAL, command_client=client)
403        coordinator, _ = _make_coordinator(leader, follower)
404
405        await coordinator.refresh_leader(leader, force=True)
406        assert coordinator.members_of(LEADER_ID) == [LEADER_ID, FOLLOWER_ID]
407
408        await coordinator.refresh_leader(leader, force=True)  # read fails
409        assert coordinator.members_of(LEADER_ID) == [LEADER_ID, FOLLOWER_ID]
410
411    async def test_solo_with_failing_slave_list_keeps_members(self) -> None:
412        """A master reported as 'solo' by a failed slave-list read keeps its members."""
413        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC)
414        client = MagicMock()
415        # get_device_group_info swallows a failed slave-list read and reports solo
416        client.get_device_group_info = AsyncMock(return_value=_ginfo("solo", None))
417        client.get_slaves_info = AsyncMock(side_effect=WiiMError("slave list offline"))
418        leader = _make_player(LEADER_ID, BACKEND_OFFICIAL, command_client=client)
419        coordinator, _ = _make_coordinator(leader, follower)
420        coordinator.set_leader_slaves(LEADER_ID, [FOLLOWER_HTTP_UUID])
421        await coordinator.reconcile()
422        assert coordinator.members_of(LEADER_ID) == [LEADER_ID, FOLLOWER_ID]
423
424        # the strict slave-list read fails, so the refresh is non-destructive
425        assert await coordinator.refresh_leader(leader, force=True) is False
426        assert coordinator.members_of(LEADER_ID) == [LEADER_ID, FOLLOWER_ID]
427
428    async def test_genuine_solo_clears_members(self) -> None:
429        """A leader whose strict slave-list read is truly empty drops its members."""
430        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC)
431        client = MagicMock()
432        client.get_device_group_info = AsyncMock(return_value=_ginfo("solo", None))
433        client.get_slaves_info = AsyncMock(return_value=[])
434        leader = _make_player(LEADER_ID, BACKEND_OFFICIAL, command_client=client)
435        coordinator, _ = _make_coordinator(leader, follower)
436        coordinator.set_leader_slaves(LEADER_ID, [FOLLOWER_HTTP_UUID])
437        await coordinator.reconcile()
438        assert coordinator.members_of(LEADER_ID) == [LEADER_ID, FOLLOWER_ID]
439
440        assert await coordinator.refresh_leader(leader, force=True) is True
441        assert coordinator.members_of(LEADER_ID) == []
442
443    async def test_concurrent_reconcile_is_serialized(self) -> None:
444        """Concurrent reconciles are serialized and leave a consistent snapshot."""
445        leader = _make_player(LEADER_ID, BACKEND_GENERIC)
446        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC)
447        coordinator, _ = _make_coordinator(leader, follower)
448        coordinator.set_leader_slaves(LEADER_ID, [FOLLOWER_HTTP_UUID])
449
450        await asyncio.gather(coordinator.reconcile(), coordinator.reconcile())
451
452        assert coordinator.members_of(LEADER_ID) == [LEADER_ID, FOLLOWER_ID]
453        assert coordinator.leader_of(FOLLOWER_ID) == LEADER_ID
454
455
456class TestCanGroupWith:
457    """can_group_with widens to available peers of either backend, minus self."""
458
459    def test_lists_available_peers_of_both_backends(self) -> None:
460        """An official player can group with available official and generic peers."""
461        me = _make_player(LEADER_ID, BACKEND_OFFICIAL)
462        generic_peer = _make_player(FOLLOWER_ID, BACKEND_GENERIC)
463        official_peer = _make_player("wiim_uuid:official-peer", BACKEND_OFFICIAL)
464        coordinator, _ = _make_coordinator(me, generic_peer, official_peer)
465
466        assert coordinator.can_group_with(me) == {FOLLOWER_ID, "wiim_uuid:official-peer"}
467
468    def test_excludes_unavailable_and_self(self) -> None:
469        """Unavailable peers and the player itself are excluded from the candidates."""
470        me = _make_player(LEADER_ID, BACKEND_OFFICIAL)
471        offline_peer = _make_player(FOLLOWER_ID, BACKEND_GENERIC, available=False)
472        coordinator, _ = _make_coordinator(me, offline_peer)
473
474        assert coordinator.can_group_with(me) == set()
475
476    async def test_unavailable_member_retained_in_topology(self) -> None:
477        """A registered but unavailable follower stays a member yet is not groupable."""
478        leader = _make_player(LEADER_ID, BACKEND_GENERIC)
479        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC, available=False)
480        coordinator, _ = _make_coordinator(leader, follower)
481
482        coordinator.set_leader_slaves(LEADER_ID, [FOLLOWER_HTTP_UUID])
483        await coordinator.reconcile()
484
485        assert coordinator.members_of(LEADER_ID) == [LEADER_ID, FOLLOWER_ID]
486        assert FOLLOWER_ID not in coordinator.can_group_with(leader)
487
488    async def test_known_follower_can_be_moved_to_another_leader(self) -> None:
489        """A follower with a known leader stays groupable (core auto-ungroups it)."""
490        me = _make_player("wiim_uuid:official-peer", BACKEND_OFFICIAL)
491        leader = _make_player(LEADER_ID, BACKEND_GENERIC)
492        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC)
493        coordinator, _ = _make_coordinator(me, leader, follower)
494        coordinator.set_leader_slaves(LEADER_ID, [FOLLOWER_HTTP_UUID])
495        await coordinator.reconcile()
496        assert coordinator.role_of(FOLLOWER_ID) == NativeGroupRole.FOLLOWER
497
498        # the follower has a known leader, so it may still be offered as a target
499        assert FOLLOWER_ID in coordinator.can_group_with(me)
500
501    async def test_unknown_leader_follower_is_excluded(self) -> None:
502        """A follower of a leader MA has not discovered cannot be cleanly moved."""
503        me = _make_player("wiim_uuid:official-peer", BACKEND_OFFICIAL)
504        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC)
505        coordinator, _ = _make_coordinator(me, follower)
506        coordinator.set_self_role(FOLLOWER_ID, True)  # follows an undiscovered leader
507        await coordinator.reconcile()
508
509        assert coordinator.can_group_with(me) == set()
510
511    async def test_unknown_leader_follower_offered_no_candidates(self) -> None:
512        """A device that itself follows an undiscovered leader gets no candidates."""
513        me = _make_player(LEADER_ID, BACKEND_GENERIC)
514        peer = _make_player("wiim_uuid:official-peer", BACKEND_OFFICIAL)
515        coordinator, _ = _make_coordinator(me, peer)
516        coordinator.set_self_role(LEADER_ID, True)  # follows an undiscovered leader
517        await coordinator.reconcile()
518
519        assert coordinator.can_group_with(me) == set()
520
521
522class TestSameBackendGrouping:
523    """Same-backend grouping keeps each backend's native path, verified leader-side."""
524
525    async def test_official_add_uses_sdk_controller(self) -> None:
526        """Two official players are joined via the WiiM SDK controller and verified."""
527        slaves: list[str] = []
528        leader = _make_player(LEADER_ID, BACKEND_OFFICIAL, command_client=_leader_client(slaves))
529        follower = _make_player(FOLLOWER_ID, BACKEND_OFFICIAL)
530        coordinator, provider = _make_coordinator(leader, follower)
531
532        async def _join(*_args: object, **_kwargs: object) -> None:
533            slaves.append(FOLLOWER_HTTP_UUID)  # the device joins the follower
534
535        provider.wiim_controller.async_join_group = AsyncMock(side_effect=_join)
536
537        await coordinator.set_members(leader, [FOLLOWER_ID], None)
538
539        provider.wiim_controller.async_join_group.assert_awaited_once_with(
540            leader.native_device_udn, [follower.native_device_udn]
541        )
542
543    async def test_official_remove_uses_sdk_controller(self) -> None:
544        """A managed official member is ungrouped via the SDK, no leader-side kick."""
545        slaves = [FOLLOWER_HTTP_UUID]
546        leader_client = _leader_client(slaves)
547        leader = _make_player(LEADER_ID, BACKEND_OFFICIAL, command_client=leader_client)
548        follower = _make_player(FOLLOWER_ID, BACKEND_OFFICIAL)
549        coordinator, provider = _make_coordinator(leader, follower)
550
551        async def _ungroup(*_args: object, **_kwargs: object) -> None:
552            slaves.clear()  # the SDK removes the follower
553
554        provider.wiim_controller.async_ungroup_device = AsyncMock(side_effect=_ungroup)
555
556        await coordinator.set_members(leader, None, [FOLLOWER_ID])
557
558        provider.wiim_controller.async_ungroup_device.assert_awaited_once_with(
559            follower.native_device_udn
560        )
561        leader_client.kick_slave.assert_not_called()  # SDK removal took effect
562
563    async def test_official_recovered_remove_issues_leader_side_kick(self) -> None:
564        """When the SDK no-ops on a recovered group, the leader kicks the follower."""
565        slaves = [FOLLOWER_HTTP_UUID]
566        leader_client = _leader_client(slaves)  # its kick_slave removes the follower
567        leader = _make_player(LEADER_ID, BACKEND_OFFICIAL, command_client=leader_client)
568        follower = _make_player(FOLLOWER_ID, BACKEND_OFFICIAL)
569        coordinator, provider = _make_coordinator(leader, follower)
570        provider.wiim_controller.async_ungroup_device = AsyncMock()  # SDK no-op
571
572        await coordinator.set_members(leader, None, [FOLLOWER_ID])
573
574        provider.wiim_controller.async_ungroup_device.assert_awaited_once()
575        leader_client.kick_slave.assert_awaited_once_with("192.168.1.50")
576
577    async def test_official_join_no_op_raises(self) -> None:
578        """An SDK join the device ignored (leader never lists it) surfaces a typed failure."""
579        slaves: list[str] = []
580        leader = _make_player(LEADER_ID, BACKEND_OFFICIAL, command_client=_leader_client(slaves))
581        follower = _make_player(FOLLOWER_ID, BACKEND_OFFICIAL)
582        coordinator, provider = _make_coordinator(leader, follower)
583        provider.wiim_controller.async_join_group = AsyncMock()  # device ignores the join
584
585        with pytest.raises(PlayerCommandFailed):
586            await coordinator.set_members(leader, [FOLLOWER_ID], None)
587
588    async def test_official_leave_no_op_raises(self) -> None:
589        """When neither the SDK nor the leader-side kick takes effect, a failure is raised."""
590        slaves = [FOLLOWER_HTTP_UUID]
591        leader_client = _leader_client(slaves)
592        leader_client.kick_slave = AsyncMock()  # the kick also no-ops (does not remove)
593        leader = _make_player(LEADER_ID, BACKEND_OFFICIAL, command_client=leader_client)
594        follower = _make_player(FOLLOWER_ID, BACKEND_OFFICIAL)
595        coordinator, provider = _make_coordinator(leader, follower)
596        provider.wiim_controller.async_ungroup_device = AsyncMock()  # SDK no-op
597
598        with pytest.raises(PlayerCommandFailed):
599            await coordinator.set_members(leader, None, [FOLLOWER_ID])
600        leader_client.kick_slave.assert_awaited_once()  # the leader-side kick was attempted
601
602    async def test_official_join_failure_raises_typed(self) -> None:
603        """An SDK error (WiimException, not pywiim's) is mapped to PlayerCommandFailed."""
604        leader = _make_player(LEADER_ID, BACKEND_OFFICIAL, command_client=_leader_client([]))
605        follower = _make_player(FOLLOWER_ID, BACKEND_OFFICIAL)
606        coordinator, provider = _make_coordinator(leader, follower)
607        provider.wiim_controller.async_join_group = AsyncMock(
608            side_effect=WiimException("HTTP API not available")
609        )
610
611        with pytest.raises(PlayerCommandFailed) as exc_info:
612            await coordinator.set_members(leader, [FOLLOWER_ID], None)
613        assert isinstance(exc_info.value.__cause__, WiimException)
614
615    async def test_official_remove_failure_raises_typed(self) -> None:
616        """An SDK ungroup error is mapped to PlayerCommandFailed with chaining."""
617        leader = _make_player(
618            LEADER_ID, BACKEND_OFFICIAL, command_client=_leader_client([FOLLOWER_HTTP_UUID])
619        )
620        follower = _make_player(FOLLOWER_ID, BACKEND_OFFICIAL)
621        coordinator, provider = _make_coordinator(leader, follower)
622        provider.wiim_controller.async_ungroup_device = AsyncMock(
623            side_effect=WiimException("device gone")
624        )
625
626        with pytest.raises(PlayerCommandFailed) as exc_info:
627            await coordinator.set_members(leader, None, [FOLLOWER_ID])
628        assert isinstance(exc_info.value.__cause__, WiimException)
629
630    async def test_generic_add_uses_low_level_join_and_verifies(self) -> None:
631        """Two generic players use the low-level join_slave and confirm leader-side."""
632        slaves: list[str] = []
633        member_client = _member_client()
634
635        async def _join(*_args: object, **_kwargs: object) -> None:
636            slaves.append(FOLLOWER_HTTP_UUID)
637
638        member_client.join_slave = AsyncMock(side_effect=_join)
639        leader = _make_player(
640            LEADER_ID, BACKEND_GENERIC, ip="192.168.1.30", command_client=_leader_client(slaves)
641        )
642        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC, command_client=member_client)
643        coordinator, _ = _make_coordinator(leader, follower)
644
645        await coordinator.set_members(leader, [FOLLOWER_ID], None)
646
647        member_client.join_slave.assert_awaited_once()
648        assert member_client.join_slave.await_args.args[0] == "192.168.1.30"
649
650    async def test_generic_join_no_op_raises(self) -> None:
651        """A low-level join the device ignores raises PlayerCommandFailed."""
652        member_client = _member_client()
653        member_client.join_slave = AsyncMock()  # device ignores it
654        leader = _make_player(LEADER_ID, BACKEND_GENERIC, command_client=_leader_client([]))
655        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC, command_client=member_client)
656        coordinator, _ = _make_coordinator(leader, follower)
657
658        with pytest.raises(PlayerCommandFailed):
659            await coordinator.set_members(leader, [FOLLOWER_ID], None)
660
661    async def test_generic_remove_uses_low_level_leave(self) -> None:
662        """A generic member leaves via the low-level client and is verified."""
663        slaves = [FOLLOWER_HTTP_UUID]
664        member_client = _member_client()
665
666        async def _leave() -> None:
667            slaves.clear()
668
669        member_client.leave_group = AsyncMock(side_effect=_leave)
670        leader = _make_player(LEADER_ID, BACKEND_GENERIC, command_client=_leader_client(slaves))
671        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC, command_client=member_client)
672        coordinator, _ = _make_coordinator(leader, follower)
673
674        await coordinator.set_members(leader, None, [FOLLOWER_ID])
675
676        member_client.leave_group.assert_awaited_once()
677
678    async def test_generic_incompatible_group_raises(self) -> None:
679        """An incompatible same-backend group surfaces a typed failure, not a no-op."""
680        member_client = _member_client()
681        member_client.join_slave = AsyncMock()
682        # the member is a legacy Wi-Fi Direct device: the compatibility gate refuses it
683        member_client.get_device_info_model = AsyncMock(
684            return_value=MagicMock(needs_wifi_direct_multiroom=True, wmrm_version="4.2")
685        )
686        leader = _make_player(LEADER_ID, BACKEND_GENERIC, command_client=_leader_client([]))
687        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC, command_client=member_client)
688        coordinator, _ = _make_coordinator(leader, follower)
689
690        with pytest.raises(PlayerCommandFailed):
691            await coordinator.set_members(leader, [FOLLOWER_ID], None)
692        member_client.join_slave.assert_not_called()
693
694    async def test_join_clears_joined_members_cached_slaves(self) -> None:
695        """A member that was a leader has its cached slave list cleared once it joins."""
696        slaves: list[str] = []
697        leader = _make_player(LEADER_ID, BACKEND_OFFICIAL, command_client=_leader_client(slaves))
698        member = _make_player(FOLLOWER_ID, BACKEND_OFFICIAL)
699        coordinator, provider = _make_coordinator(leader, member)
700        # the member had its own follower cached from when it was itself a leader
701        coordinator.set_leader_slaves(FOLLOWER_ID, ["A1B2C3D4E5F6A7B8C9D0E1F2"])
702
703        async def _join(*_args: object, **_kwargs: object) -> None:
704            slaves.append(FOLLOWER_HTTP_UUID)
705
706        provider.wiim_controller.async_join_group = AsyncMock(side_effect=_join)
707
708        await coordinator.set_members(leader, [FOLLOWER_ID], None)
709
710        # the stale list is cleared so a later leave cannot resurrect a ghost group
711        assert coordinator._raw_slaves.get(FOLLOWER_ID) == []
712
713
714class TestExpectedLeaderVerification:
715    """Grouping is verified against the leader the command targeted."""
716
717    async def test_leave_clears_self_follower_immediately(self) -> None:
718        """A confirmed leave clears the member's stale self-follower flag at once."""
719        slaves = [FOLLOWER_HTTP_UUID]
720        member_client = _member_client()
721
722        async def _leave() -> None:
723            slaves.clear()
724
725        member_client.leave_group = AsyncMock(side_effect=_leave)
726        leader = _make_player(LEADER_ID, BACKEND_GENERIC, command_client=_leader_client(slaves))
727        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC, command_client=member_client)
728        coordinator, _ = _make_coordinator(leader, follower)
729        # the follower reported itself a slave on its last live read
730        coordinator.set_self_role(FOLLOWER_ID, True)
731        await coordinator.reconcile()
732        assert coordinator.role_of(FOLLOWER_ID) == NativeGroupRole.FOLLOWER
733
734        await coordinator.set_members(leader, None, [FOLLOWER_ID])
735
736        assert coordinator.role_of(FOLLOWER_ID) == NativeGroupRole.STANDALONE
737
738    async def test_leave_keeps_follower_when_new_reverse_leader_wins(self) -> None:
739        """Clearing the self-flag on leave still yields FOLLOWER under a new known leader."""
740        other_leader_id = "wiim_uuid:22222222-3333-4444-5555-666666666666"
741        slaves_a = [FOLLOWER_HTTP_UUID]
742        member_client = _member_client()
743
744        async def _leave() -> None:
745            slaves_a.clear()
746
747        member_client.leave_group = AsyncMock(side_effect=_leave)
748        leader_a = _make_player(LEADER_ID, BACKEND_GENERIC, command_client=_leader_client(slaves_a))
749        leader_b = _make_player(other_leader_id, BACKEND_GENERIC)
750        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC, command_client=member_client)
751        coordinator, _ = _make_coordinator(leader_a, leader_b, follower)
752        coordinator.set_self_role(FOLLOWER_ID, True)
753        # concurrently, leader B has discovered the follower as its slave
754        coordinator.set_leader_slaves(other_leader_id, [FOLLOWER_HTTP_UUID])
755        await coordinator.reconcile()
756
757        await coordinator.set_members(leader_a, None, [FOLLOWER_ID])
758
759        assert coordinator.role_of(FOLLOWER_ID) == NativeGroupRole.FOLLOWER
760        assert coordinator.leader_of(FOLLOWER_ID) == other_leader_id
761
762    async def test_join_landing_on_other_leader_raises(self) -> None:
763        """A join that never appears in this leader's slave list is a no-op failure."""
764        # the device "joined" but not this leader, so the leader never lists it
765        leader = _make_player(LEADER_ID, BACKEND_OFFICIAL, command_client=_leader_client([]))
766        follower = _make_player(FOLLOWER_ID, BACKEND_OFFICIAL)
767        coordinator, provider = _make_coordinator(leader, follower)
768        provider.wiim_controller.async_join_group = AsyncMock()
769
770        with pytest.raises(PlayerCommandFailed):
771            await coordinator.set_members(leader, [FOLLOWER_ID], None)
772
773    async def test_remove_of_member_not_under_leader_is_idempotent(self) -> None:
774        """A member the leader does not list is not detached (solo or another leader)."""
775        leader = _make_player(LEADER_ID, BACKEND_OFFICIAL, command_client=_leader_client([]))
776        member_client = _member_client()
777        member_client.leave_group = AsyncMock()
778        follower = _make_player(FOLLOWER_ID, BACKEND_OFFICIAL, command_client=member_client)
779        coordinator, provider = _make_coordinator(leader, follower)
780
781        await coordinator.set_members(leader, None, [FOLLOWER_ID])
782
783        provider.wiim_controller.async_ungroup_device.assert_not_called()
784        member_client.leave_group.assert_not_called()
785
786    async def test_join_fails_closed_when_leader_unreadable(self) -> None:
787        """If the leader's live list cannot be read, the join is not confirmed."""
788        leader_client = MagicMock()
789        leader_client.get_device_group_info = AsyncMock(side_effect=WiiMError("offline"))
790        leader_client.get_device_info_model = AsyncMock(return_value=MagicMock())
791        leader = _make_player(LEADER_ID, BACKEND_OFFICIAL, command_client=leader_client)
792        follower = _make_player(FOLLOWER_ID, BACKEND_OFFICIAL)
793        coordinator, provider = _make_coordinator(leader, follower)
794        provider.wiim_controller.async_join_group = AsyncMock()
795
796        with pytest.raises(PlayerCommandFailed):
797            await coordinator.set_members(leader, [FOLLOWER_ID], None)
798
799
800class TestMixedGrouping:
801    """Cross-backend grouping commands the follower, verified from the leader's list."""
802
803    async def test_mixed_add_joins_follower_to_leader_ip(self) -> None:
804        """A cross-backend add sends a low-level join to the follower's leader IP."""
805        slaves: list[str] = []
806        member_client = _member_client()
807
808        async def _join_slave(*_args: object, **_kwargs: object) -> None:
809            slaves.append(FOLLOWER_HTTP_UUID)
810
811        member_client.join_slave = AsyncMock(side_effect=_join_slave)
812        leader = _make_player(
813            LEADER_ID, BACKEND_OFFICIAL, ip="192.168.1.20", command_client=_leader_client(slaves)
814        )
815        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC, command_client=member_client)
816        coordinator, _ = _make_coordinator(leader, follower)
817
818        await coordinator.set_members(leader, [FOLLOWER_ID], None)
819
820        member_client.join_slave.assert_awaited_once()
821        assert member_client.join_slave.await_args.args[0] == leader.native_ip
822        member_client.close.assert_not_called()
823
824    async def test_mixed_add_of_unknown_leader_follower_raises(self) -> None:
825        """Joining a member that follows an undiscovered group fails typed at command time."""
826        member_client = _member_client()
827        member_client.join_slave = AsyncMock()
828        # the member's own live read reports it is a slave of a leader MA cannot see
829        member_client.get_device_group_info = AsyncMock(
830            return_value=MagicMock(role="slave", master_uuid=OTHER_MASTER_UUID)
831        )
832        leader = _make_player(
833            LEADER_ID, BACKEND_OFFICIAL, ip="192.168.1.20", command_client=_leader_client([])
834        )
835        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC, command_client=member_client)
836        coordinator, _ = _make_coordinator(leader, follower)
837        coordinator.set_self_role(FOLLOWER_ID, True)  # follows an undiscovered leader
838        await coordinator.reconcile()
839
840        with pytest.raises(PlayerCommandFailed):
841            await coordinator.set_members(leader, [FOLLOWER_ID], None)
842        member_client.join_slave.assert_not_called()
843
844    async def test_mixed_add_dissolves_members_existing_group_first(self) -> None:
845        """A member that leads its own group is dissolved before joining as a follower."""
846        sub_id = "wiim_uuid:B1B2C3D4-E5F6-A7B8-C9D0-E1F2B1B2C3D4"
847        sub_http_uuid = "B1B2C3D4E5F6A7B8C9D0E1F2"
848        # the member currently leads a generic sub-follower; leaving drops it from the
849        # member's own live slave list, which the dissolve verification then reads.
850        member_slaves = [sub_http_uuid]
851
852        async def _sub_leave() -> None:
853            member_slaves.remove(sub_http_uuid)
854
855        sub_client = _member_client()
856        sub_client.leave_group = AsyncMock(side_effect=_sub_leave)
857        sub_follower = _make_player(sub_id, BACKEND_GENERIC, command_client=sub_client)
858        member_client = _leader_client(member_slaves)
859        member = _make_player(
860            FOLLOWER_ID, BACKEND_GENERIC, ip="192.168.1.40", command_client=member_client
861        )
862        new_slaves: list[str] = []
863        member_client.join_slave = AsyncMock(
864            side_effect=lambda *_a, **_k: new_slaves.append(FOLLOWER_HTTP_UUID)
865        )
866        leader = _make_player(
867            LEADER_ID,
868            BACKEND_OFFICIAL,
869            ip="192.168.1.20",
870            command_client=_leader_client(new_slaves),
871        )
872        coordinator, _ = _make_coordinator(leader, member, sub_follower)
873        coordinator.set_leader_slaves(FOLLOWER_ID, [sub_http_uuid])
874        await coordinator.reconcile()
875        assert coordinator.members_of(FOLLOWER_ID) == [FOLLOWER_ID, sub_id]
876
877        await coordinator.set_members(leader, [FOLLOWER_ID], None)
878
879        sub_client.leave_group.assert_awaited_once()  # the member's group was dissolved
880        member_client.join_slave.assert_awaited_once()  # before the member itself joined
881        assert member_client.join_slave.await_args.args[0] == leader.native_ip
882
883    async def test_mixed_remove_kicks_follower_from_leader(self) -> None:
884        """A cross-backend/recovered remove kicks the follower from the leader."""
885        slaves = [FOLLOWER_HTTP_UUID]
886        leader_client = _leader_client(slaves)  # its kick_slave removes the follower
887        member_client = _member_client()
888        leader = _make_player(
889            LEADER_ID, BACKEND_OFFICIAL, ip="192.168.1.20", command_client=leader_client
890        )
891        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC, command_client=member_client)
892        coordinator, _ = _make_coordinator(leader, follower)
893
894        await coordinator.set_members(leader, None, [FOLLOWER_ID])
895
896        leader_client.kick_slave.assert_awaited_once_with("192.168.1.50")
897        # the fresh command client is primed as master so kick_slave's guard passes
898        leader_client.create_group.assert_awaited_once()
899        member_client.leave_group.assert_not_called()  # the follower is not commanded directly
900
901    async def test_mixed_add_generic_leader_official_follower(self) -> None:
902        """The follower direction also works: generic leader, official follower."""
903        slaves: list[str] = []
904        member_client = _member_client()
905
906        async def _join_slave(*_args: object, **_kwargs: object) -> None:
907            slaves.append(FOLLOWER_HTTP_UUID)
908
909        member_client.join_slave = AsyncMock(side_effect=_join_slave)
910        leader = _make_player(
911            LEADER_ID, BACKEND_GENERIC, ip="192.168.1.30", command_client=_leader_client(slaves)
912        )
913        follower = _make_player(FOLLOWER_ID, BACKEND_OFFICIAL, command_client=member_client)
914        coordinator, _ = _make_coordinator(leader, follower)
915
916        await coordinator.set_members(leader, [FOLLOWER_ID], None)
917
918        assert member_client.join_slave.await_args.args[0] == "192.168.1.30"
919        member_client.close.assert_not_called()
920
921    async def test_mixed_join_no_op_raises(self) -> None:
922        """A join the leader never lists surfaces a typed failure."""
923        member_client = _member_client()
924        member_client.join_slave = AsyncMock()  # device ignores it
925        leader = _make_player(
926            LEADER_ID, BACKEND_OFFICIAL, ip="192.168.1.20", command_client=_leader_client([])
927        )
928        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC, command_client=member_client)
929        coordinator, _ = _make_coordinator(leader, follower)
930
931        with pytest.raises(PlayerCommandFailed):
932            await coordinator.set_members(leader, [FOLLOWER_ID], None)
933
934    async def test_mixed_remove_no_op_raises(self) -> None:
935        """A follower the leader still lists after a leave surfaces a typed failure."""
936        leader_client = _leader_client([FOLLOWER_HTTP_UUID])
937        leader_client.kick_slave = AsyncMock()  # device ignores the kick (does not remove)
938        member_client = _member_client()
939        leader = _make_player(
940            LEADER_ID, BACKEND_OFFICIAL, ip="192.168.1.20", command_client=leader_client
941        )
942        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC, command_client=member_client)
943        coordinator, _ = _make_coordinator(leader, follower)
944
945        with pytest.raises(PlayerCommandFailed):
946            await coordinator.set_members(leader, None, [FOLLOWER_ID])
947        leader_client.kick_slave.assert_awaited_once()
948
949    async def test_missing_leader_ip_raises(self) -> None:
950        """A mixed add with an unknown leader address fails typed instead of guessing."""
951        member_client = _member_client()
952        member_client.join_slave = AsyncMock()
953        leader = _make_player(
954            LEADER_ID, BACKEND_OFFICIAL, ip=None, command_client=_leader_client([])
955        )
956        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC, command_client=member_client)
957        coordinator, _ = _make_coordinator(leader, follower)
958
959        with pytest.raises(PlayerCommandFailed):
960            await coordinator.set_members(leader, [FOLLOWER_ID], None)
961        member_client.join_slave.assert_not_called()
962
963    async def test_mixed_add_rejects_wifi_direct_leader(self) -> None:
964        """A mixed join to a legacy Wi-Fi Direct leader fails typed instead of stranding the follower."""
965        member_client = _member_client()
966        member_client.join_slave = AsyncMock()
967        leader_client = _leader_client([])
968        leader_client.get_device_info_model = AsyncMock(
969            return_value=MagicMock(needs_wifi_direct_multiroom=True)
970        )
971        leader = _make_player(
972            LEADER_ID, BACKEND_OFFICIAL, ip="192.168.1.20", command_client=leader_client
973        )
974        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC, command_client=member_client)
975        coordinator, _ = _make_coordinator(leader, follower)
976
977        with pytest.raises(PlayerCommandFailed):
978            await coordinator.set_members(leader, [FOLLOWER_ID], None)
979        member_client.join_slave.assert_not_called()
980
981    async def test_mixed_add_rejects_wifi_direct_member(self) -> None:
982        """A mixed join of a legacy Wi-Fi Direct follower is refused too, not only the leader."""
983        member_client = _member_client()
984        member_client.join_slave = AsyncMock()
985        # the follower itself is the legacy device; pywiim would pick Wi-Fi Direct for it
986        member_client.get_device_info_model = AsyncMock(
987            return_value=MagicMock(needs_wifi_direct_multiroom=True)
988        )
989        leader = _make_player(
990            LEADER_ID, BACKEND_OFFICIAL, ip="192.168.1.20", command_client=_leader_client([])
991        )
992        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC, command_client=member_client)
993        coordinator, _ = _make_coordinator(leader, follower)
994
995        with pytest.raises(PlayerCommandFailed):
996            await coordinator.set_members(leader, [FOLLOWER_ID], None)
997        member_client.join_slave.assert_not_called()
998
999    async def test_join_records_member_self_follower(self) -> None:
1000        """A confirmed join records the member's follower role so an unregistered leader can't strand it."""
1001        slaves: list[str] = []
1002        member_client = _member_client()
1003        member_client.join_slave = AsyncMock(
1004            side_effect=lambda *_a, **_k: slaves.append(FOLLOWER_HTTP_UUID)
1005        )
1006        leader = _make_player(
1007            LEADER_ID, BACKEND_OFFICIAL, ip="192.168.1.20", command_client=_leader_client(slaves)
1008        )
1009        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC, command_client=member_client)
1010        coordinator, provider = _make_coordinator(leader, follower)
1011
1012        await coordinator.set_members(leader, [FOLLOWER_ID], None)
1013        assert coordinator.leader_of(FOLLOWER_ID) == LEADER_ID
1014
1015        # the leader is unregistered before the member's next live read; the recorded
1016        # self-follower role keeps the member suppressed rather than groupable
1017        provider.players.remove(leader)
1018        coordinator.unregister(LEADER_ID)
1019        await coordinator.reconcile()
1020
1021        assert coordinator.role_of(FOLLOWER_ID) == NativeGroupRole.FOLLOWER
1022        assert coordinator.leader_of(FOLLOWER_ID) is None
1023        assert coordinator.can_group_with(follower) == set()
1024
1025    async def test_remove_fails_typed_when_slave_list_unreadable(self) -> None:
1026        """A remove whose leader slave-list read fails raises instead of silently no-op'ing."""
1027        client = MagicMock()
1028        # the leader reads as 'solo' only because its slave-list endpoint is failing
1029        client.get_device_group_info = AsyncMock(return_value=_ginfo("solo", None))
1030        client.get_slaves_info = AsyncMock(side_effect=WiiMError("slave list offline"))
1031        client.get_device_info_model = AsyncMock(
1032            return_value=MagicMock(needs_wifi_direct_multiroom=False, wmrm_version="4.2")
1033        )
1034        leader = _make_player(LEADER_ID, BACKEND_OFFICIAL, ip="192.168.1.20", command_client=client)
1035        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC, command_client=_member_client())
1036        coordinator, _ = _make_coordinator(leader, follower)
1037
1038        with pytest.raises(PlayerCommandFailed):
1039            await coordinator.set_members(leader, None, [FOLLOWER_ID])
1040
1041    async def test_api_error_is_chained(self) -> None:
1042        """A device API error during a mixed join is chained into the typed failure."""
1043        member_client = _member_client()
1044        member_client.join_slave = AsyncMock(side_effect=WiiMError("boom"))
1045        leader = _make_player(
1046            LEADER_ID, BACKEND_OFFICIAL, ip="192.168.1.20", command_client=_leader_client([])
1047        )
1048        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC, command_client=member_client)
1049        coordinator, _ = _make_coordinator(leader, follower)
1050
1051        with pytest.raises(PlayerCommandFailed) as exc_info:
1052            await coordinator.set_members(leader, [FOLLOWER_ID], None)
1053        assert isinstance(exc_info.value.__cause__, WiiMError)
1054
1055    async def test_mixed_join_passes_leader_device_info(self) -> None:
1056        """The leader's device info is passed so pywiim can pick the join mode."""
1057        slaves: list[str] = []
1058        leader_info = MagicMock(needs_wifi_direct_multiroom=False, wmrm_version="4.2")
1059        leader_client = _leader_client(slaves)
1060        leader_client.get_device_info_model = AsyncMock(return_value=leader_info)
1061        member_client = _member_client()
1062
1063        async def _join_slave(*_args: object, **_kwargs: object) -> None:
1064            slaves.append(FOLLOWER_HTTP_UUID)
1065
1066        member_client.join_slave = AsyncMock(side_effect=_join_slave)
1067        leader = _make_player(
1068            LEADER_ID, BACKEND_OFFICIAL, ip="192.168.1.20", command_client=leader_client
1069        )
1070        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC, command_client=member_client)
1071        coordinator, _ = _make_coordinator(leader, follower)
1072
1073        await coordinator.set_members(leader, [FOLLOWER_ID], None)
1074
1075        assert member_client.join_slave.await_args.kwargs["master_device_info"] is leader_info
1076
1077    async def test_join_waits_for_role_to_propagate(self, monkeypatch: pytest.MonkeyPatch) -> None:
1078        """A join whose membership appears a moment later is not declared a no-op."""
1079        monkeypatch.setattr("music_assistant.providers.wiim.grouping.VERIFY_MAX_WAIT", 5.0)
1080        reads = {"count": 0}
1081
1082        async def _slaves() -> list[dict[str, str]]:
1083            reads["count"] += 1
1084            # the follower only shows up in the leader's slave list on the second read
1085            return (
1086                [{"uuid": FOLLOWER_HTTP_UUID, "ip": "192.168.1.50"}] if reads["count"] >= 2 else []
1087            )
1088
1089        leader_client = MagicMock()
1090        leader_client.get_device_group_info = AsyncMock(return_value=_ginfo("solo", None))
1091        leader_client.get_slaves_info = AsyncMock(side_effect=_slaves)
1092        leader_client.get_device_info_model = AsyncMock(
1093            return_value=MagicMock(needs_wifi_direct_multiroom=False, wmrm_version="4.2")
1094        )
1095        member_client = _member_client()
1096        member_client.join_slave = AsyncMock()
1097        leader = _make_player(
1098            LEADER_ID, BACKEND_OFFICIAL, ip="192.168.1.20", command_client=leader_client
1099        )
1100        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC, command_client=member_client)
1101        coordinator, _ = _make_coordinator(leader, follower)
1102
1103        await coordinator.set_members(leader, [FOLLOWER_ID], None)
1104
1105        assert reads["count"] >= 2  # it retried until the membership propagated
1106
1107
1108class TestSetMembersLifecycle:
1109    """set_members always refreshes the leader's topology, even after a failure."""
1110
1111    async def test_forces_leader_refresh_on_success(self) -> None:
1112        """After a successful command the leader's live group state is re-read."""
1113        slaves: list[str] = []
1114        client = _leader_client(slaves)
1115        leader = _make_player(LEADER_ID, BACKEND_OFFICIAL, command_client=client)
1116        follower = _make_player(FOLLOWER_ID, BACKEND_OFFICIAL)
1117        coordinator, provider = _make_coordinator(leader, follower)
1118
1119        async def _join(*_args: object, **_kwargs: object) -> None:
1120            slaves.append(FOLLOWER_HTTP_UUID)
1121
1122        provider.wiim_controller.async_join_group = AsyncMock(side_effect=_join)
1123
1124        await coordinator.set_members(leader, [FOLLOWER_ID], None)
1125
1126        client.get_device_group_info.assert_awaited()
1127
1128    async def test_forces_leader_refresh_after_failure(self) -> None:
1129        """A failed command still refreshes the leader topology before propagating."""
1130        client = _leader_client([])
1131        leader = _make_player(LEADER_ID, BACKEND_OFFICIAL, command_client=client)
1132        member_client = _member_client()
1133        member_client.join_slave = AsyncMock(side_effect=WiiMError("boom"))
1134        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC, command_client=member_client)
1135        coordinator, _ = _make_coordinator(leader, follower)
1136
1137        with pytest.raises(PlayerCommandFailed):
1138            await coordinator.set_members(leader, [FOLLOWER_ID], None)
1139        client.get_device_group_info.assert_awaited()
1140
1141    async def test_unknown_member_raises(self) -> None:
1142        """A grouping target that is not a registered native player fails typed."""
1143        leader = _make_player(LEADER_ID, BACKEND_OFFICIAL)
1144        coordinator, provider = _make_coordinator(leader)
1145        provider.mass.players.get_player.side_effect = None
1146        provider.mass.players.get_player.return_value = None
1147
1148        with pytest.raises(PlayerCommandFailed):
1149            await coordinator.set_members(leader, ["wiim_uuid:ghost"], None)
1150
1151
1152class TestRegistrationCleanup:
1153    """Unregistering a player drops its topology cache entry."""
1154
1155    async def test_unregister_drops_stale_leadership(self) -> None:
1156        """A removed leader's cached members disappear on the next reconcile."""
1157        leader = _make_player(LEADER_ID, BACKEND_GENERIC)
1158        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC)
1159        coordinator, provider = _make_coordinator(leader, follower)
1160        coordinator.set_leader_slaves(LEADER_ID, [FOLLOWER_HTTP_UUID])
1161        await coordinator.reconcile()
1162
1163        provider.players.remove(leader)
1164        coordinator.unregister(LEADER_ID)
1165        await coordinator.reconcile()
1166
1167        assert coordinator.role_of(FOLLOWER_ID) == NativeGroupRole.STANDALONE
1168
1169    async def test_reconcile_drops_cache_for_unregistered_leader(self) -> None:
1170        """Cache entries for players no longer registered are pruned during reconcile."""
1171        leader = _make_player(LEADER_ID, BACKEND_GENERIC)
1172        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC)
1173        coordinator, provider = _make_coordinator(leader, follower)
1174        coordinator.set_leader_slaves(LEADER_ID, [FOLLOWER_HTTP_UUID])
1175        await coordinator.reconcile()
1176
1177        provider.players.remove(leader)  # gone, but no explicit unregister
1178        await coordinator.reconcile()
1179
1180        assert LEADER_ID not in coordinator._raw_slaves  # pruned
1181        assert coordinator.role_of(FOLLOWER_ID) == NativeGroupRole.STANDALONE
1182
1183    async def test_refresh_discards_response_for_replaced_player(self) -> None:
1184        """A read finishing after the player was replaced is not applied to the new one."""
1185        client = MagicMock()
1186        client.get_device_group_info = AsyncMock(return_value=_leader_info([FOLLOWER_HTTP_UUID]))
1187        client.get_slaves_info = AsyncMock(
1188            return_value=[{"uuid": FOLLOWER_HTTP_UUID, "ip": "192.168.1.50"}]
1189        )
1190        leader = _make_player(LEADER_ID, BACKEND_OFFICIAL, command_client=client)
1191        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC)
1192        coordinator, provider = _make_coordinator(leader, follower)
1193        # a fast unregister/re-register replaced the player instance during the read
1194        replacement = _make_player(LEADER_ID, BACKEND_OFFICIAL)
1195        provider.mass.players.get_player.side_effect = {
1196            LEADER_ID: replacement,
1197            FOLLOWER_ID: follower,
1198        }.get
1199
1200        result = await coordinator.refresh_leader(leader, force=True)
1201
1202        assert result is False
1203        assert coordinator.members_of(LEADER_ID) == []  # stale response not applied
1204
1205
1206class TestSelfReportedFollower:
1207    """A device following a leader MA has not discovered still reports as a follower."""
1208
1209    async def test_self_role_yields_follower_without_known_leader(self) -> None:
1210        """A self-reported follower is FOLLOWER even though its leader is unknown."""
1211        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC)
1212        coordinator, _ = _make_coordinator(follower)
1213
1214        coordinator.set_self_role(FOLLOWER_ID, True)
1215        await coordinator.reconcile()
1216
1217        assert coordinator.role_of(FOLLOWER_ID) == NativeGroupRole.FOLLOWER
1218        assert coordinator.leader_of(FOLLOWER_ID) is None
1219
1220    async def test_self_reported_follower_excluded_from_can_group_with(self) -> None:
1221        """A self-reported follower is not offered as a grouping target."""
1222        me = _make_player(LEADER_ID, BACKEND_OFFICIAL)
1223        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC)
1224        coordinator, _ = _make_coordinator(me, follower)
1225
1226        coordinator.set_self_role(FOLLOWER_ID, True)
1227        await coordinator.reconcile()
1228
1229        assert coordinator.can_group_with(me) == set()
1230
1231    async def test_known_leader_takes_precedence_over_self_role(self) -> None:
1232        """A resolved reverse leader is preferred; leader_of returns the known leader."""
1233        leader = _make_player(LEADER_ID, BACKEND_GENERIC)
1234        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC)
1235        coordinator, _ = _make_coordinator(leader, follower)
1236        coordinator.set_leader_slaves(LEADER_ID, [FOLLOWER_HTTP_UUID])
1237        coordinator.set_self_role(FOLLOWER_ID, True)
1238        await coordinator.reconcile()
1239
1240        assert coordinator.role_of(FOLLOWER_ID) == NativeGroupRole.FOLLOWER
1241        assert coordinator.leader_of(FOLLOWER_ID) == LEADER_ID
1242
1243    async def test_self_role_cleared_returns_to_standalone(self) -> None:
1244        """Clearing the self-follower flag returns the player to standalone."""
1245        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC)
1246        coordinator, _ = _make_coordinator(follower)
1247        coordinator.set_self_role(FOLLOWER_ID, True)
1248        await coordinator.reconcile()
1249
1250        coordinator.set_self_role(FOLLOWER_ID, False)
1251        await coordinator.reconcile()
1252
1253        assert coordinator.role_of(FOLLOWER_ID) == NativeGroupRole.STANDALONE
1254
1255    async def test_unregister_clears_self_role(self) -> None:
1256        """Unregistering a player drops its self-reported follower flag."""
1257        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC)
1258        coordinator, _ = _make_coordinator(follower)
1259        coordinator.set_self_role(FOLLOWER_ID, True)
1260        await coordinator.reconcile()
1261        assert coordinator.role_of(FOLLOWER_ID) == NativeGroupRole.FOLLOWER
1262
1263        coordinator.unregister(FOLLOWER_ID)
1264        await coordinator.reconcile()
1265
1266        assert coordinator.role_of(FOLLOWER_ID) == NativeGroupRole.STANDALONE
1267
1268    async def test_refresh_feeds_self_follower_from_live_role(self) -> None:
1269        """A live read reporting 'slave' under an unknown leader yields a follower role."""
1270        client = MagicMock()
1271        # follower of a leader MA does not manage (its master is not a registered player)
1272        client.get_device_group_info = AsyncMock(return_value=_ginfo("slave", OTHER_MASTER_UUID))
1273        player = _make_player(FOLLOWER_ID, BACKEND_OFFICIAL, command_client=client)
1274        coordinator, _ = _make_coordinator(player)
1275
1276        await coordinator.refresh_leader(player, force=True)
1277
1278        assert coordinator.role_of(FOLLOWER_ID) == NativeGroupRole.FOLLOWER
1279        assert coordinator.leader_of(FOLLOWER_ID) is None
1280
1281    async def test_live_self_role_retained_on_failed_refresh(self) -> None:
1282        """A failed live read keeps the last self-role instead of clearing it."""
1283        client = MagicMock()
1284        client.get_device_group_info = AsyncMock(
1285            side_effect=[_ginfo("slave", OTHER_MASTER_UUID), WiiMError("offline")]
1286        )
1287        player = _make_player(FOLLOWER_ID, BACKEND_OFFICIAL, command_client=client)
1288        coordinator, _ = _make_coordinator(player)
1289
1290        await coordinator.refresh_leader(player, force=True)
1291        assert coordinator.role_of(FOLLOWER_ID) == NativeGroupRole.FOLLOWER
1292
1293        await coordinator.refresh_leader(player, force=True)  # read fails
1294        assert coordinator.role_of(FOLLOWER_ID) == NativeGroupRole.FOLLOWER
1295
1296
1297class TestConcurrentRefresh:
1298    """Overlapping refreshes for one leader must not apply out of order."""
1299
1300    async def test_same_leader_refresh_is_serialized(self) -> None:
1301        """Concurrent forced refreshes for one leader run one at a time."""
1302        concurrency = {"current": 0, "max": 0}
1303
1304        async def _slow_read() -> MagicMock:
1305            concurrency["current"] += 1
1306            concurrency["max"] = max(concurrency["max"], concurrency["current"])
1307            await asyncio.sleep(0.01)
1308            concurrency["current"] -= 1
1309            return _ginfo("solo", None)
1310
1311        client = MagicMock()
1312        client.get_device_group_info = AsyncMock(side_effect=_slow_read)
1313        client.get_slaves_info = AsyncMock(return_value=[])
1314        leader = _make_player(LEADER_ID, BACKEND_OFFICIAL, command_client=client)
1315        coordinator, _ = _make_coordinator(leader)
1316
1317        await asyncio.gather(
1318            coordinator.refresh_leader(leader, force=True),
1319            coordinator.refresh_leader(leader, force=True),
1320        )
1321
1322        assert client.get_device_group_info.await_count == 2
1323        assert concurrency["max"] == 1  # the reads never overlapped
1324
1325
1326class TestCommandSerialization:
1327    """Grouping commands are serialized so concurrent moves cannot interleave."""
1328
1329    async def test_concurrent_commands_do_not_interleave(self) -> None:
1330        """A second command waits for the first to finish its whole sequence."""
1331        other_leader_id = "wiim_uuid:22222222-3333-4444-5555-666666666666"
1332        other_follower_http = "C1B2C3D4E5F6A7B8C9D0E1F2"
1333        other_follower_id = "wiim_uuid:C1B2C3D4-E5F6-A7B8-C9D0-E1F2C1B2C3D4"
1334        gate = asyncio.Event()
1335        order: list[str] = []
1336
1337        slaves_a: list[str] = []
1338        member_a_client = _member_client()
1339
1340        async def _join_a(*_args: object, **_kwargs: object) -> None:
1341            order.append("a")
1342            await gate.wait()  # hold the first command mid-sequence
1343            slaves_a.append(FOLLOWER_HTTP_UUID)
1344
1345        member_a_client.join_slave = AsyncMock(side_effect=_join_a)
1346
1347        slaves_b: list[str] = []
1348        member_b_client = _member_client()
1349
1350        async def _join_b(*_args: object, **_kwargs: object) -> None:
1351            order.append("b")
1352            slaves_b.append(other_follower_http)
1353
1354        member_b_client.join_slave = AsyncMock(side_effect=_join_b)
1355
1356        leader_a = _make_player(
1357            LEADER_ID, BACKEND_OFFICIAL, ip="192.168.1.20", command_client=_leader_client(slaves_a)
1358        )
1359        member_a = _make_player(FOLLOWER_ID, BACKEND_GENERIC, command_client=member_a_client)
1360        leader_b = _make_player(
1361            other_leader_id,
1362            BACKEND_OFFICIAL,
1363            ip="192.168.1.21",
1364            command_client=_leader_client(slaves_b),
1365        )
1366        member_b = _make_player(other_follower_id, BACKEND_GENERIC, command_client=member_b_client)
1367        coordinator, _ = _make_coordinator(leader_a, member_a, leader_b, member_b)
1368
1369        task_a = asyncio.create_task(coordinator.set_members(leader_a, [FOLLOWER_ID], None))
1370        for _ in range(50):
1371            await asyncio.sleep(0)
1372            if order == ["a"]:
1373                break
1374        assert order == ["a"]  # the first command is mid-sequence, holding the lock
1375
1376        task_b = asyncio.create_task(coordinator.set_members(leader_b, [other_follower_id], None))
1377        for _ in range(50):
1378            await asyncio.sleep(0)
1379        assert order == ["a"]  # the second command is blocked behind the first
1380
1381        gate.set()
1382        await asyncio.gather(task_a, task_b)
1383        assert order == ["a", "b"]  # only after the first finished did the second run
1384
1385
1386# a second follower identity for whole-batch validation tests
1387SECOND_HTTP_UUID = "D1B2C3D4E5F6A7B8C9D0E1F2"
1388SECOND_ID = "wiim_uuid:D1B2C3D4-E5F6-A7B8-C9D0-E1F2D1B2C3D4"
1389
1390
1391class TestBatchValidation:
1392    """The whole batch is validated before any speaker joins or leaves."""
1393
1394    async def test_incompatible_later_addition_aborts_before_any_join(self) -> None:
1395        """A legacy/incompatible target later in the batch stops every join up front."""
1396        good_slaves: list[str] = []
1397        good_client = _member_client()
1398        good_client.join_slave = AsyncMock(
1399            side_effect=lambda *_a, **_k: good_slaves.append(FOLLOWER_HTTP_UUID)
1400        )
1401        bad_client = _member_client()
1402        bad_client.join_slave = AsyncMock()
1403        # the second target is a legacy Wi-Fi Direct device the compatibility gate refuses
1404        bad_client.get_device_info_model = AsyncMock(
1405            return_value=MagicMock(needs_wifi_direct_multiroom=True, wmrm_version="4.2")
1406        )
1407        leader = _make_player(
1408            LEADER_ID, BACKEND_GENERIC, ip="192.168.1.20", command_client=_leader_client([])
1409        )
1410        good = _make_player(FOLLOWER_ID, BACKEND_GENERIC, command_client=good_client)
1411        bad = _make_player(SECOND_ID, BACKEND_GENERIC, command_client=bad_client)
1412        coordinator, _ = _make_coordinator(leader, good, bad)
1413
1414        with pytest.raises(PlayerCommandFailed):
1415            await coordinator.set_members(leader, [FOLLOWER_ID, SECOND_ID], None)
1416        # the whole batch was validated first, so the compatible target never joined either
1417        good_client.join_slave.assert_not_called()
1418        bad_client.join_slave.assert_not_called()
1419
1420    async def test_leader_following_unknown_group_cannot_lead(self) -> None:
1421        """A leader that itself follows an undiscovered group is refused before any mutation."""
1422        leader_client = _leader_client([])
1423        # the leader's own live read reports it is a slave of a leader MA cannot see
1424        leader_client.get_device_group_info = AsyncMock(
1425            return_value=MagicMock(role="slave", master_uuid=OTHER_MASTER_UUID)
1426        )
1427        leader = _make_player(
1428            LEADER_ID, BACKEND_GENERIC, ip="192.168.1.20", command_client=leader_client
1429        )
1430        member_client = _member_client()
1431        member_client.join_slave = AsyncMock()
1432        member = _make_player(FOLLOWER_ID, BACKEND_GENERIC, command_client=member_client)
1433        coordinator, _ = _make_coordinator(leader, member)
1434
1435        with pytest.raises(PlayerCommandFailed):
1436            await coordinator.set_members(leader, [FOLLOWER_ID], None)
1437        member_client.join_slave.assert_not_called()
1438
1439    async def test_is_unknown_leader_follower_query(self) -> None:
1440        """The public query flags a self-reported follower whose leader is undiscovered."""
1441        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC)
1442        coordinator, _ = _make_coordinator(follower)
1443        coordinator.set_self_role(FOLLOWER_ID, True)
1444        await coordinator.reconcile()
1445        assert coordinator.is_unknown_leader_follower(FOLLOWER_ID) is True
1446
1447        coordinator.set_self_role(FOLLOWER_ID, False)
1448        await coordinator.reconcile()
1449        assert coordinator.is_unknown_leader_follower(FOLLOWER_ID) is False
1450
1451    async def test_unreachable_generic_removal_aborts_batch(self) -> None:
1452        """An unreachable generic follower fails a removal batch before any detach runs."""
1453        slaves = [FOLLOWER_HTTP_UUID, SECOND_HTTP_UUID]
1454        good_client = _member_client()
1455        good_client.leave_group = AsyncMock()
1456        leader = _make_player(LEADER_ID, BACKEND_GENERIC, command_client=_leader_client(slaves))
1457        good = _make_player(FOLLOWER_ID, BACKEND_GENERIC, command_client=good_client)
1458        bad = _make_player(SECOND_ID, BACKEND_GENERIC, available=False)  # unreachable
1459        coordinator, _ = _make_coordinator(leader, good, bad)
1460
1461        with pytest.raises(PlayerCommandFailed):
1462            await coordinator.set_members(leader, None, [FOLLOWER_ID, SECOND_ID])
1463        good_client.leave_group.assert_not_called()
1464
1465    def test_incompatible_generic_peer_not_offered(self) -> None:
1466        """A legacy/incompatible generic peer is not offered as a grouping candidate."""
1467        me = _make_player(LEADER_ID, BACKEND_GENERIC)
1468        peer = _make_player(FOLLOWER_ID, BACKEND_GENERIC)
1469        # the peer is a legacy Wi-Fi Direct device, so the pair is guaranteed to fail
1470        peer.cached_device_info = MagicMock(needs_wifi_direct_multiroom=True, wmrm_version="4.2")
1471        coordinator, _ = _make_coordinator(me, peer)
1472
1473        assert coordinator.can_group_with(me) == set()
1474
1475    def test_compatible_generic_peer_is_offered(self) -> None:
1476        """A modern, matching-generation generic peer is offered as a candidate."""
1477        me = _make_player(LEADER_ID, BACKEND_GENERIC)
1478        peer = _make_player(FOLLOWER_ID, BACKEND_GENERIC)
1479        coordinator, _ = _make_coordinator(me, peer)
1480
1481        assert coordinator.can_group_with(me) == {FOLLOWER_ID}
1482
1483    async def test_join_fails_closed_when_member_leads_unmanaged_slave(self) -> None:
1484        """A member still leading an addressable slave MA cannot resolve is not orphaned."""
1485        member_client = _member_client()
1486        member_client.join_slave = AsyncMock()
1487        member_client.get_device_group_info = AsyncMock(return_value=MagicMock(role="master"))
1488        # the slave resolves to no registered player, so it cannot be cleanly dissolved
1489        member_client.get_slaves_info = AsyncMock(
1490            return_value=[{"uuid": "FFFFFFFFFFFFFFFFFFFFFFFF", "ip": "192.168.1.77"}]
1491        )
1492        leader = _make_player(
1493            LEADER_ID, BACKEND_OFFICIAL, ip="192.168.1.20", command_client=_leader_client([])
1494        )
1495        member = _make_player(FOLLOWER_ID, BACKEND_GENERIC, command_client=member_client)
1496        coordinator, _ = _make_coordinator(leader, member)
1497
1498        with pytest.raises(PlayerCommandFailed):
1499            await coordinator.set_members(leader, [FOLLOWER_ID], None)
1500        member_client.join_slave.assert_not_called()
1501
1502    async def test_join_dissolves_follower_gained_since_cache(self) -> None:
1503        """A follower discovered only in the member's live read is still dissolved first."""
1504        sub_http = "B1B2C3D4E5F6A7B8C9D0E1F2"
1505        sub_id = "wiim_uuid:B1B2C3D4-E5F6-A7B8-C9D0-E1F2B1B2C3D4"
1506        member_slaves = [sub_http]
1507
1508        async def _sub_leave() -> None:
1509            member_slaves.remove(sub_http)
1510
1511        sub_client = _member_client()
1512        sub_client.leave_group = AsyncMock(side_effect=_sub_leave)
1513        sub = _make_player(sub_id, BACKEND_GENERIC, command_client=sub_client)
1514        member_client = _leader_client(member_slaves)
1515        new_slaves: list[str] = []
1516        member_client.join_slave = AsyncMock(
1517            side_effect=lambda *_a, **_k: new_slaves.append(FOLLOWER_HTTP_UUID)
1518        )
1519        member = _make_player(
1520            FOLLOWER_ID, BACKEND_GENERIC, ip="192.168.1.40", command_client=member_client
1521        )
1522        leader = _make_player(
1523            LEADER_ID,
1524            BACKEND_OFFICIAL,
1525            ip="192.168.1.20",
1526            command_client=_leader_client(new_slaves),
1527        )
1528        # NOTE: the coordinator cache is never seeded for the member; only its live read finds sub
1529        coordinator, _ = _make_coordinator(leader, member, sub)
1530
1531        await coordinator.set_members(leader, [FOLLOWER_ID], None)
1532
1533        sub_client.leave_group.assert_awaited_once()
1534        member_client.join_slave.assert_awaited_once()
1535
1536    async def test_leader_info_read_once_per_batch(self) -> None:
1537        """The leader's device info is read once for a batch, not once per added member."""
1538        slaves: list[str] = []
1539        leader_client = _leader_client(slaves)
1540        leader = _make_player(
1541            LEADER_ID, BACKEND_GENERIC, ip="192.168.1.30", command_client=leader_client
1542        )
1543        first_client = _member_client()
1544        first_client.join_slave = AsyncMock(
1545            side_effect=lambda *_a, **_k: slaves.append(FOLLOWER_HTTP_UUID)
1546        )
1547        second_client = _member_client()
1548        second_client.join_slave = AsyncMock(
1549            side_effect=lambda *_a, **_k: slaves.append(SECOND_HTTP_UUID)
1550        )
1551        first = _make_player(FOLLOWER_ID, BACKEND_GENERIC, command_client=first_client)
1552        second = _make_player(SECOND_ID, BACKEND_GENERIC, command_client=second_client)
1553        coordinator, _ = _make_coordinator(leader, first, second)
1554
1555        await coordinator.set_members(leader, [FOLLOWER_ID, SECOND_ID], None)
1556
1557        assert leader_client.get_device_info_model.await_count == 1
1558
1559    async def test_refresh_skips_unavailable_device(self) -> None:
1560        """A device that is unavailable is not polled again; its cached topology is kept."""
1561        client = MagicMock()
1562        client.get_device_group_info = AsyncMock()
1563        player = _make_player(LEADER_ID, BACKEND_GENERIC, available=False, command_client=client)
1564        coordinator, _ = _make_coordinator(player)
1565
1566        assert await coordinator.refresh_leader(player, force=True) is False
1567        client.get_device_group_info.assert_not_called()
1568
1569    async def test_live_read_reveals_unknown_leader_follower_and_aborts(self) -> None:
1570        """A member whose cache said standalone but now reads as a slave of an unknown leader fails."""
1571        member_client = _member_client()
1572        member_client.join_slave = AsyncMock()
1573        member_client.get_device_group_info = AsyncMock(
1574            return_value=MagicMock(role="slave", master_uuid=OTHER_MASTER_UUID)
1575        )
1576        leader = _make_player(
1577            LEADER_ID, BACKEND_OFFICIAL, ip="192.168.1.20", command_client=_leader_client([])
1578        )
1579        member = _make_player(FOLLOWER_ID, BACKEND_GENERIC, command_client=member_client)
1580        coordinator, _ = _make_coordinator(leader, member)  # cache starts empty (standalone)
1581
1582        with pytest.raises(PlayerCommandFailed):
1583            await coordinator.set_members(leader, [FOLLOWER_ID], None)
1584        member_client.join_slave.assert_not_called()
1585
1586    async def test_combined_batch_aborts_before_join_when_leader_unreadable(self) -> None:
1587        """A combined add/remove batch does not join anything if the leader's list is unreadable."""
1588        leader_client = MagicMock()
1589        leader_client.capabilities = {}
1590        leader_client.get_device_group_info = AsyncMock(return_value=MagicMock(role="solo"))
1591        leader_client.get_slaves_info = AsyncMock(side_effect=WiiMError("offline"))
1592        leader_client.get_device_info_model = AsyncMock(
1593            return_value=MagicMock(needs_wifi_direct_multiroom=False, wmrm_version="4.2")
1594        )
1595        leader = _make_player(
1596            LEADER_ID, BACKEND_GENERIC, ip="192.168.1.30", command_client=leader_client
1597        )
1598        add_client = _member_client()
1599        add_client.join_slave = AsyncMock()
1600        add_member = _make_player(FOLLOWER_ID, BACKEND_GENERIC, command_client=add_client)
1601        remove_member = _make_player(SECOND_ID, BACKEND_GENERIC)
1602        coordinator, _ = _make_coordinator(leader, add_member, remove_member)
1603
1604        with pytest.raises(PlayerCommandFailed):
1605            await coordinator.set_members(leader, [FOLLOWER_ID], [SECOND_ID])
1606        add_client.join_slave.assert_not_called()
1607
1608    async def test_addition_with_unreadable_group_aborts_batch(self) -> None:
1609        """A batch fails before any join if an addition's live group cannot be read."""
1610        member_client = _member_client()
1611        member_client.join_slave = AsyncMock()
1612        member_client.get_device_group_info = AsyncMock(side_effect=WiiMError("offline"))
1613        leader = _make_player(
1614            LEADER_ID, BACKEND_GENERIC, ip="192.168.1.20", command_client=_leader_client([])
1615        )
1616        member = _make_player(FOLLOWER_ID, BACKEND_GENERIC, command_client=member_client)
1617        coordinator, _ = _make_coordinator(leader, member)
1618
1619        with pytest.raises(PlayerCommandFailed):
1620            await coordinator.set_members(leader, [FOLLOWER_ID], None)
1621        member_client.join_slave.assert_not_called()
1622
1623    async def test_join_surfaces_device_info_failure(self) -> None:
1624        """A device-info read error surfaces as a chained failure, not a false incompatibility."""
1625        member_client = _member_client()
1626        member_client.join_slave = AsyncMock()
1627        member_client.get_device_info_model = AsyncMock(side_effect=WiiMError("boom"))
1628        leader = _make_player(
1629            LEADER_ID, BACKEND_OFFICIAL, ip="192.168.1.20", command_client=_leader_client([])
1630        )
1631        member = _make_player(FOLLOWER_ID, BACKEND_GENERIC, command_client=member_client)
1632        coordinator, _ = _make_coordinator(leader, member)
1633
1634        with pytest.raises(PlayerCommandFailed) as exc_info:
1635            await coordinator.set_members(leader, [FOLLOWER_ID], None)
1636        assert isinstance(exc_info.value.__cause__, WiiMError)
1637        member_client.join_slave.assert_not_called()
1638
1639    async def test_unavailable_requester_offers_no_native_peers(self) -> None:
1640        """A device whose own grouping API is unreachable offers no native candidates."""
1641        me = _make_player(LEADER_ID, BACKEND_GENERIC, available=False)
1642        peer = _make_player(FOLLOWER_ID, BACKEND_GENERIC)
1643        coordinator, _ = _make_coordinator(me, peer)
1644
1645        assert coordinator.can_group_with(me) == set()
1646
1647    async def test_available_requester_still_offers_peers(self) -> None:
1648        """A reachable device is unaffected: it still offers its available native peers."""
1649        me = _make_player(LEADER_ID, BACKEND_GENERIC)
1650        peer = _make_player(FOLLOWER_ID, BACKEND_GENERIC)
1651        coordinator, _ = _make_coordinator(me, peer)
1652
1653        assert coordinator.can_group_with(me) == {FOLLOWER_ID}
1654
1655    async def test_add_remove_conflict_rejected(self) -> None:
1656        """The same id in both add and remove is a contradictory request and is rejected."""
1657        leader = _make_player(LEADER_ID, BACKEND_OFFICIAL, command_client=_leader_client([]))
1658        member = _make_player(FOLLOWER_ID, BACKEND_OFFICIAL)
1659        coordinator, provider = _make_coordinator(leader, member)
1660
1661        with pytest.raises(PlayerCommandFailed):
1662            await coordinator.set_members(leader, [FOLLOWER_ID], [FOLLOWER_ID])
1663        provider.wiim_controller.async_join_group.assert_not_called()
1664
1665    async def test_duplicate_ids_rejected(self) -> None:
1666        """A duplicate id within a single add/remove list is rejected up front."""
1667        leader = _make_player(LEADER_ID, BACKEND_OFFICIAL, command_client=_leader_client([]))
1668        member = _make_player(FOLLOWER_ID, BACKEND_OFFICIAL)
1669        coordinator, provider = _make_coordinator(leader, member)
1670
1671        with pytest.raises(PlayerCommandFailed):
1672            await coordinator.set_members(leader, [FOLLOWER_ID, FOLLOWER_ID], None)
1673        provider.wiim_controller.async_join_group.assert_not_called()
1674
1675    async def test_newer_observation_wins_ownership_despite_later_apply(self) -> None:
1676        """A follower belongs to the most recently OBSERVED leader, not the last one applied."""
1677        other_leader_id = "wiim_uuid:22222222-3333-4444-5555-666666666666"
1678        leader_a = _make_player(LEADER_ID, BACKEND_GENERIC)
1679        leader_b = _make_player(other_leader_id, BACKEND_GENERIC)
1680        follower = _make_player(FOLLOWER_ID, BACKEND_GENERIC)
1681        coordinator, _ = _make_coordinator(leader_a, leader_b, follower)
1682
1683        # leader B observed the follower more recently, but leader A's older observation is
1684        # applied afterwards (a longer-running read that completed late)
1685        coordinator.set_leader_slaves(other_leader_id, [FOLLOWER_HTTP_UUID], observed_at=200.0)
1686        coordinator.set_leader_slaves(LEADER_ID, [FOLLOWER_HTTP_UUID], observed_at=100.0)
1687        await coordinator.reconcile()
1688
1689        assert coordinator.leader_of(FOLLOWER_ID) == other_leader_id
1690
1691    async def test_republish_all_republishes_every_native_player(self) -> None:
1692        """schedule_republish's work re-publishes every registered native player's state."""
1693        leader = _make_player(LEADER_ID, BACKEND_GENERIC)
1694        peer = _make_player(FOLLOWER_ID, BACKEND_OFFICIAL)
1695        coordinator, _ = _make_coordinator(leader, peer)
1696
1697        coordinator._republish_all()
1698
1699        leader.on_native_group_update.assert_called_once()
1700        peer.on_native_group_update.assert_called_once()
1701
1702    async def test_removal_of_absent_unreachable_member_is_skipped_not_aborted(self) -> None:
1703        """An unreachable removal target the leader no longer lists is skipped, not fatal."""
1704        slaves = [FOLLOWER_HTTP_UUID]  # the leader currently lists only the reachable member
1705        good_client = _member_client()
1706
1707        async def _leave() -> None:
1708            slaves.remove(FOLLOWER_HTTP_UUID)
1709
1710        good_client.leave_group = AsyncMock(side_effect=_leave)
1711        leader = _make_player(LEADER_ID, BACKEND_GENERIC, command_client=_leader_client(slaves))
1712        good = _make_player(FOLLOWER_ID, BACKEND_GENERIC, command_client=good_client)
1713        gone = _make_player(SECOND_ID, BACKEND_GENERIC, available=False)  # unreachable, not owned
1714        coordinator, _ = _make_coordinator(leader, good, gone)
1715
1716        # removing both must not abort on the absent, unreachable target
1717        await coordinator.set_members(leader, None, [FOLLOWER_ID, SECOND_ID])
1718
1719        good_client.leave_group.assert_awaited_once()
1720