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
125class TestNativeCurrentMediaPosition:
126    """The published position of native current_media and the timestamp it is paired with."""
127
128    def _playing_player(
129        self,
130        mock_mass: MagicMock,
131        *,
132        media_elapsed_time: int | None,
133        media_last_updated: float | None,
134        player_elapsed_time: float | None,
135        player_last_updated: float | None,
136    ) -> MockPlayer:
137        """Build a playing player with the given media and player level position anchors."""
138        provider = MockProvider("test_provider", mass=mock_mass)
139        player = MockPlayer(provider, "player_1", "Player 1")
140        player._attr_playback_state = PlaybackState.PLAYING
141        player._attr_elapsed_time = player_elapsed_time
142        player._attr_elapsed_time_last_updated = player_last_updated
143        player.set_current_media(uri="http://test/stream", title="Test")
144        assert player._attr_current_media is not None
145        player._attr_current_media.elapsed_time = media_elapsed_time
146        player._attr_current_media.elapsed_time_last_updated = media_last_updated
147        player.update_state(signal_event=False)
148        return player
149
150    def test_zero_position_is_reported(self, mock_mass: MagicMock) -> None:
151        """A position of zero (start of a track) is reported instead of being dropped."""
152        anchor = time.time()
153        player = self._playing_player(
154            mock_mass,
155            media_elapsed_time=0,
156            media_last_updated=anchor,
157            player_elapsed_time=0,
158            player_last_updated=anchor,
159        )
160
161        assert player.state.current_media is not None
162        assert player.state.current_media.elapsed_time == 0
163        assert player.state.current_media.elapsed_time_last_updated == anchor
164
165    def test_media_position_wins_without_player_position(self, mock_mass: MagicMock) -> None:
166        """A media position is reported even when the player reports none of its own."""
167        anchor = time.time()
168        player = self._playing_player(
169            mock_mass,
170            media_elapsed_time=42,
171            media_last_updated=anchor,
172            player_elapsed_time=None,
173            player_last_updated=None,
174        )
175
176        assert player.state.current_media is not None
177        assert player.state.current_media.elapsed_time == 42
178        assert player.state.current_media.elapsed_time_last_updated == anchor
179
180    def test_media_position_is_never_paired_with_the_player_timestamp(
181        self, mock_mass: MagicMock
182    ) -> None:
183        """A media position without its own timestamp does not borrow the player's."""
184        player = self._playing_player(
185            mock_mass,
186            media_elapsed_time=42,
187            media_last_updated=None,
188            player_elapsed_time=10.0,
189            player_last_updated=time.time(),
190        )
191
192        assert player.state.current_media is not None
193        assert player.state.current_media.elapsed_time == 42
194        assert player.state.current_media.elapsed_time_last_updated is None
195
196    def test_late_arriving_position_is_adopted(self, mock_mass: MagicMock) -> None:
197        """A position reported after the media was published still reaches the state."""
198        player = self._playing_player(
199            mock_mass,
200            media_elapsed_time=None,
201            media_last_updated=None,
202            player_elapsed_time=None,
203            player_last_updated=None,
204        )
205        published = player.state.current_media
206        assert published is not None
207        assert published.elapsed_time is None
208
209        anchor = time.time()
210        player._attr_elapsed_time = 5.0
211        player._attr_elapsed_time_last_updated = anchor
212        player.update_state(signal_event=False)
213
214        published = player.state.current_media
215        assert published is not None
216        assert published.elapsed_time == 5
217        assert published.elapsed_time_last_updated == anchor
218
219    def test_steady_playback_keeps_the_previous_anchor(self, mock_mass: MagicMock) -> None:
220        """A regular playback tick keeps the anchor it already published."""
221        anchor = time.time()
222        player = self._playing_player(
223            mock_mass,
224            media_elapsed_time=None,
225            media_last_updated=None,
226            player_elapsed_time=10.0,
227            player_last_updated=anchor,
228        )
229
230        player._attr_elapsed_time = 11.0
231        player._attr_elapsed_time_last_updated = anchor + 1
232        player.update_state(signal_event=False)
233
234        assert player.state.current_media is not None
235        assert player.state.current_media.elapsed_time == 10
236        assert player.state.current_media.elapsed_time_last_updated == anchor
237
238    def test_player_position_is_paired_with_its_own_timestamp(self, mock_mass: MagicMock) -> None:
239        """Falling back to the player position also takes the player level timestamp."""
240        player_anchor = time.time()
241        player = self._playing_player(
242            mock_mass,
243            media_elapsed_time=None,
244            media_last_updated=player_anchor - 30,
245            player_elapsed_time=42.7,
246            player_last_updated=player_anchor,
247        )
248
249        assert player.state.current_media is not None
250        assert player.state.current_media.elapsed_time == 42
251        assert player.state.current_media.elapsed_time_last_updated == player_anchor
252
253    def test_no_position_reports_no_timestamp(self, mock_mass: MagicMock) -> None:
254        """Without any position, the timestamp is dropped along with it."""
255        player = self._playing_player(
256            mock_mass,
257            media_elapsed_time=None,
258            media_last_updated=time.time(),
259            player_elapsed_time=None,
260            player_last_updated=time.time(),
261        )
262
263        assert player.state.current_media is not None
264        assert player.state.current_media.elapsed_time is None
265        assert player.state.current_media.elapsed_time_last_updated is None
266
267
268class TestStreamMetadataPosition:
269    """The published position of a queue item carrying live stream metadata."""
270
271    def _playing_player(
272        self,
273        mock_mass: MagicMock,
274        *,
275        metadata_elapsed_time: int | None,
276        metadata_last_updated: float | None,
277        queue_elapsed_time: float,
278        queue_last_updated: float,
279        media_type: MediaType = MediaType.RADIO,
280    ) -> MockPlayer:
281        """Build a playing player whose active queue item carries live stream metadata."""
282        queue = PlayerQueue(
283            queue_id="player_1",
284            active=True,
285            display_name="Player 1",
286            available=True,
287            items=1,
288            elapsed_time=queue_elapsed_time,
289            elapsed_time_last_updated=queue_last_updated,
290            current_item=QueueItem(
291                queue_id="player_1",
292                queue_item_id="item_1",
293                name="Live Radio",
294                duration=None,
295                streamdetails=StreamDetails(
296                    provider="test_provider",
297                    item_id="item_1",
298                    audio_format=AudioFormat(),
299                    media_type=media_type,
300                    stream_metadata=StreamMetadata(
301                        title="Live Track",
302                        elapsed_time=metadata_elapsed_time,
303                        elapsed_time_last_updated=metadata_last_updated,
304                    ),
305                ),
306            ),
307        )
308        mock_mass.player_queues.get = MagicMock(return_value=queue)
309        provider = MockProvider("test_provider", mass=mock_mass)
310        player = MockPlayer(provider, "player_1", "Player 1")
311        player._attr_playback_state = PlaybackState.PLAYING
312        player.update_state(signal_event=False)
313        return player
314
315    def test_zero_metadata_position_is_reported(self, mock_mass: MagicMock) -> None:
316        """A live track that just started reports zero, not the elapsed stream time."""
317        metadata_anchor = time.time()
318        player = self._playing_player(
319            mock_mass,
320            metadata_elapsed_time=0,
321            metadata_last_updated=metadata_anchor,
322            queue_elapsed_time=300.0,
323            queue_last_updated=metadata_anchor - 30,
324        )
325
326        assert player.state.current_media is not None
327        assert player.state.current_media.title == "Live Track"
328        assert player.state.current_media.elapsed_time == 0
329        assert player.state.current_media.elapsed_time_last_updated == metadata_anchor
330
331    def test_queue_position_is_paired_with_its_own_timestamp(self, mock_mass: MagicMock) -> None:
332        """Without a metadata position, both values come from the queue."""
333        queue_anchor = time.time()
334        player = self._playing_player(
335            mock_mass,
336            metadata_elapsed_time=None,
337            metadata_last_updated=queue_anchor - 30,
338            queue_elapsed_time=300.5,
339            queue_last_updated=queue_anchor,
340        )
341
342        assert player.state.current_media is not None
343        assert player.state.current_media.title == "Live Track"
344        assert player.state.current_media.elapsed_time == 300
345        assert player.state.current_media.elapsed_time_last_updated == queue_anchor
346
347    def test_audio_source_position_matches_player_position(self, mock_mass: MagicMock) -> None:
348        """An upstream source position is reported identically on player and media level."""
349        metadata_anchor = time.time()
350        player = self._playing_player(
351            mock_mass,
352            metadata_elapsed_time=0,
353            metadata_last_updated=metadata_anchor,
354            queue_elapsed_time=300.0,
355            queue_last_updated=metadata_anchor - 30,
356            media_type=MediaType.AUDIO_SOURCE,
357        )
358
359        assert player.state.current_media is not None
360        assert player.state.elapsed_time == 0
361        assert player.state.current_media.elapsed_time == 0
362        assert player.state.elapsed_time_last_updated == metadata_anchor
363        assert player.state.current_media.elapsed_time_last_updated == metadata_anchor
364
365
366class TestMediaUpdatedCallback:
367    """The (debounced) media-updated callback fires on media identity changes."""
368
369    def test_palette_resolution_fires_media_updated(
370        self, mock_mass: MagicMock, player: MockPlayer
371    ) -> None:
372        """A late palette resolution re-fires the media-updated callback."""
373        player.set_current_media(uri="http://test/stream", title="Test", image_url="http://img")
374        player.update_state(signal_event=False)
375        mock_mass.call_later.reset_mock()
376
377        player.set_resolved_palette("http://img", MagicMock())
378        player.update_state(force_update=True, signal_event=False)
379
380        assert any(
381            call.kwargs.get("task_id") == f"player_media_updated_{player.player_id}"
382            for call in mock_mass.call_later.call_args_list
383        )
384
385
386class TestCacheInvalidationClasses:
387    """Config-derived cached properties survive state updates, all others refresh."""
388
389    def test_config_cached_props_survive_state_updates(self, player: MockPlayer) -> None:
390        """Config-derived cached properties are not recomputed on state updates."""
391        assert "icon" in player._cache
392        marker = player._cache["icon"]
393        player._attr_volume_level = 60
394        player.update_state(signal_event=False)
395        assert player._cache.get("icon") is marker
396
397    def test_set_config_invalidates_all_cached_props(self, player: MockPlayer) -> None:
398        """set_config invalidates every cached property, including config-derived ones."""
399        assert "icon" in player._cache
400        player.set_config(player.config)
401        assert len(player._cache) == 0
402
403    def test_player_implementation_cached_props_cleared_each_update(
404        self, player: MockPlayer
405    ) -> None:
406        """Cached properties defined by player implementations refresh on every update."""
407        player._cache["some_provider_prop"] = object()
408        player.update_state()
409        assert "some_provider_prop" not in player._cache
410
411
412class TestUpdateStateTiming:
413    """Micro-benchmarks guarding the cost of the update_state paths."""
414
415    def test_no_change_update_is_fast(self, player: MockPlayer) -> None:
416        """The no-change path completes well within the microsecond budget."""
417        for _ in range(50):  # warmup
418            player.update_state()
419        timings = []
420        for _ in range(200):
421            start = time.perf_counter()
422            player.update_state()
423            timings.append(time.perf_counter() - start)
424        duration = median(timings)
425        # target is <50us on a dev machine; assert with generous CI headroom
426        assert duration < 0.001, f"no-change update_state took {duration * 1e6:.0f}us (median)"
427
428    def test_full_update_is_fast(self, player: MockPlayer) -> None:
429        """A full recalculation completes well within the microsecond budget."""
430        for volume in range(50):  # warmup
431            player._attr_volume_level = volume
432            player.update_state(signal_event=False)
433        timings = []
434        for volume in range(200):
435            player._attr_volume_level = volume % 100
436            start = time.perf_counter()
437            player.update_state(signal_event=False)
438            timings.append(time.perf_counter() - start)
439        duration = median(timings)
440        # target is <500us on a dev machine; assert with generous CI headroom
441        assert duration < 0.005, f"full update_state took {duration * 1e6:.0f}us (median)"
442