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