/
/
/
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.audio_processing import ActiveSourceAudioDetails
15from music_assistant_models.enums import ContentType, PlaybackState, PlayerType
16from music_assistant_models.media_items import AudioFormat
17
18from music_assistant.controllers.players import PlayerController
19from tests.common import MockPlayer, MockProvider
20
21
22@pytest.fixture
23def mock_mass() -> MagicMock:
24 """Create a mock MusicAssistant instance."""
25 mass = MagicMock()
26 mass.closing = False
27 mass.loop = None
28 mass.config = MagicMock()
29 mass.config.get = MagicMock(return_value=[])
30 mass.config.get_raw_player_config_value = MagicMock(
31 side_effect=lambda _player_id, _key, default=None: default
32 )
33 mass.config.get_raw_core_config_value = MagicMock(return_value="GLOBAL")
34 mass.config.set = MagicMock()
35 mass.signal_event = MagicMock()
36 mass.get_providers = MagicMock(return_value=[])
37 mass.player_queues = MagicMock()
38 mass.player_queues.get = MagicMock(return_value=None)
39 return mass
40
41
42@pytest.fixture
43def controller(mock_mass: MagicMock) -> PlayerController:
44 """Create a PlayerController instance."""
45 ctrl = PlayerController(mock_mass)
46 mock_mass.players = ctrl
47 return ctrl
48
49
50class TestFinalPlaybackStateWithActiveProtocol:
51 """When an active output protocol is set the protocol is the source of truth."""
52
53 def test_playing_protocol_propagates_playing(
54 self,
55 provider: MockProvider,
56 controller: PlayerController,
57 ) -> None:
58 """A PLAYING protocol player makes the parent report PLAYING."""
59 protocol_player = MockPlayer(provider, "ap_1", "AirPlay", player_type=PlayerType.PROTOCOL)
60 protocol_player._attr_playback_state = PlaybackState.PLAYING
61 protocol_player._attr_elapsed_time = 20.0
62 protocol_player._attr_elapsed_time_last_updated = time.time()
63
64 player = MockPlayer(provider, "player_1", "Test Player")
65 player._attr_playback_state = PlaybackState.IDLE # native is idle
66 player.set_active_output_protocol("ap_1")
67
68 controller._players = {"player_1": player, "ap_1": protocol_player}
69
70 protocol_player.update_state(signal_event=False)
71 # refresh: the protocol player is cross-player state and the production
72 # fan-out (which marks the parent dirty) is suppressed in this test
73 player.refresh_state(signal_event=False)
74
75 assert player.state.playback_state == PlaybackState.PLAYING
76
77 def test_idle_protocol_propagates_idle(
78 self,
79 provider: MockProvider,
80 controller: PlayerController,
81 ) -> None:
82 """
83 An IDLE protocol player makes the parent report IDLE (regression guard).
84
85 Previously the parent would fall through to the parent/group state when
86 the protocol was IDLE, which could create a circular state inheritance
87 in sync groups and leave the group stuck in PLAYING forever. The
88 protocol player is now the source of truth regardless of its state.
89 """
90 protocol_player = MockPlayer(provider, "ap_1", "AirPlay", player_type=PlayerType.PROTOCOL)
91 protocol_player._attr_playback_state = PlaybackState.IDLE
92 protocol_player._attr_elapsed_time = 0.0
93 protocol_player._attr_elapsed_time_last_updated = time.time()
94
95 player = MockPlayer(provider, "player_1", "Test Player")
96 # Native player still reports PLAYING (e.g. a Sonos that's outputting
97 # someone else's AirPlay audio) â but the active output protocol is
98 # the live audio source and it has gone IDLE, which is what matters.
99 player._attr_playback_state = PlaybackState.PLAYING
100 player.set_active_output_protocol("ap_1")
101
102 controller._players = {"player_1": player, "ap_1": protocol_player}
103
104 protocol_player.update_state(signal_event=False)
105 player.refresh_state(signal_event=False)
106
107 assert player.state.playback_state == PlaybackState.IDLE
108
109 def test_protocol_player_inherits_parent_source_audio(
110 self,
111 provider: MockProvider,
112 controller: PlayerController,
113 ) -> None:
114 """A protocol player publishes the live source audio details of its parent."""
115 details = ActiveSourceAudioDetails(
116 input_format=AudioFormat(
117 content_type=ContentType.FLAC,
118 sample_rate=44100,
119 bit_depth=16,
120 channels=2,
121 )
122 )
123 parent = MockPlayer(provider, "player_1", "Player")
124 parent._state.active_source_audio = details
125 protocol_player = MockPlayer(provider, "ap_1", "AirPlay", player_type=PlayerType.PROTOCOL)
126 controller._players = {"player_1": parent, "ap_1": protocol_player}
127 protocol_player.set_protocol_parent_id("player_1")
128
129 protocol_player.refresh_state(signal_event=False)
130
131 assert protocol_player.state.active_source_audio == details
132
133 def test_no_active_protocol_uses_native_state(
134 self,
135 provider: MockProvider,
136 controller: PlayerController,
137 ) -> None:
138 """Without an active output protocol, fall through to the native state."""
139 player = MockPlayer(provider, "player_1", "Test Player")
140 player._attr_playback_state = PlaybackState.PLAYING
141 player._attr_elapsed_time = 5.0
142 player._attr_elapsed_time_last_updated = time.time()
143
144 controller._players = {"player_1": player}
145
146 player.update_state(signal_event=False)
147
148 assert player.state.playback_state == PlaybackState.PLAYING
149
150 def test_negative_elapsed_time_is_clamped(
151 self,
152 provider: MockProvider,
153 controller: PlayerController,
154 ) -> None:
155 """A negative provider elapsed time is clamped at the elapsed_time accessor."""
156 player = MockPlayer(provider, "player_1", "Test Player")
157 player._attr_playback_state = PlaybackState.PAUSED
158 player._attr_elapsed_time = -1.0
159 player._attr_elapsed_time_last_updated = time.time()
160
161 controller._players = {"player_1": player}
162
163 player.update_state(signal_event=False)
164
165 assert player.elapsed_time == 0
166 assert player.corrected_elapsed_time == 0
167 assert player.state.elapsed_time == 0
168
169
170def _source_session(elapsed: float | None, updated: float | None, player_id: str) -> MagicMock:
171 """Build a mock session whose source reports the given clock."""
172 stream_metadata = MagicMock()
173 stream_metadata.elapsed_time = elapsed
174 stream_metadata.elapsed_time_last_updated = updated
175 session = MagicMock()
176 session.player_id = player_id
177 session.stream_metadata = stream_metadata
178 session.source_uri = f"test_instance://audio_source/{player_id}"
179 session.source.image = None
180 return session
181
182
183class TestFinalPlaybackStateAudioSourceElapsed:
184 """The source-clock override for a live external source playing on a player."""
185
186 def test_standalone_uses_its_sources_clock(
187 self,
188 provider: MockProvider,
189 controller: PlayerController,
190 mock_mass: MagicMock,
191 ) -> None:
192 """A standalone player reports the position its live source reports."""
193 player = MockPlayer(provider, "player_1", "Player")
194 player._attr_playback_state = PlaybackState.PLAYING
195 player._attr_active_source = "player_1"
196 player._attr_elapsed_time = 0.0
197 player._attr_elapsed_time_last_updated = time.time()
198
199 session = _source_session(42.0, time.time(), "player_1")
200 mock_mass.players.get_audio_source_session = MagicMock(
201 side_effect=lambda pid: {"player_1": session}.get(pid)
202 )
203 controller._players = {"player_1": player}
204
205 player.update_state(signal_event=False)
206
207 assert player.state.elapsed_time == 42.0
208
209 def test_group_uses_its_sources_clock_not_the_leaders(
210 self,
211 provider: MockProvider,
212 controller: PlayerController,
213 mock_mass: MagicMock,
214 ) -> None:
215 """
216 A group reports its own source's position, not the clock copied from its leader.
217
218 The session belongs to the group that owns the source, so it is found directly
219 on the group and the leader never has to be consulted.
220 """
221 group = MockPlayer(provider, "group_1", "Group", player_type=PlayerType.GROUP)
222 group._attr_playback_state = PlaybackState.PLAYING
223 group._attr_active_source = "leader_1"
224 # stale leader byte clock that gets copied into the group
225 group._attr_elapsed_time = 999.0
226 group._attr_elapsed_time_last_updated = time.time()
227
228 session = _source_session(42.0, time.time(), "group_1")
229 mock_mass.players.get_audio_source_session = MagicMock(
230 side_effect=lambda pid: {"group_1": session}.get(pid)
231 )
232 controller._players = {"group_1": group}
233
234 group.update_state(signal_event=False)
235
236 assert group.state.elapsed_time == 42.0
237
238 def test_a_player_without_a_source_keeps_its_own_clock(
239 self,
240 provider: MockProvider,
241 controller: PlayerController,
242 mock_mass: MagicMock,
243 ) -> None:
244 """A player with no live source of its own is not given another player's position."""
245 player = MockPlayer(provider, "player_1", "Player")
246 player._attr_playback_state = PlaybackState.PLAYING
247 player._attr_active_source = "other_src"
248 player._attr_elapsed_time = 5.0
249 player._attr_elapsed_time_last_updated = time.time()
250
251 elsewhere = _source_session(42.0, time.time(), "other_player")
252 mock_mass.players.get_audio_source_session = MagicMock(
253 side_effect=lambda pid: {"other_player": elsewhere}.get(pid)
254 )
255 controller._players = {"player_1": player}
256
257 player.update_state(signal_event=False)
258
259 assert player.state.elapsed_time == 5.0
260