music-assistant-server

1"""Tests for the Player.update_state change-detection path."""
2
3from __future__ import annotations
4
5import copy
6import time
7from statistics import median
8from unittest.mock import MagicMock, patch
9
10import pytest
11from music_assistant_models.enums import MediaType, PlaybackState
12from music_assistant_models.media_items import AudioFormat
13from music_assistant_models.player_queue import PlayerQueue
14from music_assistant_models.queue_item import QueueItem
15from music_assistant_models.streamdetails import StreamDetails, StreamMetadata
16
17import music_assistant.models.player as player_module
18from tests.common import MockPlayer, MockProvider
19
20
21@pytest.fixture
22def mock_mass() -> MagicMock:
23    """Create a mock MusicAssistant instance."""
24    mass = MagicMock()
25    mass.closing = False
26    mass.config.get_raw_player_config_value = MagicMock(
27        side_effect=lambda _player_id, _key, default=None: default
28    )
29    # no queue registered: current_media resolves from the player's native media
30    mass.player_queues.get = MagicMock(return_value=None)
31    mass.players.scale_volume_from_device = MagicMock(side_effect=lambda _player_id, volume: volume)
32    return mass
33
34
35@pytest.fixture
36def player(mock_mass: MagicMock) -> MockPlayer:
37    """Create a playing player with its state calculated once."""
38    provider = MockProvider("test_provider", mass=mock_mass)
39    player = MockPlayer(provider, "player_1", "Player 1")
40    now = time.time()
41    player._attr_playback_state = PlaybackState.PLAYING
42    player._attr_volume_level = 20
43    player._attr_elapsed_time = 10.0
44    player._attr_elapsed_time_last_updated = now
45    player.set_current_media(uri="http://test/stream", title="Test")
46    player.update_state(signal_event=False)
47    return player
48
49
50class TestUpdateStateChangeDetection:
51    """The no-change update_state path does no work; changed inputs recalculate."""
52
53    def test_no_change_update_does_no_work(self, player: MockPlayer) -> None:
54        """An update_state call without any changed input builds nothing."""
55        with (
56            patch.object(
57                player_module, "PlayerState", wraps=player_module.PlayerState
58            ) as state_cls,
59            patch.object(copy, "deepcopy", wraps=copy.deepcopy) as deepcopy_mock,
60        ):
61            player.update_state()
62
63        state_cls.assert_not_called()
64        deepcopy_mock.assert_not_called()
65
66    def test_position_tick_does_not_rebuild(self, player: MockPlayer) -> None:
67        """A regular playback tick keeps the anchor and builds nothing."""
68        assert player._attr_elapsed_time is not None
69        assert player._attr_elapsed_time_last_updated is not None
70        player._attr_elapsed_time += 1
71        player._attr_elapsed_time_last_updated += 1
72
73        with patch.object(
74            player_module, "PlayerState", wraps=player_module.PlayerState
75        ) as state_cls:
76            player.update_state()
77
78        state_cls.assert_not_called()
79        assert player.state.elapsed_time == 10.0
80
81    def test_position_jump_rebuilds_state(self, player: MockPlayer) -> None:
82        """A corrected-position jump (seek) adopts the new anchor."""
83        player._attr_elapsed_time = 61.0
84        player._attr_elapsed_time_last_updated = time.time()
85
86        player.update_state(signal_event=False)
87
88        assert player.state.elapsed_time == 61.0
89
90    def test_changed_input_rebuilds_state(self, player: MockPlayer) -> None:
91        """A changed player attribute is picked up by the next update_state call."""
92        player._attr_volume_level = 55
93
94        with patch.object(
95            player_module, "PlayerState", wraps=player_module.PlayerState
96        ) as state_cls:
97            player.update_state(signal_event=False)
98
99        state_cls.assert_called_once()
100        assert player.state.volume_level == 55
101
102    def test_mark_state_dirty_forces_recalculation(self, player: MockPlayer) -> None:
103        """mark_state_dirty recalculates even when no own input changed."""
104        with patch.object(
105            player_module, "PlayerState", wraps=player_module.PlayerState
106        ) as state_cls:
107            player.mark_state_dirty()
108            player.update_state(signal_event=False)
109
110        state_cls.assert_called_once()
111
112
113class TestNativeCurrentMediaPosition:
114    """The published position of native current_media and the timestamp it is paired with."""
115
116    def _playing_player(
117        self,
118        mock_mass: MagicMock,
119        *,
120        media_elapsed_time: int | None,
121        media_last_updated: float | None,
122        player_elapsed_time: float | None,
123        player_last_updated: float | None,
124    ) -> MockPlayer:
125        """Build a playing player with the given media and player level position anchors."""
126        provider = MockProvider("test_provider", mass=mock_mass)
127        player = MockPlayer(provider, "player_1", "Player 1")
128        player._attr_playback_state = PlaybackState.PLAYING
129        player._attr_elapsed_time = player_elapsed_time
130        player._attr_elapsed_time_last_updated = player_last_updated
131        player.set_current_media(uri="http://test/stream", title="Test")
132        assert player._attr_current_media is not None
133        player._attr_current_media.elapsed_time = media_elapsed_time
134        player._attr_current_media.elapsed_time_last_updated = media_last_updated
135        player.update_state(signal_event=False)
136        return player
137
138    def test_zero_position_is_reported(self, mock_mass: MagicMock) -> None:
139        """A position of zero (start of a track) is reported instead of being dropped."""
140        anchor = time.time()
141        player = self._playing_player(
142            mock_mass,
143            media_elapsed_time=0,
144            media_last_updated=anchor,
145            player_elapsed_time=0,
146            player_last_updated=anchor,
147        )
148
149        assert player.state.current_media is not None
150        assert player.state.current_media.elapsed_time == 0
151        assert player.state.current_media.elapsed_time_last_updated == anchor
152
153    def test_media_position_wins_without_player_position(self, mock_mass: MagicMock) -> None:
154        """A media position is reported even when the player reports none of its own."""
155        anchor = time.time()
156        player = self._playing_player(
157            mock_mass,
158            media_elapsed_time=42,
159            media_last_updated=anchor,
160            player_elapsed_time=None,
161            player_last_updated=None,
162        )
163
164        assert player.state.current_media is not None
165        assert player.state.current_media.elapsed_time == 42
166        assert player.state.current_media.elapsed_time_last_updated == anchor
167
168    def test_media_position_is_never_paired_with_the_player_timestamp(
169        self, mock_mass: MagicMock
170    ) -> None:
171        """A media position without its own timestamp does not borrow the player's."""
172        player = self._playing_player(
173            mock_mass,
174            media_elapsed_time=42,
175            media_last_updated=None,
176            player_elapsed_time=10.0,
177            player_last_updated=time.time(),
178        )
179
180        assert player.state.current_media is not None
181        assert player.state.current_media.elapsed_time == 42
182        assert player.state.current_media.elapsed_time_last_updated is None
183
184    def test_late_arriving_position_is_adopted(self, mock_mass: MagicMock) -> None:
185        """A position reported after the media was published still reaches the state."""
186        player = self._playing_player(
187            mock_mass,
188            media_elapsed_time=None,
189            media_last_updated=None,
190            player_elapsed_time=None,
191            player_last_updated=None,
192        )
193        published = player.state.current_media
194        assert published is not None
195        assert published.elapsed_time is None
196
197        anchor = time.time()
198        player._attr_elapsed_time = 5.0
199        player._attr_elapsed_time_last_updated = anchor
200        player.update_state(signal_event=False)
201
202        published = player.state.current_media
203        assert published is not None
204        assert published.elapsed_time == 5
205        assert published.elapsed_time_last_updated == anchor
206
207    def test_steady_playback_keeps_the_previous_anchor(self, mock_mass: MagicMock) -> None:
208        """A regular playback tick keeps the anchor it already published."""
209        anchor = time.time()
210        player = self._playing_player(
211            mock_mass,
212            media_elapsed_time=None,
213            media_last_updated=None,
214            player_elapsed_time=10.0,
215            player_last_updated=anchor,
216        )
217
218        player._attr_elapsed_time = 11.0
219        player._attr_elapsed_time_last_updated = anchor + 1
220        player.update_state(signal_event=False)
221
222        assert player.state.current_media is not None
223        assert player.state.current_media.elapsed_time == 10
224        assert player.state.current_media.elapsed_time_last_updated == anchor
225
226    def test_player_position_is_paired_with_its_own_timestamp(self, mock_mass: MagicMock) -> None:
227        """Falling back to the player position also takes the player level timestamp."""
228        player_anchor = time.time()
229        player = self._playing_player(
230            mock_mass,
231            media_elapsed_time=None,
232            media_last_updated=player_anchor - 30,
233            player_elapsed_time=42.7,
234            player_last_updated=player_anchor,
235        )
236
237        assert player.state.current_media is not None
238        assert player.state.current_media.elapsed_time == 42
239        assert player.state.current_media.elapsed_time_last_updated == player_anchor
240
241    def test_no_position_reports_no_timestamp(self, mock_mass: MagicMock) -> None:
242        """Without any position, the timestamp is dropped along with it."""
243        player = self._playing_player(
244            mock_mass,
245            media_elapsed_time=None,
246            media_last_updated=time.time(),
247            player_elapsed_time=None,
248            player_last_updated=time.time(),
249        )
250
251        assert player.state.current_media is not None
252        assert player.state.current_media.elapsed_time is None
253        assert player.state.current_media.elapsed_time_last_updated is None
254
255
256class TestStreamMetadataPosition:
257    """The published position of a queue item carrying live stream metadata."""
258
259    def _playing_player(
260        self,
261        mock_mass: MagicMock,
262        *,
263        metadata_elapsed_time: int | None,
264        metadata_last_updated: float | None,
265        queue_elapsed_time: float,
266        queue_last_updated: float,
267        media_type: MediaType = MediaType.RADIO,
268    ) -> MockPlayer:
269        """Build a playing player whose active queue item carries live stream metadata."""
270        queue = PlayerQueue(
271            queue_id="player_1",
272            active=True,
273            display_name="Player 1",
274            available=True,
275            items=1,
276            elapsed_time=queue_elapsed_time,
277            elapsed_time_last_updated=queue_last_updated,
278            current_item=QueueItem(
279                queue_id="player_1",
280                queue_item_id="item_1",
281                name="Live Radio",
282                duration=None,
283                streamdetails=StreamDetails(
284                    provider="test_provider",
285                    item_id="item_1",
286                    audio_format=AudioFormat(),
287                    media_type=media_type,
288                    stream_metadata=StreamMetadata(
289                        title="Live Track",
290                        elapsed_time=metadata_elapsed_time,
291                        elapsed_time_last_updated=metadata_last_updated,
292                    ),
293                ),
294            ),
295        )
296        mock_mass.player_queues.get = MagicMock(return_value=queue)
297        provider = MockProvider("test_provider", mass=mock_mass)
298        player = MockPlayer(provider, "player_1", "Player 1")
299        player._attr_playback_state = PlaybackState.PLAYING
300        player.update_state(signal_event=False)
301        return player
302
303    def test_zero_metadata_position_is_reported(self, mock_mass: MagicMock) -> None:
304        """A live track that just started reports zero, not the elapsed stream time."""
305        metadata_anchor = time.time()
306        player = self._playing_player(
307            mock_mass,
308            metadata_elapsed_time=0,
309            metadata_last_updated=metadata_anchor,
310            queue_elapsed_time=300.0,
311            queue_last_updated=metadata_anchor - 30,
312        )
313
314        assert player.state.current_media is not None
315        assert player.state.current_media.title == "Live Track"
316        assert player.state.current_media.elapsed_time == 0
317        assert player.state.current_media.elapsed_time_last_updated == metadata_anchor
318
319    def test_queue_position_is_paired_with_its_own_timestamp(self, mock_mass: MagicMock) -> None:
320        """Without a metadata position, both values come from the queue."""
321        queue_anchor = time.time()
322        player = self._playing_player(
323            mock_mass,
324            metadata_elapsed_time=None,
325            metadata_last_updated=queue_anchor - 30,
326            queue_elapsed_time=300.5,
327            queue_last_updated=queue_anchor,
328        )
329
330        assert player.state.current_media is not None
331        assert player.state.current_media.title == "Live Track"
332        assert player.state.current_media.elapsed_time == 300
333        assert player.state.current_media.elapsed_time_last_updated == queue_anchor
334
335    def test_audio_source_position_matches_player_position(self, mock_mass: MagicMock) -> None:
336        """An upstream source position is reported identically on player and media level."""
337        metadata_anchor = time.time()
338        player = self._playing_player(
339            mock_mass,
340            metadata_elapsed_time=0,
341            metadata_last_updated=metadata_anchor,
342            queue_elapsed_time=300.0,
343            queue_last_updated=metadata_anchor - 30,
344            media_type=MediaType.AUDIO_SOURCE,
345        )
346
347        assert player.state.current_media is not None
348        assert player.state.elapsed_time == 0
349        assert player.state.current_media.elapsed_time == 0
350        assert player.state.elapsed_time_last_updated == metadata_anchor
351        assert player.state.current_media.elapsed_time_last_updated == metadata_anchor
352
353
354class TestMediaUpdatedCallback:
355    """The (debounced) media-updated callback fires on media identity changes."""
356
357    def test_palette_resolution_fires_media_updated(
358        self, mock_mass: MagicMock, player: MockPlayer
359    ) -> None:
360        """A late palette resolution re-fires the media-updated callback."""
361        player.set_current_media(uri="http://test/stream", title="Test", image_url="http://img")
362        player.update_state(signal_event=False)
363        mock_mass.call_later.reset_mock()
364
365        player.set_resolved_palette("http://img", MagicMock())
366        player.update_state(force_update=True, signal_event=False)
367
368        assert any(
369            call.kwargs.get("task_id") == f"player_media_updated_{player.player_id}"
370            for call in mock_mass.call_later.call_args_list
371        )
372
373
374class TestCacheInvalidationClasses:
375    """Config-derived cached properties survive state updates, all others refresh."""
376
377    def test_config_cached_props_survive_state_updates(self, player: MockPlayer) -> None:
378        """Config-derived cached properties are not recomputed on state updates."""
379        assert "icon" in player._cache
380        marker = player._cache["icon"]
381        player._attr_volume_level = 60
382        player.update_state(signal_event=False)
383        assert player._cache.get("icon") is marker
384
385    def test_set_config_invalidates_all_cached_props(self, player: MockPlayer) -> None:
386        """set_config invalidates every cached property, including config-derived ones."""
387        assert "icon" in player._cache
388        player.set_config(player.config)
389        assert len(player._cache) == 0
390
391    def test_player_implementation_cached_props_cleared_each_update(
392        self, player: MockPlayer
393    ) -> None:
394        """Cached properties defined by player implementations refresh on every update."""
395        player._cache["some_provider_prop"] = object()
396        player.update_state()
397        assert "some_provider_prop" not in player._cache
398
399
400class TestUpdateStateTiming:
401    """Micro-benchmarks guarding the cost of the update_state paths."""
402
403    def test_no_change_update_is_fast(self, player: MockPlayer) -> None:
404        """The no-change path completes well within the microsecond budget."""
405        for _ in range(50):  # warmup
406            player.update_state()
407        timings = []
408        for _ in range(200):
409            start = time.perf_counter()
410            player.update_state()
411            timings.append(time.perf_counter() - start)
412        duration = median(timings)
413        # target is <50us on a dev machine; assert with generous CI headroom
414        assert duration < 0.001, f"no-change update_state took {duration * 1e6:.0f}us (median)"
415
416    def test_full_update_is_fast(self, player: MockPlayer) -> None:
417        """A full recalculation completes well within the microsecond budget."""
418        for volume in range(50):  # warmup
419            player._attr_volume_level = volume
420            player.update_state(signal_event=False)
421        timings = []
422        for volume in range(200):
423            player._attr_volume_level = volume % 100
424            start = time.perf_counter()
425            player.update_state(signal_event=False)
426            timings.append(time.perf_counter() - start)
427        duration = median(timings)
428        # target is <500us on a dev machine; assert with generous CI headroom
429        assert duration < 0.005, f"full update_state took {duration * 1e6:.0f}us (median)"
430