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