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, AudioSource
13from music_assistant_models.media_items.provider_mapping import ProviderMapping
14from music_assistant_models.player_queue import PlayerQueue
15from music_assistant_models.queue_item import QueueItem
16from music_assistant_models.streamdetails import StreamDetails, StreamMetadata
17
18import music_assistant.models.player as player_module
19from music_assistant.controllers.players.audio_sources import AudioSourceSession
20from tests.common import MockPlayer, MockProvider
21
22
23@pytest.fixture
24def mock_mass() -> MagicMock:
25    """Create a mock MusicAssistant instance."""
26    mass = MagicMock()
27    mass.closing = False
28    mass.config.get_raw_player_config_value = MagicMock(
29        side_effect=lambda _player_id, _key, default=None: default
30    )
31    # no queue registered: current_media resolves from the player's native media
32    mass.player_queues.get = MagicMock(return_value=None)
33    mass.players.get_audio_source_session = MagicMock(return_value=None)
34    mass.players.scale_volume_from_device = MagicMock(side_effect=lambda _player_id, volume: volume)
35    return mass
36
37
38@pytest.fixture
39def player(mock_mass: MagicMock) -> MockPlayer:
40    """Create a playing player with its state calculated once."""
41    provider = MockProvider("test_provider", mass=mock_mass)
42    player = MockPlayer(provider, "player_1", "Player 1")
43    now = time.time()
44    player._attr_playback_state = PlaybackState.PLAYING
45    player._attr_volume_level = 20
46    player._attr_elapsed_time = 10.0
47    player._attr_elapsed_time_last_updated = now
48    player.set_current_media(uri="http://test/stream", title="Test")
49    player.update_state(signal_event=False)
50    return player
51
52
53class TestUpdateStateChangeDetection:
54    """The no-change update_state path does no work; changed inputs recalculate."""
55
56    def test_no_change_update_does_no_work(self, player: MockPlayer) -> None:
57        """An update_state call without any changed input builds nothing."""
58        with (
59            patch.object(
60                player_module, "PlayerState", wraps=player_module.PlayerState
61            ) as state_cls,
62            patch.object(copy, "deepcopy", wraps=copy.deepcopy) as deepcopy_mock,
63        ):
64            player.update_state()
65
66        state_cls.assert_not_called()
67        deepcopy_mock.assert_not_called()
68
69    def test_position_tick_does_not_rebuild(self, player: MockPlayer) -> None:
70        """A regular playback tick keeps the anchor and builds nothing."""
71        assert player._attr_elapsed_time is not None
72        assert player._attr_elapsed_time_last_updated is not None
73        player._attr_elapsed_time += 1
74        player._attr_elapsed_time_last_updated += 1
75
76        with patch.object(
77            player_module, "PlayerState", wraps=player_module.PlayerState
78        ) as state_cls:
79            player.update_state()
80
81        state_cls.assert_not_called()
82        assert player.state.elapsed_time == 10.0
83
84    def test_position_jump_rebuilds_state(self, player: MockPlayer) -> None:
85        """A corrected-position jump (seek) adopts the new anchor."""
86        player._attr_elapsed_time = 61.0
87        player._attr_elapsed_time_last_updated = time.time()
88
89        player.update_state(signal_event=False)
90
91        assert player.state.elapsed_time == 61.0
92
93    def test_changed_input_rebuilds_state(self, player: MockPlayer) -> None:
94        """A changed player attribute is picked up by the next update_state call."""
95        player._attr_volume_level = 55
96
97        with patch.object(
98            player_module, "PlayerState", wraps=player_module.PlayerState
99        ) as state_cls:
100            player.update_state(signal_event=False)
101
102        state_cls.assert_called_once()
103        assert player.state.volume_level == 55
104
105    def test_changed_privacy_rebuilds_state(self, player: MockPlayer) -> None:
106        """A player turning private must reach clients, which decide where to show it."""
107        player._attr_private = True
108
109        with patch.object(
110            player_module, "PlayerState", wraps=player_module.PlayerState
111        ) as state_cls:
112            player.update_state(signal_event=False)
113
114        state_cls.assert_called_once()
115        assert player.state.private is True
116
117    def test_mark_state_dirty_forces_recalculation(self, player: MockPlayer) -> None:
118        """mark_state_dirty recalculates even when no own input changed."""
119        with patch.object(
120            player_module, "PlayerState", wraps=player_module.PlayerState
121        ) as state_cls:
122            player.mark_state_dirty()
123            player.update_state(signal_event=False)
124
125        state_cls.assert_called_once()
126
127    def test_position_appearing_signals_a_jump(
128        self, player: MockPlayer, mock_mass: MagicMock
129    ) -> None:
130        """
131        The first elapsed write after an anchor reset is signalled as a position jump.
132
133        A playback session restart (seek) resets the anchor to None; the session's
134        first elapsed write is its only propagation opportunity, so swallowing it
135        strands wrapper players on the stale None anchor (a frozen queue position).
136        """
137        player._attr_elapsed_time = None
138        player._attr_elapsed_time_last_updated = None
139        player.update_state(signal_event=False)
140        assert player.state.elapsed_time is None
141
142        player._attr_elapsed_time = 0.5
143        player._attr_elapsed_time_last_updated = time.time()
144        player.update_state()
145
146        mock_mass.players.on_player_position_jumped.assert_called_once_with(player)
147        assert player.state.elapsed_time == 0.5
148
149
150class TestReconcilePositionAnchor:
151    """The anchor reconcile helper's handling of incomplete anchors."""
152
153    def test_incomplete_to_complete_reports_a_jump(self) -> None:
154        """A position appearing on a None anchor is a discontinuity, not a quiet adopt."""
155        position, _, jumped = player_module._reconcile_position_anchor(
156            None, None, 0.5, time.time(), prev_playing=True, new_playing=True
157        )
158        assert position == 0.5
159        assert jumped
160
161    def test_complete_to_incomplete_adopts_silently(self) -> None:
162        """A position disappearing (stop/reset) is adopted without a jump signal."""
163        position, _, jumped = player_module._reconcile_position_anchor(
164            10.0, time.time(), None, None, prev_playing=True, new_playing=False
165        )
166        assert position is None
167        assert not jumped
168
169    def test_incomplete_to_incomplete_stays_silent(self) -> None:
170        """No position on either side has nothing to signal."""
171        _, _, jumped = player_module._reconcile_position_anchor(
172            None, None, None, None, prev_playing=False, new_playing=False
173        )
174        assert not jumped
175
176    def test_steady_playback_still_keeps_the_previous_anchor(self) -> None:
177        """A tick extrapolating to the same corrected position keeps the anchor unchanged."""
178        now = time.time()
179        position, timestamp, jumped = player_module._reconcile_position_anchor(
180            10.0, now - 5, 15.2, now, prev_playing=True, new_playing=True
181        )
182        assert position == 10.0
183        assert timestamp == now - 5
184        assert not jumped
185
186
187class TestNativeCurrentMediaPosition:
188    """The published position of native current_media and the timestamp it is paired with."""
189
190    def _playing_player(
191        self,
192        mock_mass: MagicMock,
193        *,
194        media_elapsed_time: int | None,
195        media_last_updated: float | None,
196        player_elapsed_time: float | None,
197        player_last_updated: float | None,
198    ) -> MockPlayer:
199        """Build a playing player with the given media and player level position anchors."""
200        provider = MockProvider("test_provider", mass=mock_mass)
201        player = MockPlayer(provider, "player_1", "Player 1")
202        player._attr_playback_state = PlaybackState.PLAYING
203        player._attr_elapsed_time = player_elapsed_time
204        player._attr_elapsed_time_last_updated = player_last_updated
205        player.set_current_media(uri="http://test/stream", title="Test")
206        assert player._attr_current_media is not None
207        player._attr_current_media.elapsed_time = media_elapsed_time
208        player._attr_current_media.elapsed_time_last_updated = media_last_updated
209        player.update_state(signal_event=False)
210        return player
211
212    def test_zero_position_is_reported(self, mock_mass: MagicMock) -> None:
213        """A position of zero (start of a track) is reported instead of being dropped."""
214        anchor = time.time()
215        player = self._playing_player(
216            mock_mass,
217            media_elapsed_time=0,
218            media_last_updated=anchor,
219            player_elapsed_time=0,
220            player_last_updated=anchor,
221        )
222
223        assert player.state.current_media is not None
224        assert player.state.current_media.elapsed_time == 0
225        assert player.state.current_media.elapsed_time_last_updated == anchor
226
227    def test_media_position_wins_without_player_position(self, mock_mass: MagicMock) -> None:
228        """A media position is reported even when the player reports none of its own."""
229        anchor = time.time()
230        player = self._playing_player(
231            mock_mass,
232            media_elapsed_time=42,
233            media_last_updated=anchor,
234            player_elapsed_time=None,
235            player_last_updated=None,
236        )
237
238        assert player.state.current_media is not None
239        assert player.state.current_media.elapsed_time == 42
240        assert player.state.current_media.elapsed_time_last_updated == anchor
241
242    def test_media_position_is_never_paired_with_the_player_timestamp(
243        self, mock_mass: MagicMock
244    ) -> None:
245        """A media position without its own timestamp does not borrow the player's."""
246        player = self._playing_player(
247            mock_mass,
248            media_elapsed_time=42,
249            media_last_updated=None,
250            player_elapsed_time=10.0,
251            player_last_updated=time.time(),
252        )
253
254        assert player.state.current_media is not None
255        assert player.state.current_media.elapsed_time == 42
256        assert player.state.current_media.elapsed_time_last_updated is None
257
258    def test_late_arriving_position_is_adopted(self, mock_mass: MagicMock) -> None:
259        """A position reported after the media was published still reaches the state."""
260        player = self._playing_player(
261            mock_mass,
262            media_elapsed_time=None,
263            media_last_updated=None,
264            player_elapsed_time=None,
265            player_last_updated=None,
266        )
267        published = player.state.current_media
268        assert published is not None
269        assert published.elapsed_time is None
270
271        anchor = time.time()
272        player._attr_elapsed_time = 5.0
273        player._attr_elapsed_time_last_updated = anchor
274        player.update_state(signal_event=False)
275
276        published = player.state.current_media
277        assert published is not None
278        assert published.elapsed_time == 5
279        assert published.elapsed_time_last_updated == anchor
280
281    def test_steady_playback_keeps_the_previous_anchor(self, mock_mass: MagicMock) -> None:
282        """A regular playback tick keeps the anchor it already published."""
283        anchor = time.time()
284        player = self._playing_player(
285            mock_mass,
286            media_elapsed_time=None,
287            media_last_updated=None,
288            player_elapsed_time=10.0,
289            player_last_updated=anchor,
290        )
291
292        player._attr_elapsed_time = 11.0
293        player._attr_elapsed_time_last_updated = anchor + 1
294        player.update_state(signal_event=False)
295
296        assert player.state.current_media is not None
297        assert player.state.current_media.elapsed_time == 10
298        assert player.state.current_media.elapsed_time_last_updated == anchor
299
300    def test_player_position_is_paired_with_its_own_timestamp(self, mock_mass: MagicMock) -> None:
301        """Falling back to the player position also takes the player level timestamp."""
302        player_anchor = time.time()
303        player = self._playing_player(
304            mock_mass,
305            media_elapsed_time=None,
306            media_last_updated=player_anchor - 30,
307            player_elapsed_time=42.7,
308            player_last_updated=player_anchor,
309        )
310
311        assert player.state.current_media is not None
312        assert player.state.current_media.elapsed_time == 42
313        assert player.state.current_media.elapsed_time_last_updated == player_anchor
314
315    def test_no_position_reports_no_timestamp(self, mock_mass: MagicMock) -> None:
316        """Without any position, the timestamp is dropped along with it."""
317        player = self._playing_player(
318            mock_mass,
319            media_elapsed_time=None,
320            media_last_updated=time.time(),
321            player_elapsed_time=None,
322            player_last_updated=time.time(),
323        )
324
325        assert player.state.current_media is not None
326        assert player.state.current_media.elapsed_time is None
327        assert player.state.current_media.elapsed_time_last_updated is None
328
329
330class TestStreamMetadataPosition:
331    """The published position of a queue item carrying live stream metadata."""
332
333    def _playing_player(
334        self,
335        mock_mass: MagicMock,
336        *,
337        metadata_elapsed_time: int | None,
338        metadata_last_updated: float | None,
339        queue_elapsed_time: float,
340        queue_last_updated: float,
341        media_type: MediaType = MediaType.RADIO,
342    ) -> MockPlayer:
343        """Build a playing player whose active queue item carries live stream metadata."""
344        queue = PlayerQueue(
345            queue_id="player_1",
346            active=True,
347            display_name="Player 1",
348            available=True,
349            items=1,
350            elapsed_time=queue_elapsed_time,
351            elapsed_time_last_updated=queue_last_updated,
352            current_item=QueueItem(
353                queue_id="player_1",
354                queue_item_id="item_1",
355                name="Live Radio",
356                duration=None,
357                streamdetails=StreamDetails(
358                    provider="test_provider",
359                    item_id="item_1",
360                    audio_format=AudioFormat(),
361                    media_type=media_type,
362                    stream_metadata=StreamMetadata(
363                        title="Live Track",
364                        elapsed_time=metadata_elapsed_time,
365                        elapsed_time_last_updated=metadata_last_updated,
366                    ),
367                ),
368            ),
369        )
370        mock_mass.player_queues.get = MagicMock(return_value=queue)
371        provider = MockProvider("test_provider", mass=mock_mass)
372        player = MockPlayer(provider, "player_1", "Player 1")
373        player._attr_playback_state = PlaybackState.PLAYING
374        player.update_state(signal_event=False)
375        return player
376
377    def test_zero_metadata_position_is_reported(self, mock_mass: MagicMock) -> None:
378        """A live track that just started reports zero, not the elapsed stream time."""
379        metadata_anchor = time.time()
380        player = self._playing_player(
381            mock_mass,
382            metadata_elapsed_time=0,
383            metadata_last_updated=metadata_anchor,
384            queue_elapsed_time=300.0,
385            queue_last_updated=metadata_anchor - 30,
386        )
387
388        assert player.state.current_media is not None
389        assert player.state.current_media.title == "Live Track"
390        assert player.state.current_media.elapsed_time == 0
391        assert player.state.current_media.elapsed_time_last_updated == metadata_anchor
392
393    def test_queue_position_is_paired_with_its_own_timestamp(self, mock_mass: MagicMock) -> None:
394        """Without a metadata position, both values come from the queue."""
395        queue_anchor = time.time()
396        player = self._playing_player(
397            mock_mass,
398            metadata_elapsed_time=None,
399            metadata_last_updated=queue_anchor - 30,
400            queue_elapsed_time=300.5,
401            queue_last_updated=queue_anchor,
402        )
403
404        assert player.state.current_media is not None
405        assert player.state.current_media.title == "Live Track"
406        assert player.state.current_media.elapsed_time == 300
407        assert player.state.current_media.elapsed_time_last_updated == queue_anchor
408
409    def test_a_radio_station_keeps_its_own_position(self, mock_mass: MagicMock) -> None:
410        """
411        A radio station's position is untouched by the live-source override.
412
413        Radio is a queue item and always was: the override this replaced was gated on
414        the media type being an audio source, so radio never went through it. Pinned
415        here because the override lost that gate, and a station reporting the byte
416        clock instead of its stream position would be a regression with nothing to do
417        with external sources.
418        """
419        metadata_anchor = time.time()
420        player = self._playing_player(
421            mock_mass,
422            metadata_elapsed_time=42,
423            metadata_last_updated=metadata_anchor,
424            queue_elapsed_time=999.0,
425            queue_last_updated=metadata_anchor - 30,
426            media_type=MediaType.RADIO,
427        )
428
429        assert player.state.current_media is not None
430        assert player.state.current_media.elapsed_time == 42
431        assert player.state.current_media.elapsed_time_last_updated == metadata_anchor
432
433    def test_a_live_source_elsewhere_does_not_move_this_players_clock(
434        self, mock_mass: MagicMock
435    ) -> None:
436        """
437        A source playing on another player has no say in this player's position.
438
439        Asserts the player-level clock, which is what the override writes - the
440        media-level position for a radio item comes from its stream metadata by a
441        different route, so it would not notice a leak.
442        """
443        anchor = time.time()
444        player = self._playing_player(
445            mock_mass,
446            metadata_elapsed_time=None,
447            metadata_last_updated=None,
448            queue_elapsed_time=0.0,
449            queue_last_updated=anchor,
450            media_type=MediaType.RADIO,
451        )
452        player._attr_elapsed_time = 123.0
453        player._attr_elapsed_time_last_updated = anchor
454        other = MagicMock()
455        other.stream_metadata.elapsed_time = 7
456        other.stream_metadata.elapsed_time_last_updated = anchor
457        mock_mass.players.get_audio_source_session = MagicMock(
458            side_effect=lambda pid: other if pid == "other_player" else None
459        )
460        player.update_state(signal_event=False)
461
462        assert player.state.elapsed_time == 123.0
463
464
465class TestLiveSourcePosition:
466    """The published position of a live external source playing on a player."""
467
468    def _player_with_source(
469        self,
470        mock_mass: MagicMock,
471        *,
472        metadata_elapsed_time: int | None,
473        metadata_last_updated: float | None,
474    ) -> MockPlayer:
475        """Build a playing player with a live source reporting its own position."""
476        source = AudioSource(
477            item_id="main",
478            provider="test_instance",
479            name="Spotify Connect",
480            provider_mappings={
481                ProviderMapping(
482                    item_id="main",
483                    provider_domain="spotify_connect",
484                    provider_instance="test_instance",
485                )
486            },
487        )
488        session = AudioSourceSession(
489            player_id="player_1",
490            source=source,
491            provider_instance_id="test_instance",
492            stream_metadata=StreamMetadata(
493                title="Live Track",
494                elapsed_time=metadata_elapsed_time,
495                elapsed_time_last_updated=metadata_last_updated,
496            ),
497        )
498        mock_mass.players.get_audio_source_session = MagicMock(return_value=session)
499        provider = MockProvider("test_provider", mass=mock_mass)
500        player = MockPlayer(provider, "player_1", "Player 1")
501        player._attr_playback_state = PlaybackState.PLAYING
502        player._attr_elapsed_time = 300.0
503        player._attr_elapsed_time_last_updated = (metadata_last_updated or time.time()) - 30
504        player.update_state(signal_event=False)
505        return player
506
507    def test_source_position_matches_player_position(self, mock_mass: MagicMock) -> None:
508        """An upstream source position is reported identically on player and media level."""
509        metadata_anchor = time.time()
510        player = self._player_with_source(
511            mock_mass,
512            metadata_elapsed_time=0,
513            metadata_last_updated=metadata_anchor,
514        )
515
516        assert player.state.current_media is not None
517        assert player.state.elapsed_time == 0
518        assert player.state.current_media.elapsed_time == 0
519        assert player.state.elapsed_time_last_updated == metadata_anchor
520        assert player.state.current_media.elapsed_time_last_updated == metadata_anchor
521
522    def test_source_without_a_position_falls_back_to_the_player(self, mock_mass: MagicMock) -> None:
523        """A source that reports no position leaves the player's own clock in charge."""
524        anchor = time.time()
525        player = self._player_with_source(
526            mock_mass,
527            metadata_elapsed_time=None,
528            metadata_last_updated=anchor,
529        )
530
531        assert player.state.elapsed_time == 300.0
532        assert player.state.current_media is not None
533        assert player.state.current_media.elapsed_time == 300
534
535    def test_the_source_names_what_is_playing(self, mock_mass: MagicMock) -> None:
536        """The media on screen is the track the source reports, on the source's uri."""
537        player = self._player_with_source(
538            mock_mass, metadata_elapsed_time=12, metadata_last_updated=time.time()
539        )
540
541        media = player.state.current_media
542        assert media is not None
543        assert media.title == "Live Track"
544        assert media.media_type is MediaType.AUDIO_SOURCE
545        assert media.uri == "test_instance://audio_source/main"
546        # the owner of the session, which its stream url is keyed on
547        assert media.source_id == "player_1"
548        # no queue item: this is not playing out of a queue
549        assert media.queue_item_id is None
550
551
552class TestMediaUpdatedCallback:
553    """The (debounced) media-updated callback fires on media identity changes."""
554
555    def test_palette_resolution_fires_media_updated(
556        self, mock_mass: MagicMock, player: MockPlayer
557    ) -> None:
558        """A late palette resolution re-fires the media-updated callback."""
559        player.set_current_media(uri="http://test/stream", title="Test", image_url="http://img")
560        player.update_state(signal_event=False)
561        mock_mass.call_later.reset_mock()
562
563        player.set_resolved_palette("http://img", MagicMock())
564        player.update_state(force_update=True, signal_event=False)
565
566        assert any(
567            call.kwargs.get("task_id") == f"player_media_updated_{player.player_id}"
568            for call in mock_mass.call_later.call_args_list
569        )
570
571
572class TestCacheInvalidationClasses:
573    """Config-derived cached properties survive state updates, all others refresh."""
574
575    def test_config_cached_props_survive_state_updates(self, player: MockPlayer) -> None:
576        """Config-derived cached properties are not recomputed on state updates."""
577        assert "icon" in player._cache
578        marker = player._cache["icon"]
579        player._attr_volume_level = 60
580        player.update_state(signal_event=False)
581        assert player._cache.get("icon") is marker
582
583    def test_set_config_invalidates_all_cached_props(self, player: MockPlayer) -> None:
584        """set_config invalidates every cached property, including config-derived ones."""
585        assert "icon" in player._cache
586        player.set_config(player.config)
587        assert len(player._cache) == 0
588
589    def test_player_implementation_cached_props_cleared_each_update(
590        self, player: MockPlayer
591    ) -> None:
592        """Cached properties defined by player implementations refresh on every update."""
593        player._cache["some_provider_prop"] = object()
594        player.update_state()
595        assert "some_provider_prop" not in player._cache
596
597
598class TestUpdateStateTiming:
599    """Micro-benchmarks guarding the cost of the update_state paths."""
600
601    def test_no_change_update_is_fast(self, player: MockPlayer) -> None:
602        """The no-change path completes well within the microsecond budget."""
603        for _ in range(50):  # warmup
604            player.update_state()
605        timings = []
606        for _ in range(200):
607            start = time.perf_counter()
608            player.update_state()
609            timings.append(time.perf_counter() - start)
610        duration = median(timings)
611        # target is <50us on a dev machine; assert with generous CI headroom
612        assert duration < 0.001, f"no-change update_state took {duration * 1e6:.0f}us (median)"
613
614    def test_full_update_is_fast(self, player: MockPlayer) -> None:
615        """A full recalculation completes well within the microsecond budget."""
616        for volume in range(50):  # warmup
617            player._attr_volume_level = volume
618            player.update_state(signal_event=False)
619        timings = []
620        for volume in range(200):
621            player._attr_volume_level = volume % 100
622            start = time.perf_counter()
623            player.update_state(signal_event=False)
624            timings.append(time.perf_counter() - start)
625        duration = median(timings)
626        # target is <500us on a dev machine; assert with generous CI headroom
627        assert duration < 0.005, f"full update_state took {duration * 1e6:.0f}us (median)"
628