music-assistant-server

1"""
2Tests for __final_playback_state when an output protocol is active.
3
4Covers the propagation of the protocol player's state (including IDLE) up to
5the parent player that has it set as ``active_output_protocol``.
6"""
7
8from __future__ import annotations
9
10import time
11from unittest.mock import MagicMock
12
13import pytest
14from music_assistant_models.enums import MediaType, PlaybackState, PlayerType
15
16from music_assistant.controllers.players import PlayerController
17from tests.common import MockPlayer, MockProvider
18
19
20@pytest.fixture
21def mock_mass() -> MagicMock:
22    """Create a mock MusicAssistant instance."""
23    mass = MagicMock()
24    mass.closing = False
25    mass.loop = None
26    mass.config = MagicMock()
27    mass.config.get = MagicMock(return_value=[])
28    mass.config.get_raw_player_config_value = MagicMock(
29        side_effect=lambda _player_id, _key, default=None: default
30    )
31    mass.config.get_raw_core_config_value = MagicMock(return_value="GLOBAL")
32    mass.config.set = MagicMock()
33    mass.signal_event = MagicMock()
34    mass.get_providers = MagicMock(return_value=[])
35    mass.player_queues = MagicMock()
36    mass.player_queues.get = MagicMock(return_value=None)
37    return mass
38
39
40@pytest.fixture
41def controller(mock_mass: MagicMock) -> PlayerController:
42    """Create a PlayerController instance."""
43    ctrl = PlayerController(mock_mass)
44    mock_mass.players = ctrl
45    return ctrl
46
47
48class TestFinalPlaybackStateWithActiveProtocol:
49    """When an active output protocol is set the protocol is the source of truth."""
50
51    def test_playing_protocol_propagates_playing(
52        self,
53        provider: MockProvider,
54        controller: PlayerController,
55    ) -> None:
56        """A PLAYING protocol player makes the parent report PLAYING."""
57        protocol_player = MockPlayer(provider, "ap_1", "AirPlay", player_type=PlayerType.PROTOCOL)
58        protocol_player._attr_playback_state = PlaybackState.PLAYING
59        protocol_player._attr_elapsed_time = 20.0
60        protocol_player._attr_elapsed_time_last_updated = time.time()
61
62        player = MockPlayer(provider, "player_1", "Test Player")
63        player._attr_playback_state = PlaybackState.IDLE  # native is idle
64        player.set_active_output_protocol("ap_1")
65
66        controller._players = {"player_1": player, "ap_1": protocol_player}
67
68        protocol_player.update_state(signal_event=False)
69        # refresh: the protocol player is cross-player state and the production
70        # fan-out (which marks the parent dirty) is suppressed in this test
71        player.refresh_state(signal_event=False)
72
73        assert player.state.playback_state == PlaybackState.PLAYING
74
75    def test_idle_protocol_propagates_idle(
76        self,
77        provider: MockProvider,
78        controller: PlayerController,
79    ) -> None:
80        """
81        An IDLE protocol player makes the parent report IDLE (regression guard).
82
83        Previously the parent would fall through to the parent/group state when
84        the protocol was IDLE, which could create a circular state inheritance
85        in sync groups and leave the group stuck in PLAYING forever. The
86        protocol player is now the source of truth regardless of its state.
87        """
88        protocol_player = MockPlayer(provider, "ap_1", "AirPlay", player_type=PlayerType.PROTOCOL)
89        protocol_player._attr_playback_state = PlaybackState.IDLE
90        protocol_player._attr_elapsed_time = 0.0
91        protocol_player._attr_elapsed_time_last_updated = time.time()
92
93        player = MockPlayer(provider, "player_1", "Test Player")
94        # Native player still reports PLAYING (e.g. a Sonos that's outputting
95        # someone else's AirPlay audio) — but the active output protocol is
96        # the live audio source and it has gone IDLE, which is what matters.
97        player._attr_playback_state = PlaybackState.PLAYING
98        player.set_active_output_protocol("ap_1")
99
100        controller._players = {"player_1": player, "ap_1": protocol_player}
101
102        protocol_player.update_state(signal_event=False)
103        player.refresh_state(signal_event=False)
104
105        assert player.state.playback_state == PlaybackState.IDLE
106
107    def test_no_active_protocol_uses_native_state(
108        self,
109        provider: MockProvider,
110        controller: PlayerController,
111    ) -> None:
112        """Without an active output protocol, fall through to the native state."""
113        player = MockPlayer(provider, "player_1", "Test Player")
114        player._attr_playback_state = PlaybackState.PLAYING
115        player._attr_elapsed_time = 5.0
116        player._attr_elapsed_time_last_updated = time.time()
117
118        controller._players = {"player_1": player}
119
120        player.update_state(signal_event=False)
121
122        assert player.state.playback_state == PlaybackState.PLAYING
123
124    def test_negative_elapsed_time_is_clamped(
125        self,
126        provider: MockProvider,
127        controller: PlayerController,
128    ) -> None:
129        """A negative provider elapsed time is clamped at the elapsed_time accessor."""
130        player = MockPlayer(provider, "player_1", "Test Player")
131        player._attr_playback_state = PlaybackState.PAUSED
132        player._attr_elapsed_time = -1.0
133        player._attr_elapsed_time_last_updated = time.time()
134
135        controller._players = {"player_1": player}
136
137        player.update_state(signal_event=False)
138
139        assert player.elapsed_time == 0
140        assert player.corrected_elapsed_time == 0
141        assert player.state.elapsed_time == 0
142
143
144def _audio_source_queue(elapsed: float, updated: float) -> MagicMock:
145    """Build a mock queue whose current item is an AudioSource carrying a source clock."""
146    stream_metadata = MagicMock()
147    stream_metadata.elapsed_time = elapsed
148    stream_metadata.elapsed_time_last_updated = updated
149    streamdetails = MagicMock()
150    streamdetails.media_type = MediaType.AUDIO_SOURCE
151    streamdetails.stream_metadata = stream_metadata
152    current_item = MagicMock()
153    current_item.streamdetails = streamdetails
154    queue = MagicMock()
155    queue.current_item = current_item
156    return queue
157
158
159def _empty_queue() -> MagicMock:
160    """Build a mock queue with no current item (no AudioSource source clock)."""
161    queue = MagicMock()
162    queue.current_item = None
163    return queue
164
165
166class TestFinalPlaybackStateAudioSourceElapsed:
167    """The AudioSource source-clock override, including the group own-queue fallback."""
168
169    def test_standalone_uses_active_source_audio_source_elapsed(
170        self,
171        provider: MockProvider,
172        controller: PlayerController,
173        mock_mass: MagicMock,
174    ) -> None:
175        """A standalone player reports the AudioSource source clock from its active queue."""
176        player = MockPlayer(provider, "player_1", "Player")
177        player._attr_playback_state = PlaybackState.PLAYING
178        player._attr_active_source = "player_1"
179        player._attr_elapsed_time = 0.0
180        player._attr_elapsed_time_last_updated = time.time()
181
182        own_queue = _audio_source_queue(42.0, time.time())
183        mock_mass.player_queues.get = MagicMock(
184            side_effect=lambda qid: {"player_1": own_queue}.get(qid)
185        )
186        controller._players = {"player_1": player}
187
188        player.update_state(signal_event=False)
189
190        assert player.state.elapsed_time == 42.0
191
192    def test_group_falls_back_to_own_queue_audio_source_elapsed(
193        self,
194        provider: MockProvider,
195        controller: PlayerController,
196        mock_mass: MagicMock,
197    ) -> None:
198        """
199        A group reports the AudioSource source clock from its own queue.
200
201        The group's active source resolves to the sync leader, whose queue does not
202        carry the AudioSource, so without the own-queue fallback the group would report
203        the leader's byte clock instead of the source's logical position.
204        """
205        group = MockPlayer(provider, "group_1", "Group", player_type=PlayerType.GROUP)
206        group._attr_playback_state = PlaybackState.PLAYING
207        group._attr_active_source = "leader_1"
208        # stale leader byte clock that gets copied into the group
209        group._attr_elapsed_time = 999.0
210        group._attr_elapsed_time_last_updated = time.time()
211
212        leader_queue = _empty_queue()
213        own_queue = _audio_source_queue(42.0, time.time())
214        mock_mass.player_queues.get = MagicMock(
215            side_effect=lambda qid: {"leader_1": leader_queue, "group_1": own_queue}.get(qid)
216        )
217        controller._players = {"group_1": group}
218
219        group.update_state(signal_event=False)
220
221        assert group.state.elapsed_time == 42.0
222
223    def test_non_group_does_not_use_own_queue_fallback(
224        self,
225        provider: MockProvider,
226        controller: PlayerController,
227        mock_mass: MagicMock,
228    ) -> None:
229        """A non-group player ignores an AudioSource sitting on its own (inactive) queue."""
230        player = MockPlayer(provider, "player_1", "Player")
231        player._attr_playback_state = PlaybackState.PLAYING
232        player._attr_active_source = "other_src"
233        player._attr_elapsed_time = 5.0
234        player._attr_elapsed_time_last_updated = time.time()
235
236        other_queue = _empty_queue()
237        own_queue = _audio_source_queue(42.0, time.time())
238        mock_mass.player_queues.get = MagicMock(
239            side_effect=lambda qid: {"other_src": other_queue, "player_1": own_queue}.get(qid)
240        )
241        controller._players = {"player_1": player}
242
243        player.update_state(signal_event=False)
244
245        assert player.state.elapsed_time == 5.0
246