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