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