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_changed_privacy_rebuilds_state(self, player: MockPlayer) -> None:
103        """A player turning private must reach clients, which decide where to show it."""
104        player._attr_private = True
105
106        with patch.object(
107            player_module, "PlayerState", wraps=player_module.PlayerState
108        ) as state_cls:
109            player.update_state(signal_event=False)
110
111        state_cls.assert_called_once()
112        assert player.state.private is True
113
114    def test_mark_state_dirty_forces_recalculation(self, player: MockPlayer) -> None:
115        """mark_state_dirty recalculates even when no own input changed."""
116        with patch.object(
117            player_module, "PlayerState", wraps=player_module.PlayerState
118        ) as state_cls:
119            player.mark_state_dirty()
120            player.update_state(signal_event=False)
121
122        state_cls.assert_called_once()
123
124    def test_position_appearing_signals_a_jump(
125        self, player: MockPlayer, mock_mass: MagicMock
126    ) -> None:
127        """
128        The first elapsed write after an anchor reset is signalled as a position jump.
129
130        A playback session restart (seek) resets the anchor to None; the session's
131        first elapsed write is its only propagation opportunity, so swallowing it
132        strands wrapper players on the stale None anchor (a frozen queue position).
133        """
134        player._attr_elapsed_time = None
135        player._attr_elapsed_time_last_updated = None
136        player.update_state(signal_event=False)
137        assert player.state.elapsed_time is None
138
139        player._attr_elapsed_time = 0.5
140        player._attr_elapsed_time_last_updated = time.time()
141        player.update_state()
142
143        mock_mass.players.on_player_position_jumped.assert_called_once_with(player)
144        assert player.state.elapsed_time == 0.5
145
146
147class TestReconcilePositionAnchor:
148    """The anchor reconcile helper's handling of incomplete anchors."""
149
150    def test_incomplete_to_complete_reports_a_jump(self) -> None:
151        """A position appearing on a None anchor is a discontinuity, not a quiet adopt."""
152        position, _, jumped = player_module._reconcile_position_anchor(
153            None, None, 0.5, time.time(), prev_playing=True, new_playing=True
154        )
155        assert position == 0.5
156        assert jumped
157
158    def test_complete_to_incomplete_adopts_silently(self) -> None:
159        """A position disappearing (stop/reset) is adopted without a jump signal."""
160        position, _, jumped = player_module._reconcile_position_anchor(
161            10.0, time.time(), None, None, prev_playing=True, new_playing=False
162        )
163        assert position is None
164        assert not jumped
165
166    def test_incomplete_to_incomplete_stays_silent(self) -> None:
167        """No position on either side has nothing to signal."""
168        _, _, jumped = player_module._reconcile_position_anchor(
169            None, None, None, None, prev_playing=False, new_playing=False
170        )
171        assert not jumped
172
173    def test_steady_playback_still_keeps_the_previous_anchor(self) -> None:
174        """A tick extrapolating to the same corrected position keeps the anchor unchanged."""
175        now = time.time()
176        position, timestamp, jumped = player_module._reconcile_position_anchor(
177            10.0, now - 5, 15.2, now, prev_playing=True, new_playing=True
178        )
179        assert position == 10.0
180        assert timestamp == now - 5
181        assert not jumped
182
183
184class TestNativeCurrentMediaPosition:
185    """The published position of native current_media and the timestamp it is paired with."""
186
187    def _playing_player(
188        self,
189        mock_mass: MagicMock,
190        *,
191        media_elapsed_time: int | None,
192        media_last_updated: float | None,
193        player_elapsed_time: float | None,
194        player_last_updated: float | None,
195    ) -> MockPlayer:
196        """Build a playing player with the given media and player level position anchors."""
197        provider = MockProvider("test_provider", mass=mock_mass)
198        player = MockPlayer(provider, "player_1", "Player 1")
199        player._attr_playback_state = PlaybackState.PLAYING
200        player._attr_elapsed_time = player_elapsed_time
201        player._attr_elapsed_time_last_updated = player_last_updated
202        player.set_current_media(uri="http://test/stream", title="Test")
203        assert player._attr_current_media is not None
204        player._attr_current_media.elapsed_time = media_elapsed_time
205        player._attr_current_media.elapsed_time_last_updated = media_last_updated
206        player.update_state(signal_event=False)
207        return player
208
209    def test_zero_position_is_reported(self, mock_mass: MagicMock) -> None:
210        """A position of zero (start of a track) is reported instead of being dropped."""
211        anchor = time.time()
212        player = self._playing_player(
213            mock_mass,
214            media_elapsed_time=0,
215            media_last_updated=anchor,
216            player_elapsed_time=0,
217            player_last_updated=anchor,
218        )
219
220        assert player.state.current_media is not None
221        assert player.state.current_media.elapsed_time == 0
222        assert player.state.current_media.elapsed_time_last_updated == anchor
223
224    def test_media_position_wins_without_player_position(self, mock_mass: MagicMock) -> None:
225        """A media position is reported even when the player reports none of its own."""
226        anchor = time.time()
227        player = self._playing_player(
228            mock_mass,
229            media_elapsed_time=42,
230            media_last_updated=anchor,
231            player_elapsed_time=None,
232            player_last_updated=None,
233        )
234
235        assert player.state.current_media is not None
236        assert player.state.current_media.elapsed_time == 42
237        assert player.state.current_media.elapsed_time_last_updated == anchor
238
239    def test_media_position_is_never_paired_with_the_player_timestamp(
240        self, mock_mass: MagicMock
241    ) -> None:
242        """A media position without its own timestamp does not borrow the player's."""
243        player = self._playing_player(
244            mock_mass,
245            media_elapsed_time=42,
246            media_last_updated=None,
247            player_elapsed_time=10.0,
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 == 42
253        assert player.state.current_media.elapsed_time_last_updated is None
254
255    def test_late_arriving_position_is_adopted(self, mock_mass: MagicMock) -> None:
256        """A position reported after the media was published still reaches the state."""
257        player = self._playing_player(
258            mock_mass,
259            media_elapsed_time=None,
260            media_last_updated=None,
261            player_elapsed_time=None,
262            player_last_updated=None,
263        )
264        published = player.state.current_media
265        assert published is not None
266        assert published.elapsed_time is None
267
268        anchor = time.time()
269        player._attr_elapsed_time = 5.0
270        player._attr_elapsed_time_last_updated = anchor
271        player.update_state(signal_event=False)
272
273        published = player.state.current_media
274        assert published is not None
275        assert published.elapsed_time == 5
276        assert published.elapsed_time_last_updated == anchor
277
278    def test_steady_playback_keeps_the_previous_anchor(self, mock_mass: MagicMock) -> None:
279        """A regular playback tick keeps the anchor it already published."""
280        anchor = time.time()
281        player = self._playing_player(
282            mock_mass,
283            media_elapsed_time=None,
284            media_last_updated=None,
285            player_elapsed_time=10.0,
286            player_last_updated=anchor,
287        )
288
289        player._attr_elapsed_time = 11.0
290        player._attr_elapsed_time_last_updated = anchor + 1
291        player.update_state(signal_event=False)
292
293        assert player.state.current_media is not None
294        assert player.state.current_media.elapsed_time == 10
295        assert player.state.current_media.elapsed_time_last_updated == anchor
296
297    def test_player_position_is_paired_with_its_own_timestamp(self, mock_mass: MagicMock) -> None:
298        """Falling back to the player position also takes the player level timestamp."""
299        player_anchor = time.time()
300        player = self._playing_player(
301            mock_mass,
302            media_elapsed_time=None,
303            media_last_updated=player_anchor - 30,
304            player_elapsed_time=42.7,
305            player_last_updated=player_anchor,
306        )
307
308        assert player.state.current_media is not None
309        assert player.state.current_media.elapsed_time == 42
310        assert player.state.current_media.elapsed_time_last_updated == player_anchor
311
312    def test_no_position_reports_no_timestamp(self, mock_mass: MagicMock) -> None:
313        """Without any position, the timestamp is dropped along with it."""
314        player = self._playing_player(
315            mock_mass,
316            media_elapsed_time=None,
317            media_last_updated=time.time(),
318            player_elapsed_time=None,
319            player_last_updated=time.time(),
320        )
321
322        assert player.state.current_media is not None
323        assert player.state.current_media.elapsed_time is None
324        assert player.state.current_media.elapsed_time_last_updated is None
325
326
327class TestStreamMetadataPosition:
328    """The published position of a queue item carrying live stream metadata."""
329
330    def _playing_player(
331        self,
332        mock_mass: MagicMock,
333        *,
334        metadata_elapsed_time: int | None,
335        metadata_last_updated: float | None,
336        queue_elapsed_time: float,
337        queue_last_updated: float,
338        media_type: MediaType = MediaType.RADIO,
339    ) -> MockPlayer:
340        """Build a playing player whose active queue item carries live stream metadata."""
341        queue = PlayerQueue(
342            queue_id="player_1",
343            active=True,
344            display_name="Player 1",
345            available=True,
346            items=1,
347            elapsed_time=queue_elapsed_time,
348            elapsed_time_last_updated=queue_last_updated,
349            current_item=QueueItem(
350                queue_id="player_1",
351                queue_item_id="item_1",
352                name="Live Radio",
353                duration=None,
354                streamdetails=StreamDetails(
355                    provider="test_provider",
356                    item_id="item_1",
357                    audio_format=AudioFormat(),
358                    media_type=media_type,
359                    stream_metadata=StreamMetadata(
360                        title="Live Track",
361                        elapsed_time=metadata_elapsed_time,
362                        elapsed_time_last_updated=metadata_last_updated,
363                    ),
364                ),
365            ),
366        )
367        mock_mass.player_queues.get = MagicMock(return_value=queue)
368        provider = MockProvider("test_provider", mass=mock_mass)
369        player = MockPlayer(provider, "player_1", "Player 1")
370        player._attr_playback_state = PlaybackState.PLAYING
371        player.update_state(signal_event=False)
372        return player
373
374    def test_zero_metadata_position_is_reported(self, mock_mass: MagicMock) -> None:
375        """A live track that just started reports zero, not the elapsed stream time."""
376        metadata_anchor = time.time()
377        player = self._playing_player(
378            mock_mass,
379            metadata_elapsed_time=0,
380            metadata_last_updated=metadata_anchor,
381            queue_elapsed_time=300.0,
382            queue_last_updated=metadata_anchor - 30,
383        )
384
385        assert player.state.current_media is not None
386        assert player.state.current_media.title == "Live Track"
387        assert player.state.current_media.elapsed_time == 0
388        assert player.state.current_media.elapsed_time_last_updated == metadata_anchor
389
390    def test_queue_position_is_paired_with_its_own_timestamp(self, mock_mass: MagicMock) -> None:
391        """Without a metadata position, both values come from the queue."""
392        queue_anchor = time.time()
393        player = self._playing_player(
394            mock_mass,
395            metadata_elapsed_time=None,
396            metadata_last_updated=queue_anchor - 30,
397            queue_elapsed_time=300.5,
398            queue_last_updated=queue_anchor,
399        )
400
401        assert player.state.current_media is not None
402        assert player.state.current_media.title == "Live Track"
403        assert player.state.current_media.elapsed_time == 300
404        assert player.state.current_media.elapsed_time_last_updated == queue_anchor
405
406    def test_audio_source_position_matches_player_position(self, mock_mass: MagicMock) -> None:
407        """An upstream source position is reported identically on player and media level."""
408        metadata_anchor = time.time()
409        player = self._playing_player(
410            mock_mass,
411            metadata_elapsed_time=0,
412            metadata_last_updated=metadata_anchor,
413            queue_elapsed_time=300.0,
414            queue_last_updated=metadata_anchor - 30,
415            media_type=MediaType.AUDIO_SOURCE,
416        )
417
418        assert player.state.current_media is not None
419        assert player.state.elapsed_time == 0
420        assert player.state.current_media.elapsed_time == 0
421        assert player.state.elapsed_time_last_updated == metadata_anchor
422        assert player.state.current_media.elapsed_time_last_updated == metadata_anchor
423
424
425class TestMediaUpdatedCallback:
426    """The (debounced) media-updated callback fires on media identity changes."""
427
428    def test_palette_resolution_fires_media_updated(
429        self, mock_mass: MagicMock, player: MockPlayer
430    ) -> None:
431        """A late palette resolution re-fires the media-updated callback."""
432        player.set_current_media(uri="http://test/stream", title="Test", image_url="http://img")
433        player.update_state(signal_event=False)
434        mock_mass.call_later.reset_mock()
435
436        player.set_resolved_palette("http://img", MagicMock())
437        player.update_state(force_update=True, signal_event=False)
438
439        assert any(
440            call.kwargs.get("task_id") == f"player_media_updated_{player.player_id}"
441            for call in mock_mass.call_later.call_args_list
442        )
443
444
445class TestCacheInvalidationClasses:
446    """Config-derived cached properties survive state updates, all others refresh."""
447
448    def test_config_cached_props_survive_state_updates(self, player: MockPlayer) -> None:
449        """Config-derived cached properties are not recomputed on state updates."""
450        assert "icon" in player._cache
451        marker = player._cache["icon"]
452        player._attr_volume_level = 60
453        player.update_state(signal_event=False)
454        assert player._cache.get("icon") is marker
455
456    def test_set_config_invalidates_all_cached_props(self, player: MockPlayer) -> None:
457        """set_config invalidates every cached property, including config-derived ones."""
458        assert "icon" in player._cache
459        player.set_config(player.config)
460        assert len(player._cache) == 0
461
462    def test_player_implementation_cached_props_cleared_each_update(
463        self, player: MockPlayer
464    ) -> None:
465        """Cached properties defined by player implementations refresh on every update."""
466        player._cache["some_provider_prop"] = object()
467        player.update_state()
468        assert "some_provider_prop" not in player._cache
469
470
471class TestUpdateStateTiming:
472    """Micro-benchmarks guarding the cost of the update_state paths."""
473
474    def test_no_change_update_is_fast(self, player: MockPlayer) -> None:
475        """The no-change path completes well within the microsecond budget."""
476        for _ in range(50):  # warmup
477            player.update_state()
478        timings = []
479        for _ in range(200):
480            start = time.perf_counter()
481            player.update_state()
482            timings.append(time.perf_counter() - start)
483        duration = median(timings)
484        # target is <50us on a dev machine; assert with generous CI headroom
485        assert duration < 0.001, f"no-change update_state took {duration * 1e6:.0f}us (median)"
486
487    def test_full_update_is_fast(self, player: MockPlayer) -> None:
488        """A full recalculation completes well within the microsecond budget."""
489        for volume in range(50):  # warmup
490            player._attr_volume_level = volume
491            player.update_state(signal_event=False)
492        timings = []
493        for volume in range(200):
494            player._attr_volume_level = volume % 100
495            start = time.perf_counter()
496            player.update_state(signal_event=False)
497            timings.append(time.perf_counter() - start)
498        duration = median(timings)
499        # target is <500us on a dev machine; assert with generous CI headroom
500        assert duration < 0.005, f"full update_state took {duration * 1e6:.0f}us (median)"
501