music-assistant-server

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