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