music-assistant-server

1"""Tests for MSXPlayer."""
2
3from __future__ import annotations
4
5import asyncio
6from typing import Any
7from unittest.mock import AsyncMock, Mock, patch
8
9from music_assistant_models.enums import PlaybackState, PlayerFeature, PlayerType
10from music_assistant_models.player import PlayerMedia
11
12from music_assistant.controllers.players.constants import PlayerLockPurpose
13from music_assistant.providers.msx_bridge.player import MSXPlayer
14
15# --- Initialization and properties ---
16
17
18def test_init_defaults(player: MSXPlayer) -> None:
19    """MSXPlayer should have correct default attributes."""
20    assert player._attr_name == "Test TV"
21    assert player._attr_type == PlayerType.PLAYER
22    assert PlayerFeature.PAUSE in player._attr_supported_features
23    assert PlayerFeature.SET_MEMBERS in player._attr_supported_features
24    assert PlayerFeature.VOLUME_SET in player._attr_supported_features
25    assert player._attr_available is True
26    assert player._attr_powered is True
27    assert player._attr_volume_level == 100
28    assert player.output_format == "mp3"
29    assert player.requires_flow_mode is False
30
31
32def test_init_custom_params(provider: Any) -> None:
33    """MSXPlayer should accept custom name and output_format."""
34    p = MSXPlayer(provider, "msx_custom", name="Living Room TV", output_format="flac")
35    p.update_state = Mock()  # type: ignore[misc,method-assign]
36    assert p._attr_name == "Living Room TV"
37    assert p.output_format == "flac"
38
39
40def test_needs_poll_always_true(player: MSXPlayer) -> None:
41    """needs_poll should always return True."""
42    assert player.needs_poll is True
43
44
45def test_poll_interval_playing(player: MSXPlayer) -> None:
46    """poll_interval should return 5 when playing."""
47    player._attr_playback_state = PlaybackState.PLAYING
48    assert player.poll_interval == 5
49
50
51def test_poll_interval_not_playing(player: MSXPlayer) -> None:
52    """poll_interval should return 30 when idle or paused."""
53    player._attr_playback_state = PlaybackState.IDLE
54    assert player.poll_interval == 30
55    player._attr_playback_state = PlaybackState.PAUSED
56    assert player.poll_interval == 30
57
58
59# --- Playback ---
60
61
62async def test_play_media(player: MSXPlayer) -> None:
63    """play_media should store stream URL, set state to PLAYING, and reset elapsed."""
64    media = Mock(spec=PlayerMedia)
65    media.uri = "http://ma-server/stream/12345"
66
67    await player.play_media(media)
68
69    assert player.current_stream_url == "http://ma-server/stream/12345"
70    assert player._attr_playback_state == PlaybackState.PLAYING
71    assert player._attr_elapsed_time == 0
72    assert player._attr_elapsed_time_last_updated is not None
73    assert player._attr_current_media is media
74    player.update_state.assert_called()  # type: ignore[attr-defined]
75
76
77async def test_play_media_sets_media_ready_event(player: MSXPlayer) -> None:
78    """play_media should set _media_ready event so wait_for_media returns immediately."""
79    media = Mock(spec=PlayerMedia)
80    media.uri = "http://ma-server/stream/12345"
81
82    assert not player._media_ready.is_set()
83    await player.play_media(media)
84    assert player._media_ready.is_set()
85
86
87async def test_wait_for_media_returns_on_play(player: MSXPlayer) -> None:
88    """wait_for_media should return the media once play_media sets the event."""
89    media = Mock(spec=PlayerMedia)
90    media.uri = "http://ma-server/stream/12345"
91
92    async def delayed_play() -> None:
93        await asyncio.sleep(0.05)
94        await player.play_media(media)
95
96    task = asyncio.create_task(delayed_play())
97    result = await player.wait_for_media(timeout=2.0)
98    assert result is media
99    await task
100
101
102async def test_wait_for_media_fast_path(player: MSXPlayer) -> None:
103    """wait_for_media should return immediately if play_media already ran."""
104    media = Mock(spec=PlayerMedia)
105    media.uri = "http://ma-server/stream/12345"
106
107    # Simulate: queue.play_media already called player.play_media
108    await player.play_media(media)
109    assert player._media_ready.is_set()
110
111    # Fast path — should return instantly without clearing the event
112    result = await player.wait_for_media(timeout=0.1)
113    assert result is media
114
115
116async def test_wait_for_media_timeout(player: MSXPlayer) -> None:
117    """wait_for_media should return None on timeout."""
118    result = await player.wait_for_media(timeout=0.1)
119    assert result is None
120
121
122async def test_stop_does_not_clear_media_ready_event(player: MSXPlayer) -> None:
123    """
124    stop() must NOT clear _media_ready (C1 fix).
125
126    Clearing it in stop() would race with a concurrent wait_for_media() call.
127    The wait_for_media() fast-path already guards on _attr_current_media, so
128    leaving the event set is safe.
129    """
130    player._media_ready.set()
131    await player.stop()
132    # _attr_current_media is None after stop — wait_for_media returns None even
133    # though the event may still be set.
134    assert player._attr_current_media is None
135    result = await player.wait_for_media(timeout=0.05)
136    assert result is None
137
138
139async def test_expect_new_media_arms_wait_for_media(player: MSXPlayer) -> None:
140    """
141    After expect_new_media(), wait_for_media must wait for the NEXT play_media.
142
143    Without arming, the event left set by a previous track would make
144    wait_for_media return the stale current_media immediately — serving the
145    previous track's stream to the TV.
146    """
147    old_media = Mock(spec=PlayerMedia)
148    old_media.uri = "library://track/1"
149    await player.play_media(old_media)
150
151    new_media = Mock(spec=PlayerMedia)
152    new_media.uri = "library://track/2"
153    player.expect_new_media()
154
155    async def delayed_play() -> None:
156        await asyncio.sleep(0.05)
157        await player.play_media(new_media)
158
159    task = asyncio.create_task(delayed_play())
160    result = await player.wait_for_media(timeout=2.0)
161    assert result is new_media
162    await task
163
164
165async def test_expect_new_media_timeout_returns_none(player: MSXPlayer) -> None:
166    """After expect_new_media(), wait_for_media times out with None if no play_media arrives."""
167    old_media = Mock(spec=PlayerMedia)
168    old_media.uri = "library://track/1"
169    await player.play_media(old_media)
170
171    player.expect_new_media()
172    result = await player.wait_for_media(timeout=0.05)
173    assert result is None
174
175
176async def test_play_resume(player: MSXPlayer) -> None:
177    """play() when PAUSED should notify MSX to resume and set state to PLAYING."""
178    player._attr_playback_state = PlaybackState.PAUSED
179    player._attr_elapsed_time = 42.0
180
181    with patch.object(player.provider, "notify_play_resumed") as mock_notify:
182        await player.play()
183
184    assert player._attr_playback_state == PlaybackState.PLAYING
185    mock_notify.assert_called_once_with(player.player_id)
186
187
188async def test_pause_accumulates_time(player: MSXPlayer) -> None:
189    """pause() should accumulate elapsed time from last update."""
190    player._attr_playback_state = PlaybackState.PLAYING
191    player._attr_elapsed_time = 10.0
192    player._attr_elapsed_time_last_updated = 100.0
193
194    with patch("music_assistant.providers.msx_bridge.player.time") as mock_time:
195        mock_time.time.return_value = 115.0
196        await player.pause()
197
198    assert player._attr_playback_state == PlaybackState.PAUSED
199    assert player._attr_elapsed_time == 25.0  # 10 + (115 - 100)
200    player.update_state.assert_called()  # type: ignore[attr-defined]
201
202
203async def test_pause_none_elapsed(player: MSXPlayer) -> None:
204    """pause() should not crash when elapsed_time is None."""
205    player._attr_playback_state = PlaybackState.PLAYING
206    player._attr_elapsed_time = None
207    player._attr_elapsed_time_last_updated = None
208
209    await player.pause()
210
211    assert player._attr_playback_state == PlaybackState.PAUSED
212    # elapsed stays None since there was nothing to accumulate
213    assert player._attr_elapsed_time is None
214
215
216async def test_pause_notifies_pause_on_msx(player: MSXPlayer) -> None:
217    """pause() should call provider.notify_play_paused so MSX pauses the player."""
218    player._attr_playback_state = PlaybackState.PLAYING
219    player._attr_elapsed_time = 10.0
220    player._attr_elapsed_time_last_updated = 100.0
221
222    with patch.object(player.provider, "notify_play_paused") as mock_notify:
223        await player.pause()
224
225    assert player._attr_playback_state == PlaybackState.PAUSED
226    mock_notify.assert_called_once_with(player.player_id)
227
228
229async def test_stop_clears_all(player: MSXPlayer) -> None:
230    """stop() should reset state, media, elapsed, and stream URL."""
231    player._attr_playback_state = PlaybackState.PLAYING
232    player._attr_current_media = Mock()
233    player._attr_elapsed_time = 42.0
234    player.current_stream_url = "http://something"
235
236    await player.stop()
237
238    assert player._attr_playback_state == PlaybackState.IDLE
239    assert player._attr_current_media is None
240    assert player._attr_elapsed_time is None  # type: ignore[unreachable]
241    assert player._attr_elapsed_time_last_updated is None
242    assert player.current_stream_url is None
243    player.update_state.assert_called()
244
245
246async def test_stop_idempotent(player: MSXPlayer) -> None:
247    """Calling stop() on an idle player should not raise."""
248    player._attr_playback_state = PlaybackState.IDLE
249    await player.stop()
250    assert player._attr_playback_state == PlaybackState.IDLE
251
252
253# --- Volume and polling ---
254
255
256async def test_volume_set(player: MSXPlayer) -> None:
257    """volume_set should update volume level and call update_state."""
258    await player.volume_set(75)
259    assert player._attr_volume_level == 75
260    player.update_state.assert_called()  # type: ignore[attr-defined]
261
262
263async def test_poll_updates_elapsed(player: MSXPlayer) -> None:
264    """poll() should accumulate elapsed time during PLAYING."""
265    player._attr_playback_state = PlaybackState.PLAYING
266    player._attr_elapsed_time = 10.0
267    player._attr_elapsed_time_last_updated = 200.0
268
269    with patch("music_assistant.providers.msx_bridge.player.time") as mock_time:
270        mock_time.time.return_value = 205.0
271        await player.poll()
272
273    assert player._attr_elapsed_time == 15.0  # 10 + (205 - 200)
274    assert player._attr_elapsed_time_last_updated == 205.0
275    player.update_state.assert_called()  # type: ignore[attr-defined]
276
277
278async def test_poll_noop_when_paused(player: MSXPlayer) -> None:
279    """poll() should not update anything when paused."""
280    player._attr_playback_state = PlaybackState.PAUSED
281    player._attr_elapsed_time = 42.0
282    player.update_state.reset_mock()  # type: ignore[attr-defined]
283
284    await player.poll()
285
286    assert player._attr_elapsed_time == 42.0
287    player.update_state.assert_not_called()  # type: ignore[attr-defined]
288
289
290async def test_poll_noop_when_idle(player: MSXPlayer) -> None:
291    """poll() should not update anything when idle."""
292    player._attr_playback_state = PlaybackState.IDLE
293    player.update_state.reset_mock()  # type: ignore[attr-defined]
294
295    await player.poll()
296
297    player.update_state.assert_not_called()  # type: ignore[attr-defined]
298
299
300# --- Grouping ---
301
302
303async def test_set_members_add_and_remove(provider: Any, mass_mock: Mock) -> None:
304    """set_members should add and remove group members."""
305    leader = MSXPlayer(provider, "msx_leader", name="Leader TV", output_format="mp3")
306    leader.update_state = Mock()  # type: ignore[misc,method-assign]
307    member = MSXPlayer(provider, "msx_member", name="Member TV", output_format="mp3")
308    member.update_state = Mock()  # type: ignore[misc,method-assign]
309    mass_mock.players.get = mass_mock.players.get_player = Mock(
310        side_effect=lambda pid: member if pid == "msx_member" else None
311    )
312
313    await leader.set_members(player_ids_to_add=["msx_member"])
314
315    # Leader must be first, member second
316    assert leader._attr_group_members == ["msx_leader", "msx_member"]
317    leader.update_state.assert_called()
318
319    await leader.set_members(player_ids_to_remove=["msx_member"])
320
321    # No other members: list must be empty (solo player)
322    assert leader._attr_group_members == []
323
324
325async def test_set_members_ignores_self_and_non_msx(provider: Any, mass_mock: Mock) -> None:
326    """set_members should not add self or non-MSX players."""
327    leader = MSXPlayer(provider, "msx_leader", name="Leader TV", output_format="mp3")
328    leader.update_state = Mock()  # type: ignore[misc,method-assign]
329    mass_mock.players.get = mass_mock.players.get_player = Mock(return_value=None)
330
331    await leader.set_members(player_ids_to_add=["msx_leader", "msx_other", "sendspin_123"])
332
333    assert leader._attr_group_members == []
334
335
336async def test_play_media_propagates_to_group_members(provider: Any, mass_mock: Mock) -> None:
337    """play_media should propagate to group members through the internal handler."""
338    leader = MSXPlayer(provider, "msx_leader", name="Leader TV", output_format="mp3")
339    leader.update_state = Mock()  # type: ignore[misc,method-assign]
340    leader._attr_group_members = ["msx_leader", "msx_member"]
341    member = MSXPlayer(provider, "msx_member", name="Member TV", output_format="mp3")
342    member.update_state = Mock()  # type: ignore[misc,method-assign]
343    member.play_media = AsyncMock()  # type: ignore[method-assign]
344    mass_mock.players.get = mass_mock.players.get_player = Mock(return_value=member)
345
346    media = Mock(spec=PlayerMedia)
347    media.uri = "library://track/123"
348    media.title = None
349    media.artist = None
350    media.image_url = None
351    media.duration = None
352    media.source_id = None
353    media.queue_item_id = None
354
355    with patch.object(leader.provider, "notify_play_started", Mock()):
356        await leader.play_media(media)
357
358    mass_mock.players._handle_play_media.assert_awaited_once_with("msx_member", media)
359    member.play_media.assert_not_called()
360    mass_mock.players.get_player_lock.assert_called_once_with(
361        "msx_member", PlayerLockPurpose.PLAYBACK
362    )
363
364
365async def test_play_media_no_propagation_when_empty_group(provider: Any, mass_mock: Mock) -> None:
366    """play_media with empty group_members should not call mass.players.play_media."""
367    leader = MSXPlayer(provider, "msx_leader", name="Leader TV", output_format="mp3")
368    leader.update_state = Mock()  # type: ignore[misc,method-assign]
369    leader._attr_group_members = []
370    mass_mock.players.play_media = AsyncMock()
371
372    media = Mock(spec=PlayerMedia)
373    media.uri = "library://track/123"
374    media.title = None
375    media.artist = None
376    media.image_url = None
377    media.duration = None
378    media.source_id = None
379    media.queue_item_id = None
380
381    with patch.object(leader.provider, "notify_play_started", Mock()):
382        await leader.play_media(media)
383
384    mass_mock.players.play_media.assert_not_called()
385
386
387async def test_stop_propagates_to_group_members(provider: Any, mass_mock: Mock) -> None:
388    """stop() should propagate to group members when leader."""
389    leader = MSXPlayer(provider, "msx_leader", name="Leader TV", output_format="mp3")
390    leader.update_state = Mock()  # type: ignore[misc,method-assign]
391    leader._attr_group_members = ["msx_leader", "msx_member"]
392    member = MSXPlayer(provider, "msx_member", name="Member TV", output_format="mp3")
393    member.stop = AsyncMock()  # type: ignore[method-assign]
394    mass_mock.players.get = mass_mock.players.get_player = Mock(return_value=member)
395
396    with patch.object(leader.provider, "notify_play_stopped", Mock()):
397        await leader.stop()
398
399    mass_mock.players._handle_cmd_stop.assert_awaited_once_with("msx_member")
400    member.stop.assert_not_called()
401
402
403# --- Grouping: disable and recursion guard ---
404
405
406async def test_propagation_skipped_when_grouping_disabled(provider: Any, mass_mock: Mock) -> None:
407    """play_media should NOT propagate to members when grouping is disabled."""
408    provider.grouping_enabled = False
409    leader = MSXPlayer(
410        provider,
411        "msx_leader",
412        name="Leader TV",
413        output_format="mp3",
414        grouping_enabled=False,
415    )
416    leader.update_state = Mock()  # type: ignore[misc,method-assign]
417    leader._attr_group_members = ["msx_leader", "msx_member"]
418    member = MSXPlayer(
419        provider,
420        "msx_member",
421        name="Member TV",
422        output_format="mp3",
423        grouping_enabled=False,
424    )
425    member.play_media = AsyncMock()  # type: ignore[method-assign]
426    mass_mock.players.get = mass_mock.players.get_player = Mock(return_value=member)
427
428    media = Mock(spec=PlayerMedia)
429    media.uri = "library://track/123"
430    media.title = None
431    media.artist = None
432    media.image_url = None
433    media.duration = None
434    media.source_id = None
435    media.queue_item_id = None
436
437    with patch.object(leader.provider, "notify_play_started", Mock()):
438        await leader.play_media(media)
439
440    member.play_media.assert_not_called()
441
442
443def test_no_set_members_feature_when_grouping_disabled(provider: Any) -> None:
444    """MSXPlayer should not declare SET_MEMBERS when grouping is disabled."""
445    p = MSXPlayer(
446        provider,
447        "msx_nogrouping",
448        name="Solo TV",
449        output_format="mp3",
450        grouping_enabled=False,
451    )
452    p.update_state = Mock()  # type: ignore[misc,method-assign]
453    assert PlayerFeature.SET_MEMBERS not in p._attr_supported_features
454    assert p._attr_can_group_with == set()
455
456
457async def test_propagation_recursion_guard(provider: Any, mass_mock: Mock) -> None:
458    """Propagation should not recurse when member.play_media triggers propagation."""
459    leader = MSXPlayer(provider, "msx_leader", name="Leader TV", output_format="mp3")
460    leader.update_state = Mock()  # type: ignore[misc,method-assign]
461    leader._attr_group_members = ["msx_leader", "msx_member"]
462
463    # Create a member whose play_media calls back into leader's propagation
464    member = MSXPlayer(provider, "msx_member", name="Member TV", output_format="mp3")
465    member.update_state = Mock()  # type: ignore[misc,method-assign]
466    member._attr_group_members = [
467        "msx_member",
468        "msx_leader",
469    ]  # would cause recursion without guard
470
471    mass_mock.players.get = mass_mock.players.get_player = Mock(
472        side_effect=lambda pid: (
473            member if pid == "msx_member" else leader if pid == "msx_leader" else None
474        )
475    )
476
477    async def play_member(player_id: str, propagated_media: PlayerMedia) -> None:
478        await mass_mock.players.get_player(player_id).play_media(propagated_media)
479
480    mass_mock.players._handle_play_media.side_effect = play_member
481
482    media = Mock(spec=PlayerMedia)
483    media.uri = "library://track/123"
484    media.title = None
485    media.artist = None
486    media.image_url = None
487    media.duration = None
488    media.source_id = None
489    media.queue_item_id = None
490
491    with patch.object(leader.provider, "notify_play_started", Mock()):
492        # This should NOT infinitely recurse
493        await leader.play_media(media)
494
495    # Leader played successfully (no exception from recursion)
496    assert leader._attr_playback_state == PlaybackState.PLAYING
497
498
499# --- Queue-backed playlist playback ---
500
501
502async def test_play_media_queue_sends_playlist(player: MSXPlayer, mass_mock: Mock) -> None:
503    """play_media with queue context should send playlist via WS instead of stream."""
504    media = Mock(spec=PlayerMedia)
505    media.uri = "http://ma-server/stream/12345"
506    media.title = "Track 1"
507    media.artist = "Artist 1"
508    media.image_url = None
509    media.duration = 180
510    media.source_id = "msx_test"
511    media.queue_item_id = "qi1"
512
513    queue = Mock()
514    queue.current_index = 2
515
516    mass_mock.player_queues.get.return_value = queue
517    mass_mock.player_queues.get_item.return_value = None
518    mass_mock.player_queues.items.return_value = [
519        Mock(),
520        Mock(),
521        Mock(),
522        Mock(),
523        Mock(),
524    ]
525
526    with (
527        patch.object(player.provider, "notify_play_playlist") as mock_playlist,
528        patch.object(player.provider, "notify_play_started") as mock_play,
529    ):
530        await player.play_media(media)
531
532    mock_playlist.assert_called_once_with("msx_test", 2, queue_id="msx_test")
533    mock_play.assert_not_called()
534    assert player._playing_from_queue is True
535    assert player._playlist_offset == 2
536    assert player._playlist_size == 5
537
538
539async def test_play_media_sends_goto_index_when_playing_from_queue(
540    player: MSXPlayer, mass_mock: Mock
541) -> None:
542    """play_media should send translated goto_index when _playing_from_queue is True."""
543    player._playing_from_queue = True
544    player._queue_source_id = "msx_test"
545    player._playlist_offset = 2  # playlist was rotated by 2
546    player._playlist_size = 5  # 5 items in playlist
547
548    media = Mock(spec=PlayerMedia)
549    media.uri = "http://ma-server/stream/12345"
550    media.title = None
551    media.artist = None
552    media.image_url = None
553    media.duration = None
554    media.source_id = "msx_test"
555    media.queue_item_id = "qi2"
556
557    queue = Mock()
558    queue.current_index = 3  # MA index 3 → MSX index (3-2)%5 = 1
559
560    mass_mock.player_queues.get.return_value = queue
561    mass_mock.player_queues.get_item.return_value = None
562    # Return same size as _playlist_size to avoid "queue changed" re-send
563    mass_mock.player_queues.items.return_value = [Mock()] * 5
564
565    with (
566        patch.object(player.provider, "notify_goto_index") as mock_goto,
567        patch.object(player.provider, "notify_play_playlist") as mock_playlist,
568        patch.object(player.provider, "notify_play_started") as mock_play,
569    ):
570        await player.play_media(media)
571
572    # Index translated: (3 - 2) % 5 = 1
573    mock_goto.assert_called_once_with("msx_test", 1)
574    mock_playlist.assert_not_called()
575    mock_play.assert_not_called()
576
577
578async def test_play_media_skips_ws_when_skip_notify_set(player: MSXPlayer, mass_mock: Mock) -> None:
579    """play_media should skip all WS notifications when _skip_ws_notify is True."""
580    player._playing_from_queue = True
581    player._skip_ws_notify = True
582
583    media = Mock(spec=PlayerMedia)
584    media.uri = "http://ma-server/stream/12345"
585    media.title = None
586    media.artist = None
587    media.image_url = None
588    media.duration = None
589    media.source_id = "msx_test"
590    media.queue_item_id = "qi2"
591
592    mass_mock.player_queues.get_item.return_value = None
593
594    with (
595        patch.object(player.provider, "notify_goto_index") as mock_goto,
596        patch.object(player.provider, "notify_play_playlist") as mock_playlist,
597        patch.object(player.provider, "notify_play_started") as mock_play,
598    ):
599        await player.play_media(media)
600
601    mock_goto.assert_not_called()
602    mock_playlist.assert_not_called()
603    mock_play.assert_not_called()
604
605    # Clean up
606    player._skip_ws_notify = False
607
608
609async def test_play_media_non_queue_sends_broadcast_play(
610    player: MSXPlayer,
611) -> None:
612    """play_media without queue context should push the media metadata via broadcast_play."""
613    media = Mock(spec=PlayerMedia)
614    media.uri = "http://ma-server/stream/12345"
615    media.title = "Track 1"
616    media.artist = "Artist 1"
617    media.image_url = "http://ma-server/image.png"
618    media.duration = 180
619    media.stream_duration = None
620    media.source_id = None
621    media.queue_item_id = None
622
623    with (
624        patch.object(player.provider, "notify_play_playlist") as mock_playlist,
625        patch.object(player.provider, "notify_play_started") as mock_play,
626    ):
627        await player.play_media(media)
628
629    mock_playlist.assert_not_called()
630    mock_play.assert_called_once_with(
631        player.player_id,
632        title="Track 1",
633        artist="Artist 1",
634        image_url="http://ma-server/image.png",
635        duration=180,
636        next_action=f"request:interaction:/api/next/{player.player_id}",
637        prev_action=f"request:interaction:/api/previous/{player.player_id}",
638    )
639
640
641async def test_stop_resets_playing_from_queue(player: MSXPlayer) -> None:
642    """stop() should reset _playing_from_queue flag."""
643    player._playing_from_queue = True
644    await player.stop()
645    assert player._playing_from_queue is False
646
647
648# --- WebSocket position reporting ---
649
650
651def test_update_position(player: MSXPlayer) -> None:
652    """update_position should set elapsed_time and mark WS timestamp when PLAYING."""
653    player._attr_playback_state = PlaybackState.PLAYING
654    player.update_position(42.5)
655    assert player._attr_elapsed_time == 42.5
656    assert player._attr_elapsed_time_last_updated is not None
657    assert player._last_ws_position is not None
658    player.update_state.assert_called()  # type: ignore[attr-defined]
659
660
661def test_update_position_clamps_to_served_stream_duration(player: MSXPlayer) -> None:
662    """Position reports must not exceed the shortened stream served after a seek."""
663    media = Mock(spec=PlayerMedia)
664    media.duration = 300
665    media.stream_duration = 120
666    player._attr_current_media = media
667    player._attr_playback_state = PlaybackState.PLAYING
668
669    player.update_position(150)
670
671    assert player._attr_elapsed_time == 120
672
673
674def test_update_position_ignored_when_paused(player: MSXPlayer) -> None:
675    """update_position should be ignored when PAUSED to protect accumulated time."""
676    player._attr_playback_state = PlaybackState.PAUSED
677    player._attr_elapsed_time = 45.0
678    player.update_state.reset_mock()  # type: ignore[attr-defined]
679
680    player.update_position(99.0)
681
682    # elapsed_time should remain at 45.0, not be overwritten to 99.0
683    assert player._attr_elapsed_time == 45.0
684    player.update_state.assert_not_called()  # type: ignore[attr-defined]
685
686
687async def test_poll_skips_when_ws_position_recent(player: MSXPlayer) -> None:
688    """poll() should skip wall-clock increment when WS position was reported recently."""
689    player._attr_playback_state = PlaybackState.PLAYING
690    player._attr_elapsed_time = 30.0
691    player._attr_elapsed_time_last_updated = 200.0
692    player._last_ws_position = 200.0  # very recent (monotonic)
693
694    player.update_state.reset_mock()  # type: ignore[attr-defined]
695
696    with patch("music_assistant.providers.msx_bridge.player.time") as mock_time:
697        mock_time.time.return_value = 205.0
698        mock_time.monotonic.return_value = 205.0  # only 5s since last WS (< 10s threshold)
699        await player.poll()
700
701    # Should NOT have updated elapsed_time
702    assert player._attr_elapsed_time == 30.0
703    player.update_state.assert_not_called()  # type: ignore[attr-defined]
704
705
706async def test_poll_uses_wall_clock_when_ws_stale(player: MSXPlayer) -> None:
707    """poll() should use wall-clock when WS position is stale (>10s ago)."""
708    player._attr_playback_state = PlaybackState.PLAYING
709    player._attr_elapsed_time = 30.0
710    player._attr_elapsed_time_last_updated = 200.0
711    player._last_ws_position = 180.0  # 25s ago (monotonic)
712
713    with patch("music_assistant.providers.msx_bridge.player.time") as mock_time:
714        mock_time.time.return_value = 205.0
715        mock_time.monotonic.return_value = 205.0
716        await player.poll()
717
718    assert player._attr_elapsed_time == 35.0  # 30 + (205 - 200)
719
720
721async def test_poll_clamps_to_served_stream_duration(player: MSXPlayer) -> None:
722    """Wall-clock progress must stop at the shortened stream served after a seek."""
723    media = Mock(spec=PlayerMedia)
724    media.duration = 300
725    media.stream_duration = 120
726    player._attr_current_media = media
727    player._attr_playback_state = PlaybackState.PLAYING
728    player._attr_elapsed_time = 115.0
729    player._attr_elapsed_time_last_updated = 200.0
730    player._last_ws_position = None
731
732    with patch("music_assistant.providers.msx_bridge.player.time") as mock_time:
733        mock_time.time.return_value = 210.0
734        await player.poll()
735
736    assert player._attr_elapsed_time == 120
737
738
739async def test_poll_ws_staleness_immune_to_wall_clock_jump(player: MSXPlayer) -> None:
740    """
741    A wall-clock jump (NTP step) must not make a fresh WS position look stale.
742
743    The WS staleness check must use the monotonic clock: with wall-clock, an
744    NTP correction of +1h right after a WS report makes poll() fall back to
745    the wall-clock delta and corrupt elapsed_time by hours.
746    """
747    player._attr_playback_state = PlaybackState.PLAYING
748    player._attr_elapsed_time = 30.0
749
750    with patch("music_assistant.providers.msx_bridge.player.time") as mock_time:
751        mock_time.time.return_value = 200.0
752        mock_time.monotonic.return_value = 1000.0
753        player.update_position(42.0)
754
755        # NTP jumps wall clock forward 1 hour; monotonic advances only 5s
756        mock_time.time.return_value = 200.0 + 3600.0
757        mock_time.monotonic.return_value = 1005.0
758        await player.poll()
759
760    # WS report is 5s old (monotonic) — still fresh, elapsed must be untouched
761    assert player._attr_elapsed_time == 42.0
762
763
764async def test_stop_clears_ws_position(player: MSXPlayer) -> None:
765    """stop() should clear _last_ws_position."""
766    player._last_ws_position = 100.0
767    await player.stop()
768    assert player._last_ws_position is None
769
770
771# --- Resume from pause ---
772
773
774async def test_resume_sends_ws_resume(player: MSXPlayer) -> None:
775    """play() when PAUSED should notify MSX to resume native player."""
776    player._attr_playback_state = PlaybackState.PAUSED
777    player._attr_elapsed_time = 42.0
778
779    with patch.object(player.provider, "notify_play_resumed") as mock_notify:
780        await player.play()
781
782    assert player._attr_playback_state == PlaybackState.PLAYING
783    assert player._attr_elapsed_time_last_updated is not None
784    mock_notify.assert_called_once_with(player.player_id)
785
786
787async def test_resume_skips_ws_when_skip_notify(player: MSXPlayer) -> None:
788    """play() when PAUSED with _skip_ws_notify should not broadcast to MSX."""
789    player._attr_playback_state = PlaybackState.PAUSED
790    player._attr_elapsed_time = 10.0
791    player._skip_ws_notify = True
792
793    with patch.object(player.provider, "notify_play_resumed") as mock_notify:
794        await player.play()
795
796    assert player._attr_playback_state == PlaybackState.PLAYING
797    mock_notify.assert_not_called()
798    player._skip_ws_notify = False
799
800
801async def test_pause_skips_ws_when_skip_notify(player: MSXPlayer) -> None:
802    """pause() with _skip_ws_notify should not broadcast to MSX."""
803    player._attr_playback_state = PlaybackState.PLAYING
804    player._attr_elapsed_time = 10.0
805    player._attr_elapsed_time_last_updated = 100.0
806    player._skip_ws_notify = True
807
808    with patch.object(player.provider, "notify_play_paused") as mock_notify:
809        await player.pause()
810
811    assert player._attr_playback_state == PlaybackState.PAUSED
812    mock_notify.assert_not_called()
813    player._skip_ws_notify = False
814