music-assistant-server

1"""Tests for Sync Group Player protocol awareness and locking."""
2
3from __future__ import annotations
4
5import asyncio
6import inspect
7import time
8from typing import Any
9from unittest.mock import AsyncMock, MagicMock, patch
10
11import pytest
12from music_assistant_models.enums import PlaybackState, PlayerFeature, PlayerType
13from music_assistant_models.player import OutputProtocol
14
15from music_assistant.constants import CONF_GROUP_MEMBERS, CONF_PLAYERS
16from music_assistant.providers.sync_group.player import SyncGroupPlayer
17
18
19def _player_lookup(players: dict[str, MagicMock]) -> MagicMock:
20    """Create a get_player side_effect from a player dict."""
21    return MagicMock(side_effect=lambda pid: players.get(pid))
22
23
24def _make_mock_mass() -> MagicMock:
25    """Create a minimal mock MusicAssistant instance."""
26    mass = MagicMock()
27    mass.players = MagicMock()
28    mass.players.get_player = MagicMock(return_value=None)
29    mass.players._handle_cmd_stop = AsyncMock()
30    mass.players._handle_cmd_resume = AsyncMock()
31    mass.players._handle_play_media = AsyncMock()
32    mass.players.cmd_set_members = AsyncMock()
33    mass.players._handle_set_members = AsyncMock()
34    # get_player_lock is an async context manager (used by syncgroup to lock the sync leader)
35    lock_ctx = AsyncMock()
36    lock_ctx.__aenter__.return_value = None
37    lock_ctx.__aexit__.return_value = False
38    mass.players.get_player_lock = MagicMock(return_value=lock_ctx)
39    # wait_for_player_update is an async context manager — return one that no-ops.
40    # __aexit__ must explicitly return False so exceptions inside the `async with`
41    # body propagate (an unconfigured AsyncMock returns a truthy MagicMock and
42    # would silently swallow real test failures).
43    wait_ctx = AsyncMock()
44    wait_ctx.__aenter__.return_value = None
45    wait_ctx.__aexit__.return_value = False
46    mass.players.wait_for_player_update = MagicMock(return_value=wait_ctx)
47    mass.players.trigger_player_update = MagicMock()
48    mass.call_later = MagicMock()
49    mass.cancel_timer = MagicMock()
50    mass.config = MagicMock()
51    mass.config.get_base_player_config.return_value = MagicMock(
52        name=None, default_name="Test Group", get_value=MagicMock(return_value=True)
53    )
54    return mass
55
56
57def _make_mock_player(
58    player_id: str,
59    provider_domain: str = "sonos",
60    available: bool = True,
61    protocol_domains: list[str] | None = None,
62    active_output_protocol: str | None = None,
63    playback_state: PlaybackState = PlaybackState.IDLE,
64) -> MagicMock:
65    """Create a mock player with configurable protocol support."""
66    player = MagicMock()
67    player.player_id = player_id
68    player.display_name = player_id
69    player.available = available
70    player.active_output_protocol = active_output_protocol
71    player.playback_state = playback_state
72    player.protocol_parent_id = None
73    player.provider = MagicMock()
74    player.provider.domain = provider_domain
75
76    # Build linked_output_protocols
77    protocols = []
78    for domain in protocol_domains or []:
79        proto = MagicMock(spec=OutputProtocol)
80        proto.protocol_domain = domain
81        proto.available = True
82        # default to a synthetic protocol id; tests that need a specific id can override
83        proto.output_protocol_id = f"{player_id}_{domain}_proto"
84        protocols.append(proto)
85    player.linked_output_protocols = protocols
86
87    # State mock
88    player.state = MagicMock()
89    player.state.available = available
90    player.state.playback_state = playback_state
91    player.state.can_group_with = set()
92    player.state.group_members = []
93    player.state.supported_features = {PlayerFeature.SET_MEMBERS}
94    # default to "not synced" so the syncgroup form path doesn't enter the
95    # stale-state wait loop. Tests that want to assert on the stale-state
96    # behavior can override this explicitly.
97    player.state.synced_to = None
98    player.synced_to = None
99    player.set_members = AsyncMock()
100
101    return player
102
103
104def _make_sync_group(mass: MagicMock, player_id: str = "syncgroup_test") -> SyncGroupPlayer:
105    """Create a SyncGroupPlayer with mock provider."""
106    provider = MagicMock()
107    provider.domain = "sync_group"
108    provider.instance_id = "sync_group_test"
109    provider.name = "Sync Group"
110    provider.mass = mass
111
112    def _config_get_value(key: str, default: object = None) -> object:
113        # CONF_DYNAMIC_GROUP_MEMBERS resolves to "dynamic_members"
114        if key == "dynamic_members":
115            return True
116        if key == "members_filter":
117            return []
118        return default
119
120    mass.config.get_base_player_config.return_value = MagicMock(
121        name=None, default_name="Test Group", get_value=_config_get_value
122    )
123
124    sgp = SyncGroupPlayer(provider, player_id)
125    sgp._cache.clear()
126    return sgp
127
128
129class TestProtocolAwareLeaderSelection:
130    """Test that leader selection prefers protocol continuity."""
131
132    def test_select_leader_prefers_active_protocol(self) -> None:
133        """When preferred protocol is given, prefer members supporting it."""
134        mass = _make_mock_mass()
135        sgp = _make_sync_group(mass)
136
137        # Player A: sonos native only
138        player_a = _make_mock_player("player_a", provider_domain="sonos")
139        # Player B: has airplay protocol
140        player_b = _make_mock_player(
141            "player_b", provider_domain="sonos", protocol_domains=["airplay"]
142        )
143
144        mass.players.get_player = _player_lookup({"player_a": player_a, "player_b": player_b})
145
146        sgp._attr_group_members = ["player_a", "player_b"]
147
148        leader = sgp._select_sync_leader(preferred_protocol_domain="airplay")
149        assert leader == player_b
150
151    def test_select_leader_fallback_when_no_protocol_match(self) -> None:
152        """When no member supports the preferred protocol, fall back to first available."""
153        mass = _make_mock_mass()
154        sgp = _make_sync_group(mass)
155
156        player_a = _make_mock_player("player_a", provider_domain="sonos")
157        mass.players.get_player = _player_lookup({"player_a": player_a})
158
159        sgp._attr_group_members = ["player_a"]
160
161        leader = sgp._select_sync_leader(preferred_protocol_domain="airplay")
162        assert leader == player_a
163
164    def test_select_leader_no_protocol_uses_first_available(self) -> None:
165        """When no preferred protocol, pick first available."""
166        mass = _make_mock_mass()
167        sgp = _make_sync_group(mass)
168
169        player_a = _make_mock_player("player_a")
170        player_b = _make_mock_player("player_b")
171        mass.players.get_player = _player_lookup({"player_a": player_a, "player_b": player_b})
172
173        sgp._attr_group_members = ["player_a", "player_b"]
174
175        leader = sgp._select_sync_leader()
176        assert leader == player_a
177
178
179class TestMemberSupportsProtocol:
180    """Test protocol domain checking for members."""
181
182    def test_native_provider_match(self) -> None:
183        """Player's own provider domain matches."""
184        mass = _make_mock_mass()
185        sgp = _make_sync_group(mass)
186
187        player = _make_mock_player("p1", provider_domain="airplay")
188        assert sgp._member_supports_protocol_domain(player, "airplay") is True
189
190    def test_linked_protocol_match(self) -> None:
191        """Player has a linked output protocol matching the domain."""
192        mass = _make_mock_mass()
193        sgp = _make_sync_group(mass)
194
195        player = _make_mock_player("p1", provider_domain="sonos", protocol_domains=["airplay"])
196        assert sgp._member_supports_protocol_domain(player, "airplay") is True
197
198    def test_no_match(self) -> None:
199        """Player doesn't support the requested protocol."""
200        mass = _make_mock_mass()
201        sgp = _make_sync_group(mass)
202
203        player = _make_mock_player("p1", provider_domain="sonos")
204        assert sgp._member_supports_protocol_domain(player, "airplay") is False
205
206
207class TestActiveProtocolDomain:
208    """Test that active_protocol_domain is derived correctly from live state."""
209
210    def test_no_leader_returns_none(self) -> None:
211        """With no sync leader, active_protocol_domain is None."""
212        mass = _make_mock_mass()
213        sgp = _make_sync_group(mass)
214        sgp.sync_leader = None
215        assert sgp.active_protocol_domain is None
216
217    def test_native_leader_returns_native_domain(self) -> None:
218        """With a native leader (no active output protocol) return the leader's domain."""
219        mass = _make_mock_mass()
220        sgp = _make_sync_group(mass)
221        leader = _make_mock_player("leader", provider_domain="sonos")
222        mass.players.get_player = _player_lookup({"leader": leader})
223        sgp.sync_leader = leader
224        sgp._attr_group_members = ["leader"]
225        assert sgp.active_protocol_domain == "sonos"
226
227    def test_active_protocol_with_requiring_member(self) -> None:
228        """Non-native protocol stays active while a member still requires it."""
229        mass = _make_mock_mass()
230        sgp = _make_sync_group(mass)
231        leader = _make_mock_player(
232            "leader", provider_domain="sonos", active_output_protocol="ap_leader"
233        )
234        ap_protocol = _make_mock_player("ap_leader", provider_domain="airplay")
235        # AirPlay-only member (only linked protocol is airplay)
236        ap_only = _make_mock_player(
237            "ap_only", provider_domain="universal_player", protocol_domains=["airplay"]
238        )
239        ap_only.is_native_player = False
240        mass.players.get_player = _player_lookup(
241            {"leader": leader, "ap_leader": ap_protocol, "ap_only": ap_only}
242        )
243        sgp.sync_leader = leader
244        sgp._attr_group_members = ["leader", "ap_only"]
245        assert sgp.active_protocol_domain == "airplay"
246
247    def test_active_protocol_downshifts_when_no_member_requires_it(self) -> None:
248        """Non-native protocol downshifts to native when no member still requires it."""
249        mass = _make_mock_mass()
250        sgp = _make_sync_group(mass)
251        leader = _make_mock_player(
252            "leader", provider_domain="sonos", active_output_protocol="ap_leader"
253        )
254        ap_protocol = _make_mock_player("ap_leader", provider_domain="airplay")
255        mass.players.get_player = _player_lookup({"leader": leader, "ap_leader": ap_protocol})
256        sgp.sync_leader = leader
257        # Only a native-capable Sonos leader remains; no one requires airplay
258        sgp._attr_group_members = ["leader"]
259        assert sgp.active_protocol_domain == "sonos"
260
261    def test_dissolve_clears_sync_leader(self) -> None:
262        """Verify _dissolve_syncgroup clears the sync leader."""
263        mass = _make_mock_mass()
264        sgp = _make_sync_group(mass)
265        source = inspect.getsource(sgp._dissolve_syncgroup)
266        assert "self.sync_leader = None" in source
267
268    @pytest.mark.asyncio
269    async def test_dissolve_schedules_protocol_clear_while_leader_still_playing(self) -> None:
270        """
271        Dissolve must schedule the delayed protocol clear while the leader plays.
272
273        Real devices report new state with a delay after a stop, so a playback
274        state check here would skip the clear and leave a stale active protocol
275        on the leader.
276        """
277        mass = _make_mock_mass()
278        sgp = _make_sync_group(mass)
279        leader = _make_mock_player(
280            "leader",
281            provider_domain="wiim",
282            active_output_protocol="ap_leader",
283            playback_state=PlaybackState.PLAYING,
284        )
285        mass.players.get_player = _player_lookup({"leader": leader})
286        sgp.sync_leader = leader
287
288        with patch.object(sgp, "update_state"):
289            await sgp._dissolve_syncgroup()
290
291        # use getattr to defeat mypy's narrowing after the earlier assignment,
292        # since it can't see that _dissolve_syncgroup mutates sync_leader.
293        assert getattr(sgp, "sync_leader") is None  # noqa: B009
294        mass.players.schedule_active_output_protocol_clear.assert_called_once_with(leader)
295
296    @pytest.mark.asyncio
297    async def test_dissolve_ungroups_from_the_member_holding_the_group(self) -> None:
298        """
299        Dissolve must ungroup from the member that actually holds the group members.
300
301        A provider can promote a different member to protocol leader than the one we
302        track; ungrouping from our leader would then be a no-op and leave the members
303        grouped and streaming with no way back.
304        """
305        mass = _make_mock_mass()
306        sgp = _make_sync_group(mass)
307
308        # our tracked leader no longer holds any members
309        living_room = _make_mock_player("living_room", provider_domain="sendspin")
310        bathroom = _make_mock_player("bathroom", provider_domain="sendspin")
311        kitchen = _make_mock_player("kitchen", provider_domain="sendspin")
312        bathroom.state.group_members = ["bathroom", "living_room", "kitchen"]
313
314        mass.players.get_player = _player_lookup(
315            {"living_room": living_room, "bathroom": bathroom, "kitchen": kitchen}
316        )
317
318        sgp.sync_leader = living_room
319        sgp._attr_group_members = ["living_room", "bathroom", "kitchen"]
320
321        with patch.object(sgp, "update_state"):
322            await sgp._dissolve_syncgroup()
323
324        mass.players._handle_set_members.assert_awaited_once_with(
325            bathroom, player_ids_to_remove=["living_room", "kitchen"]
326        )
327        # the promoted owner must be stopped too, our callers only stop the tracked leader
328        mass.players._handle_cmd_stop.assert_awaited_once_with("bathroom")
329        assert getattr(sgp, "sync_leader") is None  # noqa: B009
330
331    @pytest.mark.asyncio
332    async def test_dissolve_keeps_using_the_sync_leader_when_it_holds_the_group(self) -> None:
333        """A leader that still holds its members must stay the dissolve target."""
334        mass = _make_mock_mass()
335        sgp = _make_sync_group(mass)
336
337        leader = _make_mock_player("leader", provider_domain="sendspin")
338        member = _make_mock_player("member", provider_domain="sendspin")
339        leader.state.group_members = ["leader", "member"]
340
341        mass.players.get_player = _player_lookup({"leader": leader, "member": member})
342        sgp.sync_leader = leader
343        sgp._attr_group_members = ["leader", "member"]
344
345        with patch.object(sgp, "update_state"):
346            await sgp._dissolve_syncgroup()
347
348        mass.players._handle_set_members.assert_awaited_once_with(
349            leader, player_ids_to_remove=["member"]
350        )
351        mass.players._handle_cmd_stop.assert_not_awaited()
352
353    @pytest.mark.asyncio
354    async def test_dissolve_ignores_a_member_leading_an_unrelated_group(self) -> None:
355        """A member grouped outside of MA must never be adopted as the dissolve target."""
356        mass = _make_mock_mass()
357        sgp = _make_sync_group(mass)
358
359        living_room = _make_mock_player("living_room", provider_domain="sendspin")
360        member = _make_mock_player("member", provider_domain="sendspin")
361        # member leads a group that shares no player with us
362        member.state.group_members = ["member", "outsider"]
363
364        mass.players.get_player = _player_lookup({"living_room": living_room, "member": member})
365        sgp.sync_leader = living_room
366        sgp._attr_group_members = ["living_room", "member"]
367
368        with patch.object(sgp, "update_state"):
369            await sgp._dissolve_syncgroup()
370
371        mass.players._handle_set_members.assert_not_awaited()
372        mass.players._handle_cmd_stop.assert_not_awaited()
373
374
375class TestControllerLockCategory:
376    """Test that the controller's lock categories serialize correctly."""
377
378    def test_play_lock_key_format(self) -> None:
379        """Lock key for play category uses 'play_{player_id}' format."""
380        # The decorator uses lock category string as prefix
381        # play_media, set_members, enqueue_next_media all use "play" category
382        lock_key = "play_test_player_123"
383        assert lock_key.startswith("play_")
384
385
386class TestDynamicLeaderSwitch:
387    """Test dynamic leader switching behaviour."""
388
389    @pytest.mark.asyncio
390    async def test_dynamic_leader_switch_hands_off_to_new_leader(self) -> None:
391        """When the new leader IS in the live session, seamless protocol-level handoff is used."""
392        mass = _make_mock_mass()
393        sgp = _make_sync_group(mass)
394
395        # AirPlay-only protocol member ensures active_protocol_domain resolves
396        # to "airplay" and drives new-leader selection toward a member that
397        # supports AirPlay.
398        ap_only = _make_mock_player(
399            "ap_only", provider_domain="universal_player", protocol_domains=["airplay"]
400        )
401        ap_only.is_native_player = False
402
403        # new_leader's AirPlay protocol player (must be in the session for handoff)
404        ap_new = _make_mock_player("ap_new", provider_domain="airplay")
405
406        old_leader = _make_mock_player(
407            "old_leader",
408            provider_domain="sonos",
409            active_output_protocol="ap_old",
410            protocol_domains=["airplay"],
411        )
412        new_leader = _make_mock_player(
413            "new_leader",
414            provider_domain="sonos",
415            protocol_domains=["airplay"],
416            active_output_protocol="ap_new",
417        )
418        # Wire the linked airplay protocols on the parents to point to the
419        # corresponding protocol player ids so _resolve_session_target can find them.
420        new_leader.linked_output_protocols[0].output_protocol_id = "ap_new"
421        ap_only.linked_output_protocols[0].output_protocol_id = "ap_only_proto"
422        # ap_only's airplay protocol player needs to exist for the protocol-id resolution
423        ap_only_proto = _make_mock_player("ap_only_proto", provider_domain="airplay")
424
425        ap_protocol = _make_mock_player("ap_old", provider_domain="airplay")
426        # Set up the live session with new_leader's protocol player in sync_clients
427        mock_session = MagicMock()
428        mock_session.sync_clients = [ap_protocol, ap_new, ap_only]
429        ap_protocol.stream = MagicMock()
430        ap_protocol.stream.session = mock_session
431
432        mass.players.get_player = _player_lookup(
433            {
434                "old_leader": old_leader,
435                "new_leader": new_leader,
436                "ap_old": ap_protocol,
437                "ap_new": ap_new,
438                "ap_only": ap_only,
439                "ap_only_proto": ap_only_proto,
440            }
441        )
442
443        sgp.sync_leader = old_leader
444        sgp._attr_group_members = ["old_leader", "new_leader", "ap_only"]
445
446        with patch.object(sgp, "update_state"):
447            await sgp._dynamic_leader_switch("old_leader")
448
449        assert sgp.sync_leader == new_leader
450        assert "old_leader" not in sgp._attr_group_members
451
452        # 1. Old leader's session protocol player got told to step out (self-remove)
453        ap_protocol.set_members.assert_any_await(player_ids_to_remove=["ap_old"])
454
455        # 2. New leader's protocol player got the remaining members added.
456        # ap_only is already a protocol player on the airplay domain (its own
457        # provider is universal_player but its linked airplay protocol resolves
458        # via _resolve_session_target). The exact id depends on the mock wiring,
459        # but the call must have happened.
460        ap_new.set_members.assert_awaited()
461        add_call = ap_new.set_members.await_args
462        assert add_call.kwargs.get("player_ids_to_add"), (
463            "expected new leader's protocol player to receive the remaining members"
464        )
465
466    @pytest.mark.asyncio
467    async def test_dynamic_leader_switch_dissolves_when_new_leader_not_in_session(
468        self,
469    ) -> None:
470        """When the new leader is NOT in the live session, fall back to dissolve+reform."""
471        mass = _make_mock_mass()
472        mass.players.cmd_resume = AsyncMock()
473        sgp = _make_sync_group(mass)
474
475        old_leader = _make_mock_player(
476            "old_leader",
477            provider_domain="sonos",
478            active_output_protocol="ap_old",
479            protocol_domains=["airplay"],
480        )
481        # Freshly-added player — NOT in the live session
482        fresh_player = _make_mock_player(
483            "fresh_player", provider_domain="sonos", protocol_domains=["airplay"]
484        )
485        ap_protocol = _make_mock_player("ap_old", provider_domain="airplay")
486        # Session does NOT contain fresh_player's protocol player
487        mock_session = MagicMock()
488        mock_session.sync_clients = [ap_protocol]
489        ap_protocol.stream = MagicMock()
490        ap_protocol.stream.session = mock_session
491
492        mass.players.get_player = _player_lookup(
493            {
494                "old_leader": old_leader,
495                "fresh_player": fresh_player,
496                "ap_old": ap_protocol,
497            }
498        )
499
500        sgp.sync_leader = old_leader
501        sgp._attr_group_members = ["old_leader", "fresh_player"]
502
503        with patch.object(sgp, "update_state"):
504            await sgp._dynamic_leader_switch("old_leader")
505
506        # The old protocol player must NOT have been told to self-remove
507        # (that path is only for the seamless handoff). Instead, dissolve+reform
508        # happened: wait_for_player_update was used to wrap the stop.
509        ap_protocol.set_members.assert_not_awaited()
510        mass.players.wait_for_player_update.assert_called()
511        assert "old_leader" not in sgp._attr_group_members
512
513    @pytest.mark.asyncio
514    async def test_dynamic_leader_switch_follows_live_session_order(self) -> None:
515        """
516        The new leader must be the member that inherits the live session.
517
518        Providers promote their own first remaining member, so a leader picked from a
519        drifted member order would leave the group tracking a different player than the
520        one actually holding the session.
521        """
522        mass = _make_mock_mass()
523        sgp = _make_sync_group(mass)
524
525        old_leader = _make_mock_player("old_leader", provider_domain="sendspin")
526        living_room = _make_mock_player("living_room", provider_domain="sendspin")
527        bathroom = _make_mock_player("bathroom", provider_domain="sendspin")
528        # no stream/session object (Sendspin, Snapcast) -> handoff is assumed safe
529        old_leader.stream = None
530        # the live session hands over to bathroom, our own order says living_room.
531        # state.group_members is deliberately left unordered: only the raw attribute
532        # carries the provider's member order.
533        old_leader.group_members = ["old_leader", "bathroom", "living_room"]
534        old_leader.state.group_members = ["old_leader", "living_room", "bathroom"]
535
536        mass.players.get_player = _player_lookup(
537            {
538                "old_leader": old_leader,
539                "living_room": living_room,
540                "bathroom": bathroom,
541            }
542        )
543
544        sgp.sync_leader = old_leader
545        sgp._attr_group_members = ["old_leader", "living_room", "bathroom"]
546
547        with patch.object(sgp, "update_state"):
548            await sgp._dynamic_leader_switch("old_leader")
549
550        assert sgp.sync_leader == bathroom
551        assert sgp._attr_group_members == ["bathroom", "living_room"]
552        old_leader.set_members.assert_any_await(player_ids_to_remove=["old_leader"])
553        bathroom.set_members.assert_any_await(player_ids_to_add=["living_room"])
554
555    def test_align_members_translates_protocol_ids_and_keeps_outsiders_last(self) -> None:
556        """A protocol session player's order maps onto the parent members that we track."""
557        mass = _make_mock_mass()
558        sgp = _make_sync_group(mass)
559
560        living_room = _make_mock_player("living_room")
561        bathroom = _make_mock_player("bathroom")
562        # tracked but not part of the live session
563        offline = _make_mock_player("offline")
564        ap_living_room = _make_mock_player("ap_living_room", provider_domain="airplay")
565        ap_bathroom = _make_mock_player("ap_bathroom", provider_domain="airplay")
566        ap_living_room.protocol_parent_id = "living_room"
567        ap_bathroom.protocol_parent_id = "bathroom"
568
569        session_player = _make_mock_player("ap_old", provider_domain="airplay")
570        session_player.group_members = ["ap_old", "ap_bathroom", "ap_living_room"]
571
572        mass.players.get_player = _player_lookup(
573            {
574                "living_room": living_room,
575                "bathroom": bathroom,
576                "offline": offline,
577                "ap_living_room": ap_living_room,
578                "ap_bathroom": ap_bathroom,
579                "ap_old": session_player,
580            }
581        )
582        sgp._attr_group_members = ["living_room", "bathroom", "offline"]
583
584        sgp._align_members_with_session(session_player)
585
586        assert sgp._attr_group_members == ["bathroom", "living_room", "offline"]
587
588    def test_align_members_leaves_order_untouched_without_a_live_session(self) -> None:
589        """Without a session (or with one holding none of our members) the order stands."""
590        mass = _make_mock_mass()
591        sgp = _make_sync_group(mass)
592
593        stranger = _make_mock_player("stranger", provider_domain="airplay")
594        stranger.group_members = ["stranger"]
595        mass.players.get_player = _player_lookup({"stranger": stranger})
596        sgp._attr_group_members = ["living_room", "bathroom"]
597
598        sgp._align_members_with_session(None)
599        assert sgp._attr_group_members == ["living_room", "bathroom"]
600
601        sgp._align_members_with_session(stranger)
602        assert sgp._attr_group_members == ["living_room", "bathroom"]
603
604
605class TestPowerLifecycle:
606    """Test that power(True/False) drives the group's form/dissolve lifecycle."""
607
608    @pytest.mark.asyncio
609    async def test_power_on_forms_group_and_picks_leader(self) -> None:
610        """power(True) should select a sync leader and mark the group powered."""
611        mass = _make_mock_mass()
612        sgp = _make_sync_group(mass)
613
614        leader = _make_mock_player("leader", provider_domain="sonos")
615        mass.players.get_player = _player_lookup({"leader": leader})
616        sgp._attr_group_members = ["leader"]
617
618        with patch.object(sgp, "update_state"):
619            await sgp.power(True)
620
621        assert sgp.sync_leader == leader
622        assert sgp._attr_powered is True
623
624    @pytest.mark.asyncio
625    async def test_power_on_with_no_members_stays_unformed(self) -> None:
626        """power(True) on an empty group should leave sync_leader as None but mark powered."""
627        mass = _make_mock_mass()
628        sgp = _make_sync_group(mass)
629
630        with patch.object(sgp, "update_state"):
631            await sgp.power(True)
632
633        assert sgp.sync_leader is None
634        # group is powered (intent to be active) even though no leader could be picked
635        assert sgp._attr_powered is True
636
637    @pytest.mark.asyncio
638    async def test_power_off_dissolves_group(self) -> None:
639        """power(False) should clear the sync leader and mark unpowered."""
640        mass = _make_mock_mass()
641        sgp = _make_sync_group(mass)
642
643        leader = _make_mock_player("leader", provider_domain="sonos")
644        mass.players.get_player = _player_lookup({"leader": leader})
645        sgp.sync_leader = leader
646        sgp._attr_group_members = ["leader"]
647        sgp._attr_powered = True
648
649        with patch.object(sgp, "update_state"):
650            await sgp.power(False)
651
652        # use getattr to defeat mypy's narrowing of these attributes after the
653        # earlier assignments, since it can't see that power(False) mutates them.
654        assert getattr(sgp, "sync_leader") is None  # noqa: B009
655        assert getattr(sgp, "_attr_powered") is False  # noqa: B009
656
657    @pytest.mark.asyncio
658    async def test_stop_does_not_dissolve_group(self) -> None:
659        """stop() should stop the leader but leave the group formed and powered."""
660        mass = _make_mock_mass()
661        sgp = _make_sync_group(mass)
662
663        leader = _make_mock_player("leader", provider_domain="sonos")
664        mass.players.get_player = _player_lookup({"leader": leader})
665        sgp.sync_leader = leader
666        sgp._attr_group_members = ["leader"]
667        sgp._attr_powered = True
668
669        await sgp.stop()
670
671        # leader was stopped via the internal handler
672        mass.players._handle_cmd_stop.assert_awaited_once_with("leader")
673        # but the group is still formed: sync_leader and powered are unchanged
674        assert sgp.sync_leader == leader
675        assert sgp._attr_powered is True
676
677
678class TestSetMembersDoesNotRegisterIncompatible:
679    """Regression test for: incompatible members must NOT be added to _attr_group_members."""
680
681    @pytest.mark.asyncio
682    async def test_incompatible_member_is_not_registered(self) -> None:
683        """A member that fails the can_group_with check must not be appended."""
684        mass = _make_mock_mass()
685        sgp = _make_sync_group(mass)
686
687        leader = _make_mock_player("leader", provider_domain="sonos")
688        # leader's can_group_with does NOT include the incompatible member
689        leader.state.can_group_with = {"leader"}
690        incompatible = _make_mock_player("incompatible", provider_domain="alien_protocol")
691
692        mass.players.get_player = _player_lookup({"leader": leader, "incompatible": incompatible})
693        sgp.sync_leader = leader
694        sgp._attr_group_members = ["leader"]
695
696        await sgp.set_members(player_ids_to_add=["incompatible"])
697
698        # incompatible must NOT linger in the internal member list
699        assert "incompatible" not in sgp._attr_group_members
700        # and the leader was never asked to add it (the call may still happen
701        # with empty lists since the member-changed path is taken, but the
702        # incompatible id must not appear in either add or remove)
703        for call in mass.players._handle_set_members.await_args_list:
704            assert "incompatible" not in (call.kwargs.get("player_ids_to_add") or [])
705            assert "incompatible" not in (call.kwargs.get("player_ids_to_remove") or [])
706
707    @pytest.mark.asyncio
708    async def test_compatible_member_is_registered_and_forwarded(self) -> None:
709        """A compatible member should be appended and forwarded to the leader."""
710        mass = _make_mock_mass()
711        sgp = _make_sync_group(mass)
712
713        leader = _make_mock_player("leader", provider_domain="sonos")
714        leader.state.can_group_with = {"compatible"}
715        compatible = _make_mock_player("compatible", provider_domain="sonos")
716
717        mass.players.get_player = _player_lookup({"leader": leader, "compatible": compatible})
718        sgp.sync_leader = leader
719        sgp._attr_group_members = ["leader"]
720
721        await sgp.set_members(player_ids_to_add=["compatible"])
722
723        assert "compatible" in sgp._attr_group_members
724        mass.players._handle_set_members.assert_awaited_once()
725        kwargs = mass.players._handle_set_members.await_args.kwargs
726        assert kwargs.get("player_ids_to_add") == ["compatible"]
727
728    @pytest.mark.asyncio
729    async def test_member_added_when_no_leader_yet(self) -> None:
730        """Adding to an empty/unformed group must register the member regardless."""
731        mass = _make_mock_mass()
732        sgp = _make_sync_group(mass)
733
734        member = _make_mock_player("member", provider_domain="sonos")
735        mass.players.get_player = _player_lookup({"member": member})
736
737        # no sync leader, empty group
738        sgp.sync_leader = None
739        sgp._attr_group_members = []
740
741        await sgp.set_members(player_ids_to_add=["member"])
742
743        # member is registered so a future _form_syncgroup can pick it as leader
744        assert "member" in sgp._attr_group_members
745        # but _handle_set_members on the leader is not called (no leader yet)
746        mass.players._handle_set_members.assert_not_awaited()
747
748
749class TestSetMembersRemovesExternalJoiner:
750    """Regression: a member joined to the leader outside MA must be removable."""
751
752    @pytest.mark.asyncio
753    async def test_external_joiner_removal_forwards_to_leader(self) -> None:
754        """
755        Removing an external joiner must forward straight to the sync leader.
756
757        A player grouped directly to the sync leader (e.g. via the Sonos app)
758        appears in group_members via the leader's live state but is not in the
759        group's tracked member list, so it must not be silently skipped.
760        """
761        mass = _make_mock_mass()
762        sgp = _make_sync_group(mass)
763
764        leader = _make_mock_player("leader", provider_domain="sonos")
765        # the leader's live native group includes an external joiner the group
766        # itself never registered
767        leader.state.group_members = ["leader", "external_x"]
768        external = _make_mock_player("external_x", provider_domain="sonos")
769
770        mass.players.get_player = _player_lookup({"leader": leader, "external_x": external})
771        sgp.sync_leader = leader
772        sgp._attr_group_members = ["leader"]
773
774        await sgp.set_members(player_ids_to_remove=["external_x"])
775
776        # the removal must be forwarded to the sync leader
777        mass.players._handle_set_members.assert_awaited_once()
778        call = mass.players._handle_set_members.await_args
779        assert call.args[0] is leader
780        assert call.kwargs.get("player_ids_to_remove") == ["external_x"]
781        # the group's tracked member list is untouched (external was never in it)
782        assert sgp._attr_group_members == ["leader"]
783
784    @pytest.mark.asyncio
785    async def test_tracked_member_removal_uses_managed_path(self) -> None:
786        """A member the group actually manages is removed via the managed path."""
787        mass = _make_mock_mass()
788        sgp = _make_sync_group(mass)
789
790        leader = _make_mock_player("leader", provider_domain="sonos")
791        leader.state.group_members = ["leader", "member_b"]
792        member_b = _make_mock_player("member_b", provider_domain="sonos")
793
794        mass.players.get_player = _player_lookup({"leader": leader, "member_b": member_b})
795        sgp.sync_leader = leader
796        sgp._attr_group_members = ["leader", "member_b"]
797
798        await sgp.set_members(player_ids_to_remove=["member_b"])
799
800        # tracked member is dropped from the internal list and forwarded
801        assert "member_b" not in sgp._attr_group_members
802        mass.players._handle_set_members.assert_awaited_once()
803        call = mass.players._handle_set_members.await_args
804        assert call.kwargs.get("player_ids_to_remove") == ["member_b"]
805
806
807def _make_sync_group_with_filters(
808    mass: MagicMock,
809    allowed_members: list[str] | None = None,
810) -> SyncGroupPlayer:
811    """Create a SyncGroupPlayer where the allowed_members filter can be configured."""
812    provider = MagicMock()
813    provider.domain = "sync_group"
814    provider.instance_id = "sync_group_test"
815    provider.name = "Sync Group"
816    provider.mass = mass
817
818    def _config_get_value(key: str, default: object = None) -> object:
819        if key == "dynamic_members":
820            return True
821        if key == "allowed_members":
822            return allowed_members or []
823        return default
824
825    mass.config.get_base_player_config.return_value = MagicMock(
826        name=None, default_name="Test Group", get_value=_config_get_value
827    )
828
829    sgp = SyncGroupPlayer(provider, "syncgroup_test")
830    sgp._cache.clear()
831    return sgp
832
833
834class TestPlayerFilters:
835    """Test the allowed_members filter semantics."""
836
837    def test_no_filter_allows_anything(self) -> None:
838        """Empty filter: any player is allowed."""
839        mass = _make_mock_mass()
840        sgp = _make_sync_group_with_filters(mass)
841        assert sgp._is_member_allowed("p1") is True
842
843    def test_allowed_members_restricts_to_listed(self) -> None:
844        """allowed_members populated: only listed IDs pass."""
845        mass = _make_mock_mass()
846        sgp = _make_sync_group_with_filters(mass, allowed_members=["p1"])
847        assert sgp._is_member_allowed("p1") is True
848        assert sgp._is_member_allowed("p2") is False
849
850    @pytest.mark.asyncio
851    async def test_set_members_rejects_filtered_player(self) -> None:
852        """set_members must skip a candidate that fails the filter."""
853        mass = _make_mock_mass()
854        sgp = _make_sync_group_with_filters(mass, allowed_members=["allowed"])
855
856        leader = _make_mock_player("leader", provider_domain="sonos")
857        leader.state.can_group_with = {"allowed", "blocked"}
858        allowed = _make_mock_player("allowed", provider_domain="sonos")
859        blocked = _make_mock_player("blocked", provider_domain="sonos")
860
861        mass.players.get_player = _player_lookup(
862            {"leader": leader, "allowed": allowed, "blocked": blocked}
863        )
864        sgp.sync_leader = leader
865        sgp._attr_group_members = ["leader"]
866
867        await sgp.set_members(player_ids_to_add=["allowed", "blocked"])
868
869        assert "allowed" in sgp._attr_group_members
870        assert "blocked" not in sgp._attr_group_members
871
872    def test_can_group_with_current_members_exempt_from_filter(self) -> None:
873        """
874        A current member not listed in allowed_members must still seed candidates.
875
876        Regression: filter must constrain JOINERS only, not gate the seed of
877        current members. Otherwise enabling a filter that doesn't list a current
878        member makes the candidate set empty (no joinable players show up).
879        """
880        mass = _make_mock_mass()
881        # current member is "leader", but the allow-list only lists "candidate".
882        sgp = _make_sync_group_with_filters(mass, allowed_members=["candidate"])
883
884        leader = _make_mock_player("leader", provider_domain="sonos")
885        leader.state.can_group_with = {"candidate", "outsider"}
886
887        candidate = _make_mock_player("candidate", provider_domain="sonos")
888        outsider = _make_mock_player("outsider", provider_domain="sonos")
889
890        mass.players.get_player = _player_lookup(
891            {"leader": leader, "candidate": candidate, "outsider": outsider}
892        )
893        sgp._attr_group_members = ["leader"]
894
895        result = sgp.can_group_with
896        # The current member (leader) is exempt and stays in the set.
897        assert "leader" in result
898        # An allowed candidate passes the filter.
899        assert "candidate" in result
900        # A player not in the allow-list is rejected.
901        assert "outsider" not in result
902
903
904class TestPresetMembersInDynamicGroup:
905    """In a dynamic group, configured members act as a preset, not as a lock."""
906
907    def _make_dynamic_group_with_preset(
908        self, mass: MagicMock, preset_members: list[str]
909    ) -> SyncGroupPlayer:
910        """Create a dynamic sync group whose CONF_GROUP_MEMBERS is the given preset."""
911        provider = MagicMock()
912        provider.domain = "sync_group"
913        provider.instance_id = "sync_group_test"
914        provider.name = "Sync Group"
915        provider.mass = mass
916
917        def _config_get_value(key: str, default: object = None) -> object:
918            if key == "dynamic_members":
919                return True
920            if key == "group_members":
921                return list(preset_members)
922            return default
923
924        mass.config.get_base_player_config.return_value = MagicMock(
925            name=None, default_name="Test Group", get_value=_config_get_value
926        )
927        sgp = SyncGroupPlayer(provider, "syncgroup_test")
928        sgp._cache.clear()
929        return sgp
930
931    @pytest.mark.asyncio
932    async def test_preset_member_can_be_unjoined(self) -> None:
933        """A preset member in a dynamic group can be unjoined for the session."""
934        mass = _make_mock_mass()
935        sgp = self._make_dynamic_group_with_preset(mass, ["leader", "preset_b"])
936        # Re-trigger on_config_updated so the new config is applied.
937        await sgp.on_config_updated()
938
939        leader = _make_mock_player("leader", provider_domain="sonos")
940        leader.state.can_group_with = {"preset_b"}
941        preset_b = _make_mock_player("preset_b", provider_domain="sonos")
942        mass.players.get_player = _player_lookup({"leader": leader, "preset_b": preset_b})
943        sgp.sync_leader = leader
944
945        await sgp.set_members(player_ids_to_remove=["preset_b"])
946
947        assert "preset_b" not in sgp._attr_group_members
948
949    @pytest.mark.asyncio
950    async def test_static_group_members_empty_in_dynamic_mode(self) -> None:
951        """In dynamic mode static_group_members is empty (frontend gating)."""
952        mass = _make_mock_mass()
953        sgp = self._make_dynamic_group_with_preset(mass, ["a", "b", "c"])
954        await sgp.on_config_updated()
955        assert sgp._attr_static_group_members == []
956        # but the configured members are seeded into group_members
957        assert set(sgp._attr_group_members) == {"a", "b", "c"}
958
959    @pytest.mark.asyncio
960    async def test_power_on_re_applies_preset_members(self) -> None:
961        """Powering on re-applies the configured preset, restoring any previously unjoined member."""
962        mass = _make_mock_mass()
963        sgp = self._make_dynamic_group_with_preset(mass, ["leader", "preset_b"])
964        await sgp.on_config_updated()
965
966        leader = _make_mock_player("leader", provider_domain="sonos")
967        preset_b = _make_mock_player("preset_b", provider_domain="sonos")
968        mass.players.get_player = _player_lookup({"leader": leader, "preset_b": preset_b})
969        sgp._attr_group_members = ["leader"]  # preset_b was unjoined in a previous session
970
971        with patch.object(sgp, "update_state"):
972            await sgp.power(True)
973
974        assert "preset_b" in sgp._attr_group_members
975
976    @pytest.mark.asyncio
977    async def test_form_syncgroup_does_not_re_add_unjoined_preset(self) -> None:
978        """Mid-session unjoins must stick: _form_syncgroup must NOT re-add preset members."""
979        mass = _make_mock_mass()
980        sgp = self._make_dynamic_group_with_preset(mass, ["leader", "preset_b"])
981        await sgp.on_config_updated()
982
983        leader = _make_mock_player("leader", provider_domain="sonos")
984        leader.state.group_members = ["leader"]
985        preset_b = _make_mock_player("preset_b", provider_domain="sonos")
986        mass.players.get_player = _player_lookup({"leader": leader, "preset_b": preset_b})
987        sgp.sync_leader = leader
988        sgp._attr_group_members = ["leader"]  # preset_b was unjoined during this session
989
990        with (
991            patch.object(sgp, "update_state"),
992            patch.object(sgp, "_translate_to_parent_ids", side_effect=lambda x: list(x)),
993        ):
994            await sgp._form_syncgroup()
995
996        assert "preset_b" not in sgp._attr_group_members
997
998    def test_preset_member_bypasses_allow_list_filter(self) -> None:
999        """
1000        A preset member that was unjoined must still pass the allow-list filter.
1001
1002        Regression: after unjoining a preset member, the user must be able to
1003        re-add it via the OSD even when allowed_members doesn't list it.
1004        """
1005        mass = _make_mock_mass()
1006        provider = MagicMock()
1007        provider.domain = "sync_group"
1008        provider.instance_id = "sync_group_test"
1009        provider.name = "Sync Group"
1010        provider.mass = mass
1011
1012        def _config_get_value(key: str, default: object = None) -> object:
1013            if key == "dynamic_members":
1014                return True
1015            if key == "group_members":
1016                return ["preset_a", "preset_b"]
1017            if key == "allowed_members":
1018                return ["other"]
1019            return default
1020
1021        mass.config.get_base_player_config.return_value = MagicMock(
1022            name=None, default_name="Test Group", get_value=_config_get_value
1023        )
1024        sgp = SyncGroupPlayer(provider, "syncgroup_test")
1025        sgp._cache.clear()
1026
1027        # preset members bypass the allow-list filter
1028        assert sgp._is_member_allowed("preset_a") is True
1029        assert sgp._is_member_allowed("preset_b") is True
1030        # the explicit allow-list entry still passes
1031        assert sgp._is_member_allowed("other") is True
1032        # an unrelated player is rejected
1033        assert sgp._is_member_allowed("outsider") is False
1034
1035    @pytest.mark.asyncio
1036    async def test_can_group_with_falls_back_when_all_members_offline(self) -> None:
1037        """
1038        An offline preset member must not block adding other players.
1039
1040        Regression: with only offline members in the list, can_group_with
1041        returned an empty set instead of offering compatible players, so
1042        nothing could be added to the group until the member came back.
1043        """
1044        mass = _make_mock_mass()
1045        sgp = self._make_dynamic_group_with_preset(mass, ["offline_member"])
1046        await sgp.on_config_updated()
1047
1048        offline_member = _make_mock_player("offline_member", available=False)
1049        candidate = _make_mock_player("candidate")
1050        candidate.type = PlayerType.PLAYER
1051        candidate.state.can_group_with = {"other"}
1052        candidate.state.active_group = None
1053        mass.players.get_player = _player_lookup(
1054            {"offline_member": offline_member, "candidate": candidate}
1055        )
1056        mass.players.iter_players = MagicMock(return_value=[candidate])
1057
1058        assert "candidate" in sgp.can_group_with
1059
1060    @pytest.mark.asyncio
1061    async def test_can_group_with_aggregates_from_available_members(self) -> None:
1062        """With at least one available member, candidates come from the members only."""
1063        mass = _make_mock_mass()
1064        sgp = self._make_dynamic_group_with_preset(mass, ["offline_member", "online_member"])
1065        await sgp.on_config_updated()
1066
1067        offline_member = _make_mock_player("offline_member", available=False)
1068        online_member = _make_mock_player("online_member")
1069        online_member.state.can_group_with = {"online_member", "friend"}
1070        mass.players.get_player = _player_lookup(
1071            {"offline_member": offline_member, "online_member": online_member}
1072        )
1073
1074        result = sgp.can_group_with
1075
1076        assert {"online_member", "friend"} <= result
1077        mass.players.iter_players.assert_not_called()
1078
1079
1080class TestGetConfigEntriesMemberPicker:
1081    """Test the member options offered in the group settings dropdown."""
1082
1083    @pytest.mark.asyncio
1084    async def test_slaved_follower_is_still_offered(self) -> None:
1085        """
1086        A synced follower must stay selectable in the member dropdown.
1087
1088        Regression: a member removed from the config during active playback
1089        remains slaved at the protocol level. Its can_group_with is then empty,
1090        and since it is no longer in the saved ids it vanished from the
1091        dropdown, making it impossible to re-add without stopping playback.
1092        """
1093        mass = _make_mock_mass()
1094        leader = _make_mock_player("leader")
1095        leader.type = PlayerType.PLAYER
1096        leader.state.can_group_with = {"follower"}
1097        # slaved follower: empty can_group_with while synced_to is set
1098        follower = _make_mock_player("follower")
1099        follower.type = PlayerType.PLAYER
1100        follower.state.synced_to = "leader"
1101        # idle player that cannot group with anything must NOT be offered
1102        solo = _make_mock_player("solo")
1103        solo.type = PlayerType.PLAYER
1104        mass.players.all_players = MagicMock(return_value=[leader, follower, solo])
1105        sgp = _make_sync_group(mass)
1106
1107        entries = await sgp.get_config_entries()
1108        members_entry = next(x for x in entries if x.key == CONF_GROUP_MEMBERS)
1109        option_ids = {option.value for option in members_entry.options}
1110
1111        assert option_ids == {"leader", "follower"}
1112
1113
1114class TestMembersFilterMigration:
1115    """Test the members_filter (exclusion) -> allowed_members (inclusion) migration."""
1116
1117    def _run_migrate(self, data: dict[str, Any]) -> dict[str, Any]:
1118        """Run the migration block against a copy of the provided data dict."""
1119        all_player_configs = data.get(CONF_PLAYERS, {})
1120        if isinstance(all_player_configs, dict):
1121            group_provider_domains = {"sync_group", "universal_group"}
1122            universe = {
1123                pid
1124                for pid, cfg in all_player_configs.items()
1125                if isinstance(cfg, dict) and cfg.get("provider") not in group_provider_domains
1126            }
1127            for player_cfg in all_player_configs.values():
1128                if not isinstance(player_cfg, dict):
1129                    continue
1130                if player_cfg.get("provider") != "sync_group":
1131                    continue
1132                values = player_cfg.setdefault("values", {})
1133                old_exclude = values.get("members_filter") or []
1134                if not old_exclude or values.get("allowed_members") is not None:
1135                    continue
1136                values["allowed_members"] = sorted(universe - set(old_exclude))
1137                values["members_filter"] = []
1138        return data
1139
1140    def test_inverts_exclusion_to_inclusion(self) -> None:
1141        """An exclusion list of [X] becomes an inclusion list of universe - {X}."""
1142        data = {
1143            CONF_PLAYERS: {
1144                "a": {"provider": "sonos", "player_id": "a"},
1145                "b": {"provider": "airplay", "player_id": "b"},
1146                "c": {"provider": "chromecast", "player_id": "c"},
1147                "syncgroup_1": {
1148                    "provider": "sync_group",
1149                    "player_id": "syncgroup_1",
1150                    "values": {"members_filter": ["b"]},
1151                },
1152            }
1153        }
1154        out = self._run_migrate(data)
1155        sg = out[CONF_PLAYERS]["syncgroup_1"]["values"]
1156        assert sg["allowed_members"] == ["a", "c"]
1157        assert sg["members_filter"] == []
1158
1159    def test_idempotent_when_already_migrated(self) -> None:
1160        """Re-running migration on already-migrated config is a no-op."""
1161        data = {
1162            CONF_PLAYERS: {
1163                "a": {"provider": "sonos", "player_id": "a"},
1164                "syncgroup_1": {
1165                    "provider": "sync_group",
1166                    "player_id": "syncgroup_1",
1167                    "values": {"allowed_members": ["a"], "members_filter": []},
1168                },
1169            }
1170        }
1171        out = self._run_migrate(data)
1172        sg = out[CONF_PLAYERS]["syncgroup_1"]["values"]
1173        assert sg["allowed_members"] == ["a"]
1174        assert sg["members_filter"] == []
1175
1176    def test_skips_when_no_exclusion_list(self) -> None:
1177        """A sync_group without a members_filter is left untouched."""
1178        data = {
1179            CONF_PLAYERS: {
1180                "syncgroup_1": {
1181                    "provider": "sync_group",
1182                    "player_id": "syncgroup_1",
1183                    "values": {},
1184                },
1185            }
1186        }
1187        out = self._run_migrate(data)
1188        sg = out[CONF_PLAYERS]["syncgroup_1"]["values"]
1189        assert "allowed_members" not in sg
1190
1191    def test_excludes_group_providers_from_universe(self) -> None:
1192        """Group-providing players (sync_group/universal_group) are not in the universe."""
1193        data = {
1194            CONF_PLAYERS: {
1195                "a": {"provider": "sonos", "player_id": "a"},
1196                "ug": {"provider": "universal_group", "player_id": "ug"},
1197                "sg": {"provider": "sync_group", "player_id": "sg"},
1198                "syncgroup_1": {
1199                    "provider": "sync_group",
1200                    "player_id": "syncgroup_1",
1201                    "values": {"members_filter": ["a"]},
1202                },
1203            }
1204        }
1205        out = self._run_migrate(data)
1206        sg = out[CONF_PLAYERS]["syncgroup_1"]["values"]
1207        # Universe = {a} (sg/ug/syncgroup_1 are group providers excluded).
1208        # Inversion removes a, so allowed_members is empty.
1209        assert sg["allowed_members"] == []
1210
1211
1212class TestIsActiveSession:
1213    """The is_active_session property gates whether children report active_group."""
1214
1215    def test_dormant_group_is_not_active(self) -> None:
1216        """A freshly-initialized group has no leader and no grace timer."""
1217        mass = _make_mock_mass()
1218        sgp = _make_sync_group(mass)
1219        assert sgp.is_active_session is False
1220
1221    def test_group_with_sync_leader_is_active(self) -> None:
1222        """A group with a sync leader is considered active even without playback."""
1223        mass = _make_mock_mass()
1224        sgp = _make_sync_group(mass)
1225        sgp.sync_leader = _make_mock_player("leader")
1226        assert sgp.is_active_session is True
1227
1228    def test_group_in_grace_window_is_active(self) -> None:
1229        """While the idle-grace task is pending, the group still claims captured members."""
1230        mass = _make_mock_mass()
1231        sgp = _make_sync_group(mass)
1232        # simulate a pending grace task without actually running asyncio
1233        sgp._idle_grace_task = MagicMock()
1234        assert sgp.is_active_session is True
1235
1236
1237class TestPowerlessLifecycle:
1238    """Form-on-play / deform-on-stop semantics when no power control is assigned."""
1239
1240    @pytest.mark.asyncio
1241    async def test_play_media_forms_group_when_dormant(self) -> None:
1242        """play_media on a dormant group should select a leader and forward to it."""
1243        mass = _make_mock_mass()
1244        sgp = _make_sync_group(mass)
1245        leader = _make_mock_player("leader", provider_domain="sonos")
1246        mass.players.get_player = _player_lookup({"leader": leader})
1247        sgp._attr_group_members = ["leader"]
1248        assert sgp.sync_leader is None
1249
1250        with patch.object(sgp, "update_state"):
1251            await sgp.play_media(MagicMock(source_id="src", uri="x"))
1252
1253        assert sgp.sync_leader == leader
1254        # mypy narrows sync_leader to None from the earlier assert; the
1255        # play_media call is what re-populates it, but mypy can't track that
1256        # mutation through the patch context — hence the ignore.
1257        mass.players._handle_play_media.assert_awaited_once()  # type: ignore[unreachable]
1258
1259    @pytest.mark.asyncio
1260    async def test_stop_dissolves_group_when_powerless(self) -> None:
1261        """stop() on a group without power control should dissolve the session."""
1262        mass = _make_mock_mass()
1263        sgp = _make_sync_group(mass)
1264        leader = _make_mock_player("leader", provider_domain="sonos")
1265        mass.players.get_player = _player_lookup({"leader": leader})
1266        sgp.sync_leader = leader
1267        sgp._attr_group_members = ["leader"]
1268        # _attr_powered is None by default (no power control) so stop dissolves
1269        assert sgp._attr_powered is None
1270
1271        with patch.object(sgp, "update_state"):
1272            await sgp.stop()
1273
1274        # leader was stopped via the internal handler ...
1275        mass.players._handle_cmd_stop.assert_any_await("leader")
1276        # ... and the session was dissolved (sync_leader cleared)
1277        assert sgp.sync_leader is None
1278
1279    @pytest.mark.asyncio
1280    async def test_stop_preserves_group_when_pinned_with_fake_power(self) -> None:
1281        """stop() should NOT dissolve when the user has pinned the group on."""
1282        mass = _make_mock_mass()
1283        sgp = _make_sync_group(mass)
1284        leader = _make_mock_player("leader", provider_domain="sonos")
1285        mass.players.get_player = _player_lookup({"leader": leader})
1286        sgp.sync_leader = leader
1287        sgp._attr_group_members = ["leader"]
1288        sgp._attr_powered = True  # user explicitly pinned via Fake control
1289
1290        await sgp.stop()
1291
1292        mass.players._handle_cmd_stop.assert_any_await("leader")
1293        # session stays intact
1294        assert sgp.sync_leader == leader
1295
1296
1297class TestIdleGraceTimer:
1298    """The idle-grace timer absorbs natural PLAYING→IDLE transitions."""
1299
1300    def test_grace_scheduled_on_playing_to_idle(self) -> None:
1301        """_update_attributes schedules a grace task when the leader transitions to IDLE."""
1302        mass = _make_mock_mass()
1303        # create_task should return a sentinel we can inspect
1304        sentinel = MagicMock()
1305        mass.create_task = MagicMock(return_value=sentinel)
1306        sgp = _make_sync_group(mass)
1307        leader = _make_mock_player("leader", playback_state=PlaybackState.IDLE)
1308        sgp.sync_leader = leader
1309        # previous state was PLAYING
1310        sgp._attr_playback_state = PlaybackState.PLAYING
1311        sgp._attr_powered = None  # no pin → debounce applies
1312
1313        sgp._update_attributes()
1314
1315        assert sgp._idle_grace_task is sentinel
1316        mass.create_task.assert_called_once()
1317
1318    def test_grace_not_scheduled_when_pinned(self) -> None:
1319        """No grace task when the user has pinned the group with Fake power."""
1320        mass = _make_mock_mass()
1321        mass.create_task = MagicMock()
1322        sgp = _make_sync_group(mass)
1323        leader = _make_mock_player("leader", playback_state=PlaybackState.IDLE)
1324        sgp.sync_leader = leader
1325        sgp._attr_playback_state = PlaybackState.PLAYING
1326        sgp._attr_powered = True  # explicit pin
1327
1328        sgp._update_attributes()
1329
1330        assert sgp._idle_grace_task is None
1331        mass.create_task.assert_not_called()
1332
1333    def test_grace_cancelled_on_resume(self) -> None:
1334        """A pending grace task is cancelled when the leader resumes playback."""
1335        mass = _make_mock_mass()
1336        mass.create_task = MagicMock()
1337        sgp = _make_sync_group(mass)
1338        leader = _make_mock_player("leader", playback_state=PlaybackState.PLAYING)
1339        sgp.sync_leader = leader
1340        # pretend a grace task is pending
1341        prior_task = MagicMock()
1342        prior_task.done.return_value = False
1343        sgp._idle_grace_task = prior_task
1344        sgp._attr_playback_state = PlaybackState.IDLE
1345
1346        sgp._update_attributes()
1347
1348        prior_task.cancel.assert_called_once()
1349        assert sgp._idle_grace_task is None
1350
1351    def test_grace_cancelled_on_explicit_stop(self) -> None:
1352        """An explicit stop cancels the pending grace task immediately."""
1353        mass = _make_mock_mass()
1354        sgp = _make_sync_group(mass)
1355        leader = _make_mock_player("leader")
1356        mass.players.get_player = _player_lookup({"leader": leader})
1357        sgp.sync_leader = leader
1358        prior_task = MagicMock()
1359        prior_task.done.return_value = False
1360        sgp._idle_grace_task = prior_task
1361
1362        async def _run() -> None:
1363            with patch.object(sgp, "update_state"):
1364                await sgp.stop()
1365
1366        import asyncio  # noqa: PLC0415
1367
1368        asyncio.run(_run())
1369
1370        prior_task.cancel.assert_called_once()
1371        assert sgp._idle_grace_task is None
1372
1373
1374class TestFormWaitsForLeaderUnsynced:
1375    """The form path must wait for a stale leader to report synced_to=None."""
1376
1377    @pytest.mark.asyncio
1378    async def test_form_waits_when_leader_still_synced(self) -> None:
1379        """If the new leader still reports synced_to, we wait before proceeding."""
1380        mass = _make_mock_mass()
1381        sgp = _make_sync_group(mass)
1382        leader = _make_mock_player("leader", provider_domain="sonos")
1383        # leader reports stale synced_to
1384        leader.state.synced_to = "old_leader"
1385        # but make _wait_member_unsynced succeed (clears state) so the form proceeds
1386        mass.players.get_player = _player_lookup({"leader": leader})
1387        sgp._attr_group_members = ["leader"]
1388
1389        async def fake_wait(
1390            _member_id: str,
1391            _timeout: float = 5.0,
1392        ) -> bool:
1393            leader.state.synced_to = None  # simulate provider catching up
1394            return True
1395
1396        with (
1397            patch.object(sgp, "update_state"),
1398            patch.object(sgp, "_wait_member_unsynced", side_effect=fake_wait) as wait_mock,
1399        ):
1400            await sgp._form_syncgroup()
1401
1402        wait_mock.assert_awaited_once_with("leader")
1403        assert sgp.sync_leader == leader
1404
1405    @pytest.mark.asyncio
1406    async def test_form_aborts_when_leader_stuck(self) -> None:
1407        """If the wait returns False (leader genuinely stuck), abort the form."""
1408        mass = _make_mock_mass()
1409        sgp = _make_sync_group(mass)
1410        leader = _make_mock_player("leader", provider_domain="sonos")
1411        leader.state.synced_to = "old_leader"
1412        mass.players.get_player = _player_lookup({"leader": leader})
1413        sgp._attr_group_members = ["leader"]
1414
1415        with (
1416            patch.object(sgp, "update_state"),
1417            patch.object(sgp, "_wait_member_unsynced", return_value=False),
1418        ):
1419            await sgp._form_syncgroup()
1420
1421        # form must NOT have left a sync_leader set, so the caller's play_media
1422        # won't be issued against a stuck player (the original Poolhouse race).
1423        assert sgp.sync_leader is None
1424
1425    @pytest.mark.asyncio
1426    async def test_form_aborts_when_leader_cleared_during_wait(self) -> None:
1427        """If a concurrent dissolve clears sync_leader during the wait, abort the stale form."""
1428        mass = _make_mock_mass()
1429        sgp = _make_sync_group(mass)
1430        leader = _make_mock_player("leader", provider_domain="sonos")
1431        leader.state.synced_to = "old_leader"
1432        member = _make_mock_player("m2", provider_domain="sonos")
1433        mass.players.get_player = _player_lookup({"leader": leader, "m2": member})
1434        sgp._attr_group_members = ["leader", "m2"]
1435
1436        async def fake_wait(_member_id: str, _timeout: float = 5.0) -> bool:
1437            leader.state.synced_to = None
1438            sgp.sync_leader = None  # simulate a concurrent dissolve while waiting
1439            return True
1440
1441        with (
1442            patch.object(sgp, "update_state"),
1443            patch.object(sgp, "_wait_member_unsynced", side_effect=fake_wait),
1444        ):
1445            await sgp._form_syncgroup()
1446
1447        # the stale form attempt must not (re)sync any members
1448        mass.players._handle_set_members.assert_not_awaited()
1449        assert sgp.sync_leader is None
1450
1451
1452class TestWaitMemberUnsynced:
1453    """The helper that waits for a member's synced_to to clear (with recovery)."""
1454
1455    @pytest.mark.asyncio
1456    async def test_returns_true_when_already_unsynced(self) -> None:
1457        """Member already reports synced_to=None → return True without recovery."""
1458        mass = _make_mock_mass()
1459        sgp = _make_sync_group(mass)
1460        member = _make_mock_player("m1")
1461        member.synced_to = None
1462        mass.players.get_player = _player_lookup({"m1": member})
1463        mass.players.cmd_ungroup = AsyncMock()
1464
1465        ok = await sgp._wait_member_unsynced("m1")
1466
1467        assert ok is True
1468        mass.players.cmd_ungroup.assert_not_awaited()
1469
1470    @pytest.mark.asyncio
1471    async def test_attempts_recovery_when_stuck(self) -> None:
1472        """If the first wait doesn't clear it, kick the member from its stale parent directly."""
1473        mass = _make_mock_mass()
1474        sgp = _make_sync_group(mass)
1475        member = _make_mock_player("m1")
1476        member.synced_to = "old_leader"  # still stale after first wait
1477        old_leader = _make_mock_player("old_leader")
1478        mass.players.get_player = _player_lookup({"m1": member, "old_leader": old_leader})
1479
1480        # _handle_set_members on the stale parent is the expected recovery action.
1481        # We make it succeed by side-effect-clearing synced_to.
1482        async def _kick(_parent: Any, player_ids_to_remove: list[str]) -> None:
1483            assert player_ids_to_remove == ["m1"]
1484            member.synced_to = None
1485
1486        mass.players._handle_set_members = AsyncMock(side_effect=_kick)
1487        mass.players.cmd_ungroup = AsyncMock()
1488
1489        ok = await sgp._wait_member_unsynced("m1")
1490
1491        assert ok is True
1492        mass.players._handle_set_members.assert_awaited_once_with(
1493            old_leader, player_ids_to_remove=["m1"]
1494        )
1495        mass.players.cmd_ungroup.assert_not_awaited()
1496
1497    @pytest.mark.asyncio
1498    async def test_returns_false_when_genuinely_stuck(self) -> None:
1499        """If recovery doesn't help either, return False so callers can abort."""
1500        mass = _make_mock_mass()
1501        sgp = _make_sync_group(mass)
1502        member = _make_mock_player("m1")
1503        member.synced_to = "old_leader"  # stays stuck even after recovery
1504        old_leader = _make_mock_player("old_leader")
1505        mass.players.get_player = _player_lookup({"m1": member, "old_leader": old_leader})
1506        mass.players._handle_set_members = AsyncMock()  # no-op: state stays stale
1507
1508        ok = await sgp._wait_member_unsynced("m1")
1509
1510        assert ok is False
1511        mass.players._handle_set_members.assert_awaited_once_with(
1512            old_leader, player_ids_to_remove=["m1"]
1513        )
1514
1515    @pytest.mark.asyncio
1516    async def test_returns_false_when_stale_parent_gone(self) -> None:
1517        """If the stale parent no longer exists, skip the kick and report stuck."""
1518        mass = _make_mock_mass()
1519        sgp = _make_sync_group(mass)
1520        member = _make_mock_player("m1")
1521        member.synced_to = "old_leader"  # parent is no longer registered
1522        mass.players.get_player = _player_lookup({"m1": member})
1523
1524        ok = await sgp._wait_member_unsynced("m1")
1525
1526        assert ok is False
1527        mass.players._handle_set_members.assert_not_awaited()
1528
1529
1530class TestSupportedFeaturesPower:
1531    """POWER feature is only advertised when the user opts in via Fake control."""
1532
1533    def test_power_not_in_base_features_by_default(self) -> None:
1534        """No power_control config → POWER feature not advertised."""
1535        mass = _make_mock_mass()
1536        # default config: no CONF_POWER_CONTROL
1537        mass.config.get_raw_player_config_value = MagicMock(return_value=None)
1538        sgp = _make_sync_group(mass)
1539        assert PlayerFeature.POWER not in sgp.supported_features
1540
1541    def test_power_advertised_when_fake_control_assigned(self) -> None:
1542        """User assigns Fake power control → POWER feature shows up."""
1543        mass = _make_mock_mass()
1544
1545        # CONF_POWER_CONTROL is the only key we care about; for everything
1546        # else return whatever the default config helper returns.
1547        def _get_raw(_player_id: str, key: str, default: object = None) -> object:
1548            if key == "power_control":
1549                return "fake"
1550            return default
1551
1552        mass.config.get_raw_player_config_value = MagicMock(side_effect=_get_raw)
1553        sgp = _make_sync_group(mass)
1554        assert PlayerFeature.POWER in sgp.supported_features
1555
1556
1557def _recording_wait(order: list[str]) -> MagicMock:
1558    """
1559    Build a wait_for_player_update replacement that records call/enter/exit order.
1560
1561    The returned mock records ``wait_for:<player_id>:<value>`` when invoked,
1562    ``subscribe`` on context enter and ``await`` on context exit, so a test can
1563    assert that the playback start was wrapped (subscribe before the command is
1564    issued) rather than awaited after the fact.
1565    """
1566
1567    class _RecordingWait:
1568        async def __aenter__(self) -> None:
1569            order.append("subscribe")
1570
1571        async def __aexit__(self, *_exc: object) -> bool:
1572            order.append("await")
1573            return False
1574
1575    def _wait(player_id: str, **kwargs: Any) -> _RecordingWait:
1576        order.append(f"wait_for:{player_id}:{kwargs.get('attribute_value')}")
1577        return _RecordingWait()
1578
1579    return MagicMock(side_effect=_wait)
1580
1581
1582class TestLeaderPlaybackAwaited:
1583    """A (re)form must not return until the leader confirms it has started playing."""
1584
1585    @pytest.mark.asyncio
1586    async def test_play_waits_for_leader_before_returning(self) -> None:
1587        """play() wraps the resume in a wait so the group lock isn't released early."""
1588        mass = _make_mock_mass()
1589        sgp = _make_sync_group(mass)
1590        leader = _make_mock_player("leader", playback_state=PlaybackState.IDLE)
1591        mass.players.get_player = _player_lookup({"leader": leader})
1592        sgp.sync_leader = leader
1593        sgp._attr_group_members = ["leader"]
1594
1595        order: list[str] = []
1596        mass.players.wait_for_player_update = _recording_wait(order)
1597        mass.players.cmd_resume = AsyncMock(side_effect=lambda *_a, **_k: order.append("resume"))
1598
1599        with patch.object(sgp, "_form_syncgroup", new=AsyncMock()):
1600            await sgp.play()
1601
1602        # subscribe happens before the resume command, and the wait completes
1603        # after — i.e. the start is wrapped, not fire-and-forget.
1604        assert order == [
1605            f"wait_for:leader:{PlaybackState.PLAYING}",
1606            "subscribe",
1607            "resume",
1608            "await",
1609        ]
1610
1611    @pytest.mark.asyncio
1612    async def test_play_media_waits_for_leader_before_returning(self) -> None:
1613        """play_media() wraps the leader start so a concurrent (un)group can't race it."""
1614        mass = _make_mock_mass()
1615        sgp = _make_sync_group(mass)
1616        leader = _make_mock_player("leader", playback_state=PlaybackState.IDLE)
1617        mass.players.get_player = _player_lookup({"leader": leader})
1618        sgp.sync_leader = leader
1619        sgp._attr_group_members = ["leader"]
1620
1621        order: list[str] = []
1622        mass.players.wait_for_player_update = _recording_wait(order)
1623        mass.players._handle_play_media = AsyncMock(
1624            side_effect=lambda *_a, **_k: order.append("play_media")
1625        )
1626
1627        media = MagicMock()
1628        media.source_id = "syncgroup_test"
1629        with patch.object(sgp, "_form_syncgroup", new=AsyncMock()):
1630            await sgp.play_media(media)
1631
1632        assert order == [
1633            f"wait_for:leader:{PlaybackState.PLAYING}",
1634            "subscribe",
1635            "play_media",
1636            "await",
1637        ]
1638
1639    @pytest.mark.asyncio
1640    async def test_no_wait_when_group_has_no_leader(self) -> None:
1641        """With no leader to wait on, play() resumes without arming a playback wait."""
1642        mass = _make_mock_mass()
1643        sgp = _make_sync_group(mass)
1644        sgp.sync_leader = None
1645
1646        mass.players.cmd_resume = AsyncMock()
1647
1648        with patch.object(sgp, "_form_syncgroup", new=AsyncMock()):
1649            await sgp.play()
1650
1651        mass.players.wait_for_player_update.assert_not_called()
1652        mass.players.cmd_resume.assert_awaited_once()
1653
1654
1655class TestPlaybackStartMarker:
1656    """A just-issued playback start counts as playing for group-command decisions."""
1657
1658    def _setup_group(self, mass: MagicMock, group_state: PlaybackState) -> SyncGroupPlayer:
1659        sgp = _make_sync_group(mass)
1660        # a provider without dynamic leader switching, so leader removal takes the
1661        # dissolve+reform path
1662        leader = _make_mock_player("leader", provider_domain="wiim")
1663        other = _make_mock_player("other", provider_domain="wiim")
1664        mass.players.get_player = _player_lookup({"leader": leader, "other": other})
1665        sgp.sync_leader = leader
1666        sgp._attr_group_members = ["leader", "other"]
1667        sgp._attr_playback_state = group_state
1668        # close scheduled background coroutines (idle-grace) instead of leaking them
1669        mass.create_task = MagicMock(side_effect=lambda coro, *_a, **_k: coro.close())
1670        return sgp
1671
1672    @pytest.mark.asyncio
1673    async def test_leader_removal_during_startup_window_resumes(self) -> None:
1674        """
1675        Leader removal right after a start command must still resume.
1676
1677        Devices report transient states (a false PLAYING, bounces through IDLE)
1678        while a group session starts, so the group state alone misreads the
1679        startup window as idle and would skip the post-reform resume.
1680        """
1681        mass = _make_mock_mass()
1682        sgp = self._setup_group(mass, group_state=PlaybackState.IDLE)
1683        sgp._playback_start_at = time.monotonic()
1684
1685        with (
1686            patch.object(sgp, "_dissolve_and_reform", new=AsyncMock()) as reform,
1687            patch.object(sgp, "_dynamic_leader_switch", new=AsyncMock()) as dyn_switch,
1688        ):
1689            await sgp.set_members(player_ids_to_remove=["leader"])
1690
1691        dyn_switch.assert_not_awaited()
1692        reform.assert_awaited_once()
1693        assert reform.await_args is not None
1694        assert reform.await_args.kwargs.get("resume_playback") is True
1695
1696    @pytest.mark.asyncio
1697    async def test_paused_group_never_resumes_despite_recent_start(self) -> None:
1698        """PAUSED is deliberate user intent and always wins over the startup marker."""
1699        mass = _make_mock_mass()
1700        sgp = self._setup_group(mass, group_state=PlaybackState.PAUSED)
1701        sgp._playback_start_at = time.monotonic()
1702
1703        with (
1704            patch.object(sgp, "_dissolve_and_reform", new=AsyncMock()) as reform,
1705            patch.object(sgp, "_dynamic_leader_switch", new=AsyncMock()),
1706        ):
1707            await sgp.set_members(player_ids_to_remove=["leader"])
1708
1709        reform.assert_awaited_once()
1710        assert reform.await_args is not None
1711        assert reform.await_args.kwargs.get("resume_playback") is False
1712
1713    @pytest.mark.asyncio
1714    async def test_await_leader_playback_stamps_the_marker(self) -> None:
1715        """The playback wait stamps the start so decision logic can see the window."""
1716        mass = _make_mock_mass()
1717        sgp = _make_sync_group(mass)
1718        sgp.sync_leader = _make_mock_player("leader")
1719        assert sgp._playback_recently_started is False
1720
1721        async with sgp._await_leader_playback():
1722            pass
1723
1724        assert sgp._playback_recently_started is True
1725
1726    @pytest.mark.asyncio
1727    async def test_stop_voids_the_marker(self) -> None:
1728        """An explicit stop clears the marker so a later unjoin won't resume."""
1729        mass = _make_mock_mass()
1730        sgp = _make_sync_group(mass)
1731        leader = _make_mock_player("leader")
1732        mass.players.get_player = _player_lookup({"leader": leader})
1733        sgp.sync_leader = leader
1734        sgp._playback_start_at = time.monotonic()
1735
1736        with patch.object(sgp, "update_state"):
1737            await sgp.stop()
1738
1739        assert sgp._playback_recently_started is False
1740
1741
1742class TestDebouncedReform:
1743    """Leader removal re-forms debounced so cascaded unjoins coalesce."""
1744
1745    def _setup_group(self, mass: MagicMock, members: list[str]) -> SyncGroupPlayer:
1746        sgp = _make_sync_group(mass)
1747        players = {
1748            member_id: _make_mock_player(member_id, provider_domain="wiim") for member_id in members
1749        }
1750        mass.players.get_player = _player_lookup(players)
1751        sgp.sync_leader = players[members[0]]
1752        sgp._attr_group_members = list(members)
1753        sgp._attr_playback_state = PlaybackState.PLAYING
1754        # run scheduled coroutines for real so the debounced re-form actually fires;
1755        # eager_start mirrors mass.create_task so the runner enters its sleep at
1756        # schedule time (a later cancel then always lands inside the body)
1757        mass.create_task = MagicMock(
1758            side_effect=lambda coro, *_a, **_k: asyncio.Task(
1759                coro, loop=asyncio.get_running_loop(), eager_start=True
1760            )
1761        )
1762        return sgp
1763
1764    @pytest.mark.asyncio
1765    async def test_leader_removal_schedules_debounced_reform(self) -> None:
1766        """The stop/dissolve happens immediately; the re-form (with resume) is debounced."""
1767        mass = _make_mock_mass()
1768        sgp = self._setup_group(mass, members=["leader", "m2"])
1769
1770        with (
1771            patch("music_assistant.providers.sync_group.player.REFORM_DEBOUNCE_SECONDS", 0.05),
1772            patch.object(sgp, "update_state"),
1773            patch.object(sgp, "play", new=AsyncMock()) as play,
1774        ):
1775            await sgp.set_members(player_ids_to_remove=["leader"])
1776
1777            # immediate half: leader stopped, group dissolved, re-form pending
1778            mass.players._handle_cmd_stop.assert_any_await("leader")
1779            assert sgp.sync_leader is None
1780            task = sgp._reform_task
1781            assert task is not None
1782            play.assert_not_awaited()
1783            # the group keeps claiming its members while the re-form is pending
1784            assert sgp.is_active_session is True
1785
1786            await task
1787
1788        play.assert_awaited_once()
1789        assert sgp._reform_task is None
1790
1791    @pytest.mark.asyncio
1792    async def test_cascaded_unjoins_coalesce_into_single_reform(self) -> None:
1793        """A second unjoin within the window re-arms it: one re-form, final member list."""
1794        mass = _make_mock_mass()
1795        sgp = self._setup_group(mass, members=["leader", "m2", "m3"])
1796
1797        with (
1798            patch("music_assistant.providers.sync_group.player.REFORM_DEBOUNCE_SECONDS", 0.05),
1799            patch.object(sgp, "update_state"),
1800            patch.object(sgp, "play", new=AsyncMock()) as play,
1801        ):
1802            await sgp.set_members(player_ids_to_remove=["leader"])
1803            first_task = sgp._reform_task
1804            assert first_task is not None
1805
1806            # second unjoin lands inside the debounce window
1807            await sgp.set_members(player_ids_to_remove=["m2"])
1808            second_task = sgp._reform_task
1809            assert second_task is not None
1810            assert second_task is not first_task
1811
1812            await asyncio.gather(first_task, second_task)
1813
1814        play.assert_awaited_once()
1815        assert sgp._attr_group_members == ["m3"]
1816
1817    @pytest.mark.asyncio
1818    async def test_explicit_stop_cancels_pending_reform(self) -> None:
1819        """An explicit stop during the window means silence — no resume may follow."""
1820        mass = _make_mock_mass()
1821        sgp = self._setup_group(mass, members=["leader", "m2"])
1822
1823        with (
1824            patch("music_assistant.providers.sync_group.player.REFORM_DEBOUNCE_SECONDS", 0.05),
1825            patch.object(sgp, "update_state"),
1826            patch.object(sgp, "play", new=AsyncMock()) as play,
1827        ):
1828            await sgp.set_members(player_ids_to_remove=["leader"])
1829            task = sgp._reform_task
1830            assert task is not None
1831
1832            await sgp.stop()
1833            assert sgp._reform_task is None
1834            await task
1835
1836        play.assert_not_awaited()
1837        assert sgp.is_active_session is False
1838
1839    @pytest.mark.asyncio
1840    async def test_reform_aborts_when_all_members_left(self) -> None:
1841        """When the window drains the whole group, the re-form fires as a no-op."""
1842        mass = _make_mock_mass()
1843        sgp = self._setup_group(mass, members=["leader", "m2"])
1844
1845        with (
1846            patch("music_assistant.providers.sync_group.player.REFORM_DEBOUNCE_SECONDS", 0.05),
1847            patch.object(sgp, "update_state"),
1848            patch.object(sgp, "play", new=AsyncMock()) as play,
1849        ):
1850            await sgp.set_members(player_ids_to_remove=["leader"])
1851            await sgp.set_members(player_ids_to_remove=["m2"])
1852            task = sgp._reform_task
1853            assert task is not None
1854
1855            await task
1856
1857        play.assert_not_awaited()
1858        assert sgp._reform_task is None
1859        assert sgp.is_active_session is False
1860
1861    @pytest.mark.asyncio
1862    async def test_explicit_play_supersedes_pending_reform(self) -> None:
1863        """A user play during the window forms right away; the pending re-form is dropped."""
1864        mass = _make_mock_mass()
1865        sgp = self._setup_group(mass, members=["leader", "m2"])
1866        mass.players.cmd_resume = AsyncMock()
1867
1868        with (
1869            patch("music_assistant.providers.sync_group.player.REFORM_DEBOUNCE_SECONDS", 0.05),
1870            patch.object(sgp, "update_state"),
1871        ):
1872            await sgp.set_members(player_ids_to_remove=["leader"])
1873            task = sgp._reform_task
1874            assert task is not None
1875
1876            await sgp.play()
1877            assert sgp._reform_task is None
1878            await task
1879
1880        mass.players.cmd_resume.assert_awaited_once()
1881        assert sgp.sync_leader is not None
1882        assert sgp.sync_leader.player_id == "m2"
1883
1884    @pytest.mark.asyncio
1885    async def test_reform_survives_cancel_from_its_own_form(self) -> None:
1886        """
1887        The firing re-form must not cancel itself.
1888
1889        The runner resumes via play() -> _form_syncgroup, which cancels any pending
1890        re-form timer — including, without the identity guard, the very task that
1891        is executing it.
1892        """
1893        mass = _make_mock_mass()
1894        sgp = self._setup_group(mass, members=["leader", "m2"])
1895        mass.players.cmd_resume = AsyncMock()
1896
1897        with (
1898            patch("music_assistant.providers.sync_group.player.REFORM_DEBOUNCE_SECONDS", 0.05),
1899            patch.object(sgp, "update_state"),
1900        ):
1901            await sgp.set_members(player_ids_to_remove=["leader"])
1902            task = sgp._reform_task
1903            assert task is not None
1904
1905            await task
1906
1907        mass.players.cmd_resume.assert_awaited_once()
1908        assert sgp.sync_leader is not None
1909        assert sgp.sync_leader.player_id == "m2"
1910        assert sgp._reform_task is None
1911
1912
1913class TestReformResumeGate:
1914    """Removing the sync leader resumes playback if (and only if) the group was playing."""
1915
1916    def _setup_group(self, mass: MagicMock, group_state: PlaybackState) -> SyncGroupPlayer:
1917        sgp = _make_sync_group(mass)
1918        # a provider without dynamic leader switching, so leader removal takes the
1919        # dissolve+reform path
1920        leader = _make_mock_player("leader", provider_domain="wiim")
1921        other = _make_mock_player("other", provider_domain="wiim")
1922        mass.players.get_player = _player_lookup({"leader": leader, "other": other})
1923        sgp.sync_leader = leader
1924        sgp._attr_group_members = ["leader", "other"]
1925        sgp._attr_playback_state = group_state
1926        # close scheduled background coroutines (idle-grace) instead of leaking them
1927        mass.create_task = MagicMock(side_effect=lambda coro, *_a, **_k: coro.close())
1928        return sgp
1929
1930    @pytest.mark.asyncio
1931    async def test_leader_removal_resumes_when_playing(self) -> None:
1932        """Leader removal on a playing group must reform with a resume."""
1933        mass = _make_mock_mass()
1934        sgp = self._setup_group(mass, group_state=PlaybackState.PLAYING)
1935
1936        with (
1937            patch.object(sgp, "_dissolve_and_reform", new=AsyncMock()) as reform,
1938            patch.object(sgp, "_dynamic_leader_switch", new=AsyncMock()) as dyn_switch,
1939            patch.object(sgp, "_active_session_player", return_value=None),
1940        ):
1941            await sgp.set_members(player_ids_to_remove=["leader"])
1942
1943        dyn_switch.assert_not_awaited()
1944        reform.assert_awaited_once()
1945        assert reform.await_args is not None
1946        assert reform.await_args.kwargs.get("resume_playback") is True
1947
1948    @pytest.mark.asyncio
1949    async def test_leader_removal_does_not_resume_when_idle(self) -> None:
1950        """A genuinely idle group must not spuriously start playing on leader removal."""
1951        mass = _make_mock_mass()
1952        sgp = self._setup_group(mass, group_state=PlaybackState.IDLE)
1953
1954        with (
1955            patch.object(sgp, "_dissolve_and_reform", new=AsyncMock()) as reform,
1956            patch.object(sgp, "_dynamic_leader_switch", new=AsyncMock()),
1957            patch.object(sgp, "_active_session_player", return_value=None),
1958        ):
1959            await sgp.set_members(player_ids_to_remove=["leader"])
1960
1961        reform.assert_awaited_once()
1962        assert reform.await_args is not None
1963        assert reform.await_args.kwargs.get("resume_playback") is False
1964