music-assistant-server

1"""
2Tests for Player.playback_domains.
3
4Covers the live availability view: native playback, linked protocol players that
5are (un)available, and wrapper players that only contribute their linked protocols.
6"""
7
8from __future__ import annotations
9
10from unittest.mock import MagicMock
11
12import pytest
13from music_assistant_models.enums import PlayerFeature, PlayerType
14from music_assistant_models.player import OutputProtocol
15
16from tests.common import MockPlayer, MockProvider
17
18
19@pytest.fixture
20def mock_mass() -> MagicMock:
21    """Create a mock MusicAssistant instance."""
22    mass = MagicMock()
23    mass.closing = False
24    mass.config.get = MagicMock(return_value=[])
25    mass.config.get_raw_player_config_value = MagicMock(return_value=None)
26    mass.config.get_raw_core_config_value = MagicMock(return_value="GLOBAL")
27    mass.config.set = MagicMock()
28    mass.players.get_player = MagicMock(return_value=None)
29    return mass
30
31
32def _make_native_player(mock_mass: MagicMock, domain: str = "sonos") -> MockPlayer:
33    """Create a native player (own playback path) of the given provider domain."""
34    provider = MockProvider(domain, instance_id=domain, mass=mock_mass)
35    player = MockPlayer(provider, "parent", "Parent")
36    player._attr_supported_features = {PlayerFeature.PLAY_MEDIA}
37    player._cache.clear()
38    return player
39
40
41def _link_protocol(
42    mock_mass: MagicMock,
43    parent: MockPlayer,
44    protocol_id: str,
45    domain: str,
46    available: bool,
47    needs_setup: bool = False,
48) -> MockPlayer:
49    """Link a protocol player of the given domain to the parent and register it."""
50    provider = MockProvider(domain, instance_id=domain, mass=mock_mass)
51    protocol_player = MockPlayer(provider, protocol_id, domain, player_type=PlayerType.PROTOCOL)
52    protocol_player._attr_available = available
53    protocol_player._attr_needs_setup = needs_setup
54    protocol_player._cache.clear()
55    parent.set_linked_output_protocols(
56        [
57            *parent.linked_output_protocols,
58            # no `available` argument: production links protocols without one, so the
59            # flag stays at its dataclass default of True forever
60            OutputProtocol(
61                output_protocol_id=protocol_id,
62                name=domain,
63                protocol_domain=domain,
64                priority=10,
65            ),
66        ]
67    )
68    registered = {protocol_player.player_id: protocol_player}
69    mock_mass.players.get_player = MagicMock(side_effect=lambda pid: registered.get(pid))
70    parent._cache.clear()
71    return protocol_player
72
73
74class TestPlaybackDomains:
75    """Tests for Player.playback_domains."""
76
77    def test_native_player_reports_own_domain(self, mock_mass: MagicMock) -> None:
78        """Test that a native player reports its own provider domain."""
79        player = _make_native_player(mock_mass)
80        assert player.playback_domains == {"sonos"}
81
82    def test_unavailable_native_player_reports_nothing(self, mock_mass: MagicMock) -> None:
83        """Test that an offline native player no longer offers its own domain."""
84        player = _make_native_player(mock_mass)
85        player._attr_available = False
86        player._cache.clear()
87        assert player.playback_domains == set()
88
89    def test_native_player_needing_setup_reports_nothing(self, mock_mass: MagicMock) -> None:
90        """Test that a player that still needs setup cannot accept a stream."""
91        player = _make_native_player(mock_mass)
92        player._attr_needs_setup = True
93        player._cache.clear()
94        assert player.playback_domains == set()
95
96    def test_linked_protocol_adds_its_domain(self, mock_mass: MagicMock) -> None:
97        """Test that an available linked protocol adds its domain."""
98        player = _make_native_player(mock_mass)
99        _link_protocol(mock_mass, player, "ap_1", "airplay", available=True)
100        assert player.playback_domains == {"sonos", "airplay"}
101
102    def test_offline_linked_protocol_is_left_out(self, mock_mass: MagicMock) -> None:
103        """Test that a linked protocol whose player went offline is not reported."""
104        player = _make_native_player(mock_mass)
105        _link_protocol(mock_mass, player, "ap_1", "airplay", available=False)
106        # the link itself still claims to be available; only the live player counts
107        assert player.linked_output_protocols[0].available is True
108        assert player.playback_domains == {"sonos"}
109
110    def test_linked_protocol_needing_setup_is_left_out(self, mock_mass: MagicMock) -> None:
111        """Test that an unpaired protocol receiver cannot accept a stream."""
112        player = _make_native_player(mock_mass)
113        _link_protocol(mock_mass, player, "ap_1", "airplay", available=True, needs_setup=True)
114        assert player.playback_domains == {"sonos"}
115
116    def test_unregistered_linked_protocol_is_left_out(self, mock_mass: MagicMock) -> None:
117        """Test that a linked protocol without a live player is not reported."""
118        player = _make_native_player(mock_mass)
119        _link_protocol(mock_mass, player, "ap_1", "airplay", available=True)
120        mock_mass.players.get_player = MagicMock(return_value=None)
121        player._cache.clear()
122        assert player.playback_domains == {"sonos"}
123
124    def test_wrapper_player_only_reports_linked_protocols(self, mock_mass: MagicMock) -> None:
125        """Test that a UniversalPlayer never contributes its own domain."""
126        provider = MockProvider("universal_player", instance_id="universal_player", mass=mock_mass)
127        player = MockPlayer(provider, "up_1", "Universal")
128        player._attr_supported_features = {PlayerFeature.PLAY_MEDIA}
129        player._cache.clear()
130        _link_protocol(mock_mass, player, "ap_1", "airplay", available=True)
131        assert player.playback_domains == {"airplay"}
132
133    def test_protocol_endpoint_reports_own_domain(self, mock_mass: MagicMock) -> None:
134        """Test that a player that is itself a protocol endpoint reports its domain."""
135        provider = MockProvider("airplay", instance_id="airplay", mass=mock_mass)
136        player = MockPlayer(provider, "ap_1", "AirPlay", player_type=PlayerType.PROTOCOL)
137        player._attr_supported_features = {PlayerFeature.SET_MEMBERS}
138        player._cache.clear()
139        assert player.playback_domains == {"airplay"}
140
141        player._attr_available = False
142        player._cache.clear()
143        assert player.playback_domains == set()
144
145    def test_recomputed_after_state_update(self, mock_mass: MagicMock) -> None:
146        """Test that the cached value follows the protocol player going offline."""
147        player = _make_native_player(mock_mass)
148        protocol_player = _link_protocol(mock_mass, player, "ap_1", "airplay", available=True)
149        assert player.playback_domains == {"sonos", "airplay"}
150
151        protocol_player._attr_available = False
152        player.update_state(signal_event=False)
153        assert player.playback_domains == {"sonos"}
154