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