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