music-assistant-server

1"""
2Tests for PlayerController high-level operations.
3
4This module tests:
5- cmd_set_members validation and execution
6- Group/ungroup commands
7- Player state management
8- Cache invalidation after grouping operations
9"""
10
11from __future__ import annotations
12
13import asyncio
14import contextlib
15import time
16from collections.abc import AsyncIterator, Callable
17from types import SimpleNamespace
18from typing import cast
19from unittest.mock import AsyncMock, MagicMock, call, patch
20
21import pytest
22from music_assistant_models.auth import User, UserRole
23from music_assistant_models.constants import (
24    PLAYER_CONTROL_FAKE,
25    PLAYER_CONTROL_NATIVE,
26    PLAYER_CONTROL_NONE,
27)
28from music_assistant_models.enums import (
29    EventType,
30    MediaType,
31    PlaybackState,
32    PlayerFeature,
33    PlayerType,
34)
35from music_assistant_models.errors import (
36    InvalidDataError,
37    PlayerCommandFailed,
38    UnsupportedFeaturedException,
39)
40from music_assistant_models.player import OutputProtocol, PlayerMedia, PlayerSource
41from music_assistant_models.player_control import PlayerControl
42
43from music_assistant.constants import (
44    ATTR_FAKE_MUTE,
45    ATTR_MUTE_LOCK,
46    ATTR_PREVIOUS_VOLUME,
47    CONF_AUTO_PLAY,
48    CONF_MAX_VOLUME,
49    CONF_MIN_VOLUME,
50    CONF_MUTE_CONTROL,
51    CONF_POWER_CONTROL,
52    CONF_VOLUME_CONTROL,
53)
54from music_assistant.controllers.players import PlayerController
55from music_assistant.controllers.webserver.helpers.auth_middleware import current_user
56from tests.common import MockPlayer, MockProvider
57
58
59def _player_config_stub(
60    values: dict[str, object] | None = None,
61    *,
62    min_volume: int = 0,
63    max_volume: int = 100,
64) -> Callable[..., object]:
65    """
66    Build a ``get_raw_player_config_value`` side effect.
67
68    :param values: Extra config keys to answer, e.g. ``{CONF_MUTE_CONTROL: PLAYER_CONTROL_FAKE}``.
69    :param min_volume: Value returned for the ``min_volume`` key.
70    :param max_volume: Value returned for the ``max_volume`` key.
71    """
72    config: dict[str, object] = {
73        CONF_MIN_VOLUME: min_volume,
74        CONF_MAX_VOLUME: max_volume,
75        **(values or {}),
76    }
77
78    def _conf(_player_id: str, key: str, default: object = None) -> object:
79        if key in config:
80            return config[key]
81        return default
82
83    return _conf
84
85
86@pytest.fixture
87def mock_mass() -> MagicMock:
88    """Create a mock MusicAssistant instance."""
89    mass = MagicMock()
90    mass.closing = False
91    mass.loop = None
92    mass.config = MagicMock()
93    mass.config.get = MagicMock(return_value=[])
94    mass.config.get_raw_player_config_value = MagicMock(side_effect=_player_config_stub())
95    # Return "GLOBAL" for log level config (standard default)
96    mass.config.get_raw_core_config_value = MagicMock(return_value="GLOBAL")
97    mass.config.set = MagicMock()
98    mass.signal_event = MagicMock()
99    mass.get_providers = MagicMock(return_value=[])
100    return mass
101
102
103@pytest.fixture
104def controller(mock_mass: MagicMock) -> PlayerController:
105    """Create a PlayerController instance."""
106    return PlayerController(mock_mass)
107
108
109class TestSetMembersValidation:
110    """Test cmd_set_members validation logic."""
111
112    def test_set_members_requires_feature(self, mock_mass: MagicMock) -> None:
113        """Test that set_members requires SET_MEMBERS feature."""
114        controller = PlayerController(mock_mass)
115        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
116
117        leader = MockPlayer(provider, "leader", "Leader")
118        # Note: NOT adding SET_MEMBERS feature
119
120        member = MockPlayer(provider, "member", "Member")
121
122        controller._players = {"leader": leader, "member": member}
123        mock_mass.players = controller
124
125        # Should raise exception because leader doesn't support SET_MEMBERS
126        with pytest.raises(UnsupportedFeaturedException):
127            asyncio.run(controller.cmd_set_members("leader", player_ids_to_add=["member"]))
128
129    def test_cannot_group_incompatible_players(self, mock_mass: MagicMock) -> None:
130        """Test that incompatible players cannot be grouped."""
131        controller = PlayerController(mock_mass)
132        provider_a = MockProvider("provider_a", instance_id="provider_a", mass=mock_mass)
133        provider_b = MockProvider("provider_b", instance_id="provider_b", mass=mock_mass)
134
135        player_a = MockPlayer(provider_a, "player_a", "Player A")
136        player_a._attr_supported_features.add(PlayerFeature.SET_MEMBERS)
137        player_a._attr_can_group_with = {"provider_a"}  # Only same provider
138
139        player_b = MockPlayer(provider_b, "player_b", "Player B")
140
141        controller._players = {"player_a": player_a, "player_b": player_b}
142        mock_mass.players = controller
143
144        # Should raise exception because players are incompatible
145        with pytest.raises(UnsupportedFeaturedException):
146            asyncio.run(controller.cmd_set_members("player_a", player_ids_to_add=["player_b"]))
147
148
149class TestCacheInvalidationAfterGrouping:
150    """Test that caches are invalidated after grouping operations."""
151
152    async def test_all_players_cache_cleared_after_set_members(self, mock_mass: MagicMock) -> None:
153        """
154        Test that all players' caches are cleared after set_members.
155
156        Regression test for: Stale can_group_with cache after grouping changes.
157        """
158        controller = PlayerController(mock_mass)
159        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
160
161        leader = MockPlayer(provider, "leader", "Leader")
162        leader._attr_supported_features.add(PlayerFeature.SET_MEMBERS)
163        leader._attr_can_group_with = {"test"}
164        leader._attr_group_members = []
165
166        member = MockPlayer(provider, "member", "Member")
167
168        other = MockPlayer(provider, "other", "Other")
169        other._attr_supported_features.add(PlayerFeature.SET_MEMBERS)
170        other._attr_can_group_with = {"test"}
171
172        controller._players = {"leader": leader, "member": member, "other": other}
173        mock_mass.players = controller
174
175        # Populate caches
176        _ = leader.state.can_group_with
177        _ = other.state.can_group_with
178
179        # Simulate grouping (normally done by provider's set_members implementation)
180        leader._attr_group_members = ["leader", "member"]
181
182        # Call set_members to trigger cache invalidation
183        await controller._handle_set_members_with_protocols(
184            leader, player_ids_to_add=["member"], player_ids_to_remove=[]
185        )
186
187        # Note: The actual cache clearing happens via trigger_player_update
188        # which schedules update_state to be called later
189        # In a real scenario, this would clear all players' caches
190
191
192class TestNativeSetMembersGuard:
193    """Test the SET_MEMBERS feature guard on native set_members forwarding."""
194
195    async def test_native_set_members_skipped_without_feature(self, mock_mass: MagicMock) -> None:
196        """
197        Test that set_members is not called on a player without SET_MEMBERS support.
198
199        Regression test for: NotImplementedError raised from
200        _cleanup_player_memberships when removing a member from a native player
201        whose group membership is managed externally (e.g. a Google Cast group,
202        which never advertises SET_MEMBERS).
203        """
204        controller = PlayerController(mock_mass)
205        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
206
207        # a native group-like player WITHOUT SET_MEMBERS in supported_features,
208        # whose set_members behaves like the Player base class (raises)
209        parent = MockPlayer(provider, "cast_group", "Cast Group")
210        parent._attr_group_members = ["cast_group", "member"]
211        parent.set_members = AsyncMock(  # type: ignore[method-assign]
212            side_effect=NotImplementedError(
213                "set_members needs to be implemented when PlayerFeature.SET_MEMBERS is set"
214            )
215        )
216        member = MockPlayer(provider, "member", "Member")
217
218        controller._players = {"cast_group": parent, "member": member}
219        mock_mass.players = controller
220
221        # must complete without raising NotImplementedError
222        await controller._handle_set_members_with_protocols(
223            parent, player_ids_to_add=[], player_ids_to_remove=["member"]
224        )
225        parent.set_members.assert_not_called()
226
227
228class TestGroupUngroup:
229    """Test group and ungroup commands."""
230
231    async def test_group_command(self, mock_mass: MagicMock) -> None:
232        """Test the group command (cmd_group)."""
233        controller = PlayerController(mock_mass)
234        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
235
236        leader = MockPlayer(provider, "leader", "Leader")
237        leader._attr_supported_features.add(PlayerFeature.SET_MEMBERS)
238        leader._attr_can_group_with = {"member"}  # Leader can group with member
239
240        member = MockPlayer(provider, "member", "Member")
241        # Make sure member is already powered on to skip power handling
242        member._attr_powered = True
243
244        controller._players = {"leader": leader, "member": member}
245        mock_mass.players = controller
246
247        # Update state after modifying attributes and registering with controller
248        leader.update_state(signal_event=False)
249        member.update_state(signal_event=False)
250
251        # Track if set_members was called
252        set_members_called = False
253        original_set_members = leader.set_members
254
255        async def mock_set_members(
256            player_ids_to_add: list[str] | None = None,
257            player_ids_to_remove: list[str] | None = None,
258        ) -> None:
259            nonlocal set_members_called
260            set_members_called = True
261            # Call the original to update group_members
262            await original_set_members(player_ids_to_add, player_ids_to_remove)
263
264        leader.set_members = mock_set_members  # type: ignore[method-assign]
265
266        # Mock power handling to skip power control (focus is on grouping logic)
267        async def mock_handle_cmd_power(
268            player_id: str, powered: bool, skip_auto_play: bool = False
269        ) -> None:
270            pass
271
272        controller._handle_cmd_power = mock_handle_cmd_power  # type: ignore[method-assign]
273
274        # Execute group command
275        await controller.cmd_group("member", "leader")
276
277        # Verify set_members was called
278        assert set_members_called
279        # Verify member was added to leader's group
280        assert "member" in leader._attr_group_members
281
282
283class TestPlayerAvailability:
284    """Test player availability checks in grouping."""
285
286    def test_unavailable_player_rejected(self, mock_mass: MagicMock) -> None:
287        """Test that unavailable players are rejected when grouping."""
288        controller = PlayerController(mock_mass)
289        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
290
291        leader = MockPlayer(provider, "leader", "Leader")
292        leader._attr_supported_features.add(PlayerFeature.SET_MEMBERS)
293        leader._attr_can_group_with = {"test"}
294
295        member = MockPlayer(provider, "member", "Member")
296        member._attr_available = False  # Mark as unavailable
297
298        controller._players = {"leader": leader, "member": member}
299        mock_mass.players = controller
300
301        # Attempting to group with unavailable player should be handled
302        # (either silently ignored or raise exception depending on implementation)
303        # This should either skip the unavailable player or raise an exception
304        with contextlib.suppress(Exception):
305            asyncio.run(controller.cmd_set_members("leader", player_ids_to_add=["member"]))
306
307
308def _group_with_member(
309    mock_mass: MagicMock, *, initialize_group: bool = True
310) -> tuple[PlayerController, MockPlayer, MockPlayer]:
311    """
312    Build a controller holding one group player with a single member.
313
314    :param mock_mass: The mocked MusicAssistant instance to attach the controller to.
315    :param initialize_group: Whether the group player is marked as fully registered.
316
317    :return: The controller, the group player and its member.
318    """
319    controller = PlayerController(mock_mass)
320    provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
321
322    group_player = MockPlayer(provider, "group", "Group", player_type=PlayerType.GROUP)
323    member = MockPlayer(provider, "member", "Member")
324
325    controller._players = {"group": group_player, "member": member}
326    mock_mass.players = controller
327
328    group_player._attr_group_members = ["member"]
329    member.initialized.set()
330    if initialize_group:
331        group_player.initialized.set()
332    for player in (group_player, member):
333        player.update_state(signal_event=False)
334    return controller, group_player, member
335
336
337class TestStateForwarding:
338    """Test forwarding of player state changes to related players."""
339
340    def test_sync_leader_updates_are_forwarded_to_sync_children(self, mock_mass: MagicMock) -> None:
341        """A regular sync leader must notify children via the sync-parent callback."""
342        controller = PlayerController(mock_mass)
343        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
344
345        leader = MockPlayer(provider, "leader", "Leader")
346        child = MockPlayer(provider, "child", "Child")
347
348        controller._players = {"leader": leader, "child": child}
349        mock_mass.players = controller
350
351        leader._attr_group_members = ["leader", "child"]
352        leader.update_state(signal_event=False)
353        child.update_state(signal_event=False)
354
355        with (
356            patch.object(child, "on_sync_parent_updated") as on_sync_parent_updated,
357            patch.object(child, "on_group_updated") as on_group_updated,
358        ):
359            changed_values = {"playback_state": (PlaybackState.IDLE, PlaybackState.PLAYING)}
360            controller._forward_state_update(leader, changed_values)
361
362        on_sync_parent_updated.assert_called_once_with(leader, changed_values)
363        on_group_updated.assert_not_called()
364
365    def test_group_updates_are_forwarded_to_children_via_group_callback(
366        self, mock_mass: MagicMock
367    ) -> None:
368        """A group player must continue to notify children via the group callback."""
369        controller = PlayerController(mock_mass)
370        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
371
372        group_player = MockPlayer(provider, "group", "Group", player_type=PlayerType.GROUP)
373        child = MockPlayer(provider, "child", "Child")
374
375        controller._players = {"group": group_player, "child": child}
376        mock_mass.players = controller
377
378        group_player._attr_group_members = ["group", "child"]
379        group_player.update_state(signal_event=False)
380        child.update_state(signal_event=False)
381
382        with (
383            patch.object(child, "on_group_updated") as on_group_updated,
384            patch.object(child, "on_sync_parent_updated") as on_sync_parent_updated,
385        ):
386            changed_values = {"playback_state": (PlaybackState.IDLE, PlaybackState.PLAYING)}
387            controller._forward_state_update(group_player, changed_values)
388
389        on_group_updated.assert_called_once_with(
390            group_player,
391            changed_values,
392        )
393        on_sync_parent_updated.assert_not_called()
394
395    def test_sync_leader_updates_also_reach_its_group_player(self, mock_mass: MagicMock) -> None:
396        """A sync leader must notify its group player as well as its own sync children."""
397        controller = PlayerController(mock_mass)
398        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
399
400        group_player = MockPlayer(provider, "group", "Group", player_type=PlayerType.GROUP)
401        leader = MockPlayer(provider, "leader", "Leader")
402        follower = MockPlayer(provider, "follower", "Follower")
403
404        controller._players = {"group": group_player, "leader": leader, "follower": follower}
405        mock_mass.players = controller
406
407        group_player._attr_group_members = ["leader", "follower"]
408        leader._attr_group_members = ["leader", "follower"]
409        for player in (group_player, leader, follower):
410            # _get_player_groups only considers players the controller finished registering
411            player.initialized.set()
412            player.update_state(signal_event=False)
413
414        with (
415            patch.object(group_player, "on_group_member_updated") as on_group_member_updated,
416            patch.object(follower, "on_sync_parent_updated") as on_sync_parent_updated,
417        ):
418            changed_values = {"playback_state": (PlaybackState.IDLE, PlaybackState.PLAYING)}
419            controller._forward_state_update(leader, changed_values)
420
421        on_group_member_updated.assert_called_once_with(leader, changed_values)
422        on_sync_parent_updated.assert_called_once_with(leader, changed_values)
423
424    def test_group_player_is_notified_under_restricted_user_context(
425        self, mock_mass: MagicMock
426    ) -> None:
427        """The fan-out must not be narrowed by the user that triggered the update."""
428        controller, group_player, member = _group_with_member(mock_mass)
429
430        # a non-admin user that may only see the member; the contextvar is copied into
431        # the task an API command runs in, so it is live during the state fan-out
432        restricted_user = User(
433            user_id="user_1",
434            username="restricted",
435            role=UserRole.USER,
436            player_filter=["member"],
437        )
438        token = current_user.set(restricted_user)
439        try:
440            assert group_player not in controller.all_players()
441            with patch.object(group_player, "on_group_member_updated") as on_group_member_updated:
442                changed_values = {"playback_state": (PlaybackState.IDLE, PlaybackState.PLAYING)}
443                controller._forward_state_update(member, changed_values)
444        finally:
445            current_user.reset(token)
446
447        on_group_member_updated.assert_called_once_with(member, changed_values)
448
449    def test_unavailable_group_player_is_still_notified(self, mock_mass: MagicMock) -> None:
450        """A group player mirrors its members, so it must update while unavailable too."""
451        controller, group_player, member = _group_with_member(mock_mass)
452        group_player._attr_available = False
453        group_player.update_state(signal_event=False)
454        assert group_player.state.available is False
455
456        with patch.object(group_player, "on_group_member_updated") as on_group_member_updated:
457            changed_values = {"playback_state": (PlaybackState.IDLE, PlaybackState.PLAYING)}
458            controller._forward_state_update(member, changed_values)
459
460        on_group_member_updated.assert_called_once_with(member, changed_values)
461
462    def test_disabled_group_player_is_not_notified(self, mock_mass: MagicMock) -> None:
463        """A disabled player takes no part, unlike one that is merely unavailable."""
464        controller, group_player, member = _group_with_member(mock_mass)
465        group_player._config.enabled = False
466        group_player.update_state(signal_event=False, force_update=True)
467        assert group_player.state.enabled is False
468
469        with patch.object(group_player, "on_group_member_updated") as on_group_member_updated:
470            controller._forward_state_update(
471                member, {"playback_state": (PlaybackState.IDLE, PlaybackState.PLAYING)}
472            )
473
474        on_group_member_updated.assert_not_called()
475
476    def test_uninitialized_group_player_is_not_notified(self, mock_mass: MagicMock) -> None:
477        """A player the controller is still registering is not fully set up yet."""
478        controller, group_player, member = _group_with_member(mock_mass, initialize_group=False)
479
480        with patch.object(group_player, "on_group_member_updated") as on_group_member_updated:
481            controller._forward_state_update(
482                member, {"playback_state": (PlaybackState.IDLE, PlaybackState.PLAYING)}
483            )
484
485        on_group_member_updated.assert_not_called()
486
487
488class TestSleepTimer:
489    """Test native sleep timer handling."""
490
491    def test_set_and_clear_sleep_timer(self, mock_mass: MagicMock) -> None:
492        """Setting a sleep timer exposes state and schedules the stop callback."""
493        controller = PlayerController(mock_mass)
494        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
495        player = MockPlayer(provider, "player_1", "Player 1")
496        controller._players = {"player_1": player}
497        mock_mass.players = controller
498
499        expires_at = controller.set_sleep_timer("player_1", 30)
500
501        assert controller.get_sleep_timer("player_1") == expires_at
502        assert player.sleep_timer_expires_at == expires_at
503        mock_mass.call_later.assert_called_with(
504            30,
505            controller._handle_sleep_timer_expired,
506            "player_1",
507            task_id="player_sleep_timer_player_1",
508        )
509        mock_mass.signal_event.assert_any_call(
510            EventType.PLAYER_SLEEP_TIMER_UPDATED,
511            object_id="player_1",
512            data=expires_at,
513        )
514
515        controller.clear_sleep_timer("player_1")
516
517        # get_sleep_timer reads the model field directly, so this also asserts it cleared
518        assert controller.get_sleep_timer("player_1") is None
519        mock_mass.cancel_timer.assert_called_with("player_sleep_timer_player_1")
520        mock_mass.signal_event.assert_any_call(
521            EventType.PLAYER_SLEEP_TIMER_UPDATED,
522            object_id="player_1",
523            data=None,
524        )
525
526    async def test_sleep_timer_removed_when_player_unregistered(self, mock_mass: MagicMock) -> None:
527        """Unregistering a player cancels and clears its sleep timer."""
528        controller = PlayerController(mock_mass)
529        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
530        player = MockPlayer(provider, "player_1", "Player 1")
531        controller._players = {"player_1": player}
532        player.set_sleep_timer_expires_at(123.0)
533
534        await controller.unregister("player_1")
535
536        assert player.sleep_timer_expires_at is None
537        mock_mass.cancel_timer.assert_called_with("player_sleep_timer_player_1")
538
539    async def test_sleep_timer_expiry_stops_player(self, mock_mass: MagicMock) -> None:
540        """An expired sleep timer clears its state and stops playback."""
541        controller = PlayerController(mock_mass)
542        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
543        player = MockPlayer(provider, "player_1", "Player 1")
544        controller._players = {"player_1": player}
545        player.set_sleep_timer_expires_at(123.0)
546        controller.cmd_stop = AsyncMock()  # type: ignore[method-assign]
547
548        await controller._handle_sleep_timer_expired("player_1")
549
550        assert controller.get_sleep_timer("player_1") is None
551        assert player.sleep_timer_expires_at is None
552        controller.cmd_stop.assert_awaited_once_with("player_1")
553        mock_mass.signal_event.assert_any_call(
554            EventType.PLAYER_SLEEP_TIMER_UPDATED,
555            object_id="player_1",
556            data=None,
557        )
558
559    def test_set_sleep_timer_rejects_invalid_duration(self, mock_mass: MagicMock) -> None:
560        """A non-positive or float-overflowing duration raises and schedules nothing."""
561        controller = PlayerController(mock_mass)
562        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
563        player = MockPlayer(provider, "player_1", "Player 1")
564        controller._players = {"player_1": player}
565
566        # 0/-30 are non-positive; 10**400 exceeds the float range for the expiry math
567        for invalid in (0, -30, 10**400):
568            with pytest.raises(InvalidDataError):
569                controller.set_sleep_timer("player_1", invalid)
570
571        assert controller.get_sleep_timer("player_1") is None
572        mock_mass.call_later.assert_not_called()
573
574
575class TestUnregisterCleanup:
576    """Test that unregister cleans up leaked internal state."""
577
578    def test_command_locks_removed(self, mock_mass: MagicMock) -> None:
579        """Unregistering a player removes its command lock entries."""
580        controller = PlayerController(mock_mass)
581        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
582        player = MockPlayer(provider, "player_1", "Player 1")
583
584        controller._players = {"player_1": player}
585        controller._player_command_locks = {
586            "playback_player_1": asyncio.Lock(),
587            "volume_player_1": asyncio.Lock(),
588        }
589
590        asyncio.run(controller.unregister("player_1"))
591
592        assert "playback_player_1" not in controller._player_command_locks
593        assert "volume_player_1" not in controller._player_command_locks
594
595    def test_other_players_state_untouched(self, mock_mass: MagicMock) -> None:
596        """Unregistering one player does not affect another player's state."""
597        controller = PlayerController(mock_mass)
598        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
599        player_a = MockPlayer(provider, "player_a", "Player A")
600        player_b = MockPlayer(provider, "player_b", "Player B")
601
602        controller._players = {"player_a": player_a, "player_b": player_b}
603        controller._player_command_locks = {
604            "playback_player_a": asyncio.Lock(),
605            "playback_player_b": asyncio.Lock(),
606        }
607
608        asyncio.run(controller.unregister("player_a"))
609
610        assert "playback_player_b" in controller._player_command_locks
611        assert "playback_player_a" not in controller._player_command_locks
612
613    def test_suffix_player_id_not_over_matched(self, mock_mass: MagicMock) -> None:
614        """Removing player 'b' must not remove locks for player 'a_b' (no suffix matching)."""
615        controller = PlayerController(mock_mass)
616        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
617        player_b = MockPlayer(provider, "b", "Player B")
618        player_a_b = MockPlayer(provider, "a_b", "Player A_B")
619
620        controller._players = {"b": player_b, "a_b": player_a_b}
621        controller._player_command_locks = {
622            "playback_b": asyncio.Lock(),
623            "playback_a_b": asyncio.Lock(),
624        }
625
626        asyncio.run(controller.unregister("b"))
627
628        assert "playback_a_b" in controller._player_command_locks
629        assert "playback_b" not in controller._player_command_locks
630
631    def test_pending_protocol_evaluation_cancelled(self, mock_mass: MagicMock) -> None:
632        """Unregistering a player cancels and removes its pending protocol evaluation."""
633        controller = PlayerController(mock_mass)
634        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
635        player = MockPlayer(provider, "player_1", "Player 1")
636
637        mock_handle = MagicMock()
638        controller._players = {"player_1": player}
639        controller._pending_protocol_evaluations = {"player_1": mock_handle}
640
641        asyncio.run(controller.unregister("player_1"))
642
643        mock_handle.cancel.assert_called_once()
644        assert "player_1" not in controller._pending_protocol_evaluations
645
646    def test_unregister_nonexistent_player_is_noop(self, mock_mass: MagicMock) -> None:
647        """Unregistering a player that doesn't exist is silently ignored."""
648        controller = PlayerController(mock_mass)
649        controller._player_command_locks = {"set_members_other": asyncio.Lock()}
650
651        asyncio.run(controller.unregister("nonexistent"))
652
653        assert "set_members_other" in controller._player_command_locks
654
655
656def _set_play_media_override(mock_mass: MagicMock, value: bool) -> None:
657    """
658    Configure get_raw_player_config_value to return ``value`` for the play-media override key.
659
660    Other keys keep the existing defaults from the shared fixture. Use this in
661    tests for ``play_media`` override behavior so the legacy/new branch is
662    selected deterministically.
663    """
664    original = mock_mass.config.get_raw_player_config_value.side_effect
665
666    def _side_effect(player_id: str, key: str, default: object = None) -> object:
667        if key == "play_media_overrides_group":
668            return value
669        if callable(original):
670            return original(player_id, key, default)
671        return default
672
673    mock_mass.config.get_raw_player_config_value = MagicMock(side_effect=_side_effect)
674
675
676class TestCmdUngroupNewBranches:
677    """
678    Regression tests for the post-refactor cmd_ungroup flow.
679
680    The refactor changed two things:
681
682    - ``cmd_ungroup`` on a group player no longer calls ``cmd_set_members``
683      (which would hit the "Cannot remove static member" guard); instead it
684      stops or powers off the group entirely.
685    - ``cmd_ungroup`` on a static member of a group recurses to ungroup the
686      group, because static members cannot be released individually.
687    """
688
689    @pytest.mark.asyncio
690    async def test_ungroup_group_player_with_power_uses_power_off(
691        self, mock_mass: MagicMock
692    ) -> None:
693        """Ungroup on a group with explicit power control routes through cmd_power(False)."""
694        controller = PlayerController(mock_mass)
695        provider = MockProvider("test_group", instance_id="test_group", mass=mock_mass)
696        group = MockPlayer(provider, "g1", "Group", player_type=PlayerType.GROUP)
697        group._attr_powered = True
698        group._attr_group_members = ["member"]
699        group._attr_supported_features = {PlayerFeature.POWER}
700
701        # ensure power_control resolves to NATIVE so cmd_ungroup uses the power path
702        mock_mass.config.get_raw_player_config_value = MagicMock(
703            side_effect=_player_config_stub({CONF_POWER_CONTROL: PLAYER_CONTROL_NATIVE})
704        )
705
706        controller._players = {"g1": group}
707        mock_mass.players = controller
708
709        # populate state.type / state.power_control / state.group_members
710        group.set_initialized()
711        group._cache.clear()
712        group.update_state(signal_event=False)
713
714        called: dict[str, bool | str] = {}
715
716        async def _power(
717            player_id: str,
718            powered: bool,
719            skip_auto_play: bool = False,  # noqa: ARG001
720        ) -> None:
721            called["player_id"] = player_id
722            called["powered"] = powered
723
724        controller._handle_cmd_power = _power  # type: ignore[method-assign]
725
726        await controller.cmd_ungroup("g1")
727
728        assert called == {"player_id": "g1", "powered": False}
729
730    @pytest.mark.asyncio
731    async def test_ungroup_powerless_group_calls_stop(self, mock_mass: MagicMock) -> None:
732        """Ungroup on a powerless group falls through to _handle_cmd_stop."""
733        controller = PlayerController(mock_mass)
734        provider = MockProvider("test_group", instance_id="test_group", mass=mock_mass)
735        group = MockPlayer(provider, "g1", "Group", player_type=PlayerType.GROUP)
736        group._attr_powered = None  # no power control
737        group._attr_group_members = ["member"]
738        # no POWER feature → power_control auto-selects to NONE
739
740        controller._players = {"g1": group}
741        mock_mass.players = controller
742
743        group.set_initialized()
744        group._cache.clear()
745        group.update_state(signal_event=False)
746
747        stop_called: list[str] = []
748
749        async def _stop(player_id: str) -> None:
750            stop_called.append(player_id)
751
752        controller._handle_cmd_stop = _stop  # type: ignore[method-assign]
753        # also stub power to make sure we did NOT go down that branch
754        power_called: list[str] = []
755
756        async def _power(
757            player_id: str,
758            powered: bool,  # noqa: ARG001
759            skip_auto_play: bool = False,  # noqa: ARG001
760        ) -> None:
761            power_called.append(player_id)
762
763        controller._handle_cmd_power = _power  # type: ignore[method-assign]
764
765        await controller.cmd_ungroup("g1")
766
767        assert stop_called == ["g1"]
768        assert power_called == []  # powerless group → never goes through cmd_power
769
770
771class TestExternalPowerOffUnsync:
772    """
773    Tests for unsyncing a player when its power is turned off outside of MA.
774
775    When a player's (final) power state flips on->off because its linked power
776    control was switched off directly - rather than via an MA power command -
777    the player must be removed from any (sync)group it is part of.
778    """
779
780    def _make_synced_player(self, mock_mass: MagicMock) -> tuple[PlayerController, MockPlayer]:
781        """Build a controller with a player synced to a registered leader."""
782        controller = PlayerController(mock_mass)
783        provider = MockProvider("test", instance_id="test", mass=mock_mass)
784        leader = MockPlayer(provider, "leader", "Leader")
785        leader._attr_group_members = ["leader", "p1"]
786        player = MockPlayer(provider, "p1", "Player")
787        controller._players = {"leader": leader, "p1": player}
788        mock_mass.players = controller
789        for _player in (leader, player):
790            _player.set_initialized()
791            _player._cache.clear()
792            _player.update_state(signal_event=False)
793        # isolate the unsync branch from the unrelated state-forwarding machinery
794        controller._forward_state_update = MagicMock()  # type: ignore[method-assign]
795        controller.cmd_ungroup = MagicMock(return_value="ungroup-coro")  # type: ignore[method-assign]
796        return controller, player
797
798    def test_power_off_unsyncs_synced_player(self, mock_mass: MagicMock) -> None:
799        """An on->off power transition ungroups a synced player."""
800        controller, player = self._make_synced_player(mock_mass)
801        assert player.state.synced_to == "leader"
802
803        controller.signal_player_state_update(player, {"powered": (True, False)})
804
805        controller.cmd_ungroup.assert_called_once_with("p1")  # type: ignore[attr-defined]
806
807    def test_power_on_does_not_unsync(self, mock_mass: MagicMock) -> None:
808        """An off->on power transition leaves the player synced."""
809        controller, player = self._make_synced_player(mock_mass)
810
811        controller.signal_player_state_update(player, {"powered": (False, True)})
812
813        controller.cmd_ungroup.assert_not_called()  # type: ignore[attr-defined]
814
815    def test_no_power_control_is_ignored(self, mock_mass: MagicMock) -> None:
816        """A None->off transition (player without power control) is ignored."""
817        controller, player = self._make_synced_player(mock_mass)
818
819        controller.signal_player_state_update(player, {"powered": (None, False)})
820
821        controller.cmd_ungroup.assert_not_called()  # type: ignore[attr-defined]
822
823    def test_power_off_ungrouped_player_is_noop(self, mock_mass: MagicMock) -> None:
824        """Powering off a player that is not in any group does nothing."""
825        controller = PlayerController(mock_mass)
826        provider = MockProvider("test", instance_id="test", mass=mock_mass)
827        player = MockPlayer(provider, "p1", "Player")
828        controller._players = {"p1": player}
829        mock_mass.players = controller
830        player.set_initialized()
831        player._cache.clear()
832        player.update_state(signal_event=False)
833        controller._forward_state_update = MagicMock()  # type: ignore[method-assign]
834        controller.cmd_ungroup = MagicMock(return_value="ungroup-coro")  # type: ignore[method-assign]
835
836        controller.signal_player_state_update(player, {"powered": (True, False)})
837
838        controller.cmd_ungroup.assert_not_called()
839
840
841class TestPlayMediaOverride:
842    """
843    Tests for the new CONF_PLAY_MEDIA_OVERRIDES_GROUP behavior.
844
845    When a captured child player receives an explicit play_media command, the
846    default behavior is to *release* it from the active group/sync and play
847    directly on the targeted player. The legacy behavior (forward to the
848    leader) is preserved via the per-player config opt-out.
849    """
850
851    @pytest.mark.asyncio
852    async def test_override_disabled_redirects_to_group(self, mock_mass: MagicMock) -> None:
853        """With override disabled, play_media on a captured child redirects to the group."""
854        controller = PlayerController(mock_mass)
855        group_provider = MockProvider("test_group", instance_id="test_group", mass=mock_mass)
856        member_provider = MockProvider("test", instance_id="test", mass=mock_mass)
857
858        class _SessionedGroup(MockPlayer):
859            @property
860            def is_active_session(self) -> bool:
861                return True
862
863        group = _SessionedGroup(group_provider, "g1", "Group", player_type=PlayerType.GROUP)
864        group._attr_powered = None
865        group._attr_group_members = ["member"]
866
867        member = MockPlayer(member_provider, "member", "Member")
868
869        controller._players = {"g1": group, "member": member}
870        mock_mass.players = controller
871
872        group.set_initialized()
873        member.set_initialized()
874        group.update_state(signal_event=False)
875        member.update_state(signal_event=False)
876        # sanity: the member is captured by the group
877        assert member.state.active_group == "g1"
878
879        _set_play_media_override(mock_mass, False)
880
881        played_on: list[str] = []
882
883        async def _handle_play_media(player_id: str, media: object) -> None:  # noqa: ARG001
884            played_on.append(player_id)
885
886        controller._handle_play_media = _handle_play_media  # type: ignore[method-assign]
887        # the play_media wrapper acquires a playback lock; stub it out
888        controller._player_command_locks = {}
889
890        media = MagicMock(uri="x", source_id="src")
891        await controller.play_media("member", media)
892
893        # legacy behavior: redirected to the group leader
894        assert played_on == ["g1"]
895
896    @pytest.mark.asyncio
897    async def test_override_releases_dynamic_member(self, mock_mass: MagicMock) -> None:
898        """With override enabled, play_media on a dynamic group member releases it first."""
899        controller = PlayerController(mock_mass)
900        group_provider = MockProvider("test_group", instance_id="test_group", mass=mock_mass)
901        member_provider = MockProvider("test", instance_id="test", mass=mock_mass)
902
903        class _SessionedGroup(MockPlayer):
904            @property
905            def is_active_session(self) -> bool:
906                return True
907
908        group = _SessionedGroup(group_provider, "g1", "Group", player_type=PlayerType.GROUP)
909        group._attr_powered = None
910        group._attr_supported_features.add(PlayerFeature.SET_MEMBERS)
911        group._attr_group_members = ["member"]
912        # NOT a static member ⇒ dynamic — can be removed via set_members
913        group._attr_static_group_members = []
914
915        member = MockPlayer(member_provider, "member", "Member")
916
917        controller._players = {"g1": group, "member": member}
918        mock_mass.players = controller
919
920        group.set_initialized()
921        member.set_initialized()
922        group.update_state(signal_event=False)
923        member.update_state(signal_event=False)
924        assert member.state.active_group == "g1"
925
926        # default: override enabled
927        _set_play_media_override(mock_mass, True)
928
929        set_members_calls: list[dict[str, object]] = []
930
931        async def _cmd_set_members(
932            target_player: str,
933            player_ids_to_add: list[str] | None = None,  # noqa: ARG001
934            player_ids_to_remove: list[str] | None = None,
935        ) -> None:
936            set_members_calls.append(
937                {"player_id": target_player, "remove": player_ids_to_remove or []}
938            )
939
940        controller.cmd_set_members = _cmd_set_members  # type: ignore[method-assign]
941
942        played_on: list[str] = []
943
944        async def _handle_play_media(player_id: str, media: object) -> None:  # noqa: ARG001
945            played_on.append(player_id)
946
947        controller._handle_play_media = _handle_play_media  # type: ignore[method-assign]
948        controller._player_command_locks = {}
949
950        media = MagicMock(uri="x", source_id="src")
951        with patch.object(
952            controller,
953            "wait_for_player_update",
954            _skip_player_update_wait,
955        ):
956            await controller.play_media("member", media)
957
958        # the member was removed from the group ...
959        assert set_members_calls == [{"player_id": "g1", "remove": ["member"]}]
960        # ... and then play_media was issued directly on the member, NOT on the group
961        assert played_on == ["member"]
962
963    @pytest.mark.asyncio
964    async def test_override_stops_static_group(self, mock_mass: MagicMock) -> None:
965        """With override enabled, play_media on a STATIC group member stops the group."""
966        controller = PlayerController(mock_mass)
967        group_provider = MockProvider("test_group", instance_id="test_group", mass=mock_mass)
968        member_provider = MockProvider("test", instance_id="test", mass=mock_mass)
969
970        class _SessionedGroup(MockPlayer):
971            @property
972            def is_active_session(self) -> bool:
973                return True
974
975        group = _SessionedGroup(group_provider, "g1", "Group", player_type=PlayerType.GROUP)
976        group._attr_powered = None  # no power control ⇒ stop, not power-off
977        group._attr_supported_features.add(PlayerFeature.SET_MEMBERS)
978        group._attr_group_members = ["member"]
979        # static member - cannot be removed individually
980        group._attr_static_group_members = ["member"]
981
982        member = MockPlayer(member_provider, "member", "Member")
983
984        controller._players = {"g1": group, "member": member}
985        mock_mass.players = controller
986
987        group.set_initialized()
988        member.set_initialized()
989        group.update_state(signal_event=False)
990        member.update_state(signal_event=False)
991        assert member.state.active_group == "g1"
992
993        _set_play_media_override(mock_mass, True)
994
995        stop_calls: list[str] = []
996        power_calls: list[tuple[str, bool]] = []
997
998        async def _stop(player_id: str) -> None:
999            stop_calls.append(player_id)
1000
1001        async def _power(
1002            player_id: str,
1003            powered: bool,
1004            skip_auto_play: bool = False,  # noqa: ARG001
1005        ) -> None:
1006            power_calls.append((player_id, powered))
1007
1008        controller._handle_cmd_stop = _stop  # type: ignore[method-assign]
1009        controller._handle_cmd_power = _power  # type: ignore[method-assign]
1010
1011        played_on: list[str] = []
1012
1013        async def _handle_play_media(player_id: str, media: object) -> None:  # noqa: ARG001
1014            played_on.append(player_id)
1015
1016        controller._handle_play_media = _handle_play_media  # type: ignore[method-assign]
1017        controller._player_command_locks = {}
1018
1019        media = MagicMock(uri="x", source_id="src")
1020        with patch.object(
1021            controller,
1022            "wait_for_player_update",
1023            _skip_player_update_wait,
1024        ):
1025            await controller.play_media("member", media)
1026
1027        # powerless group + static member: we should have stopped the group ...
1028        assert stop_calls == ["g1"]
1029        # ... not powered it off ...
1030        assert power_calls == []
1031        # ... and play_media was issued directly on the member
1032        assert played_on == ["member"]
1033
1034
1035class TestExternalSourcePlayPause:
1036    """Pause/play handling for externally-initiated sources (no active output protocol)."""
1037
1038    @staticmethod
1039    def _make_external_source_player(
1040        provider: MockProvider,
1041        controller: PlayerController,
1042        mock_mass: MagicMock,
1043        *,
1044        playback_state: PlaybackState,
1045        can_play_pause: bool = True,
1046        supports_pause: bool = True,
1047    ) -> MockPlayer:
1048        """Build a player playing a passive external source, with no active output protocol."""
1049        player = MockPlayer(provider, "player_1", "Test Player")
1050        player._attr_supported_features = {PlayerFeature.PAUSE} if supports_pause else set()
1051        player._attr_source_list = [
1052            PlayerSource(
1053                id="spotify",
1054                name="Spotify",
1055                passive=True,
1056                can_play_pause=can_play_pause,
1057                can_next_previous=True,
1058                can_seek=True,
1059            )
1060        ]
1061        player._attr_active_source = "spotify"
1062        player._attr_playback_state = playback_state
1063        player._cache.clear()
1064        controller._players = {"player_1": player}
1065        mock_mass.players = controller
1066        mock_mass.player_queues = MagicMock()
1067        mock_mass.player_queues.get = MagicMock(return_value=None)
1068        player.update_state(signal_event=False)
1069        return player
1070
1071    def test_pause_external_source_forwards_to_player(
1072        self, mock_mass: MagicMock, controller: PlayerController, provider: MockProvider
1073    ) -> None:
1074        """Pausing a pausable external source forwards to the player, not STOP."""
1075        player = self._make_external_source_player(
1076            provider, controller, mock_mass, playback_state=PlaybackState.PLAYING
1077        )
1078        player.pause = AsyncMock()  # type: ignore[method-assign]
1079        controller._handle_cmd_stop = AsyncMock()  # type: ignore[method-assign]
1080
1081        asyncio.run(controller._handle_cmd_pause("player_1"))
1082
1083        player.pause.assert_awaited_once()
1084        controller._handle_cmd_stop.assert_not_called()
1085
1086    def test_play_external_source_unpauses_player(
1087        self, mock_mass: MagicMock, controller: PlayerController, provider: MockProvider
1088    ) -> None:
1089        """Unpausing a paused external source forwards to the player, not a restart."""
1090        player = self._make_external_source_player(
1091            provider, controller, mock_mass, playback_state=PlaybackState.PAUSED
1092        )
1093        player.play = AsyncMock()  # type: ignore[method-assign]
1094        player.play_media = AsyncMock()  # type: ignore[method-assign]
1095        controller._handle_select_source = AsyncMock()  # type: ignore[method-assign]
1096
1097        asyncio.run(controller._handle_cmd_play("player_1"))
1098
1099        player.play.assert_awaited_once()
1100        player.play_media.assert_not_called()
1101        controller._handle_select_source.assert_not_called()
1102
1103    def test_pause_falls_back_to_stop_without_pause_support(
1104        self, mock_mass: MagicMock, controller: PlayerController, provider: MockProvider
1105    ) -> None:
1106        """A player that cannot pause natively still falls back to STOP."""
1107        player = self._make_external_source_player(
1108            provider,
1109            controller,
1110            mock_mass,
1111            playback_state=PlaybackState.PLAYING,
1112            supports_pause=False,
1113        )
1114        player.pause = AsyncMock()  # type: ignore[method-assign]
1115        controller._handle_cmd_stop = AsyncMock()  # type: ignore[method-assign]
1116
1117        asyncio.run(controller._handle_cmd_pause("player_1"))
1118
1119        controller._handle_cmd_stop.assert_awaited_once()
1120        player.pause.assert_not_called()
1121
1122
1123class TestProtocolOutputPlayPause:
1124    """Play/pause on a player rendering through a linked output protocol."""
1125
1126    @staticmethod
1127    def _make_player_on_protocol(
1128        mock_mass: MagicMock,
1129        controller: PlayerController,
1130        *,
1131        playback_state: PlaybackState,
1132    ) -> MockPlayer:
1133        """Build a player playing the MA queue through a protocol that cannot pause."""
1134        native_provider = MockProvider("chromecast", mass=mock_mass)
1135        player = MockPlayer(native_provider, "player_1", "Test Player")
1136        player._attr_supported_features.add(PlayerFeature.PAUSE)
1137        player._attr_playback_state = playback_state
1138
1139        protocol_provider = MockProvider("sendspin", mass=mock_mass)
1140        protocol_player = MockPlayer(
1141            protocol_provider, "proto_1", "Test Protocol", player_type=PlayerType.PROTOCOL
1142        )
1143        protocol_player._attr_playback_state = playback_state
1144
1145        controller._players = {"player_1": player, "proto_1": protocol_player}
1146        mock_mass.players = controller
1147        mock_mass.player_queues = MagicMock()
1148        # a non-empty queue, so the MA queue source advertises play/pause support
1149        queue = MagicMock()
1150        queue.items = [MagicMock()]
1151        mock_mass.player_queues.get = MagicMock(return_value=queue)
1152        player.set_linked_output_protocols(
1153            [
1154                OutputProtocol(
1155                    output_protocol_id="proto_1",
1156                    name="Sendspin",
1157                    protocol_domain="sendspin",
1158                    priority=40,
1159                )
1160            ]
1161        )
1162        player.set_active_output_protocol("proto_1")
1163        player.set_active_mass_source("player_1")
1164        protocol_player.update_state(signal_event=False)
1165        player.refresh_state(signal_event=False)
1166        return player
1167
1168    async def test_pause_on_protocol_without_pause_falls_back_to_stop(
1169        self, mock_mass: MagicMock, controller: PlayerController
1170    ) -> None:
1171        """The native transport has no session to pause while a protocol renders the audio."""
1172        player = self._make_player_on_protocol(
1173            mock_mass, controller, playback_state=PlaybackState.PLAYING
1174        )
1175        player.pause = AsyncMock()  # type: ignore[method-assign]
1176        controller._handle_cmd_stop = AsyncMock()  # type: ignore[method-assign]
1177
1178        await controller._handle_cmd_pause("player_1")
1179
1180        player.pause.assert_not_called()
1181        # STOP goes to the visible player, not the protocol player
1182        controller._handle_cmd_stop.assert_awaited_once_with("player_1")
1183
1184    async def test_play_on_protocol_without_pause_does_not_unpause_natively(
1185        self, mock_mass: MagicMock, controller: PlayerController
1186    ) -> None:
1187        """Unpausing must not hit the native transport either; the source is restarted."""
1188        player = self._make_player_on_protocol(
1189            mock_mass, controller, playback_state=PlaybackState.PAUSED
1190        )
1191        player.play = AsyncMock()  # type: ignore[method-assign]
1192        controller._handle_select_source = AsyncMock()  # type: ignore[method-assign]
1193
1194        await controller._handle_cmd_play("player_1")
1195
1196        player.play.assert_not_called()
1197        # the MA queue source is restarted, not some other source
1198        controller._handle_select_source.assert_awaited_once_with("player_1", "player_1")
1199
1200
1201class TestMirrorsParentMedia:
1202    """Tests for _mirrors_parent_media (palette-fetch gating for grouped players)."""
1203
1204    @staticmethod
1205    def _fake_player(
1206        *,
1207        player_id: str = "p1",
1208        active_group: str | None = None,
1209        synced_to: str | None = None,
1210        player_type: PlayerType = PlayerType.PLAYER,
1211        protocol_parent_id: str | None = None,
1212    ) -> SimpleNamespace:
1213        return SimpleNamespace(
1214            player_id=player_id,
1215            state=SimpleNamespace(active_group=active_group, synced_to=synced_to, type=player_type),
1216            protocol_parent_id=protocol_parent_id,
1217        )
1218
1219    def test_standalone_player_owns_media(self, controller: PlayerController) -> None:
1220        """A standalone player resolves its own media (and palette)."""
1221        assert controller._mirrors_parent_media(self._fake_player()) is False  # type: ignore[arg-type]
1222
1223    def test_group_member_mirrors(self, controller: PlayerController) -> None:
1224        """A group member borrows its parent's media."""
1225        assert controller._mirrors_parent_media(self._fake_player(active_group="g1")) is True  # type: ignore[arg-type]
1226
1227    def test_synced_member_mirrors(self, controller: PlayerController) -> None:
1228        """A synced member borrows its leader's media."""
1229        assert controller._mirrors_parent_media(self._fake_player(synced_to="leader")) is True  # type: ignore[arg-type]
1230
1231    def test_protocol_child_mirrors(self, controller: PlayerController) -> None:
1232        """A protocol child borrows its parent's media."""
1233        player = self._fake_player(player_type=PlayerType.PROTOCOL, protocol_parent_id="parent")
1234        assert controller._mirrors_parent_media(player) is True  # type: ignore[arg-type]
1235
1236    def test_protocol_player_without_parent_owns_media(self, controller: PlayerController) -> None:
1237        """A protocol player with no parent resolves its own media."""
1238        player = self._fake_player(player_type=PlayerType.PROTOCOL)
1239        assert controller._mirrors_parent_media(player) is False  # type: ignore[arg-type]
1240
1241    def test_self_referential_parent_owns_media(self, controller: PlayerController) -> None:
1242        """A self-referential active_group/synced_to is not a real parent, so resolve locally."""
1243        player = self._fake_player(player_id="p1", synced_to="p1", active_group="p1")
1244        assert controller._mirrors_parent_media(player) is False  # type: ignore[arg-type]
1245
1246
1247class TestVolumeScalingOnRedirect:
1248    """min/max volume scaling must survive a redirect to a protocol player or external control."""
1249
1250    @staticmethod
1251    def _volume_player(
1252        player_id: str,
1253        volume_control: str,
1254        volume_set: AsyncMock | None = None,
1255    ) -> SimpleNamespace:
1256        return SimpleNamespace(
1257            player_id=player_id,
1258            type=PlayerType.PLAYER,
1259            protocol_parent_id=None,
1260            extra_data={},
1261            volume_control=volume_control,
1262            mute_control=PLAYER_CONTROL_NONE,
1263            volume_set=volume_set or AsyncMock(),
1264            update_state=MagicMock(),
1265            provider=MagicMock(),
1266            state=SimpleNamespace(
1267                name=player_id,
1268                volume_control=volume_control,
1269                volume_muted=False,
1270                mute_control=PLAYER_CONTROL_NONE,
1271            ),
1272        )
1273
1274    @pytest.mark.asyncio
1275    async def test_protocol_redirect_forwards_scaled_volume(
1276        self, controller: PlayerController, mock_mass: MagicMock
1277    ) -> None:
1278        """A volume command redirected to a protocol player honors the user-facing max_volume."""
1279
1280        def _conf(player_id: str, key: str, default: object = None) -> object:
1281            if key == "min_volume":
1282                return 0
1283            if key == "max_volume":
1284                # user-facing player caps at 50, the protocol player has no limits of its own
1285                return 50 if player_id == "user_player" else 100
1286            return default
1287
1288        mock_mass.config.get_raw_player_config_value = MagicMock(side_effect=_conf)
1289
1290        protocol = self._volume_player("protocol_player", PLAYER_CONTROL_NATIVE)
1291        user = self._volume_player("user_player", "protocol_player")
1292        players = {"user_player": user, "protocol_player": protocol}
1293
1294        with (
1295            patch.object(controller, "get_player", side_effect=players.get),
1296            patch.object(controller, "_get_active_audio_source", return_value=None),
1297        ):
1298            controller._controls = {}
1299            await controller._handle_cmd_volume_set("user_player", 100)
1300
1301        # logical 100 with a max_volume of 50 must reach the protocol player as 50, not the raw 100
1302        protocol.volume_set.assert_awaited_once_with(50)
1303
1304    @pytest.mark.asyncio
1305    async def test_external_control_redirect_forwards_scaled_volume(
1306        self, controller: PlayerController, mock_mass: MagicMock
1307    ) -> None:
1308        """A volume command redirected to an external control honors the user-facing max_volume."""
1309        mock_mass.config.get_raw_player_config_value = MagicMock(
1310            side_effect=_player_config_stub(max_volume=50)
1311        )
1312
1313        volume_set = AsyncMock()
1314        control = PlayerControl(
1315            id="ext_control",
1316            provider="test",
1317            name="External Amp",
1318            supports_volume=True,
1319            volume_set=volume_set,
1320        )
1321        user = self._volume_player("user_player", "ext_control")
1322        players = {"user_player": user}
1323
1324        with (
1325            patch.object(controller, "get_player", side_effect=players.get),
1326            patch.object(controller, "_get_active_audio_source", return_value=None),
1327        ):
1328            controller._controls = {"ext_control": control}
1329            await controller._handle_cmd_volume_set("user_player", 100)
1330
1331        volume_set.assert_awaited_once_with(50)
1332
1333    @pytest.mark.asyncio
1334    async def test_external_control_without_volume_support_raises(
1335        self, controller: PlayerController, mock_mass: MagicMock
1336    ) -> None:
1337        """A volume command redirected to a control lacking volume support is rejected."""
1338        mock_mass.config.get_raw_player_config_value = MagicMock(
1339            side_effect=_player_config_stub(max_volume=50)
1340        )
1341
1342        volume_set = AsyncMock()
1343        control = PlayerControl(
1344            id="ext_control",
1345            provider="test",
1346            name="External Amp",
1347            supports_volume=False,
1348            volume_set=volume_set,
1349        )
1350        user = self._volume_player("user_player", "ext_control")
1351        players = {"user_player": user}
1352
1353        with (
1354            patch.object(controller, "get_player", side_effect=players.get),
1355            patch.object(controller, "_get_active_audio_source", return_value=None),
1356        ):
1357            controller._controls = {"ext_control": control}
1358            with pytest.raises(UnsupportedFeaturedException):
1359                await controller._handle_cmd_volume_set("user_player", 100)
1360
1361        volume_set.assert_not_awaited()
1362
1363
1364class TestExternalPowerControl:
1365    """Power commands redirected to an external PlayerControl must forward and gate correctly."""
1366
1367    def _make_player(
1368        self, mock_mass: MagicMock, control: PlayerControl
1369    ) -> tuple[PlayerController, MockPlayer]:
1370        """Build a controller with a single player whose power control is the given control."""
1371        mock_mass.config.get_raw_player_config_value = MagicMock(
1372            side_effect=_player_config_stub({CONF_POWER_CONTROL: control.id})
1373        )
1374        controller = PlayerController(mock_mass)
1375        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
1376        player = MockPlayer(provider, "player_1", "Player 1")
1377        controller._controls = {control.id: control}
1378        controller._players = {"player_1": player}
1379        mock_mass.players = controller
1380        # auto-play would otherwise resume the (mocked) player queue on power on
1381        config_get_value = player.config.get_value
1382        player.config.get_value = MagicMock(  # type: ignore[method-assign]
1383            side_effect=lambda key, *args, **kwargs: (
1384                False if key == CONF_AUTO_PLAY else config_get_value(key, *args, **kwargs)
1385            )
1386        )
1387        player.set_initialized()
1388        player.update_state(signal_event=False)
1389        return controller, player
1390
1391    async def test_power_on_forwards_to_control(self, mock_mass: MagicMock) -> None:
1392        """Powering on a player redirects to its external control's power_on callback."""
1393        power_on = AsyncMock()
1394        power_off = AsyncMock()
1395        control = PlayerControl(
1396            id="ext_power",
1397            provider="test",
1398            name="External Power",
1399            supports_power=True,
1400            power_on=power_on,
1401            power_off=power_off,
1402        )
1403
1404        def _report_powered_on() -> None:
1405            control.power_state = True
1406
1407        # the control only reports on once switched on, which releases wait_for_power_on
1408        power_on.side_effect = _report_powered_on
1409        controller, player = self._make_player(mock_mass, control)
1410        assert player.state.powered is False
1411
1412        await controller._handle_cmd_power("player_1", True)
1413
1414        power_on.assert_awaited_once()
1415        power_off.assert_not_awaited()
1416
1417    async def test_power_on_waits_on_the_control(self, mock_mass: MagicMock) -> None:
1418        """Powering on waits for the control to report on, not for the player itself."""
1419        control = PlayerControl(
1420            id="ext_power",
1421            provider="test",
1422            name="External Power",
1423            supports_power=True,
1424            power_on=AsyncMock(),
1425            power_off=AsyncMock(),
1426        )
1427        controller, player = self._make_player(mock_mass, control)
1428        assert player.state.powered is False
1429
1430        with patch(
1431            "music_assistant.controllers.players.controller.wait_for_power_on", AsyncMock()
1432        ) as wait_for_power_on:
1433            await controller._handle_cmd_power("player_1", True)
1434
1435        wait_for_power_on.assert_awaited_once()
1436        assert wait_for_power_on.await_args is not None
1437        assert wait_for_power_on.await_args.args[2] is control
1438
1439    async def test_power_off_forwards_to_control(self, mock_mass: MagicMock) -> None:
1440        """Powering off a player redirects to its external control's power_off callback."""
1441        power_on = AsyncMock()
1442        power_off = AsyncMock()
1443        control = PlayerControl(
1444            id="ext_power",
1445            provider="test",
1446            name="External Power",
1447            supports_power=True,
1448            power_state=True,
1449            power_on=power_on,
1450            power_off=power_off,
1451        )
1452        controller, player = self._make_player(mock_mass, control)
1453        assert player.state.powered is True
1454
1455        await controller._handle_cmd_power("player_1", False)
1456
1457        power_off.assert_awaited_once()
1458        power_on.assert_not_awaited()
1459
1460    async def test_control_without_power_support_raises(self, mock_mass: MagicMock) -> None:
1461        """A power command redirected to a control lacking power support is rejected."""
1462        power_on = AsyncMock()
1463        power_off = AsyncMock()
1464        control = PlayerControl(
1465            id="ext_power",
1466            provider="test",
1467            name="External Power",
1468            supports_power=False,
1469            power_on=power_on,
1470            power_off=power_off,
1471        )
1472        controller, player = self._make_player(mock_mass, control)
1473        assert player.state.powered is False
1474
1475        with pytest.raises(UnsupportedFeaturedException):
1476            await controller._handle_cmd_power("player_1", True)
1477
1478        power_on.assert_not_awaited()
1479        power_off.assert_not_awaited()
1480
1481
1482class TestEnforceVolumeLimits:
1483    """External volume changes outside the min/max range must be corrected."""
1484
1485    @staticmethod
1486    def _set_limits(mock_mass: MagicMock, min_volume: int, max_volume: int) -> None:
1487        mock_mass.config.get_raw_player_config_value = MagicMock(
1488            side_effect=_player_config_stub(min_volume=min_volume, max_volume=max_volume)
1489        )
1490
1491    @staticmethod
1492    def _player(logical_volume: int | None) -> SimpleNamespace:
1493        return SimpleNamespace(
1494            player_id="user_player",
1495            state=SimpleNamespace(volume_level=logical_volume),
1496        )
1497
1498    def test_out_of_range_volume_is_corrected(
1499        self, controller: PlayerController, mock_mass: MagicMock
1500    ) -> None:
1501        """A device volume above max_volume (logical > 100) is clamped back to logical 100."""
1502        self._set_limits(mock_mass, 0, 80)
1503        # device volume 100 with max 80 resolves to logical 125
1504        player = self._player(125)
1505        with patch.object(controller, "_handle_cmd_volume_set", MagicMock()) as cmd:
1506            controller._enforce_volume_limits(cast("MockPlayer", player))
1507        cmd.assert_called_once_with("user_player", 100)
1508        mock_mass.create_task.assert_called_once()
1509
1510    def test_below_min_volume_is_corrected(
1511        self, controller: PlayerController, mock_mass: MagicMock
1512    ) -> None:
1513        """A device volume below min_volume (logical < 0) is clamped back to logical 0."""
1514        self._set_limits(mock_mass, 20, 100)
1515        # device volume 10 with min 20 resolves to a negative logical volume
1516        player = self._player(-13)
1517        with patch.object(controller, "_handle_cmd_volume_set", MagicMock()) as cmd:
1518            controller._enforce_volume_limits(cast("MockPlayer", player))
1519        cmd.assert_called_once_with("user_player", 0)
1520
1521    def test_in_range_volume_is_untouched(
1522        self, controller: PlayerController, mock_mass: MagicMock
1523    ) -> None:
1524        """A logical volume within 0-100 needs no correction."""
1525        self._set_limits(mock_mass, 0, 80)
1526        player = self._player(100)
1527        with patch.object(controller, "_handle_cmd_volume_set", MagicMock()) as cmd:
1528            controller._enforce_volume_limits(cast("MockPlayer", player))
1529        cmd.assert_not_called()
1530
1531    def test_no_limits_configured_is_noop(
1532        self, controller: PlayerController, mock_mass: MagicMock
1533    ) -> None:
1534        """Default 0-100 limits skip enforcement entirely."""
1535        self._set_limits(mock_mass, 0, 100)
1536        player = self._player(100)
1537        with patch.object(controller, "_handle_cmd_volume_set", MagicMock()) as cmd:
1538            controller._enforce_volume_limits(cast("MockPlayer", player))
1539        cmd.assert_not_called()
1540
1541    def test_unknown_volume_is_noop(
1542        self, controller: PlayerController, mock_mass: MagicMock
1543    ) -> None:
1544        """A player without a resolved volume level is left alone."""
1545        self._set_limits(mock_mass, 0, 80)
1546        player = self._player(None)
1547        with patch.object(controller, "_handle_cmd_volume_set", MagicMock()) as cmd:
1548            controller._enforce_volume_limits(cast("MockPlayer", player))
1549        cmd.assert_not_called()
1550
1551
1552class TestFakeMuteControl:
1553    """Fake mute must report the muted state and restore the volume on unmute."""
1554
1555    def _make_player(self, mock_mass: MagicMock) -> tuple[PlayerController, MockPlayer, AsyncMock]:
1556        """Build a controller with a single player using fake mute control."""
1557        mock_mass.config.get_raw_player_config_value = MagicMock(
1558            side_effect=_player_config_stub({CONF_MUTE_CONTROL: PLAYER_CONTROL_FAKE})
1559        )
1560        controller = PlayerController(mock_mass)
1561        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
1562        player = MockPlayer(provider, "player_1", "Player 1")
1563        controller._players = {"player_1": player}
1564        mock_mass.players = controller
1565        mock_mass.player_queues.get = MagicMock(return_value=None)
1566        player.set_initialized()
1567        player._attr_volume_level = 40
1568        # let the mocked native volume control behave like a real device
1569        volume_set = AsyncMock(
1570            side_effect=lambda volume: setattr(player, "_attr_volume_level", volume)
1571        )
1572        player.volume_set = volume_set  # type: ignore[method-assign]
1573        player.update_state(signal_event=False)
1574        return controller, player, volume_set
1575
1576    async def test_mute_then_unmute_restores_volume(self, mock_mass: MagicMock) -> None:
1577        """Muting reports volume_muted=True and unmuting restores the previous volume."""
1578        controller, player, volume_set = self._make_player(mock_mass)
1579
1580        await controller.cmd_volume_mute("player_1", True)
1581        muted_state = player.state
1582        assert muted_state.volume_muted is True
1583        assert muted_state.volume_level == 0
1584        assert player.extra_data[ATTR_PREVIOUS_VOLUME] == 40
1585
1586        await controller.cmd_volume_mute("player_1", False)
1587        volume_set.assert_awaited_with(40)
1588        # simulate the device reporting back its state after the volume command
1589        player.update_state()
1590        unmuted_state = player.state
1591        assert unmuted_state.volume_muted is False
1592        assert unmuted_state.volume_level == 40
1593
1594    async def test_repeated_mute_keeps_previous_volume(self, mock_mass: MagicMock) -> None:
1595        """A repeated mute command must not overwrite the stored volume with 0."""
1596        controller, player, volume_set = self._make_player(mock_mass)
1597
1598        await controller.cmd_volume_mute("player_1", True)
1599        await controller.cmd_volume_mute("player_1", True)
1600        assert player.extra_data[ATTR_PREVIOUS_VOLUME] == 40
1601        assert player.state.volume_muted is True
1602
1603        await controller.cmd_volume_mute("player_1", False)
1604        volume_set.assert_awaited_with(40)
1605
1606    async def test_volume_set_clears_fake_mute(self, mock_mass: MagicMock) -> None:
1607        """A regular volume change while fake muted implies an unmute."""
1608        controller, player, _volume_set = self._make_player(mock_mass)
1609
1610        await controller.cmd_volume_mute("player_1", True)
1611        muted_state = player.state
1612        assert muted_state.volume_muted is True
1613
1614        await controller.cmd_volume_set("player_1", 25)
1615        # simulate the device reporting back its state after the volume command
1616        player.update_state()
1617        unmuted_state = player.state
1618        assert unmuted_state.volume_muted is False
1619        assert unmuted_state.volume_level == 25
1620
1621
1622class TestFakeMuteInGroup:
1623    """A fake muted player in a group follows the mute lock, just like a native mute."""
1624
1625    def _make_synced_pair(
1626        self, mock_mass: MagicMock, *, member_mute_control: str = PLAYER_CONTROL_FAKE
1627    ) -> tuple[PlayerController, dict[str, MockPlayer]]:
1628        """
1629        Build a leader synced to one member.
1630
1631        :param member_mute_control: Mute control of the member, the leader always uses fake mute.
1632        """
1633
1634        def _conf(player_id: str, key: str, default: object = None) -> object:
1635            if key == CONF_MUTE_CONTROL and player_id == "member":
1636                return member_mute_control
1637            return _player_config_stub({CONF_MUTE_CONTROL: PLAYER_CONTROL_FAKE})(
1638                player_id, key, default
1639            )
1640
1641        mock_mass.config.get_raw_player_config_value = MagicMock(side_effect=_conf)
1642        controller = PlayerController(mock_mass)
1643        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
1644        players: dict[str, MockPlayer] = {}
1645        for player_id in ("leader", "member"):
1646            player = MockPlayer(provider, player_id, player_id.title())
1647            player._attr_supported_features = {
1648                PlayerFeature.VOLUME_SET,
1649                PlayerFeature.VOLUME_MUTE,
1650            }
1651            player._attr_volume_level = 50
1652            # let the mocked native volume control behave like a real device
1653            player.volume_set = AsyncMock(  # type: ignore[method-assign]
1654                side_effect=lambda volume, _player=player: setattr(
1655                    _player, "_attr_volume_level", volume
1656                )
1657            )
1658            players[player_id] = player
1659        players["leader"]._attr_group_members = ["member"]
1660        controller._players = dict(players)
1661        mock_mass.players = controller
1662        mock_mass.player_queues.get = MagicMock(return_value=None)
1663        for player in players.values():
1664            player.set_initialized()
1665            player._cache.clear()
1666            player.update_state(signal_event=False)
1667        # a second pass, so the group volume of the leader accounts for its member
1668        for player in players.values():
1669            player.update_state(signal_event=False)
1670        return controller, players
1671
1672    async def test_group_volume_keeps_a_muted_pair_muted(self, mock_mass: MagicMock) -> None:
1673        """A group volume change may not bring a muted fake mute pair back to life."""
1674        controller, players = self._make_synced_pair(mock_mass)
1675        await controller.cmd_group_volume_mute("leader", True)
1676
1677        await controller.cmd_group_volume("leader", 30)
1678
1679        for player in players.values():
1680            player.update_state()
1681            assert player.state.volume_muted is True
1682            assert player.state.volume_level == 0
1683
1684    async def test_group_volume_down_keeps_a_muted_member_muted(self, mock_mass: MagicMock) -> None:
1685        """Turning a group down leaves a single muted member silent, at its own volume."""
1686        controller, players = self._make_synced_pair(mock_mass)
1687        await controller.cmd_volume_mute("member", True)
1688        # let the group volume of the leader account for the muted member
1689        players["leader"].update_state(signal_event=False)
1690
1691        await controller.cmd_group_volume("leader", 25)
1692
1693        for player in players.values():
1694            player.update_state()
1695        member_state = players["member"].state
1696        assert member_state.volume_muted is True
1697        assert member_state.volume_level == 0
1698        # the player that is not muted follows the group volume as usual
1699        assert players["leader"].state.volume_level == 25
1700        # unmuting brings the member back at the volume it had before it was muted
1701        await controller.cmd_volume_mute("member", False)
1702        players["member"].update_state()
1703        assert players["member"].state.volume_level == 50
1704
1705    async def test_unmute_restores_the_volume_from_before_the_mute(
1706        self, mock_mass: MagicMock
1707    ) -> None:
1708        """A group volume change while muted may not alter the volume to restore."""
1709        controller, players = self._make_synced_pair(mock_mass)
1710        await controller.cmd_group_volume_mute("leader", True)
1711        await controller.cmd_group_volume("leader", 30)
1712
1713        await controller.cmd_group_volume_mute("leader", False)
1714
1715        for player in players.values():
1716            player.update_state()
1717            assert player.state.volume_muted is False
1718            assert player.state.volume_level == 50
1719
1720    async def test_group_volume_keeps_a_mixed_pair_muted(self, mock_mass: MagicMock) -> None:
1721        """Members with a different mute control stay muted alike on a group volume change."""
1722        controller, players = self._make_synced_pair(
1723            mock_mass, member_mute_control=PLAYER_CONTROL_NATIVE
1724        )
1725        mute = AsyncMock(
1726            side_effect=lambda muted: setattr(players["member"], "_attr_volume_muted", muted)
1727        )
1728        players["member"].volume_mute = mute  # type: ignore[method-assign]
1729        await controller.cmd_group_volume_mute("leader", True)
1730
1731        await controller.cmd_group_volume("leader", 30)
1732
1733        mute.assert_awaited_once_with(True)
1734        for player in players.values():
1735            player.update_state()
1736            assert player.state.volume_muted is True
1737
1738
1739class TestMuteControlGuard:
1740    """Muting is gated on the mute control, independently of the volume control."""
1741
1742    def _make_player(
1743        self,
1744        mock_mass: MagicMock,
1745        mute_control: str,
1746        volume_control: str,
1747        controls: dict[str, PlayerControl] | None = None,
1748    ) -> tuple[PlayerController, MockPlayer]:
1749        """Build a controller with a single player using the given control config."""
1750        mock_mass.config.get_raw_player_config_value = MagicMock(
1751            side_effect=_player_config_stub(
1752                {CONF_MUTE_CONTROL: mute_control, CONF_VOLUME_CONTROL: volume_control}
1753            )
1754        )
1755        controller = PlayerController(mock_mass)
1756        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
1757        player = MockPlayer(provider, "player_1", "Player 1")
1758        controller._players = {"player_1": player}
1759        controller._controls = controls or {}
1760        mock_mass.players = controller
1761        mock_mass.player_queues.get = MagicMock(return_value=None)
1762        player.set_initialized()
1763        player.update_state(signal_event=False)
1764        return controller, player
1765
1766    async def test_external_mute_control_without_volume_control(self, mock_mass: MagicMock) -> None:
1767        """A player without volume control still mutes through an external PlayerControl."""
1768        mute_set = AsyncMock()
1769        control = PlayerControl(
1770            id="ext_mute",
1771            provider="test",
1772            name="External Mute",
1773            supports_mute=True,
1774            mute_set=mute_set,
1775        )
1776        controller, player = self._make_player(
1777            mock_mass,
1778            mute_control="ext_mute",
1779            volume_control=PLAYER_CONTROL_NONE,
1780            controls={"ext_mute": control},
1781        )
1782        assert player.mute_control == "ext_mute"
1783
1784        await controller.cmd_volume_mute("player_1", True)
1785        mute_set.assert_awaited_once_with(True)
1786
1787    async def test_native_mute_control_without_volume_control(self, mock_mass: MagicMock) -> None:
1788        """A player without volume control still mutes natively."""
1789        controller, player = self._make_player(
1790            mock_mass,
1791            mute_control=PLAYER_CONTROL_NATIVE,
1792            volume_control=PLAYER_CONTROL_NONE,
1793        )
1794        volume_mute = AsyncMock()
1795        player.volume_mute = volume_mute  # type: ignore[method-assign]
1796
1797        await controller.cmd_volume_mute("player_1", True)
1798        volume_mute.assert_awaited_once_with(True)
1799
1800    async def test_mute_control_none_raises(self, mock_mass: MagicMock) -> None:
1801        """A player with volume control but no mute control rejects the command."""
1802        controller, player = self._make_player(
1803            mock_mass,
1804            mute_control=PLAYER_CONTROL_NONE,
1805            volume_control=PLAYER_CONTROL_NATIVE,
1806        )
1807        volume_mute = AsyncMock()
1808        player.volume_mute = volume_mute  # type: ignore[method-assign]
1809
1810        with pytest.raises(UnsupportedFeaturedException):
1811            await controller.cmd_volume_mute("player_1", True)
1812        volume_mute.assert_not_awaited()
1813
1814    async def test_fake_mute_without_volume_control_raises(self, mock_mass: MagicMock) -> None:
1815        """Fake mute needs a volume control to drive, so it rejects the command outright."""
1816        controller, player = self._make_player(
1817            mock_mass,
1818            mute_control=PLAYER_CONTROL_FAKE,
1819            volume_control=PLAYER_CONTROL_NONE,
1820        )
1821        player._attr_volume_level = 40
1822
1823        with pytest.raises(UnsupportedFeaturedException):
1824            await controller.cmd_volume_mute("player_1", True)
1825        assert ATTR_PREVIOUS_VOLUME not in player.extra_data
1826        assert ATTR_FAKE_MUTE not in player.extra_data
1827
1828    async def test_vanished_mute_control_raises(self, mock_mass: MagicMock) -> None:
1829        """A mute control that disappeared after being resolved is reported, not ignored."""
1830        control = PlayerControl(
1831            id="ext_mute",
1832            provider="test",
1833            name="External Mute",
1834            supports_mute=True,
1835            mute_set=AsyncMock(),
1836        )
1837        controller, player = self._make_player(
1838            mock_mass,
1839            mute_control="ext_mute",
1840            volume_control=PLAYER_CONTROL_NONE,
1841            controls={"ext_mute": control},
1842        )
1843        # the resolved control is cached on the player, so removing it here leaves
1844        # the player pointing at a control that no longer exists
1845        assert player.mute_control == "ext_mute"
1846        controller._controls = {}
1847
1848        with pytest.raises(UnsupportedFeaturedException):
1849            await controller.cmd_volume_mute("player_1", True)
1850
1851    async def test_unmute_clears_mute_lock_without_mute_control(self, mock_mass: MagicMock) -> None:
1852        """Unmuting clears a mute lock left behind by a since-removed mute control."""
1853        controller, player = self._make_player(
1854            mock_mass,
1855            mute_control=PLAYER_CONTROL_NONE,
1856            volume_control=PLAYER_CONTROL_NATIVE,
1857        )
1858        player.extra_data[ATTR_MUTE_LOCK] = True
1859
1860        with pytest.raises(UnsupportedFeaturedException):
1861            await controller.cmd_volume_mute("player_1", False)
1862        assert ATTR_MUTE_LOCK not in player.extra_data
1863
1864    async def test_failed_mute_sets_no_mute_lock(self, mock_mass: MagicMock) -> None:
1865        """A grouped player whose mute command failed is not left holding a mute lock."""
1866        control = PlayerControl(
1867            id="ext_mute",
1868            provider="test",
1869            name="External Mute",
1870            supports_mute=False,
1871        )
1872        controller, player = self._make_player(
1873            mock_mass,
1874            mute_control="ext_mute",
1875            volume_control=PLAYER_CONTROL_NONE,
1876            controls={"ext_mute": control},
1877        )
1878        player.state.synced_to = "leader"
1879
1880        with pytest.raises(UnsupportedFeaturedException):
1881            await controller.cmd_volume_mute("player_1", True)
1882        assert ATTR_MUTE_LOCK not in player.extra_data
1883
1884    async def test_failed_mute_keeps_existing_mute_lock(self, mock_mass: MagicMock) -> None:
1885        """A failed mute leaves the lock of an earlier successful mute in place."""
1886        control = PlayerControl(
1887            id="ext_mute",
1888            provider="test",
1889            name="External Mute",
1890            supports_mute=False,
1891        )
1892        controller, player = self._make_player(
1893            mock_mass,
1894            mute_control="ext_mute",
1895            volume_control=PLAYER_CONTROL_NONE,
1896            controls={"ext_mute": control},
1897        )
1898        player.state.synced_to = "leader"
1899        player.extra_data[ATTR_MUTE_LOCK] = True
1900
1901        with pytest.raises(UnsupportedFeaturedException):
1902            await controller.cmd_volume_mute("player_1", True)
1903        assert player.extra_data[ATTR_MUTE_LOCK] is True
1904
1905
1906class TestGroupMuteMemberFilter:
1907    """Group mute skips members that have no mute control of their own."""
1908
1909    async def test_member_without_mute_control_is_skipped(self, mock_mass: MagicMock) -> None:
1910        """A member without a mute control must not fail the whole group command."""
1911        controller = PlayerController(mock_mass)
1912        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
1913        leader = MockPlayer(provider, "leader", "Leader")
1914        leader._attr_supported_features = {PlayerFeature.VOLUME_SET, PlayerFeature.VOLUME_MUTE}
1915        leader._attr_group_members = ["leader", "member"]
1916        member = MockPlayer(provider, "member", "Member")
1917        member._attr_supported_features = {PlayerFeature.VOLUME_SET}
1918        controller._players = {"leader": leader, "member": member}
1919        mock_mass.players = controller
1920        mock_mass.player_queues.get = MagicMock(return_value=None)
1921        for player in (leader, member):
1922            player.set_initialized()
1923            player.update_state(signal_event=False)
1924        leader_mute = AsyncMock()
1925        leader.volume_mute = leader_mute  # type: ignore[method-assign]
1926
1927        await controller.cmd_group_volume_mute("leader", True)
1928        leader_mute.assert_awaited_once_with(True)
1929
1930
1931class TestGroupPlayerMuteRedirect:
1932    """A mute command on a group player is handled at group level."""
1933
1934    def _setup(self, mock_mass: MagicMock) -> tuple[PlayerController, MockPlayer, MockPlayer]:
1935        """Build a controller with a group player holding a single mute capable member."""
1936        controller = PlayerController(mock_mass)
1937        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
1938        group = MockPlayer(provider, "group", "Group", player_type=PlayerType.GROUP)
1939        group._attr_supported_features = {PlayerFeature.VOLUME_SET, PlayerFeature.VOLUME_MUTE}
1940        group._attr_group_members = ["member"]
1941        member = MockPlayer(provider, "member", "Member")
1942        member._attr_supported_features = {PlayerFeature.VOLUME_SET, PlayerFeature.VOLUME_MUTE}
1943        controller._players = {"group": group, "member": member}
1944        mock_mass.players = controller
1945        mock_mass.player_queues.get = MagicMock(return_value=None)
1946        for player in (group, member):
1947            player.set_initialized()
1948            player.update_state(signal_event=False)
1949        return controller, group, member
1950
1951    async def test_mute_on_group_player_is_forwarded_to_members(self, mock_mass: MagicMock) -> None:
1952        """A group player has no mute of its own, so the members must be muted instead."""
1953        controller, _group, member = self._setup(mock_mass)
1954        member_mute = AsyncMock()
1955        member.volume_mute = member_mute  # type: ignore[method-assign]
1956
1957        await controller.cmd_volume_mute("group", True)
1958
1959        member_mute.assert_awaited_once_with(True)
1960
1961    async def test_mute_on_group_player_without_own_mute_control(
1962        self, mock_mass: MagicMock
1963    ) -> None:
1964        """A group that has no mute control of its own must still mute its members."""
1965        controller, group, member = self._setup(mock_mass)
1966        group._attr_supported_features = {PlayerFeature.VOLUME_SET}
1967        group._cache.clear()
1968        group.update_state(signal_event=False)
1969        assert group.mute_control == PLAYER_CONTROL_NONE
1970        member_mute = AsyncMock()
1971        member.volume_mute = member_mute  # type: ignore[method-assign]
1972
1973        await controller.cmd_volume_mute("group", True)
1974
1975        member_mute.assert_awaited_once_with(True)
1976
1977    async def test_mute_on_group_player_without_mute_capable_members(
1978        self, mock_mass: MagicMock
1979    ) -> None:
1980        """A group whose members cannot mute must not raise, just like group mute itself."""
1981        controller, _group, member = self._setup(mock_mass)
1982        member._attr_supported_features = {PlayerFeature.VOLUME_SET}
1983        member._cache.clear()
1984        member.update_state(signal_event=False)
1985
1986        await controller.cmd_volume_mute("group", True)
1987
1988
1989class TestGroupMuteOnNonGroupPlayer:
1990    """A group mute command works on any player, just like the group volume command."""
1991
1992    def _setup(
1993        self, mock_mass: MagicMock, *members: str
1994    ) -> tuple[PlayerController, dict[str, MockPlayer]]:
1995        """Build a controller with a mute capable leader synced to the given members."""
1996        controller = PlayerController(mock_mass)
1997        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
1998        players: dict[str, MockPlayer] = {}
1999        for player_id in ("leader", *members):
2000            player = MockPlayer(provider, player_id, player_id.title())
2001            player._attr_supported_features = {
2002                PlayerFeature.VOLUME_SET,
2003                PlayerFeature.VOLUME_MUTE,
2004            }
2005            player._attr_volume_level = 50
2006            players[player_id] = player
2007        if members:
2008            # the leader is not listed as its own member here, so the tests also cover
2009            # that a sync leader is injected into its own final group_members
2010            players["leader"]._attr_group_members = list(members)
2011        controller._players = dict(players)
2012        mock_mass.players = controller
2013        mock_mass.player_queues.get = MagicMock(return_value=None)
2014        for player in players.values():
2015            player.set_initialized()
2016            player._cache.clear()
2017            player.update_state(signal_event=False)
2018        return controller, players
2019
2020    def _stub_mutes(self, players: dict[str, MockPlayer]) -> dict[str, AsyncMock]:
2021        """Replace the native mute command of every given player with a mock."""
2022        mutes: dict[str, AsyncMock] = {}
2023        for player_id, player in players.items():
2024            mutes[player_id] = AsyncMock()
2025            player.volume_mute = mutes[player_id]  # type: ignore[method-assign]
2026        return mutes
2027
2028    async def test_group_mute_on_synced_member_redirects_to_leader(
2029        self, mock_mass: MagicMock
2030    ) -> None:
2031        """A member of a sync group mutes the whole group through its sync leader."""
2032        controller, players = self._setup(mock_mass, "member")
2033        assert players["member"].state.synced_to == "leader"
2034        mutes = self._stub_mutes(players)
2035
2036        await controller.cmd_group_volume_mute("member", True)
2037
2038        mutes["leader"].assert_awaited_once_with(True)
2039        mutes["member"].assert_awaited_once_with(True)
2040        assert ATTR_MUTE_LOCK in players["member"].extra_data
2041
2042    async def test_group_mute_on_sync_leader_mutes_the_leader_once(
2043        self, mock_mass: MagicMock
2044    ) -> None:
2045        """A sync leader is part of its own member list, so it must be muted only once."""
2046        controller, players = self._setup(mock_mass, "member")
2047        mutes = self._stub_mutes(players)
2048
2049        await controller.cmd_group_volume_mute("leader", True)
2050
2051        mutes["leader"].assert_awaited_once_with(True)
2052        mutes["member"].assert_awaited_once_with(True)
2053
2054    async def test_group_mute_on_plain_player_mutes_that_player(self, mock_mass: MagicMock) -> None:
2055        """A player that is not grouped at all is muted as a normal player."""
2056        controller, players = self._setup(mock_mass)
2057        mutes = self._stub_mutes(players)
2058
2059        await controller.cmd_group_volume_mute("leader", True)
2060
2061        mutes["leader"].assert_awaited_once_with(True)
2062
2063    async def test_group_mute_on_plain_player_without_mute_control(
2064        self, mock_mass: MagicMock
2065    ) -> None:
2066        """A plain player that cannot mute reports that, just like a normal mute command."""
2067        controller, players = self._setup(mock_mass)
2068        players["leader"]._attr_supported_features = {PlayerFeature.VOLUME_SET}
2069        players["leader"]._cache.clear()
2070        players["leader"].update_state(signal_event=False)
2071
2072        with pytest.raises(UnsupportedFeaturedException):
2073            await controller.cmd_group_volume_mute("leader", True)
2074
2075    async def test_group_unmute_on_synced_member_redirects_to_leader(
2076        self, mock_mass: MagicMock
2077    ) -> None:
2078        """Unmuting through a member clears the mute (and mute lock) of every group member."""
2079        controller, players = self._setup(mock_mass, "member")
2080        mutes = self._stub_mutes(players)
2081        players["member"].extra_data[ATTR_MUTE_LOCK] = True
2082
2083        await controller.cmd_group_volume_mute("member", False)
2084
2085        mutes["leader"].assert_awaited_once_with(False)
2086        mutes["member"].assert_awaited_once_with(False)
2087        assert ATTR_MUTE_LOCK not in players["member"].extra_data
2088
2089    async def test_group_mute_locks_the_sync_leader_too(self, mock_mass: MagicMock) -> None:
2090        """A sync leader is as much part of the group as its members, so it is locked too."""
2091        controller, players = self._setup(mock_mass, "member")
2092        self._stub_mutes(players)
2093
2094        await controller.cmd_group_volume_mute("leader", True)
2095
2096        assert ATTR_MUTE_LOCK in players["leader"].extra_data
2097        assert ATTR_MUTE_LOCK in players["member"].extra_data
2098
2099    async def test_group_volume_keeps_a_muted_sync_pair_muted(self, mock_mass: MagicMock) -> None:
2100        """A group volume change may not half-unmute a muted pair of directly synced players."""
2101        controller, players = self._setup(mock_mass, "member")
2102        self._stub_mutes(players)
2103        await controller.cmd_group_volume_mute("leader", True)
2104        # the mock players do not act on the mute command, so reflect it in their state
2105        for player in players.values():
2106            player._attr_volume_muted = True
2107            player.update_state(signal_event=False)
2108            player.volume_set = AsyncMock()  # type: ignore[method-assign]
2109        # re-stub so only the mute commands of the group volume change are counted
2110        mutes = self._stub_mutes(players)
2111
2112        await controller.cmd_group_volume("leader", 30)
2113
2114        for mute in mutes.values():
2115            mute.assert_not_awaited()
2116
2117
2118class TestMuteLockAfterUngroup:
2119    """A mute lock is only honored while the player it belongs to is still grouped."""
2120
2121    def _make_synced_pair(
2122        self, mock_mass: MagicMock, member_mute_control: str
2123    ) -> tuple[PlayerController, dict[str, MockPlayer]]:
2124        """
2125        Build a leader with one synced member.
2126
2127        :param member_mute_control: Mute control to configure on both players.
2128        """
2129        mock_mass.config.get_raw_player_config_value = MagicMock(
2130            side_effect=_player_config_stub({CONF_MUTE_CONTROL: member_mute_control})
2131        )
2132        controller = PlayerController(mock_mass)
2133        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
2134        players: dict[str, MockPlayer] = {}
2135        for player_id in ("leader", "member"):
2136            player = MockPlayer(provider, player_id, player_id.title())
2137            player._attr_supported_features = {
2138                PlayerFeature.VOLUME_SET,
2139                PlayerFeature.VOLUME_MUTE,
2140            }
2141            player._attr_volume_level = 50
2142            player.volume_set = AsyncMock(  # type: ignore[method-assign]
2143                side_effect=lambda volume, _player=player: setattr(
2144                    _player, "_attr_volume_level", volume
2145                )
2146            )
2147            players[player_id] = player
2148        players["leader"]._attr_group_members = ["member"]
2149        controller._players = dict(players)
2150        mock_mass.players = controller
2151        mock_mass.player_queues.get = MagicMock(return_value=None)
2152        for player in players.values():
2153            player.set_initialized()
2154            player._cache.clear()
2155            player.update_state(signal_event=False)
2156        return controller, players
2157
2158    def _dissolve_group(self, players: dict[str, MockPlayer]) -> None:
2159        """Drop the sync group, the way a provider side topology change does."""
2160        players["leader"]._attr_group_members = []
2161        for player in players.values():
2162            player.refresh_state(signal_event=False)
2163
2164    async def test_fake_muted_player_follows_volume_again(self, mock_mass: MagicMock) -> None:
2165        """A fake muted player is no longer forced silent once its group is gone."""
2166        controller, players = self._make_synced_pair(mock_mass, PLAYER_CONTROL_FAKE)
2167        await controller.cmd_volume_mute("member", True)
2168        self._dissolve_group(players)
2169
2170        await controller.cmd_volume_set("member", 70)
2171
2172        players["member"].update_state()
2173        assert players["member"].state.volume_level == 70
2174        assert players["member"].state.volume_muted is False
2175
2176    async def test_natively_muted_player_is_unmuted_again(self, mock_mass: MagicMock) -> None:
2177        """A natively muted player is auto-unmuted by a volume change once its group is gone."""
2178        controller, players = self._make_synced_pair(mock_mass, PLAYER_CONTROL_NATIVE)
2179        mute = AsyncMock(
2180            side_effect=lambda muted: setattr(players["member"], "_attr_volume_muted", muted)
2181        )
2182        players["member"].volume_mute = mute  # type: ignore[method-assign]
2183        await controller.cmd_volume_mute("member", True)
2184        self._dissolve_group(players)
2185
2186        await controller.cmd_volume_set("member", 70)
2187
2188        assert mute.await_args_list == [call(True), call(False)]
2189        players["member"].update_state()
2190        assert players["member"].state.volume_level == 70
2191        assert players["member"].state.volume_muted is False
2192
2193    async def test_still_grouped_player_keeps_its_lock(self, mock_mass: MagicMock) -> None:
2194        """A muted player that is still grouped stays silent on a volume change."""
2195        controller, players = self._make_synced_pair(mock_mass, PLAYER_CONTROL_FAKE)
2196        await controller.cmd_volume_mute("member", True)
2197
2198        await controller.cmd_volume_set("member", 70)
2199
2200        players["member"].update_state()
2201        assert players["member"].state.volume_level == 0
2202        assert players["member"].state.volume_muted is True
2203
2204    async def test_protocol_player_follows_the_lock_of_its_parent(
2205        self, mock_mass: MagicMock
2206    ) -> None:
2207        """A protocol player inherits the lock of the parent it renders for, group and all."""
2208        controller, players = self._make_synced_pair(mock_mass, PLAYER_CONTROL_NATIVE)
2209        member = players["member"]
2210        protocol_player = MockPlayer(
2211            MockProvider("sendspin", instance_id="sendspin", mass=mock_mass),
2212            "proto_member",
2213            "Member Protocol",
2214            player_type=PlayerType.PROTOCOL,
2215        )
2216        protocol_player._attr_supported_features = {
2217            PlayerFeature.VOLUME_SET,
2218            PlayerFeature.VOLUME_MUTE,
2219        }
2220        protocol_player._attr_volume_muted = True
2221        protocol_player.set_protocol_parent_id("member")
2222        # an unmute on a protocol player is redirected to the parent it renders for
2223        mute = AsyncMock()
2224        member.volume_mute = mute  # type: ignore[method-assign]
2225        protocol_player.volume_mute = AsyncMock()  # type: ignore[method-assign]
2226        protocol_player.volume_set = AsyncMock()  # type: ignore[method-assign]
2227        controller._players["proto_member"] = protocol_player
2228        member.set_linked_output_protocols(
2229            [
2230                OutputProtocol(
2231                    output_protocol_id="proto_member",
2232                    name="Sendspin",
2233                    protocol_domain="sendspin",
2234                    priority=40,
2235                )
2236            ]
2237        )
2238        protocol_player.set_initialized()
2239        protocol_player.update_state(signal_event=False)
2240        member.refresh_state(signal_event=False)
2241        member.extra_data[ATTR_MUTE_LOCK] = True
2242
2243        # a group volume change reaches the protocol player through the internal handler
2244        await controller._handle_cmd_volume_set("proto_member", 70)
2245        mute.assert_not_awaited()
2246
2247        self._dissolve_group(players)
2248        await controller._handle_cmd_volume_set("proto_member", 70)
2249
2250        mute.assert_awaited_once_with(False)
2251
2252
2253class TestCurrentMediaTimeUpdates:
2254    """Playback-position anchor semantics of timing-only state updates."""
2255
2256    def _make_player(self, mock_mass: MagicMock) -> tuple[PlayerController, MockPlayer]:
2257        """Build a controller with a single playing player with a known position anchor."""
2258        controller = PlayerController(mock_mass)
2259        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
2260        player = MockPlayer(provider, "player_1", "Player 1")
2261        controller._players = {"player_1": player}
2262        mock_mass.players = controller
2263        # no queue registered: current_media resolves from the player's native media
2264        mock_mass.player_queues.get = MagicMock(return_value=None)
2265        player.set_initialized()
2266        now = time.time()
2267        player._attr_playback_state = PlaybackState.PLAYING
2268        player._attr_elapsed_time = 17
2269        player._attr_elapsed_time_last_updated = now
2270        player.set_current_media(uri="http://test/stream", title="Test")
2271        assert player._attr_current_media is not None
2272        player._attr_current_media.elapsed_time = 17
2273        player._attr_current_media.elapsed_time_last_updated = now
2274        player.update_state(signal_event=False)
2275        # isolate from the unrelated state-forwarding machinery
2276        controller._forward_state_update = MagicMock()  # type: ignore[method-assign]
2277        mock_mass.signal_event.reset_mock()
2278        mock_mass.player_queues.on_player_elapsed_time_corrected.reset_mock()
2279        return controller, player
2280
2281    def _player_updated_signalled(self, mock_mass: MagicMock) -> bool:
2282        """Return whether a PLAYER_UPDATED event was signalled."""
2283        return any(
2284            call.args and call.args[0] == EventType.PLAYER_UPDATED
2285            for call in mock_mass.signal_event.call_args_list
2286        )
2287
2288    def test_regular_tick_is_suppressed(self, mock_mass: MagicMock) -> None:
2289        """A regular playback tick (position and anchor advance together) emits nothing."""
2290        _controller, player = self._make_player(mock_mass)
2291        assert player._attr_current_media is not None
2292        assert player._attr_elapsed_time_last_updated is not None
2293
2294        player._attr_elapsed_time = 18
2295        player._attr_elapsed_time_last_updated += 1
2296        player._attr_current_media.elapsed_time = 18
2297        assert player._attr_current_media.elapsed_time_last_updated is not None
2298        player._attr_current_media.elapsed_time_last_updated += 1
2299        player.update_state()
2300
2301        assert not self._player_updated_signalled(mock_mass)
2302        mock_mass.player_queues.on_player_elapsed_time_corrected.assert_not_called()
2303        # the previous anchor was preserved: steady playback changes nothing
2304        assert player.state.elapsed_time == 17
2305
2306    def test_anchor_only_change_is_suppressed(self, mock_mass: MagicMock) -> None:
2307        """An anchor-only change (no significant corrected position change) emits nothing."""
2308        _controller, player = self._make_player(mock_mass)
2309        assert player._attr_current_media is not None
2310        assert player._attr_elapsed_time_last_updated is not None
2311
2312        player._attr_elapsed_time_last_updated += 0.5
2313        assert player._attr_current_media.elapsed_time_last_updated is not None
2314        player._attr_current_media.elapsed_time_last_updated += 0.5
2315        player.update_state()
2316
2317        assert not self._player_updated_signalled(mock_mass)
2318        mock_mass.player_queues.on_player_elapsed_time_corrected.assert_not_called()
2319
2320    def test_corrected_position_jump_emits_player_updated(self, mock_mass: MagicMock) -> None:
2321        """A corrected-position jump of the current media (e.g. seek) emits a player update."""
2322        _controller, player = self._make_player(mock_mass)
2323        assert player._attr_current_media is not None
2324
2325        player._attr_current_media.elapsed_time = 61
2326        player._attr_current_media.elapsed_time_last_updated = time.time()
2327        player.update_state()
2328
2329        assert self._player_updated_signalled(mock_mass)
2330        # the adopted anchor is visible to consumers
2331        assert player.state.current_media is not None
2332        assert player.state.current_media.elapsed_time == 61
2333
2334    def test_player_position_jump_corrects_queue(self, mock_mass: MagicMock) -> None:
2335        """A player-level corrected-position jump re-bases the queue timing."""
2336        controller, player = self._make_player(mock_mass)
2337
2338        player._attr_elapsed_time = 61
2339        player._attr_elapsed_time_last_updated = time.time()
2340        player.update_state()
2341
2342        # the queue is corrected and a follow-up player update is scheduled
2343        # (which re-anchors current_media onto the corrected queue time),
2344        # but no full player update is emitted for the jump itself
2345        mock_mass.player_queues.on_player_elapsed_time_corrected.assert_called_once_with(player)
2346        assert not self._player_updated_signalled(mock_mass)
2347        cast("MagicMock", controller._forward_state_update).assert_called_once()
2348        assert player.state.elapsed_time == 61
2349
2350    def test_simultaneous_player_and_media_jump_emits_immediately(
2351        self, mock_mass: MagicMock
2352    ) -> None:
2353        """A jump reaching player and current_media in one pass corrects the queue and emits."""
2354        _controller, player = self._make_player(mock_mass)
2355        assert player._attr_current_media is not None
2356        now = time.time()
2357
2358        player._attr_elapsed_time = 61
2359        player._attr_elapsed_time_last_updated = now
2360        player._attr_current_media.elapsed_time = 61
2361        player._attr_current_media.elapsed_time_last_updated = now
2362        player.update_state()
2363
2364        # the queue is re-based AND the full update is emitted right away
2365        # (current_media already holds the fresh position in the same pass)
2366        mock_mass.player_queues.on_player_elapsed_time_corrected.assert_called_once_with(player)
2367        assert self._player_updated_signalled(mock_mass)
2368
2369
2370class TestPlayAnnouncementCleanup:
2371    """Test announcement data cleanup after play_announcement."""
2372
2373    def _make_player(
2374        self, mock_mass: MagicMock, announcements: dict[str, object]
2375    ) -> tuple[PlayerController, MockPlayer, MagicMock]:
2376        """Create a controller and a player with native announcement support."""
2377        controller = PlayerController(mock_mass)
2378        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
2379        player = MockPlayer(provider, "player_1", "Player 1")
2380        player._attr_supported_features.add(PlayerFeature.PLAY_ANNOUNCEMENT)
2381        player._cache.clear()
2382        controller._players = {"player_1": player}
2383        mock_mass.players = controller
2384        render = MagicMock()
2385        render.wait_ready = AsyncMock(return_value=True)
2386        render.wait_finished = AsyncMock(return_value=3.0)
2387
2388        # mimic the real renderer: it owns which announcement each player is playing
2389        def _register(player_id: str, announce_data: object) -> MagicMock:
2390            announcements[player_id] = announce_data
2391            return render
2392
2393        async def _unregister(player_id: str, _render: object) -> None:
2394            announcements.pop(player_id, None)
2395
2396        renderer = mock_mass.streams.announcement_renderer
2397        renderer.register = MagicMock(side_effect=_register)
2398        renderer.unregister = AsyncMock(side_effect=_unregister)
2399        mock_mass.streams.get_announcement_url = MagicMock(
2400            side_effect=lambda player_id, **_kwargs: f"http://ma/announcement/{player_id}.mp3"
2401        )
2402        player.update_state(signal_event=False)
2403        return controller, player, render
2404
2405    async def test_announcement_data_removed_after_playback(self, mock_mass: MagicMock) -> None:
2406        """The registered announcement data is released once playback finished."""
2407        announcements: dict[str, object] = {}
2408        controller, player, _render = self._make_player(mock_mass, announcements)
2409
2410        async def _play_announcement(*_args: object, **_kwargs: object) -> None:
2411            # entry must exist while the announcement is being played/served
2412            assert "player_1" in announcements
2413
2414        player.play_announcement = AsyncMock(side_effect=_play_announcement)  # type: ignore[method-assign]
2415
2416        await controller.play_announcement("player_1", "http://test/announcement.mp3")
2417
2418        player.play_announcement.assert_awaited_once()
2419        assert announcements == {}
2420        mock_mass.streams.announcement_renderer.unregister.assert_awaited_once()
2421
2422    async def test_announcement_data_removed_on_error(self, mock_mass: MagicMock) -> None:
2423        """The registered announcement data is released even when playback fails."""
2424        announcements: dict[str, object] = {}
2425        controller, player, _render = self._make_player(mock_mass, announcements)
2426        player.play_announcement = AsyncMock(side_effect=RuntimeError("boom"))  # type: ignore[method-assign]
2427
2428        with pytest.raises(PlayerCommandFailed):
2429            await controller.play_announcement("player_1", "http://test/announcement.mp3")
2430
2431        assert announcements == {}
2432        mock_mass.streams.announcement_renderer.unregister.assert_awaited_once()
2433
2434    async def test_native_announcement_starts_on_first_audio(self, mock_mass: MagicMock) -> None:
2435        """A native implementation is handed the url as soon as there is audio to serve."""
2436        announcements: dict[str, object] = {}
2437        controller, player, render = self._make_player(mock_mass, announcements)
2438        player.play_announcement = AsyncMock()  # type: ignore[method-assign]
2439
2440        await controller.play_announcement("player_1", "http://test/announcement.mp3")
2441
2442        player.play_announcement.assert_awaited_once()
2443        # waiting for the whole clip here would delay the player for a slow source;
2444        # the length is resolved downstream while it plays
2445        render.wait_ready.assert_awaited_once()
2446        render.wait_finished.assert_not_awaited()
2447
2448
2449class TestPlayAnnouncementRestore:
2450    """Test the state restore of the default (fallback) announcement implementation."""
2451
2452    def _make_player(
2453        self, mock_mass: MagicMock, prev_media: PlayerMedia
2454    ) -> tuple[PlayerController, MockPlayer, AsyncMock]:
2455        """Create a controller and a playing player, returning the patched resume handler."""
2456        controller = PlayerController(mock_mass)
2457        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
2458        player = MockPlayer(provider, "player_1", "Player 1")
2459        player._attr_playback_state = PlaybackState.PLAYING
2460        player._attr_current_media = prev_media
2461        player._cache.clear()
2462        controller._players = {"player_1": player}
2463        mock_mass.players = controller
2464        # the (temporary and restored) volume commands are dispatched through the
2465        # TaskManager, so background tasks must actually run in these tests
2466        mock_mass.create_task = MagicMock(
2467            side_effect=lambda coro, **_kwargs: asyncio.ensure_future(coro)
2468        )
2469        resume_mock = AsyncMock()
2470        controller._handle_cmd_resume = resume_mock  # type: ignore[method-assign]
2471        controller._handle_cmd_stop = AsyncMock()  # type: ignore[method-assign]
2472        controller._handle_play_media = AsyncMock()  # type: ignore[method-assign]
2473        controller._wait_for_playback_state = AsyncMock()  # type: ignore[method-assign]
2474        controller.get_announcement_volume = MagicMock(return_value=None)  # type: ignore[method-assign]
2475        player.set_initialized()
2476        player.update_state(signal_event=False)
2477        return controller, player, resume_mock
2478
2479    @staticmethod
2480    def _add_group(
2481        controller: PlayerController, player: MockPlayer, *, supports_set_members: bool
2482    ) -> MockPlayer:
2483        """Register a powered group player that holds the given player as its member."""
2484        group = MockPlayer(
2485            cast("MockProvider", player.provider),
2486            "group_1",
2487            "Group 1",
2488            player_type=PlayerType.GROUP,
2489        )
2490        group._attr_powered = True
2491        group._attr_group_members = [player.player_id]
2492        if supports_set_members:
2493            group._attr_supported_features.add(PlayerFeature.SET_MEMBERS)
2494        controller._players[group.player_id] = group
2495        group.set_initialized()
2496        group._cache.clear()
2497        group.update_state(signal_event=False)
2498        # the member has no own input change to trigger a recalculation of its
2499        # (group derived) state, so force it here - just like register() does
2500        player._cache.clear()
2501        player.update_state(force_update=True, signal_event=False)
2502        assert player.state.active_group == group.player_id
2503        return group
2504
2505    @staticmethod
2506    def _announcement() -> PlayerMedia:
2507        """Return the announcement to play."""
2508        return PlayerMedia(
2509            uri="http://ma/announcement/player_1.mp3",
2510            media_type=MediaType.ANNOUNCEMENT,
2511            title="Announcement",
2512            duration=3,
2513        )
2514
2515    async def test_previous_playback_is_restored(self, mock_mass: MagicMock) -> None:
2516        """Content that was playing before the announcement is resumed afterwards."""
2517        controller, player, resume_mock = self._make_player(
2518            mock_mass, PlayerMedia(uri="http://test/track.mp3", media_type=MediaType.TRACK)
2519        )
2520
2521        await controller._play_announcement(player, self._announcement())
2522
2523        resume_mock.assert_awaited_once()
2524
2525    async def test_previous_announcement_is_not_restored(self, mock_mass: MagicMock) -> None:
2526        """A player still busy with an earlier announcement has no playback to restore."""
2527        controller, player, resume_mock = self._make_player(
2528            mock_mass,
2529            PlayerMedia(uri="http://ma/announcement/x.mp3", media_type=MediaType.ANNOUNCEMENT),
2530        )
2531
2532        await controller._play_announcement(player, self._announcement())
2533
2534        resume_mock.assert_not_awaited()
2535
2536    async def test_volume_is_restored_when_playback_fails(self, mock_mass: MagicMock) -> None:
2537        """A failing announcement never leaves the player at the raised volume."""
2538        controller, player, _ = self._make_player(
2539            mock_mass, PlayerMedia(uri="http://test/track.mp3", media_type=MediaType.TRACK)
2540        )
2541        player._attr_volume_level = 20
2542        player._cache.clear()
2543        player.update_state(force_update=True, signal_event=False)
2544        controller.get_announcement_volume = MagicMock(return_value=80)  # type: ignore[method-assign]
2545        volume_mock = AsyncMock()
2546        controller._handle_cmd_volume_set = volume_mock  # type: ignore[method-assign]
2547        controller._handle_play_media = AsyncMock(  # type: ignore[method-assign]
2548            side_effect=PlayerCommandFailed("player went away")
2549        )
2550
2551        with pytest.raises(PlayerCommandFailed):
2552            await controller._play_announcement(player, self._announcement())
2553
2554        assert volume_mock.call_args_list == [call("player_1", 80), call("player_1", 20)]
2555
2556    async def test_zero_announcement_volume_is_applied_and_restored(
2557        self, mock_mass: MagicMock
2558    ) -> None:
2559        """An announcement volume of 0 is a real volume, not an 'unset' fallback."""
2560        controller, player, _ = self._make_player(
2561            mock_mass, PlayerMedia(uri="http://test/track.mp3", media_type=MediaType.TRACK)
2562        )
2563        player._attr_volume_level = 20
2564        player._cache.clear()
2565        player.update_state(force_update=True, signal_event=False)
2566        controller.get_announcement_volume = MagicMock(return_value=0)  # type: ignore[method-assign]
2567        volume_mock = AsyncMock()
2568        controller._handle_cmd_volume_set = volume_mock  # type: ignore[method-assign]
2569
2570        await controller._play_announcement(player, self._announcement())
2571
2572        assert volume_mock.call_args_list == [call("player_1", 0), call("player_1", 20)]
2573
2574    async def test_playback_is_restored_when_duration_is_unknown(
2575        self, mock_mass: MagicMock
2576    ) -> None:
2577        """An announcement of unknown length still hands the player back to its content."""
2578        controller, player, resume_mock = self._make_player(
2579            mock_mass, PlayerMedia(uri="http://test/track.mp3", media_type=MediaType.TRACK)
2580        )
2581        announcement = self._announcement()
2582        announcement.duration = None
2583
2584        # an unknown length waits for the player to report it finished instead of failing
2585        await controller._play_announcement(player, announcement)
2586
2587        resume_mock.assert_awaited_once()
2588
2589    async def test_group_membership_is_restored_when_playback_fails(
2590        self, mock_mass: MagicMock
2591    ) -> None:
2592        """A failing announcement never leaves the player out of its group player."""
2593        controller, player, _ = self._make_player(
2594            mock_mass, PlayerMedia(uri="http://test/track.mp3", media_type=MediaType.TRACK)
2595        )
2596        group = self._add_group(controller, player, supports_set_members=True)
2597        group.set_members = AsyncMock()  # type: ignore[method-assign]
2598        controller._handle_play_media = AsyncMock(  # type: ignore[method-assign]
2599            side_effect=PlayerCommandFailed("player went away")
2600        )
2601
2602        with pytest.raises(PlayerCommandFailed):
2603            await controller._play_announcement(player, self._announcement())
2604
2605        assert group.set_members.await_args_list == [
2606            call(player_ids_to_remove=["player_1"]),
2607            call(player_ids_to_add=["player_1"]),
2608        ]
2609
2610    async def test_restore_failure_does_not_mask_the_announcement_error(
2611        self, mock_mass: MagicMock
2612    ) -> None:
2613        """A provider blowing up during the restore must not hide why the announcement failed."""
2614        controller, player, _ = self._make_player(
2615            mock_mass, PlayerMedia(uri="http://test/track.mp3", media_type=MediaType.TRACK)
2616        )
2617        group = self._add_group(controller, player, supports_set_members=True)
2618        # set_members is a raw provider call: whatever its client library raises comes
2619        # through unwrapped, so the ungroup succeeds and the regroup times out
2620        group.set_members = AsyncMock(  # type: ignore[method-assign]
2621            side_effect=[None, TimeoutError("provider timeout")]
2622        )
2623        controller._handle_play_media = AsyncMock(  # type: ignore[method-assign]
2624            side_effect=PlayerCommandFailed("player went away")
2625        )
2626
2627        with pytest.raises(PlayerCommandFailed, match="player went away"):
2628            await controller._play_announcement(player, self._announcement())
2629
2630    async def test_group_without_set_members_is_the_one_powered_off(
2631        self, mock_mass: MagicMock
2632    ) -> None:
2633        """A group that can not release members is powered off, not the announcement target."""
2634        controller, player, _ = self._make_player(
2635            mock_mass, PlayerMedia(uri="http://test/track.mp3", media_type=MediaType.TRACK)
2636        )
2637        self._add_group(controller, player, supports_set_members=False)
2638        power_mock = AsyncMock()
2639        controller._handle_cmd_power = power_mock  # type: ignore[method-assign]
2640        play_mock = AsyncMock()
2641        controller.cmd_play = play_mock  # type: ignore[method-assign]
2642
2643        await controller._play_announcement(player, self._announcement())
2644
2645        # the group is switched off for the announcement and restarted afterwards
2646        power_mock.assert_awaited_once_with("group_1", False)
2647        play_mock.assert_awaited_once_with("group_1")
2648
2649    async def test_idle_player_without_power_control_is_regrouped(
2650        self, mock_mass: MagicMock
2651    ) -> None:
2652        """An idle player that has no power state to restore is still put back in its group."""
2653        controller, player, _ = self._make_player(
2654            mock_mass, PlayerMedia(uri="http://test/track.mp3", media_type=MediaType.TRACK)
2655        )
2656        player._attr_playback_state = PlaybackState.IDLE
2657        player._attr_powered = None
2658        group = self._add_group(controller, player, supports_set_members=True)
2659        assert player.state.power_control == PLAYER_CONTROL_NONE
2660        group.set_members = AsyncMock()  # type: ignore[method-assign]
2661
2662        await controller._play_announcement(player, self._announcement())
2663
2664        assert group.set_members.await_args_list == [
2665            call(player_ids_to_remove=["player_1"]),
2666            call(player_ids_to_add=["player_1"]),
2667        ]
2668
2669
2670class TestScheduleActiveOutputProtocolClear:
2671    """Test the deferred clear of a player's active output protocol."""
2672
2673    def test_schedule_starts_cancellable_clear_task(self, mock_mass: MagicMock) -> None:
2674        """Scheduling defers the clear to a single, per-player, cancellable task."""
2675        controller = PlayerController(mock_mass)
2676        player = MagicMock()
2677        player.player_id = "player_1"
2678
2679        controller.schedule_active_output_protocol_clear(player)
2680
2681        mock_mass.create_task.assert_called_once()
2682        # close the coroutine passed to the mocked create_task to avoid a
2683        # "coroutine was never awaited" warning
2684        mock_mass.create_task.call_args.args[0].close()
2685        # no abort_existing: a duplicate schedule must reuse the pending clear
2686        # (deduped by task_id) instead of replacing it with an untracked task
2687        assert mock_mass.create_task.call_args.kwargs == {
2688            "task_id": "clear_active_protocol_player_1",
2689        }
2690
2691    @pytest.mark.asyncio
2692    async def test_clears_protocol_once_player_idle(self, mock_mass: MagicMock) -> None:
2693        """The protocol is cleared after waiting for the player to reach IDLE."""
2694        controller = PlayerController(mock_mass)
2695        player = MagicMock()
2696        player.player_id = "player_1"
2697
2698        with patch.object(controller, "_wait_for_playback_state", new=AsyncMock()) as wait_mock:
2699            await controller._clear_active_output_protocol_when_idle(player)
2700
2701        wait_mock.assert_awaited_once_with(player, PlaybackState.IDLE, timeout=10)
2702        player.set_active_output_protocol.assert_called_once_with(None)
2703
2704
2705@contextlib.asynccontextmanager
2706async def _skip_player_update_wait(
2707    *_args: object,
2708    **_kwargs: object,
2709) -> AsyncIterator[None]:
2710    """Skip provider-driven state propagation in command-routing tests."""
2711    yield
2712
2713
2714class TestRemovePlayerControl:
2715    """Test removing a registered player control."""
2716
2717    def test_removal_refreshes_the_players_that_used_it(self, mock_mass: MagicMock) -> None:
2718        """Test that only the players configured to use the removed control are refreshed."""
2719        mock_mass.loop = MagicMock()
2720        controller = PlayerController(mock_mass)
2721        using_control = MagicMock()
2722        using_control.state.power_control = "switch.amp"
2723        using_control.state.volume_control = PLAYER_CONTROL_NATIVE
2724        using_control.state.mute_control = PLAYER_CONTROL_NATIVE
2725        unrelated = MagicMock()
2726        unrelated.state.power_control = PLAYER_CONTROL_NATIVE
2727        unrelated.state.volume_control = PLAYER_CONTROL_NATIVE
2728        unrelated.state.mute_control = PLAYER_CONTROL_NATIVE
2729        controller._players = {"using_control": using_control, "unrelated": unrelated}
2730        controller._controls = {
2731            "switch.amp": PlayerControl(id="switch.amp", provider="test_prov", name="Amp")
2732        }
2733
2734        controller.remove_player_control("switch.amp")
2735
2736        assert controller.player_controls() == []
2737        mock_mass.loop.call_soon.assert_called_once_with(using_control.refresh_state)
2738
2739    async def test_a_returning_control_is_picked_back_up(self, mock_mass: MagicMock) -> None:
2740        """Test that a control removed and registered again re-attaches to its player."""
2741        # run the scheduled refresh straight away so each step is observable
2742        mock_mass.loop = MagicMock()
2743        mock_mass.loop.call_soon.side_effect = lambda callback, *args: callback(*args)
2744        mock_mass.config.get_raw_player_config_value.side_effect = _player_config_stub(
2745            {CONF_POWER_CONTROL: "switch.amp"}
2746        )
2747        controller = PlayerController(mock_mass)
2748        mock_mass.players = controller
2749        provider = MockProvider("test_provider", instance_id="test_prov", mass=mock_mass)
2750        mock_mass.get_provider.return_value = provider
2751        player = MockPlayer(provider, "player", "Player")
2752        controller._players = {"player": player}
2753        control = PlayerControl(id="switch.amp", provider="test_prov", name="Amp")
2754
2755        await controller.register_or_update_player_control(control)
2756        assert player.state.power_control == "switch.amp"
2757
2758        # the Home Assistant plugin drops and re-registers its controls around a reload
2759        controller.remove_player_control(control.id)
2760        assert player.state.power_control == PLAYER_CONTROL_NONE
2761
2762        await controller.register_or_update_player_control(control)
2763        assert player.state.power_control == "switch.amp"
2764
2765    def test_removing_an_unknown_control_does_nothing(self, mock_mass: MagicMock) -> None:
2766        """Test that removing a control that was never registered is a no-op."""
2767        mock_mass.loop = MagicMock()
2768        controller = PlayerController(mock_mass)
2769        player = MagicMock()
2770        player.state.power_control = PLAYER_CONTROL_NATIVE
2771        player.state.volume_control = PLAYER_CONTROL_NATIVE
2772        player.state.mute_control = PLAYER_CONTROL_NATIVE
2773        controller._players = {"player": player}
2774
2775        controller.remove_player_control("switch.gone")
2776
2777        mock_mass.loop.call_soon.assert_not_called()
2778
2779
2780if __name__ == "__main__":
2781    pytest.main([__file__, "-v"])
2782