music-assistant-server

1"""
2Tests for Player control auto-select and explicit config logic.
3
4Covers the power_control, volume_control, and mute_control properties
5on the Player model, including:
6- Explicit config values (native, fake, none)
7- Explicit config pointing to a protocol player or player control
8- Auto-select fallback: native -> protocol player -> none
9- Protocol player preference for active output protocol
10"""
11
12from __future__ import annotations
13
14from typing import Any
15from unittest.mock import MagicMock
16
17import pytest
18from music_assistant_models.constants import (
19    PLAYER_CONTROL_FAKE,
20    PLAYER_CONTROL_NATIVE,
21    PLAYER_CONTROL_NONE,
22)
23from music_assistant_models.enums import PlayerFeature
24
25from music_assistant.constants import (
26    CONF_MUTE_CONTROL,
27    CONF_POWER_CONTROL,
28    CONF_PREFERRED_OUTPUT_PROTOCOL,
29    CONF_VOLUME_CONTROL,
30)
31from music_assistant.models.player import LinkedOutputProtocol
32from tests.common import MockPlayer, MockProvider
33
34_PROTO_DOMAIN = "test_protocol"
35
36
37@pytest.fixture
38def mock_mass() -> MagicMock:
39    """Create a mock MusicAssistant instance."""
40    mass = MagicMock()
41    mass.closing = False
42    mass.loop = None
43    mass.config.get = MagicMock(return_value=[])
44    mass.config.get_raw_player_config_value = MagicMock(return_value=None)
45    mass.config.get_raw_core_config_value = MagicMock(return_value="GLOBAL")
46    mass.config.set = MagicMock()
47    mass.signal_event = MagicMock()
48    mass.get_providers = MagicMock(return_value=[])
49    mass.players.get_player = MagicMock(return_value=None)
50    mass.players.get_player_control = MagicMock(return_value=None)
51    return mass
52
53
54def _make_config_side_effect(
55    overrides: dict[str, str | None],
56) -> Any:
57    """Create a side_effect for get_raw_player_config_value."""
58
59    def _side_effect(_pid: str, key: str, *_args: Any) -> str | None:
60        return overrides.get(key)
61
62    return _side_effect
63
64
65def _create_player(
66    mock_mass: MagicMock,
67    features: set[PlayerFeature] | None = None,
68) -> MockPlayer:
69    """Create a MockPlayer with the given features."""
70    provider = MockProvider("test", mass=mock_mass)
71    player = MockPlayer(provider, "main_player", "Main")
72    player._attr_supported_features = features or set()
73    player._cache.clear()
74    return player
75
76
77def _create_protocol_player(
78    mock_mass: MagicMock,
79    player_id: str,
80    features: set[PlayerFeature],
81    available: bool = True,
82) -> MockPlayer:
83    """Create a mock protocol player."""
84    provider = MockProvider("protocol", mass=mock_mass)
85    player = MockPlayer(provider, player_id, f"Protocol {player_id}")
86    player._attr_supported_features = features
87    player._attr_available = available
88    player._cache.clear()
89    return player
90
91
92def _make_protocol_link(
93    protocol_id: str,
94    priority: int = 0,
95    protocol_domain: str = _PROTO_DOMAIN,
96) -> LinkedOutputProtocol:
97    """Create a protocol link for testing."""
98    return LinkedOutputProtocol(
99        output_protocol_id=protocol_id,
100        protocol_domain=protocol_domain,
101        priority=priority,
102    )
103
104
105def _link_protocols(
106    player: MockPlayer,
107    protocols: list[LinkedOutputProtocol],
108    active_protocol_id: str | None = None,
109) -> None:
110    """Link output protocols to a player and optionally set active protocol."""
111    player.set_linked_output_protocols(protocols)
112    if active_protocol_id is not None:
113        player.set_active_output_protocol(active_protocol_id)
114    player._cache.clear()
115
116
117class TestPowerControlExplicitConfig:
118    """Test power_control with explicit config values."""
119
120    def test_explicit_native(self, mock_mass: MagicMock) -> None:
121        """Power control returns native when explicitly configured and feature is supported."""
122        mock_mass.config.get_raw_player_config_value = MagicMock(
123            side_effect=_make_config_side_effect({CONF_POWER_CONTROL: PLAYER_CONTROL_NATIVE})
124        )
125        # NATIVE control requires the player to actually advertise POWER —
126        # otherwise the getter degrades to NONE (e.g. when a group player
127        # drops POWER from its feature set on upgrade but its stale config
128        # still says NATIVE).
129        player = _create_player(mock_mass, features={PlayerFeature.POWER})
130        assert player.power_control == PLAYER_CONTROL_NATIVE
131
132    def test_explicit_native_degrades_when_feature_missing(self, mock_mass: MagicMock) -> None:
133        """Stale NATIVE config falls back to NONE when the player no longer advertises POWER."""
134        mock_mass.config.get_raw_player_config_value = MagicMock(
135            side_effect=_make_config_side_effect({CONF_POWER_CONTROL: PLAYER_CONTROL_NATIVE})
136        )
137        # no POWER in features → the explicit NATIVE in config is invalid
138        player = _create_player(mock_mass)
139        assert player.power_control == PLAYER_CONTROL_NONE
140
141    def test_explicit_fake(self, mock_mass: MagicMock) -> None:
142        """Power control returns fake when explicitly configured."""
143        mock_mass.config.get_raw_player_config_value = MagicMock(
144            side_effect=_make_config_side_effect({CONF_POWER_CONTROL: PLAYER_CONTROL_FAKE})
145        )
146        player = _create_player(mock_mass)
147        assert player.power_control == PLAYER_CONTROL_FAKE
148
149    def test_explicit_none(self, mock_mass: MagicMock) -> None:
150        """Power control returns none when explicitly configured."""
151        mock_mass.config.get_raw_player_config_value = MagicMock(
152            side_effect=_make_config_side_effect({CONF_POWER_CONTROL: PLAYER_CONTROL_NONE})
153        )
154        player = _create_player(mock_mass, features={PlayerFeature.POWER})
155        assert player.power_control == PLAYER_CONTROL_NONE
156
157    def test_explicit_protocol_player_id_ignored(self, mock_mass: MagicMock) -> None:
158        """Protocol player IDs are not valid for power control; falls through to auto-select."""
159        protocol = _create_protocol_player(
160            mock_mass, "protocol_1", {PlayerFeature.POWER}, available=True
161        )
162        mock_mass.config.get_raw_player_config_value = MagicMock(
163            side_effect=_make_config_side_effect({CONF_POWER_CONTROL: "protocol_1"})
164        )
165        mock_mass.players.get_player = MagicMock(return_value=protocol)
166        mock_mass.players.get_player_control = MagicMock(return_value=None)
167        player = _create_player(mock_mass)
168        # protocol players handle their own power at the protocol level
169        assert player.power_control == PLAYER_CONTROL_NONE
170
171    def test_explicit_protocol_player_unavailable_falls_through(self, mock_mass: MagicMock) -> None:
172        """When explicitly set to unavailable protocol player, falls to auto-select."""
173        protocol = _create_protocol_player(
174            mock_mass, "protocol_1", {PlayerFeature.POWER}, available=False
175        )
176        mock_mass.config.get_raw_player_config_value = MagicMock(
177            side_effect=_make_config_side_effect({CONF_POWER_CONTROL: "protocol_1"})
178        )
179        mock_mass.players.get_player = MagicMock(return_value=protocol)
180        mock_mass.players.get_player_control = MagicMock(return_value=None)
181        player = _create_player(mock_mass)
182        assert player.power_control == PLAYER_CONTROL_NONE
183
184    def test_explicit_player_control_id(self, mock_mass: MagicMock) -> None:
185        """Power control returns player control ID when set and control exists."""
186        control = MagicMock()
187        control.id = "ext_control_1"
188        mock_mass.config.get_raw_player_config_value = MagicMock(
189            side_effect=_make_config_side_effect({CONF_POWER_CONTROL: "ext_control_1"})
190        )
191        mock_mass.players.get_player = MagicMock(return_value=None)
192        mock_mass.players.get_player_control = MagicMock(return_value=control)
193        player = _create_player(mock_mass)
194        assert player.power_control == "ext_control_1"
195
196
197class TestPowerControlAutoSelect:
198    """Test power_control auto-select logic (config is None or 'auto')."""
199
200    def test_auto_prefers_native(self, mock_mass: MagicMock) -> None:
201        """Auto-select returns native when player supports POWER feature."""
202        player = _create_player(mock_mass, features={PlayerFeature.POWER})
203        assert player.power_control == PLAYER_CONTROL_NATIVE
204
205    def test_auto_does_not_use_protocol_player(self, mock_mass: MagicMock) -> None:
206        """Auto-select does not use protocol players for power; they handle it themselves."""
207        protocol = _create_protocol_player(mock_mass, "proto_power", {PlayerFeature.POWER})
208        mock_mass.players.get_player = MagicMock(return_value=protocol)
209        player = _create_player(mock_mass)
210        _link_protocols(
211            player,
212            [_make_protocol_link("proto_power")],
213            active_protocol_id="proto_power",
214        )
215        assert player.power_control == PLAYER_CONTROL_NONE
216
217    def test_auto_returns_none_without_active_protocol(self, mock_mass: MagicMock) -> None:
218        """Auto-select returns NONE when protocol exists but is not active."""
219        protocol = _create_protocol_player(mock_mass, "proto_power", {PlayerFeature.POWER})
220        mock_mass.players.get_player = MagicMock(return_value=protocol)
221        player = _create_player(mock_mass)
222        _link_protocols(player, [_make_protocol_link("proto_power")])
223        assert player.power_control == PLAYER_CONTROL_NONE
224
225    def test_auto_returns_none_when_nothing_available(self, mock_mass: MagicMock) -> None:
226        """Auto-select returns NONE when no native and no protocol players."""
227        player = _create_player(mock_mass)
228        assert player.power_control == PLAYER_CONTROL_NONE
229
230    def test_auto_with_config_set_to_auto(self, mock_mass: MagicMock) -> None:
231        """Explicit 'auto' config value triggers the same auto-select."""
232        mock_mass.config.get_raw_player_config_value = MagicMock(
233            side_effect=_make_config_side_effect({CONF_POWER_CONTROL: "auto"})
234        )
235        player = _create_player(mock_mass, features={PlayerFeature.POWER})
236        assert player.power_control == PLAYER_CONTROL_NATIVE
237
238
239class TestVolumeControlExplicitConfig:
240    """Test volume_control with explicit config values."""
241
242    def test_explicit_native(self, mock_mass: MagicMock) -> None:
243        """Volume control returns native when explicitly configured."""
244        mock_mass.config.get_raw_player_config_value = MagicMock(
245            side_effect=_make_config_side_effect({CONF_VOLUME_CONTROL: PLAYER_CONTROL_NATIVE})
246        )
247        player = _create_player(mock_mass)
248        assert player.volume_control == PLAYER_CONTROL_NATIVE
249
250    def test_explicit_none(self, mock_mass: MagicMock) -> None:
251        """Volume control returns none even with native support."""
252        mock_mass.config.get_raw_player_config_value = MagicMock(
253            side_effect=_make_config_side_effect({CONF_VOLUME_CONTROL: PLAYER_CONTROL_NONE})
254        )
255        player = _create_player(mock_mass, features={PlayerFeature.VOLUME_SET})
256        assert player.volume_control == PLAYER_CONTROL_NONE
257
258    def test_explicit_protocol_player_id(self, mock_mass: MagicMock) -> None:
259        """Volume control returns protocol player when set and available."""
260        protocol = _create_protocol_player(
261            mock_mass, "vol_player", {PlayerFeature.VOLUME_SET}, available=True
262        )
263        mock_mass.config.get_raw_player_config_value = MagicMock(
264            side_effect=_make_config_side_effect({CONF_VOLUME_CONTROL: "vol_player"})
265        )
266        mock_mass.players.get_player = MagicMock(return_value=protocol)
267        player = _create_player(mock_mass)
268        assert player.volume_control == "vol_player"
269
270
271class TestVolumeControlAutoSelect:
272    """Test volume_control auto-select logic."""
273
274    def test_auto_prefers_native(self, mock_mass: MagicMock) -> None:
275        """Auto-select returns native when player supports VOLUME_SET."""
276        player = _create_player(mock_mass, features={PlayerFeature.VOLUME_SET})
277        assert player.volume_control == PLAYER_CONTROL_NATIVE
278
279    def test_auto_falls_back_to_protocol_player(self, mock_mass: MagicMock) -> None:
280        """Auto-select returns protocol player for volume."""
281        protocol = _create_protocol_player(mock_mass, "proto_vol", {PlayerFeature.VOLUME_SET})
282        mock_mass.players.get_player = MagicMock(return_value=protocol)
283        player = _create_player(mock_mass)
284        _link_protocols(player, [_make_protocol_link("proto_vol")])
285        assert player.volume_control == "proto_vol"
286
287    def test_auto_returns_none_when_nothing_available(self, mock_mass: MagicMock) -> None:
288        """Auto-select returns NONE when nothing supports volume."""
289        player = _create_player(mock_mass)
290        assert player.volume_control == PLAYER_CONTROL_NONE
291
292
293class TestMuteControlExplicitConfig:
294    """Test mute_control with explicit config values."""
295
296    def test_explicit_native(self, mock_mass: MagicMock) -> None:
297        """Mute control returns native when explicitly configured."""
298        mock_mass.config.get_raw_player_config_value = MagicMock(
299            side_effect=_make_config_side_effect({CONF_MUTE_CONTROL: PLAYER_CONTROL_NATIVE})
300        )
301        player = _create_player(mock_mass)
302        assert player.mute_control == PLAYER_CONTROL_NATIVE
303
304    def test_explicit_none(self, mock_mass: MagicMock) -> None:
305        """Mute control returns none when explicitly configured."""
306        mock_mass.config.get_raw_player_config_value = MagicMock(
307            side_effect=_make_config_side_effect({CONF_MUTE_CONTROL: PLAYER_CONTROL_NONE})
308        )
309        player = _create_player(mock_mass, features={PlayerFeature.VOLUME_MUTE})
310        assert player.mute_control == PLAYER_CONTROL_NONE
311
312    def test_explicit_fake(self, mock_mass: MagicMock) -> None:
313        """Mute control returns fake when explicitly configured and a volume control exists."""
314        mock_mass.config.get_raw_player_config_value = MagicMock(
315            side_effect=_make_config_side_effect({CONF_MUTE_CONTROL: PLAYER_CONTROL_FAKE})
316        )
317        player = _create_player(mock_mass, features={PlayerFeature.VOLUME_SET})
318        assert player.mute_control == PLAYER_CONTROL_FAKE
319
320    def test_explicit_fake_degrades_without_volume_control(self, mock_mass: MagicMock) -> None:
321        """Fake mute drives the volume control, so it degrades to NONE without one."""
322        mock_mass.config.get_raw_player_config_value = MagicMock(
323            side_effect=_make_config_side_effect(
324                {
325                    CONF_MUTE_CONTROL: PLAYER_CONTROL_FAKE,
326                    CONF_VOLUME_CONTROL: PLAYER_CONTROL_NONE,
327                }
328            )
329        )
330        # a native mute capability is irrelevant here: the user explicitly picked fake
331        player = _create_player(mock_mass, features={PlayerFeature.VOLUME_MUTE})
332        assert player.mute_control == PLAYER_CONTROL_NONE
333        assert PlayerFeature.VOLUME_MUTE not in player.state.supported_features
334
335
336class TestMuteControlAutoSelect:
337    """Test mute_control auto-select logic."""
338
339    def test_auto_prefers_native(self, mock_mass: MagicMock) -> None:
340        """Auto-select returns native when player supports VOLUME_MUTE."""
341        player = _create_player(mock_mass, features={PlayerFeature.VOLUME_MUTE})
342        assert player.mute_control == PLAYER_CONTROL_NATIVE
343
344    def test_auto_falls_back_to_protocol_player(self, mock_mass: MagicMock) -> None:
345        """Auto-select returns protocol player for mute."""
346        protocol = _create_protocol_player(mock_mass, "proto_mute", {PlayerFeature.VOLUME_MUTE})
347        mock_mass.players.get_player = MagicMock(return_value=protocol)
348        player = _create_player(mock_mass)
349        _link_protocols(player, [_make_protocol_link("proto_mute")])
350        assert player.mute_control == "proto_mute"
351
352    def test_auto_returns_none_when_nothing_available(self, mock_mass: MagicMock) -> None:
353        """Auto-select returns NONE when nothing supports mute."""
354        player = _create_player(mock_mass)
355        assert player.mute_control == PLAYER_CONTROL_NONE
356
357
358class TestProtocolPlayerPreferActive:
359    """Test that auto-select prefers the active output protocol."""
360
361    def _get_player_lookup(self, players: dict[str, MockPlayer]) -> Any:
362        """Create a side_effect that looks up players by ID."""
363
364        def _lookup(pid: str) -> MockPlayer | None:
365            return players.get(pid)
366
367        return _lookup
368
369    def test_prefers_active_protocol(self, mock_mass: MagicMock) -> None:
370        """Auto-select prefers the active output protocol."""
371        proto_a = _create_protocol_player(mock_mass, "proto_a", {PlayerFeature.VOLUME_SET})
372        proto_b = _create_protocol_player(mock_mass, "proto_b", {PlayerFeature.VOLUME_SET})
373        mock_mass.players.get_player = MagicMock(
374            side_effect=self._get_player_lookup({"proto_a": proto_a, "proto_b": proto_b})
375        )
376        player = _create_player(mock_mass)
377        _link_protocols(
378            player,
379            [
380                _make_protocol_link("proto_a", priority=0),
381                _make_protocol_link("proto_b", priority=1),
382            ],
383            active_protocol_id="proto_b",
384        )
385        assert player.volume_control == "proto_b"
386
387    def test_falls_back_to_any_when_no_active(self, mock_mass: MagicMock) -> None:
388        """When no active protocol, falls back to first available."""
389        proto_a = _create_protocol_player(mock_mass, "proto_a", {PlayerFeature.VOLUME_SET})
390        proto_b = _create_protocol_player(mock_mass, "proto_b", {PlayerFeature.VOLUME_SET})
391        mock_mass.players.get_player = MagicMock(
392            side_effect=self._get_player_lookup({"proto_a": proto_a, "proto_b": proto_b})
393        )
394        player = _create_player(mock_mass)
395        _link_protocols(
396            player,
397            [
398                _make_protocol_link("proto_a", priority=0),
399                _make_protocol_link("proto_b", priority=1),
400            ],
401        )
402        assert player.volume_control == "proto_a"
403
404    def test_power_requires_active_protocol(self, mock_mass: MagicMock) -> None:
405        """Power auto-select requires active protocol; returns NONE without one."""
406        proto_a = _create_protocol_player(mock_mass, "proto_a", {PlayerFeature.POWER})
407        mock_mass.players.get_player = MagicMock(
408            side_effect=self._get_player_lookup({"proto_a": proto_a})
409        )
410        player = _create_player(mock_mass)
411        _link_protocols(
412            player,
413            [_make_protocol_link("proto_a", priority=0)],
414        )
415        # no active protocol set -> power returns NONE
416        assert player.power_control == PLAYER_CONTROL_NONE
417
418    def test_power_ignores_active_protocol(self, mock_mass: MagicMock) -> None:
419        """Power auto-select does not use protocol players even when active."""
420        proto_a = _create_protocol_player(mock_mass, "proto_a", {PlayerFeature.POWER})
421        mock_mass.players.get_player = MagicMock(
422            side_effect=self._get_player_lookup({"proto_a": proto_a})
423        )
424        player = _create_player(mock_mass)
425        _link_protocols(
426            player,
427            [_make_protocol_link("proto_a", priority=0)],
428            active_protocol_id="proto_a",
429        )
430        assert player.power_control == PLAYER_CONTROL_NONE
431
432    def test_volume_skips_unavailable_protocol_player(self, mock_mass: MagicMock) -> None:
433        """Volume auto-select skips unavailable protocol players."""
434        proto_a = _create_protocol_player(
435            mock_mass, "proto_a", {PlayerFeature.VOLUME_SET}, available=False
436        )
437        proto_b = _create_protocol_player(
438            mock_mass, "proto_b", {PlayerFeature.VOLUME_SET}, available=True
439        )
440        mock_mass.players.get_player = MagicMock(
441            side_effect=self._get_player_lookup({"proto_a": proto_a, "proto_b": proto_b})
442        )
443        player = _create_player(mock_mass)
444        _link_protocols(
445            player,
446            [
447                _make_protocol_link("proto_a", priority=0),
448                _make_protocol_link("proto_b", priority=1),
449            ],
450        )
451        assert player.volume_control == "proto_b"
452
453
454class TestControlDomainPriority:
455    """Test the protocol domain order that the ambient volume/mute fallback applies."""
456
457    def _get_player_lookup(self, players: dict[str, MockPlayer]) -> Any:
458        """Create a side_effect that looks up players by ID."""
459
460        def _lookup(pid: str) -> MockPlayer | None:
461            return players.get(pid)
462
463        return _lookup
464
465    def test_prefers_domain_that_takes_commands_while_idle(self, mock_mass: MagicMock) -> None:
466        """A domain that accepts commands while idle owns the volume."""
467        airplay = _create_protocol_player(mock_mass, "airplay_child", {PlayerFeature.VOLUME_SET})
468        cast = _create_protocol_player(mock_mass, "cast_child", {PlayerFeature.VOLUME_SET})
469        mock_mass.players.get_player = MagicMock(
470            side_effect=self._get_player_lookup({"airplay_child": airplay, "cast_child": cast})
471        )
472        player = _create_player(mock_mass)
473        # airplay links first, but only cast can take a volume command while idle
474        _link_protocols(
475            player,
476            [
477                _make_protocol_link("airplay_child", protocol_domain="airplay"),
478                _make_protocol_link("cast_child", protocol_domain="chromecast"),
479            ],
480        )
481        assert player.volume_control == "cast_child"
482
483    def test_unknown_domain_sorts_last(self, mock_mass: MagicMock) -> None:
484        """A domain outside the priority table loses from a known one."""
485        other = _create_protocol_player(mock_mass, "other_child", {PlayerFeature.VOLUME_SET})
486        sendspin = _create_protocol_player(mock_mass, "sendspin_child", {PlayerFeature.VOLUME_SET})
487        mock_mass.players.get_player = MagicMock(
488            side_effect=self._get_player_lookup({"other_child": other, "sendspin_child": sendspin})
489        )
490        player = _create_player(mock_mass)
491        _link_protocols(
492            player,
493            [
494                _make_protocol_link("other_child", protocol_domain="something_else"),
495                _make_protocol_link("sendspin_child", protocol_domain="sendspin"),
496            ],
497        )
498        assert player.volume_control == "sendspin_child"
499
500    def test_volume_and_mute_may_land_on_different_siblings(self, mock_mass: MagicMock) -> None:
501        """
502        Volume and mute resolve independently, so they can pick different interfaces.
503
504        A DLNA renderer derives VOLUME_SET and VOLUME_MUTE from separate
505        RenderingControl actions, so it may advertise one without the other. The
506        ambient controls then split across siblings; only the output-scoped
507        variants keep a resolution inside one signal path.
508        """
509        dlna = _create_protocol_player(mock_mass, "dlna_child", {PlayerFeature.VOLUME_SET})
510        airplay = _create_protocol_player(
511            mock_mass,
512            "airplay_child",
513            {PlayerFeature.VOLUME_SET, PlayerFeature.VOLUME_MUTE},
514        )
515        mock_mass.players.get_player = MagicMock(
516            side_effect=self._get_player_lookup({"dlna_child": dlna, "airplay_child": airplay})
517        )
518        player = _create_player(mock_mass)
519        _link_protocols(
520            player,
521            [
522                _make_protocol_link("dlna_child", protocol_domain="dlna"),
523                _make_protocol_link("airplay_child", protocol_domain="airplay"),
524            ],
525        )
526        # dlna outranks airplay for volume, but has no mute to offer
527        assert player.volume_control == "dlna_child"
528        assert player.mute_control == "airplay_child"
529        # the output-scoped variants stay within the named signal path
530        assert player.volume_control_for_output("dlna_child") == "dlna_child"
531        assert player.mute_control_for_output("dlna_child") == PLAYER_CONTROL_NONE
532
533
534class TestPreferredOutputProtocolAutoValue:
535    """Test that 'auto' preferred output protocol is treated as no preference."""
536
537    def _get_player_lookup(self, players: dict[str, MockPlayer]) -> Any:
538        """Create a side_effect that looks up players by ID."""
539
540        def _lookup(pid: str) -> MockPlayer | None:
541            return players.get(pid)
542
543        return _lookup
544
545    def test_preferred_auto_power_ignores_protocol(self, mock_mass: MagicMock) -> None:
546        """With preferred='auto', power control does not use protocol players."""
547        proto = _create_protocol_player(mock_mass, "proto_a", {PlayerFeature.POWER})
548        mock_mass.players.get_player = MagicMock(
549            side_effect=self._get_player_lookup({"proto_a": proto})
550        )
551        mock_mass.config.get_raw_player_config_value = MagicMock(
552            side_effect=_make_config_side_effect({CONF_PREFERRED_OUTPUT_PROTOCOL: "auto"})
553        )
554        player = _create_player(mock_mass)
555        _link_protocols(
556            player,
557            [_make_protocol_link("proto_a")],
558            active_protocol_id="proto_a",
559        )
560        assert player.power_control == PLAYER_CONTROL_NONE
561
562    def test_preferred_auto_falls_back_for_volume(self, mock_mass: MagicMock) -> None:
563        """With preferred='auto', volume control falls back to any linked protocol."""
564        proto = _create_protocol_player(mock_mass, "proto_a", {PlayerFeature.VOLUME_SET})
565        mock_mass.players.get_player = MagicMock(
566            side_effect=self._get_player_lookup({"proto_a": proto})
567        )
568        mock_mass.config.get_raw_player_config_value = MagicMock(
569            side_effect=_make_config_side_effect({CONF_PREFERRED_OUTPUT_PROTOCOL: "auto"})
570        )
571        player = _create_player(mock_mass)
572        _link_protocols(
573            player,
574            [_make_protocol_link("proto_a")],
575        )
576        # volume doesn't require active, so it falls back to linked protocol
577        assert player.volume_control == "proto_a"
578
579    def test_preferred_auto_no_active_power_returns_none(self, mock_mass: MagicMock) -> None:
580        """With preferred='auto' and no active protocol, power returns NONE."""
581        proto = _create_protocol_player(mock_mass, "proto_a", {PlayerFeature.POWER})
582        mock_mass.players.get_player = MagicMock(
583            side_effect=self._get_player_lookup({"proto_a": proto})
584        )
585        mock_mass.config.get_raw_player_config_value = MagicMock(
586            side_effect=_make_config_side_effect({CONF_PREFERRED_OUTPUT_PROTOCOL: "auto"})
587        )
588        player = _create_player(mock_mass)
589        _link_protocols(
590            player,
591            [_make_protocol_link("proto_a")],
592        )
593        assert player.power_control == PLAYER_CONTROL_NONE
594
595    def test_preferred_pointing_at_another_players_output_is_ignored(
596        self, mock_mass: MagicMock
597    ) -> None:
598        """A preference left behind by a relink does not steal another speaker's output."""
599        own = _create_protocol_player(mock_mass, "proto_a", {PlayerFeature.VOLUME_SET})
600        other = _create_protocol_player(mock_mass, "other_proto", {PlayerFeature.VOLUME_SET})
601        mock_mass.players.get_player = MagicMock(
602            side_effect=self._get_player_lookup({"proto_a": own, "other_proto": other})
603        )
604        mock_mass.config.get_raw_player_config_value = MagicMock(
605            side_effect=_make_config_side_effect({CONF_PREFERRED_OUTPUT_PROTOCOL: "other_proto"})
606        )
607        player = _create_player(mock_mass)
608        _link_protocols(player, [_make_protocol_link("proto_a")])
609        assert player.volume_control == "proto_a"
610
611
612class TestActiveOutputProtocolClearCancellation:
613    """Test that setting the active output protocol cancels a pending clear."""
614
615    def test_set_cancels_pending_clear_task(self, mock_mass: MagicMock) -> None:
616        """
617        Setting a protocol cancels the scheduled clear for the same player.
618
619        This is the mechanism that stops a new session's freshly set protocol
620        from being wiped by a clear scheduled after the previous group stop.
621        """
622        player = _create_player(mock_mass)
623
624        player.set_active_output_protocol("ap_1")
625
626        mock_mass.cancel_task.assert_called_once_with("clear_active_protocol_main_player")
627
628
629class TestControlForOutput:
630    """Test the output-scoped volume/mute control resolution."""
631
632    def _get_player_lookup(self, players: dict[str, MockPlayer]) -> Any:
633        """Create a side_effect that looks up players by ID."""
634
635        def _lookup(pid: str) -> MockPlayer | None:
636            return players.get(pid)
637
638        return _lookup
639
640    def test_resolves_named_output_without_active_protocol(self, mock_mass: MagicMock) -> None:
641        """
642        The named output owns the volume even when no protocol is marked active.
643
644        This is the ordering trap the ambient volume_control falls into: with nothing
645        active it picks a sibling interface, which made a joining AirPlay speaker
646        decide another control owned the volume and play at unity gain.
647        """
648        proto_a = _create_protocol_player(mock_mass, "proto_a", {PlayerFeature.VOLUME_SET})
649        proto_b = _create_protocol_player(mock_mass, "proto_b", {PlayerFeature.VOLUME_SET})
650        mock_mass.players.get_player = MagicMock(
651            side_effect=self._get_player_lookup({"proto_a": proto_a, "proto_b": proto_b})
652        )
653        player = _create_player(mock_mass)
654        _link_protocols(
655            player,
656            [
657                _make_protocol_link("proto_a", priority=0),
658                _make_protocol_link("proto_b", priority=1),
659            ],
660        )
661        # the ambient answer picks the first linked protocol, the output-scoped one
662        # answers for the interface that is actually about to render
663        assert player.volume_control == "proto_a"
664        assert player.volume_control_for_output("proto_b") == "proto_b"
665
666    def test_ignores_active_protocol_pointing_elsewhere(self, mock_mass: MagicMock) -> None:
667        """A protocol marked active does not override the named output."""
668        proto_a = _create_protocol_player(mock_mass, "proto_a", {PlayerFeature.VOLUME_SET})
669        proto_b = _create_protocol_player(mock_mass, "proto_b", {PlayerFeature.VOLUME_SET})
670        mock_mass.players.get_player = MagicMock(
671            side_effect=self._get_player_lookup({"proto_a": proto_a, "proto_b": proto_b})
672        )
673        player = _create_player(mock_mass)
674        _link_protocols(
675            player,
676            [
677                _make_protocol_link("proto_a", priority=0),
678                _make_protocol_link("proto_b", priority=1),
679            ],
680            active_protocol_id="proto_a",
681        )
682        assert player.volume_control_for_output("proto_b") == "proto_b"
683
684    def test_returns_none_when_output_lacks_feature(self, mock_mass: MagicMock) -> None:
685        """No fallback to a sibling: an output without the feature does not own it."""
686        proto_a = _create_protocol_player(mock_mass, "proto_a", {PlayerFeature.VOLUME_SET})
687        proto_b = _create_protocol_player(mock_mass, "proto_b", set())
688        mock_mass.players.get_player = MagicMock(
689            side_effect=self._get_player_lookup({"proto_a": proto_a, "proto_b": proto_b})
690        )
691        player = _create_player(mock_mass)
692        _link_protocols(
693            player,
694            [
695                _make_protocol_link("proto_a", priority=0),
696                _make_protocol_link("proto_b", priority=1),
697            ],
698        )
699        assert player.volume_control == "proto_a"
700        assert player.volume_control_for_output("proto_b") == PLAYER_CONTROL_NONE
701
702    def test_prefers_native(self, mock_mass: MagicMock) -> None:
703        """A player with native volume owns it regardless of the output."""
704        proto_a = _create_protocol_player(mock_mass, "proto_a", {PlayerFeature.VOLUME_SET})
705        mock_mass.players.get_player = MagicMock(
706            side_effect=self._get_player_lookup({"proto_a": proto_a})
707        )
708        player = _create_player(mock_mass, features={PlayerFeature.VOLUME_SET})
709        _link_protocols(player, [_make_protocol_link("proto_a")])
710        assert player.volume_control_for_output("proto_a") == PLAYER_CONTROL_NATIVE
711
712    def test_explicit_config_wins(self, mock_mass: MagicMock) -> None:
713        """An explicitly configured control is a device-wide statement, so it still wins."""
714        proto_a = _create_protocol_player(mock_mass, "proto_a", {PlayerFeature.VOLUME_SET})
715        mock_mass.players.get_player = MagicMock(
716            side_effect=self._get_player_lookup({"proto_a": proto_a})
717        )
718        mock_mass.config.get_raw_player_config_value = MagicMock(
719            side_effect=_make_config_side_effect({CONF_VOLUME_CONTROL: PLAYER_CONTROL_FAKE})
720        )
721        player = _create_player(mock_mass)
722        _link_protocols(player, [_make_protocol_link("proto_a")])
723        assert player.volume_control_for_output("proto_a") == PLAYER_CONTROL_FAKE
724
725    def test_unknown_output_returns_none(self, mock_mass: MagicMock) -> None:
726        """An output that cannot be resolved owns nothing."""
727        player = _create_player(mock_mass)
728        assert player.volume_control_for_output("nope") == PLAYER_CONTROL_NONE
729
730    def test_mute_resolves_named_output(self, mock_mass: MagicMock) -> None:
731        """Mute resolves against the named output the same way volume does."""
732        proto_a = _create_protocol_player(mock_mass, "proto_a", {PlayerFeature.VOLUME_MUTE})
733        proto_b = _create_protocol_player(mock_mass, "proto_b", {PlayerFeature.VOLUME_MUTE})
734        mock_mass.players.get_player = MagicMock(
735            side_effect=self._get_player_lookup({"proto_a": proto_a, "proto_b": proto_b})
736        )
737        player = _create_player(mock_mass)
738        _link_protocols(
739            player,
740            [
741                _make_protocol_link("proto_a", priority=0),
742                _make_protocol_link("proto_b", priority=1),
743            ],
744        )
745        assert player.mute_control == "proto_a"
746        assert player.mute_control_for_output("proto_b") == "proto_b"
747
748    def test_mute_returns_none_when_output_lacks_feature(self, mock_mass: MagicMock) -> None:
749        """An output with volume but no mute does not own the mute."""
750        proto_a = _create_protocol_player(mock_mass, "proto_a", {PlayerFeature.VOLUME_SET})
751        mock_mass.players.get_player = MagicMock(
752            side_effect=self._get_player_lookup({"proto_a": proto_a})
753        )
754        player = _create_player(mock_mass)
755        _link_protocols(player, [_make_protocol_link("proto_a")])
756        assert player.mute_control_for_output("proto_a") == PLAYER_CONTROL_NONE
757
758    def test_explicit_player_id_config_beats_named_output(self, mock_mass: MagicMock) -> None:
759        """An explicitly configured control outranks the output carrying the audio."""
760        proto_a = _create_protocol_player(mock_mass, "proto_a", {PlayerFeature.VOLUME_SET})
761        proto_b = _create_protocol_player(mock_mass, "proto_b", {PlayerFeature.VOLUME_SET})
762        mock_mass.players.get_player = MagicMock(
763            side_effect=self._get_player_lookup({"proto_a": proto_a, "proto_b": proto_b})
764        )
765        mock_mass.config.get_raw_player_config_value = MagicMock(
766            side_effect=_make_config_side_effect({CONF_VOLUME_CONTROL: "proto_a"})
767        )
768        player = _create_player(mock_mass)
769        _link_protocols(
770            player,
771            [
772                _make_protocol_link("proto_a", priority=0),
773                _make_protocol_link("proto_b", priority=1),
774            ],
775        )
776        assert player.volume_control_for_output("proto_b") == "proto_a"
777
778    def test_does_not_require_output_to_be_available(self, mock_mass: MagicMock) -> None:
779        """
780        An unavailable reading does not hand the volume to a different interface.
781
782        The caller names the output it is about to stream to, so a stale registry
783        reading is no reason to fall back to a sibling that isn't in the signal path.
784        """
785        proto_a = _create_protocol_player(mock_mass, "proto_a", {PlayerFeature.VOLUME_SET})
786        proto_b = _create_protocol_player(
787            mock_mass, "proto_b", {PlayerFeature.VOLUME_SET}, available=False
788        )
789        mock_mass.players.get_player = MagicMock(
790            side_effect=self._get_player_lookup({"proto_a": proto_a, "proto_b": proto_b})
791        )
792        player = _create_player(mock_mass)
793        _link_protocols(
794            player,
795            [
796                _make_protocol_link("proto_a", priority=0),
797                _make_protocol_link("proto_b", priority=1),
798            ],
799        )
800        assert player.volume_control == "proto_a"
801        assert player.volume_control_for_output("proto_b") == "proto_b"
802
803    def test_fake_mute_degrades_without_volume_control_on_output(
804        self, mock_mass: MagicMock
805    ) -> None:
806        """Fake mute drives the volume to zero, so it needs a volume control to drive."""
807        proto_a = _create_protocol_player(mock_mass, "proto_a", set())
808        mock_mass.players.get_player = MagicMock(
809            side_effect=self._get_player_lookup({"proto_a": proto_a})
810        )
811        mock_mass.config.get_raw_player_config_value = MagicMock(
812            side_effect=_make_config_side_effect({CONF_MUTE_CONTROL: PLAYER_CONTROL_FAKE})
813        )
814        player = _create_player(mock_mass)
815        _link_protocols(player, [_make_protocol_link("proto_a")])
816        assert player.volume_control_for_output("proto_a") == PLAYER_CONTROL_NONE
817        assert player.mute_control_for_output("proto_a") == PLAYER_CONTROL_NONE
818