music-assistant-server

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