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, Iterator
17from types import SimpleNamespace
18from typing import Any, NamedTuple, cast
19from unittest.mock import ANY, AsyncMock, MagicMock, call, patch
20
21import pytest
22from music_assistant_models.auth import User, UserRole
23from music_assistant_models.config_entries import ConfigEntry, CoreConfig, PlayerConfig
24from music_assistant_models.constants import (
25    PLAYER_CONTROL_FAKE,
26    PLAYER_CONTROL_NATIVE,
27    PLAYER_CONTROL_NONE,
28)
29from music_assistant_models.enums import (
30    ConfigEntryType,
31    EventType,
32    MediaType,
33    PlaybackState,
34    PlayerFeature,
35    PlayerType,
36)
37from music_assistant_models.errors import (
38    InvalidDataError,
39    MusicAssistantError,
40    PlayerCommandFailed,
41    UnsupportedFeaturedException,
42)
43from music_assistant_models.player import PlayerMedia, PlayerSource
44from music_assistant_models.player_control import PlayerControl
45
46from music_assistant.constants import (
47    ANNOUNCE_ALERT_FILE,
48    ATTR_FAKE_MUTE,
49    ATTR_MUTE_LOCK,
50    ATTR_PREVIOUS_VOLUME,
51    CONF_AUTO_PLAY,
52    CONF_ENTRY_TTS_PRE_ANNOUNCE,
53    CONF_ICON,
54    CONF_MAX_VOLUME,
55    CONF_MIN_VOLUME,
56    CONF_MUTE_CONTROL,
57    CONF_OUTPUT_CODEC,
58    CONF_POWER_CONTROL,
59    CONF_VOLUME_CONTROL,
60    CONF_VOLUME_STEP,
61)
62from music_assistant.controllers.players import PlayerController
63from music_assistant.controllers.players.announcements import ANNOUNCEMENT_TTS_TIMEOUT
64from music_assistant.controllers.webserver.helpers.auth_middleware import current_user
65from music_assistant.helpers.tts import TTS_QUERY_TIMEOUT_SECONDS
66from music_assistant.models.player import LinkedOutputProtocol
67from music_assistant.models.player_provider import PlayerProvider
68from tests.common import MockPlayer, MockProvider, create_mock_config, use_real_create_task
69
70
71def _player_config_stub(
72    values: dict[str, object] | None = None,
73    *,
74    min_volume: int = 0,
75    max_volume: int = 100,
76) -> Callable[..., object]:
77    """
78    Build a ``get_raw_player_config_value`` side effect.
79
80    :param values: Extra config keys to answer, e.g. ``{CONF_MUTE_CONTROL: PLAYER_CONTROL_FAKE}``.
81    :param min_volume: Value returned for the ``min_volume`` key.
82    :param max_volume: Value returned for the ``max_volume`` key.
83    """
84    config: dict[str, object] = {
85        CONF_MIN_VOLUME: min_volume,
86        CONF_MAX_VOLUME: max_volume,
87        **(values or {}),
88    }
89
90    def _conf(_player_id: str, key: str, default: object = None) -> object:
91        if key in config:
92            return config[key]
93        return default
94
95    return _conf
96
97
98def _announcement() -> PlayerMedia:
99    """Return the announcement to play."""
100    return PlayerMedia(
101        uri="http://ma/announcement/player_1.mp3",
102        media_type=MediaType.ANNOUNCEMENT,
103        title="Announcement",
104        duration=3,
105    )
106
107
108def _volume_step_config(step: int | None) -> CoreConfig:
109    """
110    Build a "players" CoreConfig carrying the given ``volume_step`` value.
111
112    Pass ``None`` to leave the entry unresolved and genuinely exercise the default
113    (``CoreConfig.get_value`` returns the entry's ``value`` verbatim, not its
114    ``default_value``, so a real "not configured" state is ``value=None``).
115    """
116    return CoreConfig(
117        domain="players",
118        values={
119            CONF_VOLUME_STEP: ConfigEntry(
120                key=CONF_VOLUME_STEP,
121                type=ConfigEntryType.INTEGER,
122                default_value=0,
123                value=step,
124            )
125        },
126    )
127
128
129@pytest.fixture
130def running_background_tasks(mock_mass: MagicMock) -> Iterator[None]:
131    """
132    Really run the tasks that ``mass.create_task`` is handed, and raise what they raise.
133
134    The real implementation only logs the exception of a background task, which would
135    leave a test that dispatches its work through the TaskManager passing regardless.
136    """
137    tasks: list[asyncio.Task[Any]] = []
138    use_real_create_task(mock_mass)
139    real_create_task = mock_mass.create_task
140
141    def _create_task(target: Any, *args: Any, **kwargs: Any) -> Any:
142        task = real_create_task(target, *args, **kwargs)
143        if isinstance(task, asyncio.Task):
144            tasks.append(task)
145        return task
146
147    mock_mass.create_task = MagicMock(side_effect=_create_task)
148    yield
149    errors = [
150        err
151        for task in tasks
152        if task.done() and not task.cancelled() and (err := task.exception()) is not None
153    ]
154    if len(errors) == 1:
155        raise errors[0]
156    if errors:
157        raise BaseExceptionGroup("background tasks failed", errors)
158
159
160def _mute_natively(player: MockPlayer) -> AsyncMock:
161    """Give the player a native mute control, mute it and return its mute handler."""
162    player._attr_supported_features.add(PlayerFeature.VOLUME_MUTE)
163    player._attr_volume_muted = True
164    mute_mock = AsyncMock(side_effect=lambda muted: setattr(player, "_attr_volume_muted", muted))
165    player.volume_mute = mute_mock  # type: ignore[method-assign]
166    player._cache.clear()
167    player.update_state(force_update=True, signal_event=False)
168    assert player.state.volume_muted is True
169    return mute_mock
170
171
172@pytest.fixture
173def mock_mass() -> MagicMock:
174    """Create a mock MusicAssistant instance."""
175    mass = MagicMock()
176    mass.closing = False
177    mass.loop = None
178    mass.config = MagicMock()
179    mass.config.get = MagicMock(return_value=[])
180    mass.config.get_raw_player_config_value = MagicMock(side_effect=_player_config_stub())
181    # Return "GLOBAL" for log level config (standard default)
182    mass.config.get_raw_core_config_value = MagicMock(return_value="GLOBAL")
183    mass.config.set = MagicMock()
184    mass.signal_event = MagicMock()
185    mass.get_providers = MagicMock(return_value=[])
186    return mass
187
188
189@pytest.fixture
190def controller(mock_mass: MagicMock) -> PlayerController:
191    """Create a PlayerController instance."""
192    return PlayerController(mock_mass)
193
194
195class TestSetMembersValidation:
196    """Test cmd_set_members validation logic."""
197
198    def test_set_members_requires_feature(self, mock_mass: MagicMock) -> None:
199        """Test that set_members requires SET_MEMBERS feature."""
200        controller = PlayerController(mock_mass)
201        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
202
203        leader = MockPlayer(provider, "leader", "Leader")
204        # Note: NOT adding SET_MEMBERS feature
205
206        member = MockPlayer(provider, "member", "Member")
207
208        controller._players = {"leader": leader, "member": member}
209        mock_mass.players = controller
210
211        # Should raise exception because leader doesn't support SET_MEMBERS
212        with pytest.raises(UnsupportedFeaturedException):
213            asyncio.run(controller.cmd_set_members("leader", player_ids_to_add=["member"]))
214
215    def test_cannot_group_incompatible_players(self, mock_mass: MagicMock) -> None:
216        """Test that incompatible players cannot be grouped."""
217        controller = PlayerController(mock_mass)
218        provider_a = MockProvider("provider_a", instance_id="provider_a", mass=mock_mass)
219        provider_b = MockProvider("provider_b", instance_id="provider_b", mass=mock_mass)
220
221        player_a = MockPlayer(provider_a, "player_a", "Player A")
222        player_a._attr_supported_features.add(PlayerFeature.SET_MEMBERS)
223        player_a._attr_can_group_with = {"provider_a"}  # Only same provider
224
225        player_b = MockPlayer(provider_b, "player_b", "Player B")
226
227        controller._players = {"player_a": player_a, "player_b": player_b}
228        mock_mass.players = controller
229
230        # Should raise exception because players are incompatible
231        with pytest.raises(UnsupportedFeaturedException):
232            asyncio.run(controller.cmd_set_members("player_a", player_ids_to_add=["player_b"]))
233
234
235class TestCacheInvalidationAfterGrouping:
236    """Test that caches are invalidated after grouping operations."""
237
238    async def test_all_players_cache_cleared_after_set_members(self, mock_mass: MagicMock) -> None:
239        """
240        Test that all players' caches are cleared after set_members.
241
242        Regression test for: Stale can_group_with cache after grouping changes.
243        """
244        controller = PlayerController(mock_mass)
245        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
246
247        leader = MockPlayer(provider, "leader", "Leader")
248        leader._attr_supported_features.add(PlayerFeature.SET_MEMBERS)
249        leader._attr_can_group_with = {"test"}
250        leader._attr_group_members = []
251
252        member = MockPlayer(provider, "member", "Member")
253
254        other = MockPlayer(provider, "other", "Other")
255        other._attr_supported_features.add(PlayerFeature.SET_MEMBERS)
256        other._attr_can_group_with = {"test"}
257
258        controller._players = {"leader": leader, "member": member, "other": other}
259        mock_mass.players = controller
260
261        # Populate caches
262        _ = leader.state.can_group_with
263        _ = other.state.can_group_with
264
265        # Simulate grouping (normally done by provider's set_members implementation)
266        leader._attr_group_members = ["leader", "member"]
267
268        # Call set_members to trigger cache invalidation
269        await controller._handle_set_members_with_protocols(
270            leader, player_ids_to_add=["member"], player_ids_to_remove=[]
271        )
272
273        # Note: The actual cache clearing happens via trigger_player_update
274        # which schedules update_state to be called later
275        # In a real scenario, this would clear all players' caches
276
277
278class TestNativeSetMembersGuard:
279    """Test the SET_MEMBERS feature guard on native set_members forwarding."""
280
281    async def test_native_set_members_skipped_without_feature(self, mock_mass: MagicMock) -> None:
282        """
283        Test that set_members is not called on a player without SET_MEMBERS support.
284
285        Regression test for: NotImplementedError raised from
286        _cleanup_player_memberships when removing a member from a native player
287        whose group membership is managed externally (e.g. a Google Cast group,
288        which never advertises SET_MEMBERS).
289        """
290        controller = PlayerController(mock_mass)
291        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
292
293        # a native group-like player WITHOUT SET_MEMBERS in supported_features,
294        # whose set_members behaves like the Player base class (raises)
295        parent = MockPlayer(provider, "cast_group", "Cast Group")
296        parent._attr_group_members = ["cast_group", "member"]
297        parent.set_members = AsyncMock(  # type: ignore[method-assign]
298            side_effect=NotImplementedError(
299                "set_members needs to be implemented when PlayerFeature.SET_MEMBERS is set"
300            )
301        )
302        member = MockPlayer(provider, "member", "Member")
303
304        controller._players = {"cast_group": parent, "member": member}
305        mock_mass.players = controller
306
307        # must complete without raising NotImplementedError
308        await controller._handle_set_members_with_protocols(
309            parent, player_ids_to_add=[], player_ids_to_remove=["member"]
310        )
311        parent.set_members.assert_not_called()
312
313
314class TestGroupUngroup:
315    """Test group and ungroup commands."""
316
317    async def test_group_command(self, mock_mass: MagicMock) -> None:
318        """Test the group command (cmd_group)."""
319        controller = PlayerController(mock_mass)
320        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
321
322        leader = MockPlayer(provider, "leader", "Leader")
323        leader._attr_supported_features.add(PlayerFeature.SET_MEMBERS)
324        leader._attr_can_group_with = {"member"}  # Leader can group with member
325
326        member = MockPlayer(provider, "member", "Member")
327        # Make sure member is already powered on to skip power handling
328        member._attr_powered = True
329
330        controller._players = {"leader": leader, "member": member}
331        mock_mass.players = controller
332
333        # Update state after modifying attributes and registering with controller
334        leader.update_state(signal_event=False)
335        member.update_state(signal_event=False)
336
337        # Track if set_members was called
338        set_members_called = False
339        original_set_members = leader.set_members
340
341        async def mock_set_members(
342            player_ids_to_add: list[str] | None = None,
343            player_ids_to_remove: list[str] | None = None,
344        ) -> None:
345            nonlocal set_members_called
346            set_members_called = True
347            # Call the original to update group_members
348            await original_set_members(player_ids_to_add, player_ids_to_remove)
349
350        leader.set_members = mock_set_members  # type: ignore[method-assign]
351
352        # Mock power handling to skip power control (focus is on grouping logic)
353        async def mock_handle_cmd_power(
354            player_id: str, powered: bool, skip_auto_play: bool = False
355        ) -> None:
356            pass
357
358        controller._handle_cmd_power = mock_handle_cmd_power  # type: ignore[method-assign]
359
360        # Execute group command
361        await controller.cmd_group("member", "leader")
362
363        # Verify set_members was called
364        assert set_members_called
365        # Verify member was added to leader's group
366        assert "member" in leader._attr_group_members
367
368
369class TestPlayerAvailability:
370    """Test player availability checks in grouping."""
371
372    def test_unavailable_player_rejected(self, mock_mass: MagicMock) -> None:
373        """Test that unavailable players are rejected when grouping."""
374        controller = PlayerController(mock_mass)
375        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
376
377        leader = MockPlayer(provider, "leader", "Leader")
378        leader._attr_supported_features.add(PlayerFeature.SET_MEMBERS)
379        leader._attr_can_group_with = {"test"}
380
381        member = MockPlayer(provider, "member", "Member")
382        member._attr_available = False  # Mark as unavailable
383
384        controller._players = {"leader": leader, "member": member}
385        mock_mass.players = controller
386
387        # Attempting to group with unavailable player should be handled
388        # (either silently ignored or raise exception depending on implementation)
389        # This should either skip the unavailable player or raise an exception
390        with contextlib.suppress(Exception):
391            asyncio.run(controller.cmd_set_members("leader", player_ids_to_add=["member"]))
392
393
394def _group_with_member(
395    mock_mass: MagicMock, *, initialize_group: bool = True
396) -> tuple[PlayerController, MockPlayer, MockPlayer]:
397    """
398    Build a controller holding one group player with a single member.
399
400    :param mock_mass: The mocked MusicAssistant instance to attach the controller to.
401    :param initialize_group: Whether the group player is marked as fully registered.
402
403    :return: The controller, the group player and its member.
404    """
405    controller = PlayerController(mock_mass)
406    provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
407
408    group_player = MockPlayer(provider, "group", "Group", player_type=PlayerType.GROUP)
409    member = MockPlayer(provider, "member", "Member")
410
411    controller._players = {"group": group_player, "member": member}
412    mock_mass.players = controller
413
414    group_player._attr_group_members = ["member"]
415    member.initialized.set()
416    if initialize_group:
417        group_player.initialized.set()
418    for player in (group_player, member):
419        player.update_state(signal_event=False)
420    return controller, group_player, member
421
422
423@contextlib.contextmanager
424def _restricted_user(visible_player_ids: list[str]) -> Iterator[None]:
425    """
426    Run the wrapped block as a non-admin user that may only see the given players.
427
428    :param visible_player_ids: The player ids the user is allowed to see.
429    """
430    # the contextvar is copied into the task an API command runs in, so it stays
431    # live for everything that command reaches, internal bookkeeping included
432    token = current_user.set(
433        User(
434            user_id="user_1",
435            username="restricted",
436            role=UserRole.USER,
437            player_filter=visible_player_ids,
438        )
439    )
440    try:
441        yield
442    finally:
443        current_user.reset(token)
444
445
446class TestStateForwarding:
447    """Test forwarding of player state changes to related players."""
448
449    def test_sync_leader_updates_are_forwarded_to_sync_children(self, mock_mass: MagicMock) -> None:
450        """A regular sync leader must notify children via the sync-parent callback."""
451        controller = PlayerController(mock_mass)
452        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
453
454        leader = MockPlayer(provider, "leader", "Leader")
455        child = MockPlayer(provider, "child", "Child")
456
457        controller._players = {"leader": leader, "child": child}
458        mock_mass.players = controller
459
460        leader._attr_group_members = ["leader", "child"]
461        leader.update_state(signal_event=False)
462        child.update_state(signal_event=False)
463
464        with (
465            patch.object(child, "on_sync_parent_updated") as on_sync_parent_updated,
466            patch.object(child, "on_group_updated") as on_group_updated,
467        ):
468            changed_values = {"playback_state": (PlaybackState.IDLE, PlaybackState.PLAYING)}
469            controller._forward_state_update(leader, changed_values)
470
471        on_sync_parent_updated.assert_called_once_with(leader, changed_values)
472        on_group_updated.assert_not_called()
473
474    def test_group_updates_are_forwarded_to_children_via_group_callback(
475        self, mock_mass: MagicMock
476    ) -> None:
477        """A group player must continue to notify children via the group callback."""
478        controller = PlayerController(mock_mass)
479        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
480
481        group_player = MockPlayer(provider, "group", "Group", player_type=PlayerType.GROUP)
482        child = MockPlayer(provider, "child", "Child")
483
484        controller._players = {"group": group_player, "child": child}
485        mock_mass.players = controller
486
487        group_player._attr_group_members = ["group", "child"]
488        group_player.update_state(signal_event=False)
489        child.update_state(signal_event=False)
490
491        with (
492            patch.object(child, "on_group_updated") as on_group_updated,
493            patch.object(child, "on_sync_parent_updated") as on_sync_parent_updated,
494        ):
495            changed_values = {"playback_state": (PlaybackState.IDLE, PlaybackState.PLAYING)}
496            controller._forward_state_update(group_player, changed_values)
497
498        on_group_updated.assert_called_once_with(
499            group_player,
500            changed_values,
501        )
502        on_sync_parent_updated.assert_not_called()
503
504    def test_sync_leader_updates_also_reach_its_group_player(self, mock_mass: MagicMock) -> None:
505        """A sync leader must notify its group player as well as its own sync children."""
506        controller = PlayerController(mock_mass)
507        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
508
509        group_player = MockPlayer(provider, "group", "Group", player_type=PlayerType.GROUP)
510        leader = MockPlayer(provider, "leader", "Leader")
511        follower = MockPlayer(provider, "follower", "Follower")
512
513        controller._players = {"group": group_player, "leader": leader, "follower": follower}
514        mock_mass.players = controller
515
516        group_player._attr_group_members = ["leader", "follower"]
517        leader._attr_group_members = ["leader", "follower"]
518        for player in (group_player, leader, follower):
519            # _get_player_groups only considers players the controller finished registering
520            player.initialized.set()
521            player.update_state(signal_event=False)
522
523        with (
524            patch.object(group_player, "on_group_member_updated") as on_group_member_updated,
525            patch.object(follower, "on_sync_parent_updated") as on_sync_parent_updated,
526        ):
527            changed_values = {"playback_state": (PlaybackState.IDLE, PlaybackState.PLAYING)}
528            controller._forward_state_update(leader, changed_values)
529
530        on_group_member_updated.assert_called_once_with(leader, changed_values)
531        on_sync_parent_updated.assert_called_once_with(leader, changed_values)
532
533    def test_group_player_is_notified_under_restricted_user_context(
534        self, mock_mass: MagicMock
535    ) -> None:
536        """The fan-out must not be narrowed by the user that triggered the update."""
537        controller, group_player, member = _group_with_member(mock_mass)
538
539        with _restricted_user(["member"]):
540            assert group_player not in controller.all_players()
541            with patch.object(group_player, "on_group_member_updated") as on_group_member_updated:
542                changed_values = {"playback_state": (PlaybackState.IDLE, PlaybackState.PLAYING)}
543                controller._forward_state_update(member, changed_values)
544
545        on_group_member_updated.assert_called_once_with(member, changed_values)
546
547    def test_unavailable_group_player_is_still_notified(self, mock_mass: MagicMock) -> None:
548        """A group player mirrors its members, so it must update while unavailable too."""
549        controller, group_player, member = _group_with_member(mock_mass)
550        group_player._attr_available = False
551        group_player.update_state(signal_event=False)
552        assert group_player.state.available is False
553
554        with patch.object(group_player, "on_group_member_updated") as on_group_member_updated:
555            changed_values = {"playback_state": (PlaybackState.IDLE, PlaybackState.PLAYING)}
556            controller._forward_state_update(member, changed_values)
557
558        on_group_member_updated.assert_called_once_with(member, changed_values)
559
560    def test_disabled_group_player_is_not_notified(self, mock_mass: MagicMock) -> None:
561        """A disabled player takes no part, unlike one that is merely unavailable."""
562        controller, group_player, member = _group_with_member(mock_mass)
563        group_player._config.enabled = False
564        group_player.update_state(signal_event=False, force_update=True)
565        assert group_player.state.enabled is False
566
567        with patch.object(group_player, "on_group_member_updated") as on_group_member_updated:
568            controller._forward_state_update(
569                member, {"playback_state": (PlaybackState.IDLE, PlaybackState.PLAYING)}
570            )
571
572        on_group_member_updated.assert_not_called()
573
574    def test_uninitialized_group_player_is_not_notified(self, mock_mass: MagicMock) -> None:
575        """A player the controller is still registering is not fully set up yet."""
576        controller, group_player, member = _group_with_member(mock_mass, initialize_group=False)
577
578        with patch.object(group_player, "on_group_member_updated") as on_group_member_updated:
579            controller._forward_state_update(
580                member, {"playback_state": (PlaybackState.IDLE, PlaybackState.PLAYING)}
581            )
582
583        on_group_member_updated.assert_not_called()
584
585
586class TestUnscopedPlayerLookups:
587    """Derived player state must not depend on who triggered the recalculation."""
588
589    def test_iter_players_ignores_the_user_filter(self, mock_mass: MagicMock) -> None:
590        """iter_players is the internal view: every registered player, no user filter."""
591        controller, group_player, member = _group_with_member(mock_mass)
592
593        with _restricted_user(["member"]):
594            assert controller.all_players() == [member]
595            assert set(controller.iter_players()) == {group_player, member}
596
597    def test_provider_filter_still_ignores_the_user_filter(self, mock_mass: MagicMock) -> None:
598        """A lookup scoped to one provider must still ignore the user filter."""
599        controller, group_player, member = _group_with_member(mock_mass)
600
601        with _restricted_user(["member"]):
602            found = set(controller.iter_players(provider_filter="test"))
603
604        assert found == {group_player, member}
605
606    def test_provider_sees_all_of_its_own_players(self, mock_mass: MagicMock) -> None:
607        """A provider owns its players, so a user filter must never hide them from it."""
608        controller = PlayerController(mock_mass)
609        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
610
611        plain = MockPlayer(provider, "plain", "Plain")
612        # a protocol player id is picked from a list that hides protocol players, so it
613        # is never in a user's filter - the whole set used to disappear for any non-admin
614        protocol = MockPlayer(provider, "protocol", "Protocol", player_type=PlayerType.PROTOCOL)
615        controller._players = {"plain": plain, "protocol": protocol}
616        mock_mass.players = controller
617        for player in (plain, protocol):
618            player.initialized.set()
619            player.update_state(signal_event=False)
620
621        with _restricted_user(["plain"]):
622            owned = set(PlayerProvider.players.fget(provider))  # type: ignore[attr-defined]
623
624        assert owned == {plain, protocol}
625
626    def test_active_group_survives_a_restricted_user_context(self, mock_mass: MagicMock) -> None:
627        """A member must still know its group when the triggering user cannot see it."""
628        _controller, group_player, member = _group_with_member(mock_mass)
629        group_player._attr_group_members = ["member"]
630        group_player._attr_powered = True
631        group_player.update_state(signal_event=False)
632
633        with _restricted_user(["member"]):
634            member.update_state(signal_event=False, force_update=True)
635            assert member.state.active_group == "group"
636
637    def test_synced_to_survives_a_restricted_user_context(self, mock_mass: MagicMock) -> None:
638        """A follower must still resolve its sync leader across a user filter."""
639        controller = PlayerController(mock_mass)
640        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
641
642        leader = MockPlayer(provider, "leader", "Leader")
643        follower = MockPlayer(provider, "follower", "Follower")
644        controller._players = {"leader": leader, "follower": follower}
645        mock_mass.players = controller
646
647        leader._attr_group_members = ["leader", "follower"]
648        for player in (leader, follower):
649            player.initialized.set()
650            player.update_state(signal_event=False)
651
652        with _restricted_user(["follower"]):
653            follower.update_state(signal_event=False, force_update=True)
654            assert follower.state.synced_to == "leader"
655
656    def test_can_group_with_survives_a_restricted_user_context(self, mock_mass: MagicMock) -> None:
657        """Expanding a provider id into players must not drop players the user cannot see."""
658        controller = PlayerController(mock_mass)
659        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
660        mock_mass.get_provider.return_value = provider
661
662        player_a = MockPlayer(provider, "player_a", "Player A")
663        player_b = MockPlayer(provider, "player_b", "Player B")
664        controller._players = {"player_a": player_a, "player_b": player_b}
665        mock_mass.players = controller
666
667        # a provider instance id stands for "every player of that provider"
668        player_a._attr_can_group_with = {"test"}
669        for player in (player_a, player_b):
670            player.initialized.set()
671            player.update_state(signal_event=False)
672
673        with _restricted_user(["player_a"]):
674            player_a.update_state(signal_event=False, force_update=True)
675            assert "player_b" in player_a.state.can_group_with
676
677
678class TestSleepTimer:
679    """Test native sleep timer handling."""
680
681    def test_set_and_clear_sleep_timer(self, mock_mass: MagicMock) -> None:
682        """Setting a sleep timer exposes state and schedules the stop callback."""
683        controller = PlayerController(mock_mass)
684        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
685        player = MockPlayer(provider, "player_1", "Player 1")
686        controller._players = {"player_1": player}
687        mock_mass.players = controller
688
689        expires_at = controller.set_sleep_timer("player_1", 30)
690
691        assert controller.get_sleep_timer("player_1") == expires_at
692        assert player.sleep_timer_expires_at == expires_at
693        mock_mass.call_later.assert_called_with(
694            30,
695            controller._handle_sleep_timer_expired,
696            "player_1",
697            task_id="player_sleep_timer_player_1",
698        )
699        mock_mass.signal_event.assert_any_call(
700            EventType.PLAYER_SLEEP_TIMER_UPDATED,
701            object_id="player_1",
702            data=expires_at,
703        )
704
705        controller.clear_sleep_timer("player_1")
706
707        # get_sleep_timer reads the model field directly, so this also asserts it cleared
708        assert controller.get_sleep_timer("player_1") is None
709        mock_mass.cancel_timer.assert_called_with("player_sleep_timer_player_1")
710        mock_mass.signal_event.assert_any_call(
711            EventType.PLAYER_SLEEP_TIMER_UPDATED,
712            object_id="player_1",
713            data=None,
714        )
715
716    async def test_sleep_timer_removed_when_player_unregistered(self, mock_mass: MagicMock) -> None:
717        """Unregistering a player cancels and clears its sleep timer."""
718        controller = PlayerController(mock_mass)
719        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
720        player = MockPlayer(provider, "player_1", "Player 1")
721        controller._players = {"player_1": player}
722        player.set_sleep_timer_expires_at(123.0)
723
724        await controller.unregister("player_1")
725
726        assert player.sleep_timer_expires_at is None
727        mock_mass.cancel_timer.assert_called_with("player_sleep_timer_player_1")
728
729    async def test_sleep_timer_expiry_stops_player(self, mock_mass: MagicMock) -> None:
730        """An expired sleep timer clears its state and stops playback."""
731        controller = PlayerController(mock_mass)
732        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
733        player = MockPlayer(provider, "player_1", "Player 1")
734        controller._players = {"player_1": player}
735        player.set_sleep_timer_expires_at(123.0)
736        controller.cmd_stop = AsyncMock()  # type: ignore[method-assign]
737
738        await controller._handle_sleep_timer_expired("player_1")
739
740        assert controller.get_sleep_timer("player_1") is None
741        assert player.sleep_timer_expires_at is None
742        controller.cmd_stop.assert_awaited_once_with("player_1")
743        mock_mass.signal_event.assert_any_call(
744            EventType.PLAYER_SLEEP_TIMER_UPDATED,
745            object_id="player_1",
746            data=None,
747        )
748
749    def test_set_sleep_timer_rejects_invalid_duration(self, mock_mass: MagicMock) -> None:
750        """A non-positive or float-overflowing duration raises and schedules nothing."""
751        controller = PlayerController(mock_mass)
752        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
753        player = MockPlayer(provider, "player_1", "Player 1")
754        controller._players = {"player_1": player}
755
756        # 0/-30 are non-positive; 10**400 exceeds the float range for the expiry math
757        for invalid in (0, -30, 10**400):
758            with pytest.raises(InvalidDataError):
759                controller.set_sleep_timer("player_1", invalid)
760
761        assert controller.get_sleep_timer("player_1") is None
762        mock_mass.call_later.assert_not_called()
763
764
765class TestUnregisterCleanup:
766    """Test that unregister cleans up leaked internal state."""
767
768    def test_command_locks_removed(self, mock_mass: MagicMock) -> None:
769        """Unregistering a player removes its command lock entries."""
770        controller = PlayerController(mock_mass)
771        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
772        player = MockPlayer(provider, "player_1", "Player 1")
773
774        controller._players = {"player_1": player}
775        controller._player_command_locks = {
776            "playback_player_1": asyncio.Lock(),
777            "volume_player_1": asyncio.Lock(),
778        }
779
780        asyncio.run(controller.unregister("player_1"))
781
782        assert "playback_player_1" not in controller._player_command_locks
783        assert "volume_player_1" not in controller._player_command_locks
784
785    def test_other_players_state_untouched(self, mock_mass: MagicMock) -> None:
786        """Unregistering one player does not affect another player's state."""
787        controller = PlayerController(mock_mass)
788        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
789        player_a = MockPlayer(provider, "player_a", "Player A")
790        player_b = MockPlayer(provider, "player_b", "Player B")
791
792        controller._players = {"player_a": player_a, "player_b": player_b}
793        controller._player_command_locks = {
794            "playback_player_a": asyncio.Lock(),
795            "playback_player_b": asyncio.Lock(),
796        }
797
798        asyncio.run(controller.unregister("player_a"))
799
800        assert "playback_player_b" in controller._player_command_locks
801        assert "playback_player_a" not in controller._player_command_locks
802
803    def test_suffix_player_id_not_over_matched(self, mock_mass: MagicMock) -> None:
804        """Removing player 'b' must not remove locks for player 'a_b' (no suffix matching)."""
805        controller = PlayerController(mock_mass)
806        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
807        player_b = MockPlayer(provider, "b", "Player B")
808        player_a_b = MockPlayer(provider, "a_b", "Player A_B")
809
810        controller._players = {"b": player_b, "a_b": player_a_b}
811        controller._player_command_locks = {
812            "playback_b": asyncio.Lock(),
813            "playback_a_b": asyncio.Lock(),
814        }
815
816        asyncio.run(controller.unregister("b"))
817
818        assert "playback_a_b" in controller._player_command_locks
819        assert "playback_b" not in controller._player_command_locks
820
821    def test_pending_protocol_evaluation_cancelled(self, mock_mass: MagicMock) -> None:
822        """Unregistering a player cancels and removes its pending protocol evaluation."""
823        controller = PlayerController(mock_mass)
824        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
825        player = MockPlayer(provider, "player_1", "Player 1")
826
827        mock_handle = MagicMock()
828        controller._players = {"player_1": player}
829        controller._pending_protocol_evaluations = {"player_1": mock_handle}
830
831        asyncio.run(controller.unregister("player_1"))
832
833        mock_handle.cancel.assert_called_once()
834        assert "player_1" not in controller._pending_protocol_evaluations
835
836    def test_unregister_nonexistent_player_is_noop(self, mock_mass: MagicMock) -> None:
837        """Unregistering a player that doesn't exist is silently ignored."""
838        controller = PlayerController(mock_mass)
839        controller._player_command_locks = {"set_members_other": asyncio.Lock()}
840
841        asyncio.run(controller.unregister("nonexistent"))
842
843        assert "set_members_other" in controller._player_command_locks
844
845
846class TestRegisterUnregisterRace:
847    """Test registration that is interrupted by an unregister of the same player."""
848
849    @staticmethod
850    def _stub_register_calls(mock_mass: MagicMock) -> None:
851        """Stub the awaited mass calls made during register/unregister."""
852        # registration reads config keys with differently typed defaults (mapping for the
853        # player config store, str | None for the cached MAC addresses)
854        mock_mass.config.get = MagicMock(side_effect=lambda _key, default=None: default)
855        mock_mass.cache.get = AsyncMock(return_value=None)
856        mock_mass.config.get_player_config = AsyncMock(return_value=create_mock_config("Player 1"))
857        mock_mass.player_queues.on_player_register = AsyncMock()
858        mock_mass.player_queues.on_player_remove = MagicMock()
859
860    @staticmethod
861    def _player_added_signalled(mock_mass: MagicMock) -> bool:
862        """Return True if a PLAYER_ADDED event was signalled."""
863        return any(
864            call_args.args and call_args.args[0] == EventType.PLAYER_ADDED
865            for call_args in mock_mass.signal_event.call_args_list
866        )
867
868    async def test_register_aborts_when_unregistered_during_config_load(
869        self, mock_mass: MagicMock
870    ) -> None:
871        """A player unregistered while its config loads is not announced as added."""
872        controller = PlayerController(mock_mass)
873        self._stub_register_calls(mock_mass)
874        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
875        player = MockPlayer(provider, "player_1", "Player 1")
876
877        async def _unregister_midway(player_id: str) -> MagicMock:
878            # stands in for a provider unload or device disconnect landing
879            # while register awaits the player config
880            await controller.unregister(player_id)
881            return create_mock_config("Player 1")
882
883        mock_mass.config.get_player_config = AsyncMock(side_effect=_unregister_midway)
884        config_hook = AsyncMock()
885
886        with (
887            patch(
888                "music_assistant.controllers.players.controller.enrich_device_mac_address",
889                AsyncMock(),
890            ),
891            patch.object(player, "on_config_updated", config_hook),
892        ):
893            await controller.register(player)
894
895        assert "player_1" not in controller._players
896        assert not player.initialized.is_set()
897        # setup stops right away: the provider hook must not run on a player
898        # whose on_unload already ran
899        config_hook.assert_not_called()
900        mock_mass.player_queues.on_player_register.assert_not_called()
901        assert not self._player_added_signalled(mock_mass)
902
903    async def test_register_aborts_when_unregistered_during_config_hook(
904        self, mock_mass: MagicMock
905    ) -> None:
906        """A player unregistered while its on_config_updated hook runs is not announced."""
907        controller = PlayerController(mock_mass)
908        self._stub_register_calls(mock_mass)
909        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
910        player = MockPlayer(provider, "player_1", "Player 1")
911
912        async def _unregister_midway() -> None:
913            await controller.unregister("player_1")
914
915        with (
916            patch(
917                "music_assistant.controllers.players.controller.enrich_device_mac_address",
918                AsyncMock(),
919            ),
920            patch.object(player, "on_config_updated", AsyncMock(side_effect=_unregister_midway)),
921        ):
922            await controller.register(player)
923
924        assert "player_1" not in controller._players
925        assert not player.initialized.is_set()
926        mock_mass.player_queues.on_player_register.assert_not_called()
927        assert not self._player_added_signalled(mock_mass)
928
929    async def test_register_drops_queue_recreated_after_unregister(
930        self, mock_mass: MagicMock
931    ) -> None:
932        """A queue restored after the unregister already removed it is dropped again."""
933        controller = PlayerController(mock_mass)
934        self._stub_register_calls(mock_mass)
935        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
936        player = MockPlayer(provider, "player_1", "Player 1")
937
938        events: list[str] = []
939
940        async def _unregister_midway(registering_player: MockPlayer) -> None:
941            # on_player_register restores the queue from cache before storing it, so an
942            # unregister can land in between and have its cleanup undone
943            await controller.unregister(registering_player.player_id)
944            events.append("queue_created")
945
946        def _track_removal(*_args: object, **_kwargs: object) -> None:
947            events.append("queue_removed")
948
949        mock_mass.player_queues.on_player_remove = MagicMock(side_effect=_track_removal)
950        mock_mass.player_queues.on_player_register = AsyncMock(side_effect=_unregister_midway)
951
952        with patch(
953            "music_assistant.controllers.players.controller.enrich_device_mac_address",
954            AsyncMock(),
955        ):
956            await controller.register(player)
957
958        assert "player_1" not in controller._players
959        # the queue recreated for the removed player must be cleaned up again
960        assert events == ["queue_removed", "queue_created", "queue_removed"]
961
962    async def test_register_or_update_marks_replacement_initialized(
963        self, mock_mass: MagicMock
964    ) -> None:
965        """Replacing a registered player carries the initialized state to the new object."""
966        controller = PlayerController(mock_mass)
967        self._stub_register_calls(mock_mass)
968        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
969        existing = MockPlayer(provider, "player_1", "Player 1")
970        existing.set_initialized()
971        controller._players = {"player_1": existing}
972        replacement = MockPlayer(provider, "player_1", "Player 1")
973
974        await controller.register_or_update(replacement)
975
976        assert controller._players["player_1"] is replacement
977        assert replacement.initialized.is_set()
978
979    async def test_register_or_update_hands_config_to_replacement(
980        self, mock_mass: MagicMock
981    ) -> None:
982        """A replacement instance inherits the config of the player it takes over."""
983        controller = PlayerController(mock_mass)
984        self._stub_register_calls(mock_mass)
985        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
986        existing = MockPlayer(provider, "player_1", "Player 1")
987        resolved_config = create_mock_config("Player 1")
988        existing.set_config(resolved_config)
989        existing.set_initialized()
990        controller._players = {"player_1": existing}
991        replacement = MockPlayer(provider, "player_1", "Player 1")
992        config_hook = AsyncMock()
993
994        with patch.object(replacement, "on_config_updated", config_hook):
995            await controller.register_or_update(replacement)
996
997        # without the resolved config the replacement would read defaults for every
998        # config backed setting (group members, flow mode, visibility, ...)
999        assert replacement.config is resolved_config
1000        config_hook.assert_awaited_once()
1001
1002    async def test_register_or_update_aborts_when_replacement_is_unregistered(
1003        self, mock_mass: MagicMock
1004    ) -> None:
1005        """A replacement unregistered while its config hook runs is not marked ready."""
1006        controller = PlayerController(mock_mass)
1007        self._stub_register_calls(mock_mass)
1008        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
1009        existing = MockPlayer(provider, "player_1", "Player 1")
1010        existing.set_config(create_mock_config("Player 1"))
1011        existing.set_initialized()
1012        controller._players = {"player_1": existing}
1013        replacement = MockPlayer(provider, "player_1", "Player 1")
1014
1015        async def _unregister_midway() -> None:
1016            await controller.unregister("player_1")
1017
1018        with patch.object(
1019            replacement, "on_config_updated", AsyncMock(side_effect=_unregister_midway)
1020        ):
1021            await controller.register_or_update(replacement)
1022
1023        assert "player_1" not in controller._players
1024        assert not replacement.initialized.is_set()
1025
1026    async def test_register_or_update_leaves_same_instance_untouched(
1027        self, mock_mass: MagicMock
1028    ) -> None:
1029        """Re-announcing the same instance does not re-run its config hook."""
1030        controller = PlayerController(mock_mass)
1031        self._stub_register_calls(mock_mass)
1032        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
1033        player = MockPlayer(provider, "player_1", "Player 1")
1034        player.set_initialized()
1035        controller._players = {"player_1": player}
1036        config_hook = AsyncMock()
1037
1038        with patch.object(player, "on_config_updated", config_hook):
1039            await controller.register_or_update(player)
1040
1041        assert controller._players["player_1"] is player
1042        config_hook.assert_not_called()
1043
1044    async def test_register_or_update_waits_for_inflight_register(
1045        self, mock_mass: MagicMock
1046    ) -> None:
1047        """A player is never swapped out while register() is still setting it up."""
1048        controller = PlayerController(mock_mass)
1049        self._stub_register_calls(mock_mass)
1050        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
1051        player = MockPlayer(provider, "player_1", "Player 1")
1052        replacement = MockPlayer(provider, "player_1", "Player 1")
1053        release = asyncio.Event()
1054        registering = asyncio.Event()
1055
1056        async def _blocked_config(*_args: object) -> MagicMock:
1057            registering.set()
1058            await release.wait()
1059            return create_mock_config("Player 1")
1060
1061        mock_mass.config.get_player_config = AsyncMock(side_effect=_blocked_config)
1062
1063        with patch(
1064            "music_assistant.controllers.players.controller.enrich_device_mac_address",
1065            AsyncMock(),
1066        ):
1067            register_task = asyncio.create_task(controller.register(player))
1068            # only proceed once register() is provably inside its critical section
1069            await registering.wait()
1070            update_task = asyncio.create_task(controller.register_or_update(replacement))
1071            await _yield_to_loop()
1072
1073            # register() is still in flight, so the replacement must not be swapped in yet
1074            assert controller._players["player_1"] is player
1075            assert not update_task.done()
1076
1077            release.set()
1078            await register_task
1079            await update_task
1080
1081        assert controller._players["player_1"] is replacement
1082        assert player.initialized.is_set()
1083        assert replacement.initialized.is_set()
1084
1085
1086async def _yield_to_loop() -> None:
1087    """Give other pending tasks a chance to run up to their next suspension point."""
1088    for _ in range(5):
1089        await asyncio.sleep(0)
1090
1091
1092def _set_play_media_override(mock_mass: MagicMock, value: bool) -> None:
1093    """
1094    Configure get_raw_player_config_value to return ``value`` for the play-media override key.
1095
1096    Other keys keep the existing defaults from the shared fixture. Use this in
1097    tests for ``play_media`` override behavior so the legacy/new branch is
1098    selected deterministically.
1099    """
1100    original = mock_mass.config.get_raw_player_config_value.side_effect
1101
1102    def _side_effect(player_id: str, key: str, default: object = None) -> object:
1103        if key == "play_media_overrides_group":
1104            return value
1105        if callable(original):
1106            return original(player_id, key, default)
1107        return default
1108
1109    mock_mass.config.get_raw_player_config_value = MagicMock(side_effect=_side_effect)
1110
1111
1112class TestCmdUngroupNewBranches:
1113    """
1114    Regression tests for the post-refactor cmd_ungroup flow.
1115
1116    The refactor changed two things:
1117
1118    - ``cmd_ungroup`` on a group player no longer calls ``cmd_set_members``
1119      (which would hit the "Cannot remove static member" guard); instead it
1120      stops or powers off the group entirely.
1121    - ``cmd_ungroup`` on a static member of a group recurses to ungroup the
1122      group, because static members cannot be released individually.
1123    """
1124
1125    @pytest.mark.asyncio
1126    async def test_ungroup_group_player_with_power_uses_power_off(
1127        self, mock_mass: MagicMock
1128    ) -> None:
1129        """Ungroup on a group with explicit power control routes through cmd_power(False)."""
1130        controller = PlayerController(mock_mass)
1131        provider = MockProvider("test_group", instance_id="test_group", mass=mock_mass)
1132        group = MockPlayer(provider, "g1", "Group", player_type=PlayerType.GROUP)
1133        group._attr_powered = True
1134        group._attr_group_members = ["member"]
1135        group._attr_supported_features = {PlayerFeature.POWER}
1136
1137        # ensure power_control resolves to NATIVE so cmd_ungroup uses the power path
1138        mock_mass.config.get_raw_player_config_value = MagicMock(
1139            side_effect=_player_config_stub({CONF_POWER_CONTROL: PLAYER_CONTROL_NATIVE})
1140        )
1141
1142        controller._players = {"g1": group}
1143        mock_mass.players = controller
1144
1145        # populate state.type / state.power_control / state.group_members
1146        group.set_initialized()
1147        group._cache.clear()
1148        group.update_state(signal_event=False)
1149
1150        called: dict[str, bool | str] = {}
1151
1152        async def _power(
1153            player_id: str,
1154            powered: bool,
1155            skip_auto_play: bool = False,  # noqa: ARG001
1156        ) -> None:
1157            called["player_id"] = player_id
1158            called["powered"] = powered
1159
1160        controller._handle_cmd_power = _power  # type: ignore[method-assign]
1161
1162        await controller.cmd_ungroup("g1")
1163
1164        assert called == {"player_id": "g1", "powered": False}
1165
1166    @pytest.mark.asyncio
1167    async def test_ungroup_powerless_group_calls_stop(self, mock_mass: MagicMock) -> None:
1168        """Ungroup on a powerless group falls through to _handle_cmd_stop."""
1169        controller = PlayerController(mock_mass)
1170        provider = MockProvider("test_group", instance_id="test_group", mass=mock_mass)
1171        group = MockPlayer(provider, "g1", "Group", player_type=PlayerType.GROUP)
1172        group._attr_powered = None  # no power control
1173        group._attr_group_members = ["member"]
1174        # no POWER feature → power_control auto-selects to NONE
1175
1176        controller._players = {"g1": group}
1177        mock_mass.players = controller
1178
1179        group.set_initialized()
1180        group._cache.clear()
1181        group.update_state(signal_event=False)
1182
1183        stop_called: list[str] = []
1184
1185        async def _stop(player_id: str) -> None:
1186            stop_called.append(player_id)
1187
1188        controller._handle_cmd_stop = _stop  # type: ignore[method-assign]
1189        # also stub power to make sure we did NOT go down that branch
1190        power_called: list[str] = []
1191
1192        async def _power(
1193            player_id: str,
1194            powered: bool,  # noqa: ARG001
1195            skip_auto_play: bool = False,  # noqa: ARG001
1196        ) -> None:
1197            power_called.append(player_id)
1198
1199        controller._handle_cmd_power = _power  # type: ignore[method-assign]
1200
1201        await controller.cmd_ungroup("g1")
1202
1203        assert stop_called == ["g1"]
1204        assert power_called == []  # powerless group → never goes through cmd_power
1205
1206
1207class TestExternalPowerOffUnsync:
1208    """
1209    Tests for unsyncing a player when its power is turned off outside of MA.
1210
1211    When a player's (final) power state flips on->off because its linked power
1212    control was switched off directly - rather than via an MA power command -
1213    the player must be removed from any (sync)group it is part of.
1214    """
1215
1216    def _make_synced_player(self, mock_mass: MagicMock) -> tuple[PlayerController, MockPlayer]:
1217        """Build a controller with a player synced to a registered leader."""
1218        controller = PlayerController(mock_mass)
1219        provider = MockProvider("test", instance_id="test", mass=mock_mass)
1220        leader = MockPlayer(provider, "leader", "Leader")
1221        leader._attr_group_members = ["leader", "p1"]
1222        player = MockPlayer(provider, "p1", "Player")
1223        controller._players = {"leader": leader, "p1": player}
1224        mock_mass.players = controller
1225        for _player in (leader, player):
1226            _player.set_initialized()
1227            _player._cache.clear()
1228            _player.update_state(signal_event=False)
1229        # isolate the unsync branch from the unrelated state-forwarding machinery
1230        controller._forward_state_update = MagicMock()  # type: ignore[method-assign]
1231        controller.cmd_ungroup = MagicMock(return_value="ungroup-coro")  # type: ignore[method-assign]
1232        return controller, player
1233
1234    def test_power_off_unsyncs_synced_player(self, mock_mass: MagicMock) -> None:
1235        """An on->off power transition ungroups a synced player."""
1236        controller, player = self._make_synced_player(mock_mass)
1237        assert player.state.synced_to == "leader"
1238
1239        controller.signal_player_state_update(player, {"powered": (True, False)})
1240
1241        controller.cmd_ungroup.assert_called_once_with("p1")  # type: ignore[attr-defined]
1242
1243    def test_power_on_does_not_unsync(self, mock_mass: MagicMock) -> None:
1244        """An off->on power transition leaves the player synced."""
1245        controller, player = self._make_synced_player(mock_mass)
1246
1247        controller.signal_player_state_update(player, {"powered": (False, True)})
1248
1249        controller.cmd_ungroup.assert_not_called()  # type: ignore[attr-defined]
1250
1251    def test_no_power_control_is_ignored(self, mock_mass: MagicMock) -> None:
1252        """A None->off transition (player without power control) is ignored."""
1253        controller, player = self._make_synced_player(mock_mass)
1254
1255        controller.signal_player_state_update(player, {"powered": (None, False)})
1256
1257        controller.cmd_ungroup.assert_not_called()  # type: ignore[attr-defined]
1258
1259    def test_power_off_ungrouped_player_is_noop(self, mock_mass: MagicMock) -> None:
1260        """Powering off a player that is not in any group does nothing."""
1261        controller = PlayerController(mock_mass)
1262        provider = MockProvider("test", instance_id="test", mass=mock_mass)
1263        player = MockPlayer(provider, "p1", "Player")
1264        controller._players = {"p1": player}
1265        mock_mass.players = controller
1266        player.set_initialized()
1267        player._cache.clear()
1268        player.update_state(signal_event=False)
1269        controller._forward_state_update = MagicMock()  # type: ignore[method-assign]
1270        controller.cmd_ungroup = MagicMock(return_value="ungroup-coro")  # type: ignore[method-assign]
1271
1272        controller.signal_player_state_update(player, {"powered": (True, False)})
1273
1274        controller.cmd_ungroup.assert_not_called()
1275
1276
1277class TestPlayMediaOverride:
1278    """
1279    Tests for the new CONF_PLAY_MEDIA_OVERRIDES_GROUP behavior.
1280
1281    When a captured child player receives an explicit play_media command, the
1282    default behavior is to *release* it from the active group/sync and play
1283    directly on the targeted player. The legacy behavior (forward to the
1284    leader) is preserved via the per-player config opt-out.
1285    """
1286
1287    @pytest.mark.asyncio
1288    async def test_override_disabled_redirects_to_group(self, mock_mass: MagicMock) -> None:
1289        """With override disabled, play_media on a captured child redirects to the group."""
1290        controller = PlayerController(mock_mass)
1291        group_provider = MockProvider("test_group", instance_id="test_group", mass=mock_mass)
1292        member_provider = MockProvider("test", instance_id="test", mass=mock_mass)
1293
1294        class _SessionedGroup(MockPlayer):
1295            @property
1296            def is_active_session(self) -> bool:
1297                return True
1298
1299        group = _SessionedGroup(group_provider, "g1", "Group", player_type=PlayerType.GROUP)
1300        group._attr_powered = None
1301        group._attr_group_members = ["member"]
1302
1303        member = MockPlayer(member_provider, "member", "Member")
1304
1305        controller._players = {"g1": group, "member": member}
1306        mock_mass.players = controller
1307
1308        group.set_initialized()
1309        member.set_initialized()
1310        group.update_state(signal_event=False)
1311        member.update_state(signal_event=False)
1312        # sanity: the member is captured by the group
1313        assert member.state.active_group == "g1"
1314
1315        _set_play_media_override(mock_mass, False)
1316
1317        played_on: list[str] = []
1318
1319        async def _handle_play_media(player_id: str, media: object) -> None:  # noqa: ARG001
1320            played_on.append(player_id)
1321
1322        controller._handle_play_media = _handle_play_media  # type: ignore[method-assign]
1323        # the play_media wrapper acquires a playback lock; stub it out
1324        controller._player_command_locks = {}
1325
1326        media = MagicMock(uri="x", source_id="src")
1327        await controller.play_media("member", media)
1328
1329        # legacy behavior: redirected to the group leader
1330        assert played_on == ["g1"]
1331
1332    @pytest.mark.asyncio
1333    async def test_override_releases_dynamic_member(self, mock_mass: MagicMock) -> None:
1334        """With override enabled, play_media on a dynamic group member releases it first."""
1335        controller = PlayerController(mock_mass)
1336        group_provider = MockProvider("test_group", instance_id="test_group", mass=mock_mass)
1337        member_provider = MockProvider("test", instance_id="test", mass=mock_mass)
1338
1339        class _SessionedGroup(MockPlayer):
1340            @property
1341            def is_active_session(self) -> bool:
1342                return True
1343
1344        group = _SessionedGroup(group_provider, "g1", "Group", player_type=PlayerType.GROUP)
1345        group._attr_powered = None
1346        group._attr_supported_features.add(PlayerFeature.SET_MEMBERS)
1347        group._attr_group_members = ["member"]
1348        # NOT a static member ⇒ dynamic — can be removed via set_members
1349        group._attr_static_group_members = []
1350
1351        member = MockPlayer(member_provider, "member", "Member")
1352
1353        controller._players = {"g1": group, "member": member}
1354        mock_mass.players = controller
1355
1356        group.set_initialized()
1357        member.set_initialized()
1358        group.update_state(signal_event=False)
1359        member.update_state(signal_event=False)
1360        assert member.state.active_group == "g1"
1361
1362        # default: override enabled
1363        _set_play_media_override(mock_mass, True)
1364
1365        set_members_calls: list[dict[str, object]] = []
1366
1367        async def _cmd_set_members(
1368            target_player: str,
1369            player_ids_to_add: list[str] | None = None,  # noqa: ARG001
1370            player_ids_to_remove: list[str] | None = None,
1371        ) -> None:
1372            set_members_calls.append(
1373                {"player_id": target_player, "remove": player_ids_to_remove or []}
1374            )
1375
1376        controller.cmd_set_members = _cmd_set_members  # type: ignore[method-assign]
1377
1378        played_on: list[str] = []
1379
1380        async def _handle_play_media(player_id: str, media: object) -> None:  # noqa: ARG001
1381            played_on.append(player_id)
1382
1383        controller._handle_play_media = _handle_play_media  # type: ignore[method-assign]
1384        controller._player_command_locks = {}
1385
1386        media = MagicMock(uri="x", source_id="src")
1387        with patch.object(
1388            controller,
1389            "wait_for_player_update",
1390            _skip_player_update_wait,
1391        ):
1392            await controller.play_media("member", media)
1393
1394        # the member was removed from the group ...
1395        assert set_members_calls == [{"player_id": "g1", "remove": ["member"]}]
1396        # ... and then play_media was issued directly on the member, NOT on the group
1397        assert played_on == ["member"]
1398
1399    @pytest.mark.asyncio
1400    async def test_override_stops_static_group(self, mock_mass: MagicMock) -> None:
1401        """With override enabled, play_media on a STATIC group member stops the group."""
1402        controller = PlayerController(mock_mass)
1403        group_provider = MockProvider("test_group", instance_id="test_group", mass=mock_mass)
1404        member_provider = MockProvider("test", instance_id="test", mass=mock_mass)
1405
1406        class _SessionedGroup(MockPlayer):
1407            @property
1408            def is_active_session(self) -> bool:
1409                return True
1410
1411        group = _SessionedGroup(group_provider, "g1", "Group", player_type=PlayerType.GROUP)
1412        group._attr_powered = None  # no power control ⇒ stop, not power-off
1413        group._attr_supported_features.add(PlayerFeature.SET_MEMBERS)
1414        group._attr_group_members = ["member"]
1415        # static member - cannot be removed individually
1416        group._attr_static_group_members = ["member"]
1417
1418        member = MockPlayer(member_provider, "member", "Member")
1419
1420        controller._players = {"g1": group, "member": member}
1421        mock_mass.players = controller
1422
1423        group.set_initialized()
1424        member.set_initialized()
1425        group.update_state(signal_event=False)
1426        member.update_state(signal_event=False)
1427        assert member.state.active_group == "g1"
1428
1429        _set_play_media_override(mock_mass, True)
1430
1431        stop_calls: list[str] = []
1432        power_calls: list[tuple[str, bool]] = []
1433
1434        async def _stop(player_id: str) -> None:
1435            stop_calls.append(player_id)
1436
1437        async def _power(
1438            player_id: str,
1439            powered: bool,
1440            skip_auto_play: bool = False,  # noqa: ARG001
1441        ) -> None:
1442            power_calls.append((player_id, powered))
1443
1444        controller._handle_cmd_stop = _stop  # type: ignore[method-assign]
1445        controller._handle_cmd_power = _power  # type: ignore[method-assign]
1446
1447        played_on: list[str] = []
1448
1449        async def _handle_play_media(player_id: str, media: object) -> None:  # noqa: ARG001
1450            played_on.append(player_id)
1451
1452        controller._handle_play_media = _handle_play_media  # type: ignore[method-assign]
1453        controller._player_command_locks = {}
1454
1455        media = MagicMock(uri="x", source_id="src")
1456        with patch.object(
1457            controller,
1458            "wait_for_player_update",
1459            _skip_player_update_wait,
1460        ):
1461            await controller.play_media("member", media)
1462
1463        # powerless group + static member: we should have stopped the group ...
1464        assert stop_calls == ["g1"]
1465        # ... not powered it off ...
1466        assert power_calls == []
1467        # ... and play_media was issued directly on the member
1468        assert played_on == ["member"]
1469
1470
1471class TestExternalSourcePlayPause:
1472    """Pause/play handling for externally-initiated sources (no active output protocol)."""
1473
1474    @staticmethod
1475    def _make_external_source_player(
1476        provider: MockProvider,
1477        controller: PlayerController,
1478        mock_mass: MagicMock,
1479        *,
1480        playback_state: PlaybackState,
1481        can_play_pause: bool = True,
1482        supports_pause: bool = True,
1483    ) -> MockPlayer:
1484        """Build a player playing a passive external source, with no active output protocol."""
1485        player = MockPlayer(provider, "player_1", "Test Player")
1486        player._attr_supported_features = {PlayerFeature.PAUSE} if supports_pause else set()
1487        player._attr_source_list = [
1488            PlayerSource(
1489                id="spotify",
1490                name="Spotify",
1491                passive=True,
1492                can_play_pause=can_play_pause,
1493                can_next_previous=True,
1494                can_seek=True,
1495            )
1496        ]
1497        player._attr_active_source = "spotify"
1498        player._attr_playback_state = playback_state
1499        player._cache.clear()
1500        controller._players = {"player_1": player}
1501        mock_mass.players = controller
1502        mock_mass.player_queues = MagicMock()
1503        mock_mass.player_queues.get = MagicMock(return_value=None)
1504        player.update_state(signal_event=False)
1505        return player
1506
1507    def test_pause_external_source_forwards_to_player(
1508        self, mock_mass: MagicMock, controller: PlayerController, provider: MockProvider
1509    ) -> None:
1510        """Pausing a pausable external source forwards to the player, not STOP."""
1511        player = self._make_external_source_player(
1512            provider, controller, mock_mass, playback_state=PlaybackState.PLAYING
1513        )
1514        player.pause = AsyncMock()  # type: ignore[method-assign]
1515        controller._handle_cmd_stop = AsyncMock()  # type: ignore[method-assign]
1516
1517        asyncio.run(controller._handle_cmd_pause("player_1"))
1518
1519        player.pause.assert_awaited_once()
1520        controller._handle_cmd_stop.assert_not_called()
1521
1522    def test_play_external_source_unpauses_player(
1523        self, mock_mass: MagicMock, controller: PlayerController, provider: MockProvider
1524    ) -> None:
1525        """Unpausing a paused external source forwards to the player, not a restart."""
1526        player = self._make_external_source_player(
1527            provider, controller, mock_mass, playback_state=PlaybackState.PAUSED
1528        )
1529        player.play = AsyncMock()  # type: ignore[method-assign]
1530        player.play_media = AsyncMock()  # type: ignore[method-assign]
1531        controller._handle_select_source = AsyncMock()  # type: ignore[method-assign]
1532
1533        asyncio.run(controller._handle_cmd_play("player_1"))
1534
1535        player.play.assert_awaited_once()
1536        player.play_media.assert_not_called()
1537        controller._handle_select_source.assert_not_called()
1538
1539    def test_pause_falls_back_to_stop_without_pause_support(
1540        self, mock_mass: MagicMock, controller: PlayerController, provider: MockProvider
1541    ) -> None:
1542        """A player that cannot pause natively still falls back to STOP."""
1543        player = self._make_external_source_player(
1544            provider,
1545            controller,
1546            mock_mass,
1547            playback_state=PlaybackState.PLAYING,
1548            supports_pause=False,
1549        )
1550        player.pause = AsyncMock()  # type: ignore[method-assign]
1551        controller._handle_cmd_stop = AsyncMock()  # type: ignore[method-assign]
1552
1553        asyncio.run(controller._handle_cmd_pause("player_1"))
1554
1555        controller._handle_cmd_stop.assert_awaited_once()
1556        player.pause.assert_not_called()
1557
1558
1559class TestProtocolOutputPlayPause:
1560    """Play/pause on a player rendering through a linked output protocol."""
1561
1562    @staticmethod
1563    def _make_player_on_protocol(
1564        mock_mass: MagicMock,
1565        controller: PlayerController,
1566        *,
1567        playback_state: PlaybackState,
1568    ) -> MockPlayer:
1569        """Build a player playing the MA queue through a protocol that cannot pause."""
1570        native_provider = MockProvider("chromecast", mass=mock_mass)
1571        player = MockPlayer(native_provider, "player_1", "Test Player")
1572        player._attr_supported_features.add(PlayerFeature.PAUSE)
1573        player._attr_playback_state = playback_state
1574
1575        protocol_provider = MockProvider("sendspin", mass=mock_mass)
1576        protocol_player = MockPlayer(
1577            protocol_provider, "proto_1", "Test Protocol", player_type=PlayerType.PROTOCOL
1578        )
1579        protocol_player._attr_playback_state = playback_state
1580
1581        controller._players = {"player_1": player, "proto_1": protocol_player}
1582        mock_mass.players = controller
1583        mock_mass.player_queues = MagicMock()
1584        # a non-empty queue, so the MA queue source advertises play/pause support
1585        queue = MagicMock()
1586        queue.items = [MagicMock()]
1587        mock_mass.player_queues.get = MagicMock(return_value=queue)
1588        player.set_linked_output_protocols(
1589            [
1590                LinkedOutputProtocol(
1591                    output_protocol_id="proto_1",
1592                    protocol_domain="sendspin",
1593                    priority=40,
1594                )
1595            ]
1596        )
1597        player.set_active_output_protocol("proto_1")
1598        player.set_active_mass_source("player_1")
1599        protocol_player.update_state(signal_event=False)
1600        player.refresh_state(signal_event=False)
1601        return player
1602
1603    async def test_pause_on_protocol_without_pause_falls_back_to_stop(
1604        self, mock_mass: MagicMock, controller: PlayerController
1605    ) -> None:
1606        """The native transport has no session to pause while a protocol renders the audio."""
1607        player = self._make_player_on_protocol(
1608            mock_mass, controller, playback_state=PlaybackState.PLAYING
1609        )
1610        player.pause = AsyncMock()  # type: ignore[method-assign]
1611        controller._handle_cmd_stop = AsyncMock()  # type: ignore[method-assign]
1612
1613        await controller._handle_cmd_pause("player_1")
1614
1615        player.pause.assert_not_called()
1616        # STOP goes to the visible player, not the protocol player
1617        controller._handle_cmd_stop.assert_awaited_once_with("player_1")
1618
1619    async def test_play_on_protocol_without_pause_does_not_unpause_natively(
1620        self, mock_mass: MagicMock, controller: PlayerController
1621    ) -> None:
1622        """Unpausing must not hit the native transport either; the source is restarted."""
1623        player = self._make_player_on_protocol(
1624            mock_mass, controller, playback_state=PlaybackState.PAUSED
1625        )
1626        player.play = AsyncMock()  # type: ignore[method-assign]
1627        controller._handle_select_source = AsyncMock()  # type: ignore[method-assign]
1628
1629        await controller._handle_cmd_play("player_1")
1630
1631        player.play.assert_not_called()
1632        # the MA queue source is restarted, not some other source
1633        controller._handle_select_source.assert_awaited_once_with("player_1", "player_1")
1634
1635
1636class TestMirrorsParentMedia:
1637    """Tests for _mirrors_parent_media (palette-fetch gating for grouped players)."""
1638
1639    @staticmethod
1640    def _fake_player(
1641        *,
1642        player_id: str = "p1",
1643        active_group: str | None = None,
1644        synced_to: str | None = None,
1645        player_type: PlayerType = PlayerType.PLAYER,
1646        protocol_parent_id: str | None = None,
1647    ) -> SimpleNamespace:
1648        return SimpleNamespace(
1649            player_id=player_id,
1650            state=SimpleNamespace(active_group=active_group, synced_to=synced_to, type=player_type),
1651            protocol_parent_id=protocol_parent_id,
1652        )
1653
1654    def test_standalone_player_owns_media(self, controller: PlayerController) -> None:
1655        """A standalone player resolves its own media (and palette)."""
1656        assert controller._mirrors_parent_media(self._fake_player()) is False  # type: ignore[arg-type]
1657
1658    def test_group_member_mirrors(self, controller: PlayerController) -> None:
1659        """A group member borrows its parent's media."""
1660        assert controller._mirrors_parent_media(self._fake_player(active_group="g1")) is True  # type: ignore[arg-type]
1661
1662    def test_synced_member_mirrors(self, controller: PlayerController) -> None:
1663        """A synced member borrows its leader's media."""
1664        assert controller._mirrors_parent_media(self._fake_player(synced_to="leader")) is True  # type: ignore[arg-type]
1665
1666    def test_protocol_child_mirrors(self, controller: PlayerController) -> None:
1667        """A protocol child borrows its parent's media."""
1668        player = self._fake_player(player_type=PlayerType.PROTOCOL, protocol_parent_id="parent")
1669        assert controller._mirrors_parent_media(player) is True  # type: ignore[arg-type]
1670
1671    def test_protocol_player_without_parent_owns_media(self, controller: PlayerController) -> None:
1672        """A protocol player with no parent resolves its own media."""
1673        player = self._fake_player(player_type=PlayerType.PROTOCOL)
1674        assert controller._mirrors_parent_media(player) is False  # type: ignore[arg-type]
1675
1676    def test_self_referential_parent_owns_media(self, controller: PlayerController) -> None:
1677        """A self-referential active_group/synced_to is not a real parent, so resolve locally."""
1678        player = self._fake_player(player_id="p1", synced_to="p1", active_group="p1")
1679        assert controller._mirrors_parent_media(player) is False  # type: ignore[arg-type]
1680
1681
1682class TestVolumeScalingOnRedirect:
1683    """min/max volume scaling must survive a redirect to a protocol player or external control."""
1684
1685    @staticmethod
1686    def _volume_player(
1687        player_id: str,
1688        volume_control: str,
1689        volume_set: AsyncMock | None = None,
1690    ) -> SimpleNamespace:
1691        return SimpleNamespace(
1692            player_id=player_id,
1693            type=PlayerType.PLAYER,
1694            protocol_parent_id=None,
1695            extra_data={},
1696            volume_control=volume_control,
1697            mute_control=PLAYER_CONTROL_NONE,
1698            volume_set=volume_set or AsyncMock(),
1699            update_state=MagicMock(),
1700            provider=MagicMock(),
1701            state=SimpleNamespace(
1702                name=player_id,
1703                volume_control=volume_control,
1704                volume_muted=False,
1705                mute_control=PLAYER_CONTROL_NONE,
1706            ),
1707        )
1708
1709    @pytest.mark.asyncio
1710    async def test_protocol_redirect_forwards_scaled_volume(
1711        self, controller: PlayerController, mock_mass: MagicMock
1712    ) -> None:
1713        """A volume command redirected to a protocol player honors the user-facing max_volume."""
1714
1715        def _conf(player_id: str, key: str, default: object = None) -> object:
1716            if key == "min_volume":
1717                return 0
1718            if key == "max_volume":
1719                # user-facing player caps at 50, the protocol player has no limits of its own
1720                return 50 if player_id == "user_player" else 100
1721            return default
1722
1723        mock_mass.config.get_raw_player_config_value = MagicMock(side_effect=_conf)
1724
1725        protocol = self._volume_player("protocol_player", PLAYER_CONTROL_NATIVE)
1726        user = self._volume_player("user_player", "protocol_player")
1727        players = {"user_player": user, "protocol_player": protocol}
1728
1729        with (
1730            patch.object(controller, "get_player", side_effect=players.get),
1731            patch.object(controller, "_get_active_audio_source", return_value=None),
1732        ):
1733            controller._controls = {}
1734            await controller._handle_cmd_volume_set("user_player", 100)
1735
1736        # logical 100 with a max_volume of 50 must reach the protocol player as 50, not the raw 100
1737        protocol.volume_set.assert_awaited_once_with(50)
1738
1739    @pytest.mark.asyncio
1740    async def test_external_control_redirect_forwards_scaled_volume(
1741        self, controller: PlayerController, mock_mass: MagicMock
1742    ) -> None:
1743        """A volume command redirected to an external control honors the user-facing max_volume."""
1744        mock_mass.config.get_raw_player_config_value = MagicMock(
1745            side_effect=_player_config_stub(max_volume=50)
1746        )
1747
1748        volume_set = AsyncMock()
1749        control = PlayerControl(
1750            id="ext_control",
1751            provider="test",
1752            name="External Amp",
1753            supports_volume=True,
1754            volume_set=volume_set,
1755        )
1756        user = self._volume_player("user_player", "ext_control")
1757        players = {"user_player": user}
1758
1759        with (
1760            patch.object(controller, "get_player", side_effect=players.get),
1761            patch.object(controller, "_get_active_audio_source", return_value=None),
1762        ):
1763            controller._controls = {"ext_control": control}
1764            await controller._handle_cmd_volume_set("user_player", 100)
1765
1766        volume_set.assert_awaited_once_with(50)
1767
1768    @pytest.mark.asyncio
1769    async def test_external_control_without_volume_support_raises(
1770        self, controller: PlayerController, mock_mass: MagicMock
1771    ) -> None:
1772        """A volume command redirected to a control lacking volume support is rejected."""
1773        mock_mass.config.get_raw_player_config_value = MagicMock(
1774            side_effect=_player_config_stub(max_volume=50)
1775        )
1776
1777        volume_set = AsyncMock()
1778        control = PlayerControl(
1779            id="ext_control",
1780            provider="test",
1781            name="External Amp",
1782            supports_volume=False,
1783            volume_set=volume_set,
1784        )
1785        user = self._volume_player("user_player", "ext_control")
1786        players = {"user_player": user}
1787
1788        with (
1789            patch.object(controller, "get_player", side_effect=players.get),
1790            patch.object(controller, "_get_active_audio_source", return_value=None),
1791        ):
1792            controller._controls = {"ext_control": control}
1793            with pytest.raises(UnsupportedFeaturedException):
1794                await controller._handle_cmd_volume_set("user_player", 100)
1795
1796        volume_set.assert_not_awaited()
1797
1798
1799class TestExternalPowerControl:
1800    """Power commands redirected to an external PlayerControl must forward and gate correctly."""
1801
1802    def _make_player(
1803        self, mock_mass: MagicMock, control: PlayerControl
1804    ) -> tuple[PlayerController, MockPlayer]:
1805        """Build a controller with a single player whose power control is the given control."""
1806        mock_mass.config.get_raw_player_config_value = MagicMock(
1807            side_effect=_player_config_stub({CONF_POWER_CONTROL: control.id})
1808        )
1809        controller = PlayerController(mock_mass)
1810        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
1811        player = MockPlayer(provider, "player_1", "Player 1")
1812        controller._controls = {control.id: control}
1813        controller._players = {"player_1": player}
1814        mock_mass.players = controller
1815        # auto-play would otherwise resume the (mocked) player queue on power on
1816        config_get_value = player.config.get_value
1817        player.config.get_value = MagicMock(  # type: ignore[method-assign]
1818            side_effect=lambda key, *args, **kwargs: (
1819                False if key == CONF_AUTO_PLAY else config_get_value(key, *args, **kwargs)
1820            )
1821        )
1822        player.set_initialized()
1823        player.update_state(signal_event=False)
1824        return controller, player
1825
1826    async def test_power_on_forwards_to_control(self, mock_mass: MagicMock) -> None:
1827        """Powering on a player redirects to its external control's power_on callback."""
1828        power_on = AsyncMock()
1829        power_off = AsyncMock()
1830        control = PlayerControl(
1831            id="ext_power",
1832            provider="test",
1833            name="External Power",
1834            supports_power=True,
1835            power_on=power_on,
1836            power_off=power_off,
1837        )
1838
1839        def _report_powered_on() -> None:
1840            control.power_state = True
1841
1842        # the control only reports on once switched on, which releases wait_for_power_on
1843        power_on.side_effect = _report_powered_on
1844        controller, player = self._make_player(mock_mass, control)
1845        assert player.state.powered is False
1846
1847        await controller._handle_cmd_power("player_1", True)
1848
1849        power_on.assert_awaited_once()
1850        power_off.assert_not_awaited()
1851
1852    async def test_power_on_waits_on_the_control(self, mock_mass: MagicMock) -> None:
1853        """Powering on waits for the control to report on, not for the player itself."""
1854        control = PlayerControl(
1855            id="ext_power",
1856            provider="test",
1857            name="External Power",
1858            supports_power=True,
1859            power_on=AsyncMock(),
1860            power_off=AsyncMock(),
1861        )
1862        controller, player = self._make_player(mock_mass, control)
1863        assert player.state.powered is False
1864
1865        with patch(
1866            "music_assistant.controllers.players.controller.wait_for_power_on", AsyncMock()
1867        ) as wait_for_power_on:
1868            await controller._handle_cmd_power("player_1", True)
1869
1870        wait_for_power_on.assert_awaited_once()
1871        assert wait_for_power_on.await_args is not None
1872        assert wait_for_power_on.await_args.args[2] is control
1873
1874    async def test_power_off_forwards_to_control(self, mock_mass: MagicMock) -> None:
1875        """Powering off a player redirects to its external control's power_off callback."""
1876        power_on = AsyncMock()
1877        power_off = AsyncMock()
1878        control = PlayerControl(
1879            id="ext_power",
1880            provider="test",
1881            name="External Power",
1882            supports_power=True,
1883            power_state=True,
1884            power_on=power_on,
1885            power_off=power_off,
1886        )
1887        controller, player = self._make_player(mock_mass, control)
1888        assert player.state.powered is True
1889
1890        await controller._handle_cmd_power("player_1", False)
1891
1892        power_off.assert_awaited_once()
1893        power_on.assert_not_awaited()
1894
1895    async def test_control_without_power_support_raises(self, mock_mass: MagicMock) -> None:
1896        """A power command redirected to a control lacking power support is rejected."""
1897        power_on = AsyncMock()
1898        power_off = AsyncMock()
1899        control = PlayerControl(
1900            id="ext_power",
1901            provider="test",
1902            name="External Power",
1903            supports_power=False,
1904            power_on=power_on,
1905            power_off=power_off,
1906        )
1907        controller, player = self._make_player(mock_mass, control)
1908        assert player.state.powered is False
1909
1910        with pytest.raises(UnsupportedFeaturedException):
1911            await controller._handle_cmd_power("player_1", True)
1912
1913        power_on.assert_not_awaited()
1914        power_off.assert_not_awaited()
1915
1916
1917class TestEnforceVolumeLimits:
1918    """External volume changes outside the min/max range must be corrected."""
1919
1920    @staticmethod
1921    def _set_limits(mock_mass: MagicMock, min_volume: int, max_volume: int) -> None:
1922        mock_mass.config.get_raw_player_config_value = MagicMock(
1923            side_effect=_player_config_stub(min_volume=min_volume, max_volume=max_volume)
1924        )
1925
1926    @staticmethod
1927    def _player(logical_volume: int | None) -> SimpleNamespace:
1928        return SimpleNamespace(
1929            player_id="user_player",
1930            state=SimpleNamespace(volume_level=logical_volume),
1931        )
1932
1933    def test_out_of_range_volume_is_corrected(
1934        self, controller: PlayerController, mock_mass: MagicMock
1935    ) -> None:
1936        """A device volume above max_volume (logical > 100) is clamped back to logical 100."""
1937        self._set_limits(mock_mass, 0, 80)
1938        # device volume 100 with max 80 resolves to logical 125
1939        player = self._player(125)
1940        with patch.object(controller, "_handle_cmd_volume_set", MagicMock()) as cmd:
1941            controller._enforce_volume_limits(cast("MockPlayer", player))
1942        cmd.assert_called_once_with("user_player", 100)
1943        mock_mass.create_task.assert_called_once()
1944
1945    def test_below_min_volume_is_corrected(
1946        self, controller: PlayerController, mock_mass: MagicMock
1947    ) -> None:
1948        """A device volume below min_volume (logical < 0) is clamped back to logical 0."""
1949        self._set_limits(mock_mass, 20, 100)
1950        # device volume 10 with min 20 resolves to a negative logical volume
1951        player = self._player(-13)
1952        with patch.object(controller, "_handle_cmd_volume_set", MagicMock()) as cmd:
1953            controller._enforce_volume_limits(cast("MockPlayer", player))
1954        cmd.assert_called_once_with("user_player", 0)
1955
1956    def test_in_range_volume_is_untouched(
1957        self, controller: PlayerController, mock_mass: MagicMock
1958    ) -> None:
1959        """A logical volume within 0-100 needs no correction."""
1960        self._set_limits(mock_mass, 0, 80)
1961        player = self._player(100)
1962        with patch.object(controller, "_handle_cmd_volume_set", MagicMock()) as cmd:
1963            controller._enforce_volume_limits(cast("MockPlayer", player))
1964        cmd.assert_not_called()
1965
1966    def test_no_limits_configured_is_noop(
1967        self, controller: PlayerController, mock_mass: MagicMock
1968    ) -> None:
1969        """Default 0-100 limits skip enforcement entirely."""
1970        self._set_limits(mock_mass, 0, 100)
1971        player = self._player(100)
1972        with patch.object(controller, "_handle_cmd_volume_set", MagicMock()) as cmd:
1973            controller._enforce_volume_limits(cast("MockPlayer", player))
1974        cmd.assert_not_called()
1975
1976    def test_unknown_volume_is_noop(
1977        self, controller: PlayerController, mock_mass: MagicMock
1978    ) -> None:
1979        """A player without a resolved volume level is left alone."""
1980        self._set_limits(mock_mass, 0, 80)
1981        player = self._player(None)
1982        with patch.object(controller, "_handle_cmd_volume_set", MagicMock()) as cmd:
1983            controller._enforce_volume_limits(cast("MockPlayer", player))
1984        cmd.assert_not_called()
1985
1986
1987class TestFakeMuteControl:
1988    """Fake mute must report the muted state and restore the volume on unmute."""
1989
1990    def _make_player(
1991        self, mock_mass: MagicMock, volume_level: int | None = 40
1992    ) -> tuple[PlayerController, MockPlayer, AsyncMock]:
1993        """
1994        Build a controller with a single player using fake mute control.
1995
1996        :param mock_mass: the mocked MusicAssistant instance.
1997        :param volume_level: initial volume level of the player, None if unknown.
1998        """
1999        mock_mass.config.get_raw_player_config_value = MagicMock(
2000            side_effect=_player_config_stub({CONF_MUTE_CONTROL: PLAYER_CONTROL_FAKE})
2001        )
2002        controller = PlayerController(mock_mass)
2003        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
2004        player = MockPlayer(provider, "player_1", "Player 1")
2005        controller._players = {"player_1": player}
2006        mock_mass.players = controller
2007        mock_mass.player_queues.get = MagicMock(return_value=None)
2008        player.set_initialized()
2009        player._attr_volume_level = volume_level
2010        # let the mocked native volume control behave like a real device
2011        volume_set = AsyncMock(
2012            side_effect=lambda volume: setattr(player, "_attr_volume_level", volume)
2013        )
2014        player.volume_set = volume_set  # type: ignore[method-assign]
2015        player.update_state(signal_event=False)
2016        return controller, player, volume_set
2017
2018    async def test_mute_then_unmute_restores_volume(self, mock_mass: MagicMock) -> None:
2019        """Muting reports volume_muted=True and unmuting restores the previous volume."""
2020        controller, player, volume_set = self._make_player(mock_mass)
2021
2022        await controller.cmd_volume_mute("player_1", True)
2023        muted_state = player.state
2024        assert muted_state.volume_muted is True
2025        assert muted_state.volume_level == 0
2026        assert player.extra_data[ATTR_PREVIOUS_VOLUME] == 40
2027
2028        await controller.cmd_volume_mute("player_1", False)
2029        volume_set.assert_awaited_with(40)
2030        # simulate the device reporting back its state after the volume command
2031        player.update_state()
2032        unmuted_state = player.state
2033        assert unmuted_state.volume_muted is False
2034        assert unmuted_state.volume_level == 40
2035
2036    async def test_repeated_mute_keeps_previous_volume(self, mock_mass: MagicMock) -> None:
2037        """A repeated mute command must not overwrite the stored volume with 0."""
2038        controller, player, volume_set = self._make_player(mock_mass)
2039
2040        await controller.cmd_volume_mute("player_1", True)
2041        await controller.cmd_volume_mute("player_1", True)
2042        assert player.extra_data[ATTR_PREVIOUS_VOLUME] == 40
2043        assert player.state.volume_muted is True
2044
2045        await controller.cmd_volume_mute("player_1", False)
2046        volume_set.assert_awaited_with(40)
2047
2048    async def test_unmute_with_unknown_previous_volume(self, mock_mass: MagicMock) -> None:
2049        """Unmuting a player whose volume was unknown at mute time restores a low volume."""
2050        controller, player, volume_set = self._make_player(mock_mass, volume_level=None)
2051
2052        await controller.cmd_volume_mute("player_1", True)
2053        assert player.extra_data[ATTR_PREVIOUS_VOLUME] is None
2054
2055        await controller.cmd_volume_mute("player_1", False)
2056        volume_set.assert_awaited_with(1)
2057        player.update_state()
2058        assert player.state.volume_muted is False
2059
2060    async def test_unmute_of_unmuted_player_keeps_volume(self, mock_mass: MagicMock) -> None:
2061        """An unmute command for a player that is not muted may not touch the volume."""
2062        controller, player, volume_set = self._make_player(mock_mass, volume_level=50)
2063
2064        await controller.cmd_volume_mute("player_1", False)
2065        volume_set.assert_not_awaited()
2066        assert player.state.volume_level == 50
2067
2068    async def test_unmute_restores_a_stored_zero_volume(self, mock_mass: MagicMock) -> None:
2069        """A player that was already silent stays silent after mute and unmute."""
2070        controller, _player, volume_set = self._make_player(mock_mass, volume_level=0)
2071
2072        await controller.cmd_volume_mute("player_1", True)
2073        await controller.cmd_volume_mute("player_1", False)
2074        volume_set.assert_awaited_with(0)
2075
2076    async def test_volume_set_clears_fake_mute(self, mock_mass: MagicMock) -> None:
2077        """A regular volume change while fake muted implies an unmute."""
2078        controller, player, _volume_set = self._make_player(mock_mass)
2079
2080        await controller.cmd_volume_mute("player_1", True)
2081        muted_state = player.state
2082        assert muted_state.volume_muted is True
2083
2084        await controller.cmd_volume_set("player_1", 25)
2085        # simulate the device reporting back its state after the volume command
2086        player.update_state()
2087        unmuted_state = player.state
2088        assert unmuted_state.volume_muted is False
2089        assert unmuted_state.volume_level == 25
2090
2091
2092class TestVolumeStep:
2093    """The volume_step core config setting controls the size of a single volume nudge."""
2094
2095    def _make_player(
2096        self, mock_mass: MagicMock, step: int | None, volume_level: int
2097    ) -> tuple[PlayerController, MockPlayer]:
2098        """Build a controller with a single player and the given volume_step config."""
2099        controller = PlayerController(mock_mass)
2100        controller.config = _volume_step_config(step)
2101        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
2102        player = MockPlayer(provider, "player_1", "Player 1")
2103        player._attr_volume_level = volume_level
2104        # let the mocked native volume control behave like a real device
2105        player.volume_set = AsyncMock(  # type: ignore[method-assign]
2106            side_effect=lambda volume: setattr(player, "_attr_volume_level", volume)
2107        )
2108        controller._players = {"player_1": player}
2109        mock_mass.players = controller
2110        mock_mass.player_queues.get = MagicMock(return_value=None)
2111        player.set_initialized()
2112        player.update_state(signal_event=False)
2113        return controller, player
2114
2115    def _make_synced_pair(
2116        self, mock_mass: MagicMock, step: int | None
2117    ) -> tuple[PlayerController, dict[str, MockPlayer]]:
2118        """Build a leader synced to one member, both at volume 50, with a volume_step config."""
2119        controller = PlayerController(mock_mass)
2120        controller.config = _volume_step_config(step)
2121        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
2122        players: dict[str, MockPlayer] = {}
2123        for player_id in ("leader", "member"):
2124            player = MockPlayer(provider, player_id, player_id.title())
2125            player._attr_volume_level = 50
2126            # let the mocked native volume control behave like a real device
2127            player.volume_set = AsyncMock(  # type: ignore[method-assign]
2128                side_effect=lambda volume, _player=player: setattr(
2129                    _player, "_attr_volume_level", volume
2130                )
2131            )
2132            players[player_id] = player
2133        players["leader"]._attr_group_members = ["member"]
2134        controller._players = dict(players)
2135        mock_mass.players = controller
2136        mock_mass.player_queues.get = MagicMock(return_value=None)
2137        for player in players.values():
2138            player.set_initialized()
2139            player._cache.clear()
2140            player.update_state(signal_event=False)
2141        # a second, forced pass: update_state() only recalculates when a player's own
2142        # attributes changed, so the leader must be forced to re-derive its group_volume
2143        # from the now-initialized member.
2144        for player in players.values():
2145            player.update_state(force_update=True, signal_event=False)
2146        return controller, players
2147
2148    @pytest.mark.parametrize(
2149        ("start", "expected"),
2150        [(5, 6), (20, 22), (50, 53), (80, 82), (95, 96)],
2151    )
2152    async def test_default_step_up_matches_the_adaptive_ladder(
2153        self, mock_mass: MagicMock, start: int, expected: int
2154    ) -> None:
2155        """With volume_step at its default (0), volume_up keeps today's adaptive ladder."""
2156        controller, player = self._make_player(mock_mass, None, start)
2157
2158        await controller.cmd_volume_up("player_1")
2159
2160        player.update_state()
2161        assert player.state.volume_level == expected
2162
2163    @pytest.mark.parametrize(
2164        ("start", "expected"),
2165        [(5, 4), (20, 18), (50, 47), (80, 78), (95, 94)],
2166    )
2167    async def test_default_step_down_matches_the_adaptive_ladder(
2168        self, mock_mass: MagicMock, start: int, expected: int
2169    ) -> None:
2170        """With volume_step at its default (0), volume_down keeps today's adaptive ladder."""
2171        controller, player = self._make_player(mock_mass, None, start)
2172
2173        await controller.cmd_volume_down("player_1")
2174
2175        player.update_state()
2176        assert player.state.volume_level == expected
2177
2178    async def test_configured_step_moves_up_by_a_flat_amount_mid_range(
2179        self, mock_mass: MagicMock
2180    ) -> None:
2181        """A configured flat step of 5 moves by exactly 5 in the middle of the range."""
2182        controller, player = self._make_player(mock_mass, 5, 50)
2183
2184        await controller.cmd_volume_up("player_1")
2185
2186        player.update_state()
2187        assert player.state.volume_level == 55
2188
2189    async def test_configured_step_moves_up_by_a_flat_amount_near_the_extreme(
2190        self, mock_mass: MagicMock
2191    ) -> None:
2192        """A configured flat step of 5 near the extreme overrides the finer ladder step."""
2193        controller, player = self._make_player(mock_mass, 5, 5)
2194
2195        await controller.cmd_volume_up("player_1")
2196
2197        player.update_state()
2198        assert player.state.volume_level == 10
2199
2200    async def test_configured_step_moves_down_by_a_flat_amount_mid_range(
2201        self, mock_mass: MagicMock
2202    ) -> None:
2203        """A configured flat step of 5 moves down by exactly 5 in the middle of the range."""
2204        controller, player = self._make_player(mock_mass, 5, 50)
2205
2206        await controller.cmd_volume_down("player_1")
2207
2208        player.update_state()
2209        assert player.state.volume_level == 45
2210
2211    async def test_configured_step_moves_down_by_a_flat_amount_near_the_extreme(
2212        self, mock_mass: MagicMock
2213    ) -> None:
2214        """A configured flat step of 5 near the extreme overrides the finer ladder step."""
2215        controller, player = self._make_player(mock_mass, 5, 95)
2216
2217        await controller.cmd_volume_down("player_1")
2218
2219        player.update_state()
2220        assert player.state.volume_level == 90
2221
2222    async def test_large_configured_step_clamps_up_at_the_maximum(
2223        self, mock_mass: MagicMock
2224    ) -> None:
2225        """A large configured step clamps volume_up at 100."""
2226        controller, player = self._make_player(mock_mass, 10, 95)
2227
2228        await controller.cmd_volume_up("player_1")
2229
2230        player.update_state()
2231        assert player.state.volume_level == 100
2232
2233    async def test_large_configured_step_clamps_down_at_zero(self, mock_mass: MagicMock) -> None:
2234        """A large configured step clamps volume_down at 0."""
2235        controller, player = self._make_player(mock_mass, 10, 5)
2236
2237        await controller.cmd_volume_down("player_1")
2238
2239        player.update_state()
2240        assert player.state.volume_level == 0
2241
2242    async def test_group_volume_up_with_default_step_uses_the_ladder(
2243        self, mock_mass: MagicMock
2244    ) -> None:
2245        """cmd_group_volume_up honours the default (0) adaptive ladder too."""
2246        controller, players = self._make_synced_pair(mock_mass, None)
2247
2248        await controller.cmd_group_volume_up("leader")
2249
2250        for player in players.values():
2251            player.update_state()
2252            assert player.state.volume_level == 53
2253
2254    async def test_group_volume_down_with_default_step_uses_the_ladder(
2255        self, mock_mass: MagicMock
2256    ) -> None:
2257        """cmd_group_volume_down honours the default (0) adaptive ladder too."""
2258        controller, players = self._make_synced_pair(mock_mass, None)
2259
2260        await controller.cmd_group_volume_down("leader")
2261
2262        for player in players.values():
2263            player.update_state()
2264            assert player.state.volume_level == 47
2265
2266    async def test_group_volume_up_with_configured_step(self, mock_mass: MagicMock) -> None:
2267        """cmd_group_volume_up honours a configured flat step."""
2268        controller, players = self._make_synced_pair(mock_mass, 5)
2269
2270        await controller.cmd_group_volume_up("leader")
2271
2272        for player in players.values():
2273            player.update_state()
2274            assert player.state.volume_level == 55
2275
2276    async def test_group_volume_down_with_configured_step(self, mock_mass: MagicMock) -> None:
2277        """cmd_group_volume_down honours a configured flat step."""
2278        controller, players = self._make_synced_pair(mock_mass, 5)
2279
2280        await controller.cmd_group_volume_down("leader")
2281
2282        for player in players.values():
2283            player.update_state()
2284            assert player.state.volume_level == 45
2285
2286    async def test_config_entry_exposes_default_and_range(
2287        self, controller: PlayerController
2288    ) -> None:
2289        """get_config_entries returns the volume_step entry with its default and range."""
2290        entries = await controller.get_config_entries()
2291
2292        entry = next(entry for entry in entries if entry.key == CONF_VOLUME_STEP)
2293        assert entry.type == ConfigEntryType.INTEGER
2294        assert entry.default_value == 0
2295        assert entry.range == (0, 10)
2296
2297
2298class TestFakeMuteInGroup:
2299    """A fake muted player in a group follows the mute lock, just like a native mute."""
2300
2301    def _make_synced_pair(
2302        self, mock_mass: MagicMock, *, member_mute_control: str = PLAYER_CONTROL_FAKE
2303    ) -> tuple[PlayerController, dict[str, MockPlayer]]:
2304        """
2305        Build a leader synced to one member.
2306
2307        :param member_mute_control: Mute control of the member, the leader always uses fake mute.
2308        """
2309
2310        def _conf(player_id: str, key: str, default: object = None) -> object:
2311            if key == CONF_MUTE_CONTROL and player_id == "member":
2312                return member_mute_control
2313            return _player_config_stub({CONF_MUTE_CONTROL: PLAYER_CONTROL_FAKE})(
2314                player_id, key, default
2315            )
2316
2317        mock_mass.config.get_raw_player_config_value = MagicMock(side_effect=_conf)
2318        controller = PlayerController(mock_mass)
2319        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
2320        players: dict[str, MockPlayer] = {}
2321        for player_id in ("leader", "member"):
2322            player = MockPlayer(provider, player_id, player_id.title())
2323            player._attr_supported_features = {
2324                PlayerFeature.VOLUME_SET,
2325                PlayerFeature.VOLUME_MUTE,
2326            }
2327            player._attr_volume_level = 50
2328            # let the mocked native volume control behave like a real device
2329            player.volume_set = AsyncMock(  # type: ignore[method-assign]
2330                side_effect=lambda volume, _player=player: setattr(
2331                    _player, "_attr_volume_level", volume
2332                )
2333            )
2334            players[player_id] = player
2335        players["leader"]._attr_group_members = ["member"]
2336        controller._players = dict(players)
2337        mock_mass.players = controller
2338        mock_mass.player_queues.get = MagicMock(return_value=None)
2339        for player in players.values():
2340            player.set_initialized()
2341            player._cache.clear()
2342            player.update_state(signal_event=False)
2343        # a second pass, so the group volume of the leader accounts for its member
2344        for player in players.values():
2345            player.update_state(signal_event=False)
2346        return controller, players
2347
2348    async def test_group_volume_keeps_a_muted_pair_muted(self, mock_mass: MagicMock) -> None:
2349        """A group volume change may not bring a muted fake mute pair back to life."""
2350        controller, players = self._make_synced_pair(mock_mass)
2351        await controller.cmd_group_volume_mute("leader", True)
2352
2353        await controller.cmd_group_volume("leader", 30)
2354
2355        for player in players.values():
2356            player.update_state()
2357            assert player.state.volume_muted is True
2358            assert player.state.volume_level == 0
2359
2360    async def test_group_volume_down_keeps_a_muted_member_muted(self, mock_mass: MagicMock) -> None:
2361        """Turning a group down leaves a single muted member silent, at its own volume."""
2362        controller, players = self._make_synced_pair(mock_mass)
2363        await controller.cmd_volume_mute("member", True)
2364        # let the group volume of the leader account for the muted member
2365        players["leader"].update_state(signal_event=False)
2366
2367        await controller.cmd_group_volume("leader", 25)
2368
2369        for player in players.values():
2370            player.update_state()
2371        member_state = players["member"].state
2372        assert member_state.volume_muted is True
2373        assert member_state.volume_level == 0
2374        # the player that is not muted follows the group volume as usual
2375        assert players["leader"].state.volume_level == 25
2376        # unmuting brings the member back at the volume it had before it was muted
2377        await controller.cmd_volume_mute("member", False)
2378        players["member"].update_state()
2379        assert players["member"].state.volume_level == 50
2380
2381    async def test_unmute_restores_the_volume_from_before_the_mute(
2382        self, mock_mass: MagicMock
2383    ) -> None:
2384        """A group volume change while muted may not alter the volume to restore."""
2385        controller, players = self._make_synced_pair(mock_mass)
2386        await controller.cmd_group_volume_mute("leader", True)
2387        await controller.cmd_group_volume("leader", 30)
2388
2389        await controller.cmd_group_volume_mute("leader", False)
2390
2391        for player in players.values():
2392            player.update_state()
2393            assert player.state.volume_muted is False
2394            assert player.state.volume_level == 50
2395
2396    async def test_group_volume_keeps_a_mixed_pair_muted(self, mock_mass: MagicMock) -> None:
2397        """Members with a different mute control stay muted alike on a group volume change."""
2398        controller, players = self._make_synced_pair(
2399            mock_mass, member_mute_control=PLAYER_CONTROL_NATIVE
2400        )
2401        mute = AsyncMock(
2402            side_effect=lambda muted: setattr(players["member"], "_attr_volume_muted", muted)
2403        )
2404        players["member"].volume_mute = mute  # type: ignore[method-assign]
2405        await controller.cmd_group_volume_mute("leader", True)
2406
2407        await controller.cmd_group_volume("leader", 30)
2408
2409        mute.assert_awaited_once_with(True)
2410        for player in players.values():
2411            player.update_state()
2412            assert player.state.volume_muted is True
2413
2414
2415class TestMuteControlGuard:
2416    """Muting is gated on the mute control, independently of the volume control."""
2417
2418    def _make_player(
2419        self,
2420        mock_mass: MagicMock,
2421        mute_control: str,
2422        volume_control: str,
2423        controls: dict[str, PlayerControl] | None = None,
2424    ) -> tuple[PlayerController, MockPlayer]:
2425        """Build a controller with a single player using the given control config."""
2426        mock_mass.config.get_raw_player_config_value = MagicMock(
2427            side_effect=_player_config_stub(
2428                {CONF_MUTE_CONTROL: mute_control, CONF_VOLUME_CONTROL: volume_control}
2429            )
2430        )
2431        controller = PlayerController(mock_mass)
2432        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
2433        player = MockPlayer(provider, "player_1", "Player 1")
2434        controller._players = {"player_1": player}
2435        controller._controls = controls or {}
2436        mock_mass.players = controller
2437        mock_mass.player_queues.get = MagicMock(return_value=None)
2438        player.set_initialized()
2439        player.update_state(signal_event=False)
2440        return controller, player
2441
2442    async def test_external_mute_control_without_volume_control(self, mock_mass: MagicMock) -> None:
2443        """A player without volume control still mutes through an external PlayerControl."""
2444        mute_set = AsyncMock()
2445        control = PlayerControl(
2446            id="ext_mute",
2447            provider="test",
2448            name="External Mute",
2449            supports_mute=True,
2450            mute_set=mute_set,
2451        )
2452        controller, player = self._make_player(
2453            mock_mass,
2454            mute_control="ext_mute",
2455            volume_control=PLAYER_CONTROL_NONE,
2456            controls={"ext_mute": control},
2457        )
2458        assert player.mute_control == "ext_mute"
2459
2460        await controller.cmd_volume_mute("player_1", True)
2461        mute_set.assert_awaited_once_with(True)
2462
2463    async def test_native_mute_control_without_volume_control(self, mock_mass: MagicMock) -> None:
2464        """A player without volume control still mutes natively."""
2465        controller, player = self._make_player(
2466            mock_mass,
2467            mute_control=PLAYER_CONTROL_NATIVE,
2468            volume_control=PLAYER_CONTROL_NONE,
2469        )
2470        volume_mute = AsyncMock()
2471        player.volume_mute = volume_mute  # type: ignore[method-assign]
2472
2473        await controller.cmd_volume_mute("player_1", True)
2474        volume_mute.assert_awaited_once_with(True)
2475
2476    async def test_mute_control_none_raises(self, mock_mass: MagicMock) -> None:
2477        """A player with volume control but no mute control rejects the command."""
2478        controller, player = self._make_player(
2479            mock_mass,
2480            mute_control=PLAYER_CONTROL_NONE,
2481            volume_control=PLAYER_CONTROL_NATIVE,
2482        )
2483        volume_mute = AsyncMock()
2484        player.volume_mute = volume_mute  # type: ignore[method-assign]
2485
2486        with pytest.raises(UnsupportedFeaturedException):
2487            await controller.cmd_volume_mute("player_1", True)
2488        volume_mute.assert_not_awaited()
2489
2490    async def test_fake_mute_without_volume_control_raises(self, mock_mass: MagicMock) -> None:
2491        """Fake mute needs a volume control to drive, so it rejects the command outright."""
2492        controller, player = self._make_player(
2493            mock_mass,
2494            mute_control=PLAYER_CONTROL_FAKE,
2495            volume_control=PLAYER_CONTROL_NONE,
2496        )
2497        player._attr_volume_level = 40
2498
2499        with pytest.raises(UnsupportedFeaturedException):
2500            await controller.cmd_volume_mute("player_1", True)
2501        assert ATTR_PREVIOUS_VOLUME not in player.extra_data
2502        assert ATTR_FAKE_MUTE not in player.extra_data
2503
2504    async def test_vanished_mute_control_raises(self, mock_mass: MagicMock) -> None:
2505        """A mute control that disappeared after being resolved is reported, not ignored."""
2506        control = PlayerControl(
2507            id="ext_mute",
2508            provider="test",
2509            name="External Mute",
2510            supports_mute=True,
2511            mute_set=AsyncMock(),
2512        )
2513        controller, player = self._make_player(
2514            mock_mass,
2515            mute_control="ext_mute",
2516            volume_control=PLAYER_CONTROL_NONE,
2517            controls={"ext_mute": control},
2518        )
2519        # the resolved control is cached on the player, so removing it here leaves
2520        # the player pointing at a control that no longer exists
2521        assert player.mute_control == "ext_mute"
2522        controller._controls = {}
2523
2524        with pytest.raises(UnsupportedFeaturedException):
2525            await controller.cmd_volume_mute("player_1", True)
2526
2527    async def test_unmute_clears_mute_lock_without_mute_control(self, mock_mass: MagicMock) -> None:
2528        """Unmuting clears a mute lock left behind by a since-removed mute control."""
2529        controller, player = self._make_player(
2530            mock_mass,
2531            mute_control=PLAYER_CONTROL_NONE,
2532            volume_control=PLAYER_CONTROL_NATIVE,
2533        )
2534        player.extra_data[ATTR_MUTE_LOCK] = True
2535
2536        with pytest.raises(UnsupportedFeaturedException):
2537            await controller.cmd_volume_mute("player_1", False)
2538        assert ATTR_MUTE_LOCK not in player.extra_data
2539
2540    async def test_failed_mute_sets_no_mute_lock(self, mock_mass: MagicMock) -> None:
2541        """A grouped player whose mute command failed is not left holding a mute lock."""
2542        control = PlayerControl(
2543            id="ext_mute",
2544            provider="test",
2545            name="External Mute",
2546            supports_mute=False,
2547        )
2548        controller, player = self._make_player(
2549            mock_mass,
2550            mute_control="ext_mute",
2551            volume_control=PLAYER_CONTROL_NONE,
2552            controls={"ext_mute": control},
2553        )
2554        player.state.synced_to = "leader"
2555
2556        with pytest.raises(UnsupportedFeaturedException):
2557            await controller.cmd_volume_mute("player_1", True)
2558        assert ATTR_MUTE_LOCK not in player.extra_data
2559
2560    async def test_failed_mute_keeps_existing_mute_lock(self, mock_mass: MagicMock) -> None:
2561        """A failed mute leaves the lock of an earlier successful mute in place."""
2562        control = PlayerControl(
2563            id="ext_mute",
2564            provider="test",
2565            name="External Mute",
2566            supports_mute=False,
2567        )
2568        controller, player = self._make_player(
2569            mock_mass,
2570            mute_control="ext_mute",
2571            volume_control=PLAYER_CONTROL_NONE,
2572            controls={"ext_mute": control},
2573        )
2574        player.state.synced_to = "leader"
2575        player.extra_data[ATTR_MUTE_LOCK] = True
2576
2577        with pytest.raises(UnsupportedFeaturedException):
2578            await controller.cmd_volume_mute("player_1", True)
2579        assert player.extra_data[ATTR_MUTE_LOCK] is True
2580
2581
2582class TestGroupMuteMemberFilter:
2583    """Group mute skips members that have no mute control of their own."""
2584
2585    async def test_member_without_mute_control_is_skipped(self, mock_mass: MagicMock) -> None:
2586        """A member without a mute control must not fail the whole group command."""
2587        controller = PlayerController(mock_mass)
2588        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
2589        leader = MockPlayer(provider, "leader", "Leader")
2590        leader._attr_supported_features = {PlayerFeature.VOLUME_SET, PlayerFeature.VOLUME_MUTE}
2591        leader._attr_group_members = ["leader", "member"]
2592        member = MockPlayer(provider, "member", "Member")
2593        member._attr_supported_features = {PlayerFeature.VOLUME_SET}
2594        controller._players = {"leader": leader, "member": member}
2595        mock_mass.players = controller
2596        mock_mass.player_queues.get = MagicMock(return_value=None)
2597        for player in (leader, member):
2598            player.set_initialized()
2599            player.update_state(signal_event=False)
2600        leader_mute = AsyncMock()
2601        leader.volume_mute = leader_mute  # type: ignore[method-assign]
2602
2603        await controller.cmd_group_volume_mute("leader", True)
2604        leader_mute.assert_awaited_once_with(True)
2605
2606
2607class TestGroupPlayerMuteRedirect:
2608    """A mute command on a group player is handled at group level."""
2609
2610    def _setup(self, mock_mass: MagicMock) -> tuple[PlayerController, MockPlayer, MockPlayer]:
2611        """Build a controller with a group player holding a single mute capable member."""
2612        controller = PlayerController(mock_mass)
2613        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
2614        group = MockPlayer(provider, "group", "Group", player_type=PlayerType.GROUP)
2615        group._attr_supported_features = {PlayerFeature.VOLUME_SET, PlayerFeature.VOLUME_MUTE}
2616        group._attr_group_members = ["member"]
2617        member = MockPlayer(provider, "member", "Member")
2618        member._attr_supported_features = {PlayerFeature.VOLUME_SET, PlayerFeature.VOLUME_MUTE}
2619        controller._players = {"group": group, "member": member}
2620        mock_mass.players = controller
2621        mock_mass.player_queues.get = MagicMock(return_value=None)
2622        for player in (group, member):
2623            player.set_initialized()
2624            player.update_state(signal_event=False)
2625        return controller, group, member
2626
2627    async def test_mute_on_group_player_is_forwarded_to_members(self, mock_mass: MagicMock) -> None:
2628        """A group player has no mute of its own, so the members must be muted instead."""
2629        controller, _group, member = self._setup(mock_mass)
2630        member_mute = AsyncMock()
2631        member.volume_mute = member_mute  # type: ignore[method-assign]
2632
2633        await controller.cmd_volume_mute("group", True)
2634
2635        member_mute.assert_awaited_once_with(True)
2636
2637    async def test_mute_on_group_player_without_own_mute_control(
2638        self, mock_mass: MagicMock
2639    ) -> None:
2640        """A group that has no mute control of its own must still mute its members."""
2641        controller, group, member = self._setup(mock_mass)
2642        group._attr_supported_features = {PlayerFeature.VOLUME_SET}
2643        group._cache.clear()
2644        group.update_state(signal_event=False)
2645        assert group.mute_control == PLAYER_CONTROL_NONE
2646        member_mute = AsyncMock()
2647        member.volume_mute = member_mute  # type: ignore[method-assign]
2648
2649        await controller.cmd_volume_mute("group", True)
2650
2651        member_mute.assert_awaited_once_with(True)
2652
2653    async def test_mute_on_group_player_without_mute_capable_members(
2654        self, mock_mass: MagicMock
2655    ) -> None:
2656        """A group whose members cannot mute must not raise, just like group mute itself."""
2657        controller, _group, member = self._setup(mock_mass)
2658        member._attr_supported_features = {PlayerFeature.VOLUME_SET}
2659        member._cache.clear()
2660        member.update_state(signal_event=False)
2661
2662        await controller.cmd_volume_mute("group", True)
2663
2664
2665class TestGroupMuteOnNonGroupPlayer:
2666    """A group mute command works on any player, just like the group volume command."""
2667
2668    def _setup(
2669        self, mock_mass: MagicMock, *members: str
2670    ) -> tuple[PlayerController, dict[str, MockPlayer]]:
2671        """Build a controller with a mute capable leader synced to the given members."""
2672        controller = PlayerController(mock_mass)
2673        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
2674        players: dict[str, MockPlayer] = {}
2675        for player_id in ("leader", *members):
2676            player = MockPlayer(provider, player_id, player_id.title())
2677            player._attr_supported_features = {
2678                PlayerFeature.VOLUME_SET,
2679                PlayerFeature.VOLUME_MUTE,
2680            }
2681            player._attr_volume_level = 50
2682            players[player_id] = player
2683        if members:
2684            # the leader is not listed as its own member here, so the tests also cover
2685            # that a sync leader is injected into its own final group_members
2686            players["leader"]._attr_group_members = list(members)
2687        controller._players = dict(players)
2688        mock_mass.players = controller
2689        mock_mass.player_queues.get = MagicMock(return_value=None)
2690        for player in players.values():
2691            player.set_initialized()
2692            player._cache.clear()
2693            player.update_state(signal_event=False)
2694        return controller, players
2695
2696    def _stub_mutes(self, players: dict[str, MockPlayer]) -> dict[str, AsyncMock]:
2697        """Replace the native mute command of every given player with a mock."""
2698        mutes: dict[str, AsyncMock] = {}
2699        for player_id, player in players.items():
2700            mutes[player_id] = AsyncMock()
2701            player.volume_mute = mutes[player_id]  # type: ignore[method-assign]
2702        return mutes
2703
2704    async def test_group_mute_on_synced_member_redirects_to_leader(
2705        self, mock_mass: MagicMock
2706    ) -> None:
2707        """A member of a sync group mutes the whole group through its sync leader."""
2708        controller, players = self._setup(mock_mass, "member")
2709        assert players["member"].state.synced_to == "leader"
2710        mutes = self._stub_mutes(players)
2711
2712        await controller.cmd_group_volume_mute("member", True)
2713
2714        mutes["leader"].assert_awaited_once_with(True)
2715        mutes["member"].assert_awaited_once_with(True)
2716        assert ATTR_MUTE_LOCK in players["member"].extra_data
2717
2718    async def test_group_mute_on_sync_leader_mutes_the_leader_once(
2719        self, mock_mass: MagicMock
2720    ) -> None:
2721        """A sync leader is part of its own member list, so it must be muted only once."""
2722        controller, players = self._setup(mock_mass, "member")
2723        mutes = self._stub_mutes(players)
2724
2725        await controller.cmd_group_volume_mute("leader", True)
2726
2727        mutes["leader"].assert_awaited_once_with(True)
2728        mutes["member"].assert_awaited_once_with(True)
2729
2730    async def test_group_mute_on_plain_player_mutes_that_player(self, mock_mass: MagicMock) -> None:
2731        """A player that is not grouped at all is muted as a normal player."""
2732        controller, players = self._setup(mock_mass)
2733        mutes = self._stub_mutes(players)
2734
2735        await controller.cmd_group_volume_mute("leader", True)
2736
2737        mutes["leader"].assert_awaited_once_with(True)
2738
2739    async def test_group_mute_on_plain_player_without_mute_control(
2740        self, mock_mass: MagicMock
2741    ) -> None:
2742        """A plain player that cannot mute reports that, just like a normal mute command."""
2743        controller, players = self._setup(mock_mass)
2744        players["leader"]._attr_supported_features = {PlayerFeature.VOLUME_SET}
2745        players["leader"]._cache.clear()
2746        players["leader"].update_state(signal_event=False)
2747
2748        with pytest.raises(UnsupportedFeaturedException):
2749            await controller.cmd_group_volume_mute("leader", True)
2750
2751    async def test_group_unmute_on_synced_member_redirects_to_leader(
2752        self, mock_mass: MagicMock
2753    ) -> None:
2754        """Unmuting through a member clears the mute (and mute lock) of every group member."""
2755        controller, players = self._setup(mock_mass, "member")
2756        mutes = self._stub_mutes(players)
2757        players["member"].extra_data[ATTR_MUTE_LOCK] = True
2758
2759        await controller.cmd_group_volume_mute("member", False)
2760
2761        mutes["leader"].assert_awaited_once_with(False)
2762        mutes["member"].assert_awaited_once_with(False)
2763        assert ATTR_MUTE_LOCK not in players["member"].extra_data
2764
2765    async def test_group_mute_locks_the_sync_leader_too(self, mock_mass: MagicMock) -> None:
2766        """A sync leader is as much part of the group as its members, so it is locked too."""
2767        controller, players = self._setup(mock_mass, "member")
2768        self._stub_mutes(players)
2769
2770        await controller.cmd_group_volume_mute("leader", True)
2771
2772        assert ATTR_MUTE_LOCK in players["leader"].extra_data
2773        assert ATTR_MUTE_LOCK in players["member"].extra_data
2774
2775    async def test_group_volume_keeps_a_muted_sync_pair_muted(self, mock_mass: MagicMock) -> None:
2776        """A group volume change may not half-unmute a muted pair of directly synced players."""
2777        controller, players = self._setup(mock_mass, "member")
2778        self._stub_mutes(players)
2779        await controller.cmd_group_volume_mute("leader", True)
2780        # the mock players do not act on the mute command, so reflect it in their state
2781        for player in players.values():
2782            player._attr_volume_muted = True
2783            player.update_state(signal_event=False)
2784            player.volume_set = AsyncMock()  # type: ignore[method-assign]
2785        # re-stub so only the mute commands of the group volume change are counted
2786        mutes = self._stub_mutes(players)
2787
2788        await controller.cmd_group_volume("leader", 30)
2789
2790        for mute in mutes.values():
2791            mute.assert_not_awaited()
2792
2793
2794class TestMuteLockAfterUngroup:
2795    """A mute lock is only honored while the player it belongs to is still grouped."""
2796
2797    def _make_synced_pair(
2798        self, mock_mass: MagicMock, member_mute_control: str
2799    ) -> tuple[PlayerController, dict[str, MockPlayer]]:
2800        """
2801        Build a leader with one synced member.
2802
2803        :param member_mute_control: Mute control to configure on both players.
2804        """
2805        mock_mass.config.get_raw_player_config_value = MagicMock(
2806            side_effect=_player_config_stub({CONF_MUTE_CONTROL: member_mute_control})
2807        )
2808        controller = PlayerController(mock_mass)
2809        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
2810        players: dict[str, MockPlayer] = {}
2811        for player_id in ("leader", "member"):
2812            player = MockPlayer(provider, player_id, player_id.title())
2813            player._attr_supported_features = {
2814                PlayerFeature.VOLUME_SET,
2815                PlayerFeature.VOLUME_MUTE,
2816            }
2817            player._attr_volume_level = 50
2818            player.volume_set = AsyncMock(  # type: ignore[method-assign]
2819                side_effect=lambda volume, _player=player: setattr(
2820                    _player, "_attr_volume_level", volume
2821                )
2822            )
2823            players[player_id] = player
2824        players["leader"]._attr_group_members = ["member"]
2825        controller._players = dict(players)
2826        mock_mass.players = controller
2827        mock_mass.player_queues.get = MagicMock(return_value=None)
2828        for player in players.values():
2829            player.set_initialized()
2830            player._cache.clear()
2831            player.update_state(signal_event=False)
2832        return controller, players
2833
2834    def _dissolve_group(self, players: dict[str, MockPlayer]) -> None:
2835        """Drop the sync group, the way a provider side topology change does."""
2836        players["leader"]._attr_group_members = []
2837        for player in players.values():
2838            player.refresh_state(signal_event=False)
2839
2840    async def test_fake_muted_player_follows_volume_again(self, mock_mass: MagicMock) -> None:
2841        """A fake muted player is no longer forced silent once its group is gone."""
2842        controller, players = self._make_synced_pair(mock_mass, PLAYER_CONTROL_FAKE)
2843        await controller.cmd_volume_mute("member", True)
2844        self._dissolve_group(players)
2845
2846        await controller.cmd_volume_set("member", 70)
2847
2848        players["member"].update_state()
2849        assert players["member"].state.volume_level == 70
2850        assert players["member"].state.volume_muted is False
2851
2852    async def test_natively_muted_player_is_unmuted_again(self, mock_mass: MagicMock) -> None:
2853        """A natively muted player is auto-unmuted by a volume change once its group is gone."""
2854        controller, players = self._make_synced_pair(mock_mass, PLAYER_CONTROL_NATIVE)
2855        mute = AsyncMock(
2856            side_effect=lambda muted: setattr(players["member"], "_attr_volume_muted", muted)
2857        )
2858        players["member"].volume_mute = mute  # type: ignore[method-assign]
2859        await controller.cmd_volume_mute("member", True)
2860        self._dissolve_group(players)
2861
2862        await controller.cmd_volume_set("member", 70)
2863
2864        assert mute.await_args_list == [call(True), call(False)]
2865        players["member"].update_state()
2866        assert players["member"].state.volume_level == 70
2867        assert players["member"].state.volume_muted is False
2868
2869    async def test_still_grouped_player_keeps_its_lock(self, mock_mass: MagicMock) -> None:
2870        """A muted player that is still grouped stays silent on a volume change."""
2871        controller, players = self._make_synced_pair(mock_mass, PLAYER_CONTROL_FAKE)
2872        await controller.cmd_volume_mute("member", True)
2873
2874        await controller.cmd_volume_set("member", 70)
2875
2876        players["member"].update_state()
2877        assert players["member"].state.volume_level == 0
2878        assert players["member"].state.volume_muted is True
2879
2880    async def test_protocol_player_follows_the_lock_of_its_parent(
2881        self, mock_mass: MagicMock
2882    ) -> None:
2883        """A protocol player inherits the lock of the parent it renders for, group and all."""
2884        controller, players = self._make_synced_pair(mock_mass, PLAYER_CONTROL_NATIVE)
2885        member = players["member"]
2886        protocol_player = MockPlayer(
2887            MockProvider("sendspin", instance_id="sendspin", mass=mock_mass),
2888            "proto_member",
2889            "Member Protocol",
2890            player_type=PlayerType.PROTOCOL,
2891        )
2892        protocol_player._attr_supported_features = {
2893            PlayerFeature.VOLUME_SET,
2894            PlayerFeature.VOLUME_MUTE,
2895        }
2896        protocol_player._attr_volume_muted = True
2897        protocol_player.set_protocol_parent_id("member")
2898        # an unmute on a protocol player is redirected to the parent it renders for
2899        mute = AsyncMock()
2900        member.volume_mute = mute  # type: ignore[method-assign]
2901        protocol_player.volume_mute = AsyncMock()  # type: ignore[method-assign]
2902        protocol_player.volume_set = AsyncMock()  # type: ignore[method-assign]
2903        controller._players["proto_member"] = protocol_player
2904        member.set_linked_output_protocols(
2905            [
2906                LinkedOutputProtocol(
2907                    output_protocol_id="proto_member",
2908                    protocol_domain="sendspin",
2909                    priority=40,
2910                )
2911            ]
2912        )
2913        protocol_player.set_initialized()
2914        protocol_player.update_state(signal_event=False)
2915        member.refresh_state(signal_event=False)
2916        member.extra_data[ATTR_MUTE_LOCK] = True
2917
2918        # a group volume change reaches the protocol player through the internal handler
2919        await controller._handle_cmd_volume_set("proto_member", 70)
2920        mute.assert_not_awaited()
2921
2922        self._dissolve_group(players)
2923        await controller._handle_cmd_volume_set("proto_member", 70)
2924
2925        mute.assert_awaited_once_with(False)
2926
2927
2928class TestCurrentMediaTimeUpdates:
2929    """Playback-position anchor semantics of timing-only state updates."""
2930
2931    def _make_player(self, mock_mass: MagicMock) -> tuple[PlayerController, MockPlayer]:
2932        """Build a controller with a single playing player with a known position anchor."""
2933        controller = PlayerController(mock_mass)
2934        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
2935        player = MockPlayer(provider, "player_1", "Player 1")
2936        controller._players = {"player_1": player}
2937        mock_mass.players = controller
2938        # no queue registered: current_media resolves from the player's native media
2939        mock_mass.player_queues.get = MagicMock(return_value=None)
2940        player.set_initialized()
2941        now = time.time()
2942        player._attr_playback_state = PlaybackState.PLAYING
2943        player._attr_elapsed_time = 17
2944        player._attr_elapsed_time_last_updated = now
2945        player.set_current_media(uri="http://test/stream", title="Test")
2946        assert player._attr_current_media is not None
2947        player._attr_current_media.elapsed_time = 17
2948        player._attr_current_media.elapsed_time_last_updated = now
2949        player.update_state(signal_event=False)
2950        # isolate from the unrelated state-forwarding machinery
2951        controller._forward_state_update = MagicMock()  # type: ignore[method-assign]
2952        mock_mass.signal_event.reset_mock()
2953        mock_mass.player_queues.on_player_elapsed_time_corrected.reset_mock()
2954        return controller, player
2955
2956    def _player_updated_signalled(self, mock_mass: MagicMock) -> bool:
2957        """Return whether a PLAYER_UPDATED event was signalled."""
2958        return any(
2959            call.args and call.args[0] == EventType.PLAYER_UPDATED
2960            for call in mock_mass.signal_event.call_args_list
2961        )
2962
2963    def test_regular_tick_is_suppressed(self, mock_mass: MagicMock) -> None:
2964        """A regular playback tick (position and anchor advance together) emits nothing."""
2965        _controller, player = self._make_player(mock_mass)
2966        assert player._attr_current_media is not None
2967        assert player._attr_elapsed_time_last_updated is not None
2968
2969        player._attr_elapsed_time = 18
2970        player._attr_elapsed_time_last_updated += 1
2971        player._attr_current_media.elapsed_time = 18
2972        assert player._attr_current_media.elapsed_time_last_updated is not None
2973        player._attr_current_media.elapsed_time_last_updated += 1
2974        player.update_state()
2975
2976        assert not self._player_updated_signalled(mock_mass)
2977        mock_mass.player_queues.on_player_elapsed_time_corrected.assert_not_called()
2978        # the previous anchor was preserved: steady playback changes nothing
2979        assert player.state.elapsed_time == 17
2980
2981    def test_anchor_only_change_is_suppressed(self, mock_mass: MagicMock) -> None:
2982        """An anchor-only change (no significant corrected position change) emits nothing."""
2983        _controller, player = self._make_player(mock_mass)
2984        assert player._attr_current_media is not None
2985        assert player._attr_elapsed_time_last_updated is not None
2986
2987        player._attr_elapsed_time_last_updated += 0.5
2988        assert player._attr_current_media.elapsed_time_last_updated is not None
2989        player._attr_current_media.elapsed_time_last_updated += 0.5
2990        player.update_state()
2991
2992        assert not self._player_updated_signalled(mock_mass)
2993        mock_mass.player_queues.on_player_elapsed_time_corrected.assert_not_called()
2994
2995    def test_corrected_position_jump_emits_player_updated(self, mock_mass: MagicMock) -> None:
2996        """A corrected-position jump of the current media (e.g. seek) emits a player update."""
2997        _controller, player = self._make_player(mock_mass)
2998        assert player._attr_current_media is not None
2999
3000        player._attr_current_media.elapsed_time = 61
3001        player._attr_current_media.elapsed_time_last_updated = time.time()
3002        player.update_state()
3003
3004        assert self._player_updated_signalled(mock_mass)
3005        # the adopted anchor is visible to consumers
3006        assert player.state.current_media is not None
3007        assert player.state.current_media.elapsed_time == 61
3008
3009    def test_player_position_jump_corrects_queue(self, mock_mass: MagicMock) -> None:
3010        """A player-level corrected-position jump re-bases the queue timing."""
3011        controller, player = self._make_player(mock_mass)
3012
3013        player._attr_elapsed_time = 61
3014        player._attr_elapsed_time_last_updated = time.time()
3015        player.update_state()
3016
3017        # the queue is corrected and a follow-up player update is scheduled
3018        # (which re-anchors current_media onto the corrected queue time),
3019        # but no full player update is emitted for the jump itself
3020        mock_mass.player_queues.on_player_elapsed_time_corrected.assert_called_once_with(player)
3021        assert not self._player_updated_signalled(mock_mass)
3022        cast("MagicMock", controller._forward_state_update).assert_called_once()
3023        assert player.state.elapsed_time == 61
3024
3025    def test_simultaneous_player_and_media_jump_emits_immediately(
3026        self, mock_mass: MagicMock
3027    ) -> None:
3028        """A jump reaching player and current_media in one pass corrects the queue and emits."""
3029        _controller, player = self._make_player(mock_mass)
3030        assert player._attr_current_media is not None
3031        now = time.time()
3032
3033        player._attr_elapsed_time = 61
3034        player._attr_elapsed_time_last_updated = now
3035        player._attr_current_media.elapsed_time = 61
3036        player._attr_current_media.elapsed_time_last_updated = now
3037        player.update_state()
3038
3039        # the queue is re-based AND the full update is emitted right away
3040        # (current_media already holds the fresh position in the same pass)
3041        mock_mass.player_queues.on_player_elapsed_time_corrected.assert_called_once_with(player)
3042        assert self._player_updated_signalled(mock_mass)
3043
3044
3045class TestPlayAnnouncementCleanup:
3046    """Test announcement data cleanup after play_announcement."""
3047
3048    def _make_player(
3049        self, mock_mass: MagicMock, announcements: dict[str, object]
3050    ) -> tuple[PlayerController, MockPlayer, MagicMock]:
3051        """Create a controller and a player with native announcement support."""
3052        controller = PlayerController(mock_mass)
3053        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
3054        player = MockPlayer(provider, "player_1", "Player 1")
3055        player._attr_supported_features.add(PlayerFeature.PLAY_ANNOUNCEMENT)
3056        player._cache.clear()
3057        controller._players = {"player_1": player}
3058        mock_mass.players = controller
3059        render = MagicMock()
3060        render.wait_ready = AsyncMock(return_value=True)
3061        render.wait_finished = AsyncMock(return_value=3.0)
3062
3063        # mimic the real renderer: it owns which announcement each player is playing
3064        def _register(player_id: str, announce_data: object) -> MagicMock:
3065            announcements[player_id] = announce_data
3066            return render
3067
3068        async def _unregister(player_id: str, _render: object) -> None:
3069            announcements.pop(player_id, None)
3070
3071        renderer = mock_mass.streams.announcement_renderer
3072        renderer.register = MagicMock(side_effect=_register)
3073        renderer.unregister = AsyncMock(side_effect=_unregister)
3074        mock_mass.streams.get_announcement_url = MagicMock(
3075            side_effect=lambda player_id, **_kwargs: f"http://ma/announcement/{player_id}.mp3"
3076        )
3077        player.update_state(signal_event=False)
3078        return controller, player, render
3079
3080    async def test_announcement_data_removed_after_playback(self, mock_mass: MagicMock) -> None:
3081        """The registered announcement data is released once playback finished."""
3082        announcements: dict[str, object] = {}
3083        controller, player, _render = self._make_player(mock_mass, announcements)
3084
3085        async def _play_announcement(*_args: object, **_kwargs: object) -> None:
3086            # entry must exist while the announcement is being played/served
3087            assert "player_1" in announcements
3088
3089        player.play_announcement = AsyncMock(side_effect=_play_announcement)  # type: ignore[method-assign]
3090
3091        await controller.play_announcement("player_1", "http://test/announcement.mp3")
3092
3093        player.play_announcement.assert_awaited_once()
3094        assert announcements == {}
3095        mock_mass.streams.announcement_renderer.unregister.assert_awaited_once()
3096
3097    async def test_announcement_data_removed_on_error(self, mock_mass: MagicMock) -> None:
3098        """The registered announcement data is released even when playback fails."""
3099        announcements: dict[str, object] = {}
3100        controller, player, _render = self._make_player(mock_mass, announcements)
3101        player.play_announcement = AsyncMock(side_effect=RuntimeError("boom"))  # type: ignore[method-assign]
3102
3103        with pytest.raises(PlayerCommandFailed):
3104            await controller.play_announcement("player_1", "http://test/announcement.mp3")
3105
3106        assert announcements == {}
3107        mock_mass.streams.announcement_renderer.unregister.assert_awaited_once()
3108
3109    async def test_native_announcement_starts_on_first_audio(self, mock_mass: MagicMock) -> None:
3110        """A native implementation is handed the url as soon as there is audio to serve."""
3111        announcements: dict[str, object] = {}
3112        controller, player, render = self._make_player(mock_mass, announcements)
3113        player.play_announcement = AsyncMock()  # type: ignore[method-assign]
3114
3115        await controller.play_announcement("player_1", "http://test/announcement.mp3")
3116
3117        player.play_announcement.assert_awaited_once()
3118        # waiting for the whole clip here would delay the player for a slow source;
3119        # the length is resolved downstream while it plays
3120        render.wait_ready.assert_awaited_once()
3121        render.wait_finished.assert_not_awaited()
3122
3123
3124class _AnnounceSetup(NamedTuple):
3125    """A player announcing through a linked protocol output, with the calls it makes mocked."""
3126
3127    controller: PlayerController
3128    parent: MockPlayer
3129    output: MockPlayer
3130    volume_set: AsyncMock
3131    play_announcement: AsyncMock
3132
3133
3134@pytest.mark.usefixtures("running_background_tasks")
3135class TestNativeAnnouncementVolumeRouting:
3136    """The announcement volume is applied through the control that owns it."""
3137
3138    ANNOUNCE_VOLUME = 45
3139
3140    def _make_setup(self, mock_mass: MagicMock, volume_control: str) -> _AnnounceSetup:
3141        """
3142        Create a player announcing through a linked protocol output.
3143
3144        The parent holds a sibling interface and a bridge riding on the announcing
3145        output, so any of them can be named as its volume control.
3146
3147        :param volume_control: Value of the parent's volume control config entry.
3148        """
3149        controller = PlayerController(mock_mass)
3150        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
3151        parent = MockPlayer(provider, "parent", "Parent")
3152        output = MockPlayer(provider, "output", "Output")
3153        output._attr_supported_features.add(PlayerFeature.PLAY_ANNOUNCEMENT)
3154        output._attr_supported_features.add(PlayerFeature.VOLUME_SET)
3155        sibling = MockPlayer(provider, "sibling", "Sibling")
3156        sibling._attr_supported_features.add(PlayerFeature.VOLUME_SET)
3157        sibling._attr_volume_level = 20
3158        bridge = MockPlayer(provider, "bridge", "Bridge")
3159        bridge._attr_supported_features.add(PlayerFeature.VOLUME_SET)
3160        bridge._attr_underlying_player_id = "output"
3161        bridge._attr_volume_level = 20
3162        controller._players = {
3163            "parent": parent,
3164            "output": output,
3165            "sibling": sibling,
3166            "bridge": bridge,
3167        }
3168        # an external control (e.g. a Home Assistant volume entity) is not a player
3169        controller._controls = {
3170            "ha_volume": PlayerControl(
3171                id="ha_volume",
3172                provider="hass",
3173                name="Amplifier volume",
3174                supports_volume=True,
3175                volume_level=20,
3176            )
3177        }
3178        mock_mass.players = controller
3179        mock_mass.config.get_raw_player_config_value = MagicMock(
3180            side_effect=_player_config_stub({CONF_VOLUME_CONTROL: volume_control})
3181        )
3182        for player in controller._players.values():
3183            player._cache.clear()
3184            player.update_state(signal_event=False)
3185        play_announcement = AsyncMock()
3186        volume_set = AsyncMock()
3187        output.play_announcement = play_announcement  # type: ignore[method-assign]
3188        controller._handle_cmd_volume_set = volume_set  # type: ignore[method-assign]
3189        return _AnnounceSetup(controller, parent, output, volume_set, play_announcement)
3190
3191    async def _announce(self, setup: _AnnounceSetup) -> None:
3192        """Play an announcement on the parent, rendered by the linked output."""
3193        await setup.controller._play_native_announcement(
3194            setup.parent, setup.output, _announcement(), self.ANNOUNCE_VOLUME
3195        )
3196
3197    async def test_external_control_applies_the_announcement_volume(
3198        self, mock_mass: MagicMock
3199    ) -> None:
3200        """A sibling interface owning the volume gets the announcement volume, not the output."""
3201        setup = self._make_setup(mock_mass, "sibling")
3202
3203        await self._announce(setup)
3204
3205        # the output cannot attenuate what another control is already attenuating,
3206        # so the level goes through that control and the output announces at unity
3207        assert setup.volume_set.await_args_list == [
3208            call("parent", self.ANNOUNCE_VOLUME),
3209            call("parent", 20),
3210        ]
3211        setup.play_announcement.assert_awaited_once_with(ANY, None)
3212
3213    async def test_player_control_applies_the_announcement_volume(
3214        self, mock_mass: MagicMock
3215    ) -> None:
3216        """A player control owning the volume (e.g. an HA entity) gets the announcement volume."""
3217        setup = self._make_setup(mock_mass, "ha_volume")
3218
3219        await self._announce(setup)
3220
3221        assert setup.volume_set.await_args_list == [
3222            call("parent", self.ANNOUNCE_VOLUME),
3223            call("parent", 20),
3224        ]
3225        setup.play_announcement.assert_awaited_once_with(ANY, None)
3226
3227    async def test_external_control_volume_restored_when_the_announcement_fails(
3228        self, mock_mass: MagicMock
3229    ) -> None:
3230        """The temporary volume is restored even when the announcement itself fails."""
3231        setup = self._make_setup(mock_mass, "sibling")
3232        setup.play_announcement.side_effect = RuntimeError("boom")
3233
3234        with pytest.raises(RuntimeError):
3235            await self._announce(setup)
3236
3237        assert setup.volume_set.await_args_list[-1] == call("parent", 20)
3238
3239    async def test_announcing_output_keeps_the_announcement_volume(
3240        self, mock_mass: MagicMock
3241    ) -> None:
3242        """An output that owns the volume applies the announcement volume itself."""
3243        setup = self._make_setup(mock_mass, "output")
3244
3245        await self._announce(setup)
3246
3247        setup.volume_set.assert_not_awaited()
3248        setup.play_announcement.assert_awaited_once_with(ANY, self.ANNOUNCE_VOLUME)
3249
3250    async def test_bridge_on_the_announcing_output_keeps_the_announcement_volume(
3251        self, mock_mass: MagicMock
3252    ) -> None:
3253        """A bridge riding on the announcing output forwards the volume to it."""
3254        setup = self._make_setup(mock_mass, "bridge")
3255
3256        await self._announce(setup)
3257
3258        setup.volume_set.assert_not_awaited()
3259        setup.play_announcement.assert_awaited_once_with(ANY, self.ANNOUNCE_VOLUME)
3260
3261    async def test_native_volume_keeps_the_announcement_volume(self, mock_mass: MagicMock) -> None:
3262        """Native parent volume lives on the device the output talks to, so it applies it."""
3263        setup = self._make_setup(mock_mass, PLAYER_CONTROL_NATIVE)
3264
3265        await self._announce(setup)
3266
3267        setup.volume_set.assert_not_awaited()
3268        setup.play_announcement.assert_awaited_once_with(ANY, self.ANNOUNCE_VOLUME)
3269
3270    async def test_without_volume_control_no_volume_is_applied(self, mock_mass: MagicMock) -> None:
3271        """Nothing in the signal path can set a volume, so the announcement plays as-is."""
3272        setup = self._make_setup(mock_mass, PLAYER_CONTROL_NONE)
3273
3274        await self._announce(setup)
3275
3276        setup.volume_set.assert_not_awaited()
3277        setup.play_announcement.assert_awaited_once_with(ANY, None)
3278
3279
3280class TestPlayAnnouncementMessage:
3281    """A spoken message is rendered by a TTS engine and then announced like any other audio."""
3282
3283    ANNOUNCE_MODULE = "music_assistant.controllers.players.announcements"
3284
3285    def _make_engine(self, path: str = "http://speech/spoken.mp3") -> MagicMock:
3286        """Create a TTS engine that renders every message to the given path."""
3287        engine = MagicMock()
3288        engine.uid = "tts_plugin/voice"
3289        engine.id = "voice"
3290        engine.provider.get_tts_message = AsyncMock(return_value=SimpleNamespace(path=path))
3291        return engine
3292
3293    def _make_player(
3294        self, mock_mass: MagicMock, announcements: dict[str, object]
3295    ) -> tuple[PlayerController, AsyncMock]:
3296        """Create a controller and a player with native announcement support."""
3297        controller, player, _render = TestPlayAnnouncementCleanup()._make_player(
3298            mock_mass, announcements
3299        )
3300        announce = AsyncMock()
3301        player.play_announcement = announce  # type: ignore[method-assign]
3302        return controller, announce
3303
3304    async def test_message_is_spoken_by_the_configured_engine(self, mock_mass: MagicMock) -> None:
3305        """A message is rendered by the default engine and announced as the rendered audio."""
3306        announcements: dict[str, object] = {}
3307        controller, announce = self._make_player(mock_mass, announcements)
3308        engine = self._make_engine()
3309
3310        with patch(
3311            f"{self.ANNOUNCE_MODULE}.select_core_tts_engine", AsyncMock(return_value=engine)
3312        ):
3313            await controller.play_announcement("player_1", message="dinner is ready")
3314
3315        # no language is sent, so the engine speaks in the language it is configured for
3316        engine.provider.get_tts_message.assert_awaited_once_with(
3317            "dinner is ready", language=None, engine_id="voice", options=None
3318        )
3319        registered = mock_mass.streams.announcement_renderer.register.call_args.args[1]
3320        assert registered["announcement_url"] == "http://speech/spoken.mp3"
3321        announce.assert_awaited_once()
3322
3323    async def test_an_explicit_language_reaches_the_engine(self, mock_mass: MagicMock) -> None:
3324        """A message can name the language to speak it in."""
3325        announcements: dict[str, object] = {}
3326        controller, _announce = self._make_player(mock_mass, announcements)
3327        engine = self._make_engine()
3328
3329        with patch(
3330            f"{self.ANNOUNCE_MODULE}.select_core_tts_engine", AsyncMock(return_value=engine)
3331        ):
3332            await controller.play_announcement(
3333                "player_1", message="het eten is klaar", language="nl-NL"
3334            )
3335
3336        engine.provider.get_tts_message.assert_awaited_once_with(
3337            "het eten is klaar", language="nl-NL", engine_id="voice", options=None
3338        )
3339
3340    async def test_a_rejected_language_is_retried_without_it(self, mock_mass: MagicMock) -> None:
3341        """An engine that rejects the language speaks the message in its default voice."""
3342        announcements: dict[str, object] = {}
3343        controller, announce = self._make_player(mock_mass, announcements)
3344        engine = self._make_engine()
3345        engine.provider.get_tts_message = AsyncMock(
3346            side_effect=[
3347                RuntimeError("unsupported language"),
3348                SimpleNamespace(path="http://speech/spoken.mp3"),
3349            ]
3350        )
3351
3352        with patch(
3353            f"{self.ANNOUNCE_MODULE}.select_core_tts_engine", AsyncMock(return_value=engine)
3354        ):
3355            await controller.play_announcement(
3356                "player_1", message="dinner is ready", language="en-US"
3357            )
3358
3359        first_call, second_call = engine.provider.get_tts_message.await_args_list
3360        assert first_call.kwargs["language"] == "en-US"
3361        assert second_call.kwargs["language"] is None
3362        registered = mock_mass.streams.announcement_renderer.register.call_args.args[1]
3363        assert registered["announcement_url"] == "http://speech/spoken.mp3"
3364        announce.assert_awaited_once()
3365
3366    @pytest.mark.parametrize(
3367        "error", [TimeoutError(), MusicAssistantError("engine did not respond within 30s")]
3368    )
3369    async def test_a_failure_that_is_not_a_language_rejection_is_not_retried(
3370        self, mock_mass: MagicMock, error: Exception
3371    ) -> None:
3372        """A timeout or a structured failure is no language rejection, so it is not retried."""
3373        announcements: dict[str, object] = {}
3374        controller, announce = self._make_player(mock_mass, announcements)
3375        engine = self._make_engine()
3376        engine.provider.get_tts_message = AsyncMock(side_effect=error)
3377
3378        with (
3379            patch(f"{self.ANNOUNCE_MODULE}.select_core_tts_engine", AsyncMock(return_value=engine)),
3380            pytest.raises(MusicAssistantError),
3381        ):
3382            await controller.play_announcement("player_1", message="dinner is ready")
3383
3384        engine.provider.get_tts_message.assert_awaited_once()
3385        announce.assert_not_awaited()
3386
3387    async def test_an_explicit_engine_is_used(self, mock_mass: MagicMock) -> None:
3388        """A message names the engine to speak it, overriding the configured default."""
3389        announcements: dict[str, object] = {}
3390        controller, _announce = self._make_player(mock_mass, announcements)
3391        engine = self._make_engine()
3392
3393        with (
3394            patch(
3395                f"{self.ANNOUNCE_MODULE}.resolve_tts_engine", AsyncMock(return_value=engine)
3396            ) as resolve,
3397            patch(f"{self.ANNOUNCE_MODULE}.select_core_tts_engine", AsyncMock()) as select,
3398        ):
3399            await controller.play_announcement(
3400                "player_1", message="hello", tts_engine="tts_plugin/voice"
3401            )
3402
3403        resolve.assert_awaited_once_with(mock_mass, "tts_plugin/voice")
3404        select.assert_not_awaited()
3405
3406    async def test_pre_announce_follows_the_player_config(self, mock_mass: MagicMock) -> None:
3407        """A spoken message uses the player's pre-announce setting without sniffing the url."""
3408        announcements: dict[str, object] = {}
3409        controller, _announce = self._make_player(mock_mass, announcements)
3410        engine = self._make_engine()
3411        mock_mass.config.get_raw_player_config_value = MagicMock(
3412            side_effect=lambda _player_id, key, default=None: (
3413                True if key == CONF_ENTRY_TTS_PRE_ANNOUNCE.key else default
3414            )
3415        )
3416
3417        with patch(
3418            f"{self.ANNOUNCE_MODULE}.select_core_tts_engine", AsyncMock(return_value=engine)
3419        ):
3420            await controller.play_announcement("player_1", message="dinner is ready")
3421
3422        registered = mock_mass.streams.announcement_renderer.register.call_args.args[1]
3423        assert registered["pre_announce"] is True
3424
3425    async def test_the_engine_gets_the_shorter_announcement_timeout(
3426        self, mock_mass: MagicMock
3427    ) -> None:
3428        """The engine is capped well below the background default, it holds the player lock."""
3429        announcements: dict[str, object] = {}
3430        controller, _announce = self._make_player(mock_mass, announcements)
3431        engine = self._make_engine()
3432        query = AsyncMock(return_value=SimpleNamespace(path="http://speech/spoken.mp3"))
3433
3434        with (
3435            patch(f"{self.ANNOUNCE_MODULE}.select_core_tts_engine", AsyncMock(return_value=engine)),
3436            patch(f"{self.ANNOUNCE_MODULE}.query_tts_engine_with_language_fallback", query),
3437        ):
3438            await controller.play_announcement("player_1", message="hello")
3439
3440        assert query.call_args.kwargs["timeout"] == ANNOUNCEMENT_TTS_TIMEOUT
3441        assert ANNOUNCEMENT_TTS_TIMEOUT < TTS_QUERY_TIMEOUT_SECONDS
3442
3443    async def test_an_engine_without_a_message_is_rejected(self, mock_mass: MagicMock) -> None:
3444        """Naming an engine for a url announcement is rejected instead of silently ignored."""
3445        announcements: dict[str, object] = {}
3446        controller, _announce = self._make_player(mock_mass, announcements)
3447
3448        with pytest.raises(PlayerCommandFailed, match="only be used to speak a message"):
3449            await controller.play_announcement(
3450                "player_1", url="http://test/clip.mp3", tts_engine="tts_plugin/voice"
3451            )
3452
3453    async def test_a_language_without_a_message_is_rejected(self, mock_mass: MagicMock) -> None:
3454        """Naming a language for a url announcement is rejected instead of silently ignored."""
3455        announcements: dict[str, object] = {}
3456        controller, _announce = self._make_player(mock_mass, announcements)
3457
3458        with pytest.raises(PlayerCommandFailed, match="A language can only be used"):
3459            await controller.play_announcement(
3460                "player_1", url="http://test/clip.mp3", language="nl-NL"
3461            )
3462
3463    async def test_a_failing_engine_surfaces_its_error(self, mock_mass: MagicMock) -> None:
3464        """An engine that fails to speak the message fails the announcement."""
3465        announcements: dict[str, object] = {}
3466        controller, announce = self._make_player(mock_mass, announcements)
3467        engine = self._make_engine()
3468        engine.provider.get_tts_message = AsyncMock(side_effect=RuntimeError("engine down"))
3469
3470        with (
3471            patch(f"{self.ANNOUNCE_MODULE}.select_core_tts_engine", AsyncMock(return_value=engine)),
3472            pytest.raises(PlayerCommandFailed),
3473        ):
3474            await controller.play_announcement("player_1", message="hello")
3475
3476        announce.assert_not_awaited()
3477        mock_mass.streams.announcement_renderer.register.assert_not_called()
3478
3479    async def test_a_url_or_a_message_is_required(self, mock_mass: MagicMock) -> None:
3480        """An announcement with neither a url nor a message is rejected."""
3481        announcements: dict[str, object] = {}
3482        controller, _announce = self._make_player(mock_mass, announcements)
3483
3484        with pytest.raises(PlayerCommandFailed, match="Either a url or a message"):
3485            await controller.play_announcement("player_1")
3486
3487    async def test_a_url_and_a_message_are_mutually_exclusive(self, mock_mass: MagicMock) -> None:
3488        """An announcement carrying both a url and a message is rejected."""
3489        announcements: dict[str, object] = {}
3490        controller, _announce = self._make_player(mock_mass, announcements)
3491
3492        with pytest.raises(PlayerCommandFailed, match="not both"):
3493            await controller.play_announcement(
3494                "player_1", url="http://test/clip.mp3", message="hello"
3495            )
3496
3497    async def test_unknown_engine_is_rejected(self, mock_mass: MagicMock) -> None:
3498        """A message naming an engine that does not exist fails instead of using another."""
3499        announcements: dict[str, object] = {}
3500        controller, _announce = self._make_player(mock_mass, announcements)
3501
3502        with (
3503            patch(f"{self.ANNOUNCE_MODULE}.resolve_tts_engine", AsyncMock(return_value=None)),
3504            pytest.raises(PlayerCommandFailed, match="is not available"),
3505        ):
3506            await controller.play_announcement("player_1", message="hello", tts_engine="gone")
3507
3508    async def test_no_engine_available_is_rejected(self, mock_mass: MagicMock) -> None:
3509        """A message fails clearly when no TTS engine is set up at all."""
3510        announcements: dict[str, object] = {}
3511        controller, _announce = self._make_player(mock_mass, announcements)
3512
3513        with (
3514            patch(f"{self.ANNOUNCE_MODULE}.select_core_tts_engine", AsyncMock(return_value=None)),
3515            pytest.raises(PlayerCommandFailed, match="No text-to-speech engine"),
3516        ):
3517            await controller.play_announcement("player_1", message="hello")
3518
3519    async def test_audio_that_can_not_be_fetched_is_rejected(self, mock_mass: MagicMock) -> None:
3520        """An engine that only rendered to disk fails, since an announcement is fetched by url."""
3521        announcements: dict[str, object] = {}
3522        controller, _announce = self._make_player(mock_mass, announcements)
3523        engine = self._make_engine(path=str(ANNOUNCE_ALERT_FILE))
3524
3525        with (
3526            patch(f"{self.ANNOUNCE_MODULE}.select_core_tts_engine", AsyncMock(return_value=engine)),
3527            pytest.raises(PlayerCommandFailed, match="rendered the message to a local file"),
3528        ):
3529            await controller.play_announcement("player_1", message="hello")
3530
3531    async def test_group_members_play_the_rendered_audio(self, mock_mass: MagicMock) -> None:
3532        """The message is spoken once and every group member announces the resulting audio."""
3533        announcements: dict[str, object] = {}
3534        use_real_create_task(mock_mass)
3535        controller, _announce = self._make_player(mock_mass, announcements)
3536        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
3537        group = MockPlayer(provider, "group_1", "Group 1", player_type=PlayerType.GROUP)
3538        group._attr_supported_features.add(PlayerFeature.PLAY_ANNOUNCEMENT)
3539        group._attr_group_members = ["player_1"]
3540        group._cache.clear()
3541        controller._players["group_1"] = group
3542        group.update_state(signal_event=False)
3543        engine = self._make_engine()
3544
3545        with patch(
3546            f"{self.ANNOUNCE_MODULE}.select_core_tts_engine", AsyncMock(return_value=engine)
3547        ):
3548            await controller.play_announcement("group_1", message="dinner is ready")
3549
3550        # rendered once for the group, then handed to the member as plain audio
3551        engine.provider.get_tts_message.assert_awaited_once()
3552        member_call = next(
3553            call_args
3554            for call_args in mock_mass.streams.announcement_renderer.register.call_args_list
3555            if call_args.args[0] == "player_1"
3556        )
3557        assert member_call.args[1]["announcement_url"] == "http://speech/spoken.mp3"
3558
3559
3560class TestNativeAnnouncementRouting:
3561    """Announcement routing respects the player's own support and its active output."""
3562
3563    def _make_player_with_linked_child(
3564        self,
3565        mock_mass: MagicMock,
3566        playback_state: PlaybackState,
3567        *,
3568        parent_supports_announce: bool = False,
3569        active_protocol: str | None = None,
3570    ) -> tuple[PlayerController, MockPlayer, MockPlayer, AsyncMock, AsyncMock]:
3571        """Create a controller, a player, its linked protocol child and the two path mocks."""
3572        controller = PlayerController(mock_mass)
3573        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
3574        player = MockPlayer(provider, "player_1", "Player 1")
3575        player._attr_playback_state = playback_state
3576        if parent_supports_announce:
3577            player._attr_supported_features.add(PlayerFeature.PLAY_ANNOUNCEMENT)
3578        proto_provider = MockProvider("airplay", mass=mock_mass)
3579        proto = MockPlayer(
3580            proto_provider, "proto_1", "AirPlay Child", player_type=PlayerType.PROTOCOL
3581        )
3582        proto._attr_supported_features.add(PlayerFeature.PLAY_ANNOUNCEMENT)
3583        controller._players = {"player_1": player, "proto_1": proto}
3584        mock_mass.players = controller
3585        player.set_linked_output_protocols(
3586            [
3587                LinkedOutputProtocol(
3588                    output_protocol_id="proto_1",
3589                    protocol_domain="airplay",
3590                    priority=40,
3591                )
3592            ]
3593        )
3594        if active_protocol is not None:
3595            player.set_active_output_protocol(active_protocol)
3596        render = MagicMock()
3597        render.wait_ready = AsyncMock(return_value=True)
3598        renderer = mock_mass.streams.announcement_renderer
3599        renderer.register = MagicMock(return_value=render)
3600        renderer.unregister = AsyncMock()
3601        mock_mass.streams.get_announcement_url = MagicMock(
3602            side_effect=lambda player_id, **_kwargs: f"http://ma/announcement/{player_id}.mp3"
3603        )
3604        proto.update_state(signal_event=False)
3605        player.update_state(signal_event=False)
3606        native_path = AsyncMock()
3607        generic_path = AsyncMock()
3608        controller._play_native_announcement = native_path  # type: ignore[method-assign]
3609        controller._play_announcement = generic_path  # type: ignore[method-assign]
3610        return controller, player, proto, native_path, generic_path
3611
3612    async def test_playing_player_does_not_route_to_an_idle_linked_child(
3613        self, mock_mass: MagicMock
3614    ) -> None:
3615        """
3616        A player rendering through one output must not announce through another.
3617
3618        E.g. a WiiM playing natively with an idle linked AirPlay child: routing
3619        the announcement to the child would seize the device from the native
3620        output, with nothing restoring that playback afterwards.
3621        """
3622        controller, _player, _proto, native_path, generic_path = (
3623            self._make_player_with_linked_child(
3624                mock_mass, PlaybackState.PLAYING, active_protocol="native"
3625            )
3626        )
3627
3628        await controller.play_announcement("player_1", "http://test/announcement.mp3")
3629
3630        native_path.assert_not_awaited()
3631        generic_path.assert_awaited_once()
3632
3633    async def test_idle_player_routes_to_the_linked_child(self, mock_mass: MagicMock) -> None:
3634        """An idle player announces natively through any capable linked protocol."""
3635        controller, _player, proto, native_path, _generic_path = (
3636            self._make_player_with_linked_child(mock_mass, PlaybackState.IDLE)
3637        )
3638
3639        await controller.play_announcement("player_1", "http://test/announcement.mp3")
3640
3641        native_path.assert_awaited_once()
3642        assert native_path.call_args.args[1] is proto
3643
3644    async def test_active_protocol_child_beats_own_native_support(
3645        self, mock_mass: MagicMock
3646    ) -> None:
3647        """
3648        The output that is actively rendering wins over the player's own support.
3649
3650        E.g. a Sonos playing through its AirPlay child: the announcement rides
3651        the same audio path as the music (mixed into the live stream, in sync
3652        with the rest of a group) instead of a second mechanism firing beside
3653        the playback.
3654        """
3655        controller, _player, proto, native_path, _generic_path = (
3656            self._make_player_with_linked_child(
3657                mock_mass,
3658                PlaybackState.PLAYING,
3659                parent_supports_announce=True,
3660                active_protocol="proto_1",
3661            )
3662        )
3663
3664        await controller.play_announcement("player_1", "http://test/announcement.mp3")
3665
3666        native_path.assert_awaited_once()
3667        assert native_path.call_args.args[1] is proto
3668
3669    async def test_own_native_support_wins_when_playing_natively(
3670        self, mock_mass: MagicMock
3671    ) -> None:
3672        """A player rendering through its own native output announces natively."""
3673        controller, player, _proto, native_path, _generic_path = (
3674            self._make_player_with_linked_child(
3675                mock_mass,
3676                PlaybackState.PLAYING,
3677                parent_supports_announce=True,
3678                active_protocol="native",
3679            )
3680        )
3681
3682        await controller.play_announcement("player_1", "http://test/announcement.mp3")
3683
3684        native_path.assert_awaited_once()
3685        assert native_path.call_args.args[1] is player
3686
3687    async def test_idle_player_prefers_its_own_support_over_a_linked_child(
3688        self, mock_mass: MagicMock
3689    ) -> None:
3690        """Without active playback the player's own announcement support wins."""
3691        controller, player, _proto, native_path, _generic_path = (
3692            self._make_player_with_linked_child(
3693                mock_mass,
3694                PlaybackState.IDLE,
3695                parent_supports_announce=True,
3696            )
3697        )
3698
3699        await controller.play_announcement("player_1", "http://test/announcement.mp3")
3700
3701        native_path.assert_awaited_once()
3702        assert native_path.call_args.args[1] is player
3703
3704
3705@pytest.mark.usefixtures("running_background_tasks")
3706class TestPlayAnnouncementRestore:
3707    """Test the state restore of the default (fallback) announcement implementation."""
3708
3709    def _make_player(
3710        self, mock_mass: MagicMock, prev_media: PlayerMedia
3711    ) -> tuple[PlayerController, MockPlayer, AsyncMock]:
3712        """Create a controller and a playing player, returning the patched resume handler."""
3713        controller = PlayerController(mock_mass)
3714        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
3715        player = MockPlayer(provider, "player_1", "Player 1")
3716        player._attr_playback_state = PlaybackState.PLAYING
3717        player._attr_current_media = prev_media
3718        player._cache.clear()
3719        controller._players = {"player_1": player}
3720        mock_mass.players = controller
3721        mock_mass.player_queues.get = MagicMock(return_value=None)
3722        resume_mock = AsyncMock()
3723        controller._handle_cmd_resume = resume_mock  # type: ignore[method-assign]
3724        controller._handle_cmd_stop = AsyncMock()  # type: ignore[method-assign]
3725        controller._handle_play_media = AsyncMock()  # type: ignore[method-assign]
3726        controller._wait_for_playback_state = AsyncMock()  # type: ignore[method-assign]
3727        controller.get_announcement_volume = MagicMock(return_value=None)  # type: ignore[method-assign]
3728        player.set_initialized()
3729        player.update_state(signal_event=False)
3730        return controller, player, resume_mock
3731
3732    @staticmethod
3733    def _add_group(
3734        controller: PlayerController, player: MockPlayer, *, supports_set_members: bool
3735    ) -> MockPlayer:
3736        """Register a powered group player that holds the given player as its member."""
3737        group = MockPlayer(
3738            cast("MockProvider", player.provider),
3739            "group_1",
3740            "Group 1",
3741            player_type=PlayerType.GROUP,
3742        )
3743        group._attr_powered = True
3744        group._attr_group_members = [player.player_id]
3745        if supports_set_members:
3746            group._attr_supported_features.add(PlayerFeature.SET_MEMBERS)
3747        controller._players[group.player_id] = group
3748        group.set_initialized()
3749        group._cache.clear()
3750        group.update_state(signal_event=False)
3751        # the member has no own input change to trigger a recalculation of its
3752        # (group derived) state, so force it here - just like register() does
3753        player._cache.clear()
3754        player.update_state(force_update=True, signal_event=False)
3755        assert player.state.active_group == group.player_id
3756        return group
3757
3758    async def test_previous_playback_is_restored(self, mock_mass: MagicMock) -> None:
3759        """Content that was playing before the announcement is resumed afterwards."""
3760        controller, player, resume_mock = self._make_player(
3761            mock_mass, PlayerMedia(uri="http://test/track.mp3", media_type=MediaType.TRACK)
3762        )
3763
3764        await controller._play_announcement(player, _announcement())
3765
3766        resume_mock.assert_awaited_once()
3767
3768    async def test_previous_announcement_is_not_restored(self, mock_mass: MagicMock) -> None:
3769        """A player still busy with an earlier announcement has no playback to restore."""
3770        controller, player, resume_mock = self._make_player(
3771            mock_mass,
3772            PlayerMedia(uri="http://ma/announcement/x.mp3", media_type=MediaType.ANNOUNCEMENT),
3773        )
3774
3775        await controller._play_announcement(player, _announcement())
3776
3777        resume_mock.assert_not_awaited()
3778
3779    async def test_volume_is_restored_when_playback_fails(self, mock_mass: MagicMock) -> None:
3780        """A failing announcement never leaves the player at the raised volume."""
3781        controller, player, _ = self._make_player(
3782            mock_mass, PlayerMedia(uri="http://test/track.mp3", media_type=MediaType.TRACK)
3783        )
3784        player._attr_volume_level = 20
3785        player._cache.clear()
3786        player.update_state(force_update=True, signal_event=False)
3787        controller.get_announcement_volume = MagicMock(return_value=80)  # type: ignore[method-assign]
3788        volume_mock = AsyncMock()
3789        controller._handle_cmd_volume_set = volume_mock  # type: ignore[method-assign]
3790        controller._handle_play_media = AsyncMock(  # type: ignore[method-assign]
3791            side_effect=PlayerCommandFailed("player went away")
3792        )
3793
3794        with pytest.raises(PlayerCommandFailed):
3795            await controller._play_announcement(player, _announcement())
3796
3797        assert volume_mock.call_args_list == [call("player_1", 80), call("player_1", 20)]
3798
3799    async def test_zero_announcement_volume_is_applied_and_restored(
3800        self, mock_mass: MagicMock
3801    ) -> None:
3802        """An announcement volume of 0 is a real volume, not an 'unset' fallback."""
3803        controller, player, _ = self._make_player(
3804            mock_mass, PlayerMedia(uri="http://test/track.mp3", media_type=MediaType.TRACK)
3805        )
3806        player._attr_volume_level = 20
3807        player._cache.clear()
3808        player.update_state(force_update=True, signal_event=False)
3809        controller.get_announcement_volume = MagicMock(return_value=0)  # type: ignore[method-assign]
3810        volume_mock = AsyncMock()
3811        controller._handle_cmd_volume_set = volume_mock  # type: ignore[method-assign]
3812
3813        await controller._play_announcement(player, _announcement())
3814
3815        assert volume_mock.call_args_list == [call("player_1", 0), call("player_1", 20)]
3816
3817    async def test_playback_is_restored_when_duration_is_unknown(
3818        self, mock_mass: MagicMock
3819    ) -> None:
3820        """An announcement of unknown length still hands the player back to its content."""
3821        controller, player, resume_mock = self._make_player(
3822            mock_mass, PlayerMedia(uri="http://test/track.mp3", media_type=MediaType.TRACK)
3823        )
3824        announcement = _announcement()
3825        announcement.duration = None
3826
3827        # an unknown length waits for the player to report it finished instead of failing
3828        await controller._play_announcement(player, announcement)
3829
3830        resume_mock.assert_awaited_once()
3831
3832    async def test_group_membership_is_restored_when_playback_fails(
3833        self, mock_mass: MagicMock
3834    ) -> None:
3835        """A failing announcement never leaves the player out of its group player."""
3836        controller, player, _ = self._make_player(
3837            mock_mass, PlayerMedia(uri="http://test/track.mp3", media_type=MediaType.TRACK)
3838        )
3839        group = self._add_group(controller, player, supports_set_members=True)
3840        group.set_members = AsyncMock()  # type: ignore[method-assign]
3841        controller._handle_play_media = AsyncMock(  # type: ignore[method-assign]
3842            side_effect=PlayerCommandFailed("player went away")
3843        )
3844
3845        with pytest.raises(PlayerCommandFailed):
3846            await controller._play_announcement(player, _announcement())
3847
3848        assert group.set_members.await_args_list == [
3849            call(player_ids_to_remove=["player_1"]),
3850            call(player_ids_to_add=["player_1"]),
3851        ]
3852
3853    async def test_restore_failure_does_not_mask_the_announcement_error(
3854        self, mock_mass: MagicMock
3855    ) -> None:
3856        """A provider blowing up during the restore must not hide why the announcement failed."""
3857        controller, player, _ = self._make_player(
3858            mock_mass, PlayerMedia(uri="http://test/track.mp3", media_type=MediaType.TRACK)
3859        )
3860        group = self._add_group(controller, player, supports_set_members=True)
3861        # set_members is a raw provider call: whatever its client library raises comes
3862        # through unwrapped, so the ungroup succeeds and the regroup times out
3863        group.set_members = AsyncMock(  # type: ignore[method-assign]
3864            side_effect=[None, TimeoutError("provider timeout")]
3865        )
3866        controller._handle_play_media = AsyncMock(  # type: ignore[method-assign]
3867            side_effect=PlayerCommandFailed("player went away")
3868        )
3869
3870        with pytest.raises(PlayerCommandFailed, match="player went away"):
3871            await controller._play_announcement(player, _announcement())
3872
3873    async def test_group_without_set_members_is_the_one_powered_off(
3874        self, mock_mass: MagicMock
3875    ) -> None:
3876        """A group that can not release members is powered off, not the announcement target."""
3877        controller, player, _ = self._make_player(
3878            mock_mass, PlayerMedia(uri="http://test/track.mp3", media_type=MediaType.TRACK)
3879        )
3880        self._add_group(controller, player, supports_set_members=False)
3881        power_mock = AsyncMock()
3882        controller._handle_cmd_power = power_mock  # type: ignore[method-assign]
3883        play_mock = AsyncMock()
3884        controller.cmd_play = play_mock  # type: ignore[method-assign]
3885
3886        await controller._play_announcement(player, _announcement())
3887
3888        # the group is switched off for the announcement and restarted afterwards
3889        power_mock.assert_awaited_once_with("group_1", False)
3890        play_mock.assert_awaited_once_with("group_1")
3891
3892    async def test_idle_player_without_power_control_is_regrouped(
3893        self, mock_mass: MagicMock
3894    ) -> None:
3895        """An idle player that has no power state to restore is still put back in its group."""
3896        controller, player, _ = self._make_player(
3897            mock_mass, PlayerMedia(uri="http://test/track.mp3", media_type=MediaType.TRACK)
3898        )
3899        player._attr_playback_state = PlaybackState.IDLE
3900        player._attr_powered = None
3901        group = self._add_group(controller, player, supports_set_members=True)
3902        assert player.state.power_control == PLAYER_CONTROL_NONE
3903        group.set_members = AsyncMock()  # type: ignore[method-assign]
3904
3905        await controller._play_announcement(player, _announcement())
3906
3907        assert group.set_members.await_args_list == [
3908            call(player_ids_to_remove=["player_1"]),
3909            call(player_ids_to_add=["player_1"]),
3910        ]
3911
3912    async def test_muted_player_is_unmuted_and_muted_back(self, mock_mass: MagicMock) -> None:
3913        """A muted player hears the announcement and is muted again afterwards."""
3914        controller, player, _ = self._make_player(
3915            mock_mass, PlayerMedia(uri="http://test/track.mp3", media_type=MediaType.TRACK)
3916        )
3917        mute_mock = _mute_natively(player)
3918
3919        await controller._play_announcement(player, _announcement())
3920
3921        assert mute_mock.await_args_list == [call(False), call(True)]
3922
3923    async def test_player_without_volume_control_is_still_unmuted(
3924        self, mock_mass: MagicMock
3925    ) -> None:
3926        """A player that can only be muted is unmuted for the announcement all the same."""
3927        mock_mass.config.get_raw_player_config_value = MagicMock(
3928            side_effect=_player_config_stub({CONF_VOLUME_CONTROL: PLAYER_CONTROL_NONE})
3929        )
3930        controller, player, _ = self._make_player(
3931            mock_mass, PlayerMedia(uri="http://test/track.mp3", media_type=MediaType.TRACK)
3932        )
3933        mute_mock = _mute_natively(player)
3934        assert player.state.volume_control == PLAYER_CONTROL_NONE
3935
3936        await controller._play_announcement(player, _announcement())
3937
3938        assert mute_mock.await_args_list == [call(False), call(True)]
3939
3940    async def test_mute_is_restored_before_the_player_is_regrouped(
3941        self, mock_mass: MagicMock
3942    ) -> None:
3943        """A player is handed back to its group already muted, holding on to its mute lock."""
3944        controller, player, _ = self._make_player(
3945            mock_mass, PlayerMedia(uri="http://test/track.mp3", media_type=MediaType.TRACK)
3946        )
3947        group = self._add_group(controller, player, supports_set_members=True)
3948        real_set_members = group.set_members
3949
3950        async def _set_members(**kwargs: list[str]) -> None:
3951            # let the membership really change, so the player is ungrouped while the
3952            # announcement plays - just like it is in production. neither player picks
3953            # the new membership up on its own here, so publish it on both.
3954            await real_set_members(**kwargs)
3955            group.update_state(force_update=True, signal_event=False)
3956            player._cache.clear()
3957            player.update_state(force_update=True, signal_event=False)
3958
3959        set_members = AsyncMock(side_effect=_set_members)
3960        group.set_members = set_members  # type: ignore[method-assign]
3961        player.extra_data[ATTR_MUTE_LOCK] = True
3962        recorder = MagicMock()
3963        recorder.attach_mock(_mute_natively(player), "mute")
3964        recorder.attach_mock(set_members, "set_members")
3965
3966        await controller._play_announcement(player, _announcement())
3967
3968        assert recorder.mock_calls == [
3969            call.set_members(player_ids_to_remove=["player_1"]),
3970            call.mute(False),
3971            call.mute(True),
3972            call.set_members(player_ids_to_add=["player_1"]),
3973        ]
3974        # the lock survives the announcement, so the regroup does not unmute the player
3975        assert player.extra_data[ATTR_MUTE_LOCK] is True
3976
3977    async def test_unmuted_player_is_left_alone(self, mock_mass: MagicMock) -> None:
3978        """A player that was not muted is never sent a mute command."""
3979        controller, player, _ = self._make_player(
3980            mock_mass, PlayerMedia(uri="http://test/track.mp3", media_type=MediaType.TRACK)
3981        )
3982        mute_mock = _mute_natively(player)
3983        player._attr_volume_muted = False
3984        player._cache.clear()
3985        player.update_state(force_update=True, signal_event=False)
3986        mute_mock.reset_mock()
3987
3988        await controller._play_announcement(player, _announcement())
3989
3990        mute_mock.assert_not_awaited()
3991
3992    async def test_mute_is_restored_when_playback_fails(self, mock_mass: MagicMock) -> None:
3993        """A failing announcement never leaves the player unmuted."""
3994        controller, player, _ = self._make_player(
3995            mock_mass, PlayerMedia(uri="http://test/track.mp3", media_type=MediaType.TRACK)
3996        )
3997        mute_mock = _mute_natively(player)
3998        controller._handle_play_media = AsyncMock(  # type: ignore[method-assign]
3999            side_effect=PlayerCommandFailed("player went away")
4000        )
4001
4002        with pytest.raises(PlayerCommandFailed):
4003            await controller._play_announcement(player, _announcement())
4004
4005        assert mute_mock.await_args_list == [call(False), call(True)]
4006
4007    async def test_muted_sync_group_members_all_hear_the_announcement(
4008        self, mock_mass: MagicMock
4009    ) -> None:
4010        """Every member of a muted sync group is unmuted, keeping its mute lock."""
4011        controller, leader, _ = self._make_player(
4012            mock_mass, PlayerMedia(uri="http://test/track.mp3", media_type=MediaType.TRACK)
4013        )
4014        member = MockPlayer(cast("MockProvider", leader.provider), "player_2", "Player 2")
4015        controller._players["player_2"] = member
4016        member.set_initialized()
4017        leader._attr_group_members = ["player_1", "player_2"]
4018        mute_mocks = {player.player_id: _mute_natively(player) for player in (leader, member)}
4019        # both members were muted while grouped, so both hold a mute lock
4020        for player in (leader, member):
4021            player.extra_data[ATTR_MUTE_LOCK] = True
4022
4023        await controller._play_announcement(leader, _announcement())
4024
4025        for player_id, mute_mock in mute_mocks.items():
4026            assert mute_mock.await_args_list == [call(False), call(True)], player_id
4027            assert controller._players[player_id].extra_data[ATTR_MUTE_LOCK] is True
4028
4029    async def test_fake_muted_player_announces_at_its_real_volume(
4030        self, mock_mass: MagicMock
4031    ) -> None:
4032        """A fake muted player announces at its real volume, not at the zero it is parked on."""
4033        mock_mass.config.get_raw_player_config_value = MagicMock(
4034            side_effect=_player_config_stub({CONF_MUTE_CONTROL: PLAYER_CONTROL_FAKE})
4035        )
4036        controller, player, _ = self._make_player(
4037            mock_mass, PlayerMedia(uri="http://test/track.mp3", media_type=MediaType.TRACK)
4038        )
4039
4040        def _apply_volume(volume: int) -> None:
4041            player._attr_volume_level = volume
4042            player.update_state(signal_event=False)
4043
4044        player._attr_volume_level = 40
4045        volume_set = AsyncMock(side_effect=_apply_volume)
4046        player.volume_set = volume_set  # type: ignore[method-assign]
4047        player._cache.clear()
4048        player.update_state(force_update=True, signal_event=False)
4049        await controller.cmd_volume_mute("player_1", True)
4050        assert player.state.volume_muted is True
4051        controller.get_announcement_volume = MagicMock(return_value=80)  # type: ignore[method-assign]
4052
4053        await controller._play_announcement(player, _announcement())
4054
4055        # unmute to 40, announce at 80, restore 40 and park back on 0 for the fake mute
4056        assert volume_set.await_args_list == [
4057            call(0),
4058            call(40),
4059            call(80),
4060            call(40),
4061            call(0),
4062        ]
4063        assert player.state.volume_muted is True
4064        assert player.extra_data[ATTR_PREVIOUS_VOLUME] == 40
4065
4066
4067@pytest.mark.usefixtures("running_background_tasks")
4068class TestPlayNativeAnnouncement:
4069    """Test the mute handling around an announcement that a player plays natively."""
4070
4071    def _make_player(self, mock_mass: MagicMock) -> tuple[PlayerController, MockPlayer, AsyncMock]:
4072        """Create a controller and a player with native announcement support."""
4073        controller = PlayerController(mock_mass)
4074        provider = MockProvider("test_provider", instance_id="test", mass=mock_mass)
4075        player = MockPlayer(provider, "player_1", "Player 1")
4076        player._attr_supported_features.add(PlayerFeature.PLAY_ANNOUNCEMENT)
4077        player._cache.clear()
4078        controller._players = {"player_1": player}
4079        mock_mass.players = controller
4080        mock_mass.player_queues.get = MagicMock(return_value=None)
4081        controller.get_announcement_volume = MagicMock(return_value=None)  # type: ignore[method-assign]
4082        announce_mock = AsyncMock()
4083        player.play_announcement = announce_mock  # type: ignore[method-assign]
4084        player.set_initialized()
4085        player.update_state(signal_event=False)
4086        return controller, player, announce_mock
4087
4088    async def test_muted_player_is_unmuted_and_muted_back(self, mock_mass: MagicMock) -> None:
4089        """A muted player hears the announcement and is muted again afterwards."""
4090        controller, player, announce_mock = self._make_player(mock_mass)
4091        recorder = MagicMock()
4092        recorder.attach_mock(_mute_natively(player), "mute")
4093        recorder.attach_mock(announce_mock, "announce")
4094
4095        await controller._play_native_announcement(player, player, _announcement(), None)
4096
4097        assert recorder.mock_calls == [
4098            call.mute(False),
4099            call.announce(ANY, None),
4100            call.mute(True),
4101        ]
4102
4103    async def test_unmuted_player_is_left_alone(self, mock_mass: MagicMock) -> None:
4104        """A player that was not muted is never sent a mute command."""
4105        controller, player, _ = self._make_player(mock_mass)
4106        mute_mock = _mute_natively(player)
4107        player._attr_volume_muted = False
4108        player._cache.clear()
4109        player.update_state(force_update=True, signal_event=False)
4110        mute_mock.reset_mock()
4111
4112        await controller._play_native_announcement(player, player, _announcement(), None)
4113
4114        mute_mock.assert_not_awaited()
4115
4116    async def test_mute_is_restored_when_the_provider_fails(self, mock_mass: MagicMock) -> None:
4117        """A failing announcement never leaves the player unmuted."""
4118        controller, player, announce_mock = self._make_player(mock_mass)
4119        mute_mock = _mute_natively(player)
4120        announce_mock.side_effect = PlayerCommandFailed("player went away")
4121
4122        with pytest.raises(PlayerCommandFailed):
4123            await controller._play_native_announcement(player, player, _announcement(), None)
4124
4125        assert mute_mock.await_args_list == [call(False), call(True)]
4126
4127    async def test_muted_sync_group_members_all_hear_the_announcement(
4128        self, mock_mass: MagicMock
4129    ) -> None:
4130        """Every member of a muted sync group is unmuted, keeping its mute lock."""
4131        controller, leader, _ = self._make_player(mock_mass)
4132        member = MockPlayer(cast("MockProvider", leader.provider), "player_2", "Player 2")
4133        controller._players["player_2"] = member
4134        member.set_initialized()
4135        leader._attr_group_members = ["player_1", "player_2"]
4136        mute_mocks = {player.player_id: _mute_natively(player) for player in (leader, member)}
4137        # both members were muted while grouped, so both hold a mute lock
4138        for player in (leader, member):
4139            player.extra_data[ATTR_MUTE_LOCK] = True
4140
4141        await controller._play_native_announcement(leader, leader, _announcement(), None)
4142
4143        for player_id, mute_mock in mute_mocks.items():
4144            assert mute_mock.await_args_list == [call(False), call(True)], player_id
4145            assert controller._players[player_id].extra_data[ATTR_MUTE_LOCK] is True
4146
4147    async def test_fake_muted_player_announces_at_its_real_volume(
4148        self, mock_mass: MagicMock
4149    ) -> None:
4150        """The announcement volume is resolved after the unmute, not from the parked zero."""
4151        mock_mass.config.get_raw_player_config_value = MagicMock(
4152            side_effect=_player_config_stub({CONF_MUTE_CONTROL: PLAYER_CONTROL_FAKE})
4153        )
4154        controller, player, announce_mock = self._make_player(mock_mass)
4155
4156        def _apply_volume(volume: int) -> None:
4157            player._attr_volume_level = volume
4158            player.update_state(signal_event=False)
4159
4160        player._attr_volume_level = 40
4161        volume_set = AsyncMock(side_effect=_apply_volume)
4162        player.volume_set = volume_set  # type: ignore[method-assign]
4163        player._cache.clear()
4164        player.update_state(force_update=True, signal_event=False)
4165        await controller.cmd_volume_mute("player_1", True)
4166        assert player.state.volume_muted is True
4167        # stand in for the configured strategy, which reads the volume of the player
4168        controller.get_announcement_volume = MagicMock(  # type: ignore[method-assign]
4169            side_effect=lambda _player_id, _volume_level: player.state.volume_level
4170        )
4171
4172        await controller._play_native_announcement(player, player, _announcement(), None)
4173
4174        assert announce_mock.await_args == call(ANY, 40)
4175        # unmute to 40 for the announcement, park back on 0 for the fake mute
4176        assert volume_set.await_args_list == [call(0), call(40), call(0)]
4177        assert player.state.volume_muted is True
4178        assert player.extra_data[ATTR_PREVIOUS_VOLUME] == 40
4179
4180
4181class TestScheduleActiveOutputProtocolClear:
4182    """Test the deferred clear of a player's active output protocol."""
4183
4184    def test_schedule_starts_cancellable_clear_task(self, mock_mass: MagicMock) -> None:
4185        """Scheduling defers the clear to a single, per-player, cancellable task."""
4186        controller = PlayerController(mock_mass)
4187        player = MagicMock()
4188        player.player_id = "player_1"
4189
4190        controller.schedule_active_output_protocol_clear(player)
4191
4192        mock_mass.create_task.assert_called_once()
4193        # close the coroutine passed to the mocked create_task to avoid a
4194        # "coroutine was never awaited" warning
4195        mock_mass.create_task.call_args.args[0].close()
4196        # no abort_existing: a duplicate schedule must reuse the pending clear
4197        # (deduped by task_id) instead of replacing it with an untracked task
4198        assert mock_mass.create_task.call_args.kwargs == {
4199            "task_id": "clear_active_protocol_player_1",
4200        }
4201
4202    @pytest.mark.asyncio
4203    async def test_clears_protocol_once_player_idle(self, mock_mass: MagicMock) -> None:
4204        """The protocol is cleared after waiting for the player to reach IDLE."""
4205        controller = PlayerController(mock_mass)
4206        player = MagicMock()
4207        player.player_id = "player_1"
4208
4209        with patch.object(controller, "_wait_for_playback_state", new=AsyncMock()) as wait_mock:
4210            await controller._clear_active_output_protocol_when_idle(player)
4211
4212        wait_mock.assert_awaited_once_with(player, PlaybackState.IDLE, timeout=10)
4213        player.set_active_output_protocol.assert_called_once_with(None)
4214
4215
4216@contextlib.asynccontextmanager
4217async def _skip_player_update_wait(
4218    *_args: object,
4219    **_kwargs: object,
4220) -> AsyncIterator[None]:
4221    """Skip provider-driven state propagation in command-routing tests."""
4222    yield
4223
4224
4225class TestRemovePlayerControl:
4226    """Test removing a registered player control."""
4227
4228    def test_removal_refreshes_the_players_that_used_it(self, mock_mass: MagicMock) -> None:
4229        """Test that only the players configured to use the removed control are refreshed."""
4230        mock_mass.loop = MagicMock()
4231        controller = PlayerController(mock_mass)
4232        using_control = MagicMock()
4233        using_control.state.power_control = "switch.amp"
4234        using_control.state.volume_control = PLAYER_CONTROL_NATIVE
4235        using_control.state.mute_control = PLAYER_CONTROL_NATIVE
4236        unrelated = MagicMock()
4237        unrelated.state.power_control = PLAYER_CONTROL_NATIVE
4238        unrelated.state.volume_control = PLAYER_CONTROL_NATIVE
4239        unrelated.state.mute_control = PLAYER_CONTROL_NATIVE
4240        controller._players = {"using_control": using_control, "unrelated": unrelated}
4241        controller._controls = {
4242            "switch.amp": PlayerControl(id="switch.amp", provider="test_prov", name="Amp")
4243        }
4244
4245        controller.remove_player_control("switch.amp")
4246
4247        assert controller.player_controls() == []
4248        mock_mass.loop.call_soon.assert_called_once_with(using_control.refresh_state)
4249
4250    async def test_a_returning_control_is_picked_back_up(self, mock_mass: MagicMock) -> None:
4251        """Test that a control removed and registered again re-attaches to its player."""
4252        # run the scheduled refresh straight away so each step is observable
4253        mock_mass.loop = MagicMock()
4254        mock_mass.loop.call_soon.side_effect = lambda callback, *args: callback(*args)
4255        mock_mass.config.get_raw_player_config_value.side_effect = _player_config_stub(
4256            {CONF_POWER_CONTROL: "switch.amp"}
4257        )
4258        controller = PlayerController(mock_mass)
4259        mock_mass.players = controller
4260        provider = MockProvider("test_provider", instance_id="test_prov", mass=mock_mass)
4261        mock_mass.get_provider.return_value = provider
4262        player = MockPlayer(provider, "player", "Player")
4263        controller._players = {"player": player}
4264        control = PlayerControl(id="switch.amp", provider="test_prov", name="Amp")
4265
4266        await controller.register_or_update_player_control(control)
4267        assert player.state.power_control == "switch.amp"
4268
4269        # the Home Assistant plugin drops and re-registers its controls around a reload
4270        controller.remove_player_control(control.id)
4271        assert player.state.power_control == PLAYER_CONTROL_NONE
4272
4273        await controller.register_or_update_player_control(control)
4274        assert player.state.power_control == "switch.amp"
4275
4276    def test_removing_an_unknown_control_does_nothing(self, mock_mass: MagicMock) -> None:
4277        """Test that removing a control that was never registered is a no-op."""
4278        mock_mass.loop = MagicMock()
4279        controller = PlayerController(mock_mass)
4280        player = MagicMock()
4281        player.state.power_control = PLAYER_CONTROL_NATIVE
4282        player.state.volume_control = PLAYER_CONTROL_NATIVE
4283        player.state.mute_control = PLAYER_CONTROL_NATIVE
4284        controller._players = {"player": player}
4285
4286        controller.remove_player_control("switch.gone")
4287
4288        mock_mass.loop.call_soon.assert_not_called()
4289
4290
4291class _FailingTeardownPlayer(MockPlayer):
4292    """Player whose provider fails to release it."""
4293
4294    unloaded = False
4295
4296    async def on_unload(self) -> None:
4297        """Handle logic when the player is unloaded from the Player controller."""
4298        self.unloaded = True
4299        msg = "device is gone"
4300        raise RuntimeError(msg)
4301
4302
4303class TestUnregisterTeardown:
4304    """Test that a failing player teardown stays contained."""
4305
4306    async def test_failing_on_unload_still_unregisters_the_player(
4307        self, mock_mass: MagicMock
4308    ) -> None:
4309        """Test that a provider raising while releasing its player does not break unregister."""
4310        controller = PlayerController(mock_mass)
4311        mock_mass.players = controller
4312        provider = MockProvider("test_provider", instance_id="test_prov", mass=mock_mass)
4313        player = _FailingTeardownPlayer(provider, "boom", "Boom")
4314        controller._players = {"boom": player}
4315
4316        await controller.unregister("boom")
4317
4318        assert "boom" not in controller._players
4319        assert player.unloaded
4320
4321
4322class TestConfigChangeRestartsPlayback:
4323    """Test that a changed player setting which needs a reload restarts playback."""
4324
4325    @staticmethod
4326    def _config(*, requires_reload: bool) -> PlayerConfig:
4327        """Build a PlayerConfig holding a single output codec entry."""
4328        return PlayerConfig(
4329            provider="test_prov",
4330            player_id="player_1",
4331            values={
4332                CONF_OUTPUT_CODEC: ConfigEntry(
4333                    key=CONF_OUTPUT_CODEC,
4334                    type=ConfigEntryType.STRING,
4335                    label="Output codec",
4336                    value="flac",
4337                    requires_reload=requires_reload,
4338                )
4339            },
4340        )
4341
4342    @staticmethod
4343    def _prepare(mock_mass: MagicMock, queue_state: PlaybackState) -> PlayerController:
4344        """Register a player whose active queue is in the given state."""
4345        controller = PlayerController(mock_mass)
4346        player = MagicMock()
4347        player.state.active_source = "player_1"
4348        player.on_config_updated = AsyncMock()
4349        controller._players = {"player_1": player}
4350        queue = MagicMock()
4351        queue.queue_id = "player_1"
4352        queue.state = queue_state
4353        mock_mass.player_queues.get = MagicMock(return_value=queue)
4354        mock_mass.player_queues.stop = AsyncMock()
4355        return controller
4356
4357    async def test_reload_setting_restarts_playback(self, mock_mass: MagicMock) -> None:
4358        """Test that changing a reload-requiring setting stops and resumes the queue."""
4359        controller = self._prepare(mock_mass, PlaybackState.PLAYING)
4360
4361        await controller.on_player_config_change(
4362            self._config(requires_reload=True), {f"values/{CONF_OUTPUT_CODEC}"}
4363        )
4364
4365        mock_mass.player_queues.stop.assert_awaited_once_with("player_1")
4366        mock_mass.call_later.assert_called_once_with(
4367            1, mock_mass.player_queues.resume, "player_1", False
4368        )
4369
4370    async def test_plain_setting_does_not_restart_playback(self, mock_mass: MagicMock) -> None:
4371        """Test that a setting which applies on the fly leaves playback alone."""
4372        controller = self._prepare(mock_mass, PlaybackState.PLAYING)
4373
4374        await controller.on_player_config_change(
4375            self._config(requires_reload=False), {f"values/{CONF_OUTPUT_CODEC}"}
4376        )
4377
4378        mock_mass.player_queues.stop.assert_not_awaited()
4379        mock_mass.call_later.assert_not_called()
4380
4381    async def test_untouched_reload_setting_does_not_restart_playback(
4382        self, mock_mass: MagicMock
4383    ) -> None:
4384        """Test that only a changed reload-requiring setting restarts playback."""
4385        controller = self._prepare(mock_mass, PlaybackState.PLAYING)
4386
4387        await controller.on_player_config_change(
4388            self._config(requires_reload=True), {f"values/{CONF_ICON}"}
4389        )
4390
4391        mock_mass.player_queues.stop.assert_not_awaited()
4392        mock_mass.call_later.assert_not_called()
4393
4394    async def test_idle_queue_is_left_alone(self, mock_mass: MagicMock) -> None:
4395        """Test that a reload-requiring change does not start playback on an idle queue."""
4396        controller = self._prepare(mock_mass, PlaybackState.IDLE)
4397
4398        await controller.on_player_config_change(
4399            self._config(requires_reload=True), {f"values/{CONF_OUTPUT_CODEC}"}
4400        )
4401
4402        mock_mass.player_queues.stop.assert_not_awaited()
4403        mock_mass.call_later.assert_not_called()
4404
4405
4406if __name__ == "__main__":
4407    pytest.main([__file__, "-v"])
4408