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