/
/
1"""Tests that ``play_index`` resets the queue's elapsed clock with the item switch."""
2
3from __future__ import annotations
4
5import time
6from unittest.mock import AsyncMock, MagicMock, Mock
7
8from music_assistant_models.errors import MediaNotFoundError
9from music_assistant_models.player_queue import PlayerQueue
10from music_assistant_models.queue_item import QueueItem
11
12from music_assistant.controllers.player_queues import PlayerQueuesController
13from music_assistant.controllers.player_queues.state import PlayerQueueData
14
15QUEUE_ID = "q1"
16STALE_ELAPSED = 43_217.7
17
18
19def _controller_with_stale_queue() -> tuple[
20 PlayerQueuesController, PlayerQueue, list[tuple[str | None, float]]
21]:
22 """Build a bare controller mid-way through item "old", recording signal_update states."""
23 ctrl = PlayerQueuesController.__new__(PlayerQueuesController)
24 queue = PlayerQueue(queue_id=QUEUE_ID, active=True, display_name="Q1", available=True, items=2)
25 items = [
26 QueueItem(queue_id=QUEUE_ID, queue_item_id="old", name="old", duration=50_000),
27 QueueItem(queue_id=QUEUE_ID, queue_item_id="new", name="new", duration=279),
28 ]
29 queue.current_index = 0
30 queue.current_item = items[0]
31 queue.elapsed_time = STALE_ELAPSED
32 queue.elapsed_time_last_updated = time.time() - 900
33 queue_data = PlayerQueueData(queue=queue)
34 queue_data.items = items
35 ctrl._queue_data = {QUEUE_ID: queue_data}
36
37 signals: list[tuple[str | None, float]] = []
38
39 def _record_signal(_queue_id: str, items_changed: bool = False) -> None: # noqa: ARG001
40 item = queue.current_item
41 signals.append((item.queue_item_id if item else None, queue.elapsed_time))
42
43 ctrl.signal_update = Mock(side_effect=_record_signal) # type: ignore[method-assign]
44 ctrl._check_player_permission = Mock() # type: ignore[method-assign]
45 ctrl._set_transitioning = Mock() # type: ignore[method-assign]
46 ctrl._load_item = AsyncMock() # type: ignore[method-assign]
47 ctrl._get_next_index = Mock(return_value=None) # type: ignore[method-assign]
48 ctrl.player_media_from_queue_item = AsyncMock() # type: ignore[method-assign]
49 ctrl.mass = MagicMock()
50 ctrl.mass.players.play_media = AsyncMock()
51 ctrl.logger = MagicMock() # the retry-fallback path logs a warning
52 return ctrl, queue, signals
53
54
55async def test_play_index_resets_elapsed_with_item_switch() -> None:
56 """Starting a new item zeroes elapsed_time and freshens its timestamp."""
57 ctrl, queue, signals = _controller_with_stale_queue()
58 before = time.time()
59
60 await ctrl.play_index(QUEUE_ID, 1)
61
62 assert queue.current_item is not None
63 assert queue.current_item.queue_item_id == "new"
64 assert queue.elapsed_time == 0
65 assert queue.elapsed_time_last_updated >= before
66 # no signaled state may pair the new item with the previous item's elapsed
67 assert all(elapsed == 0 for item_id, elapsed in signals if item_id == "new"), signals
68
69
70async def test_play_index_sets_elapsed_to_seek_position() -> None:
71 """A seeked start (resume, audiobook resume point) anchors elapsed at the seek."""
72 ctrl, queue, signals = _controller_with_stale_queue()
73
74 await ctrl.play_index(QUEUE_ID, 1, seek_position=30)
75
76 assert queue.elapsed_time == 30
77 assert all(elapsed == 30 for item_id, elapsed in signals if item_id == "new"), signals
78
79
80async def test_seek_publishes_target_before_restarting_stream() -> None:
81 """A seek publishes its target before play_index starts rebuilding the stream."""
82 ctrl, queue, _signals = _controller_with_stale_queue()
83 events: list[tuple[str, float]] = []
84 before = time.time()
85
86 def _record_signal(_queue_id: str, items_changed: bool = False) -> None: # noqa: ARG001
87 events.append(("signal", queue.elapsed_time))
88
89 async def _record_restart(*_args: object, **_kwargs: object) -> None:
90 events.append(("restart", queue.elapsed_time))
91
92 ctrl.signal_update = Mock(side_effect=_record_signal) # type: ignore[method-assign]
93 play_index = AsyncMock(side_effect=_record_restart)
94 ctrl.play_index = play_index # type: ignore[method-assign]
95
96 await ctrl.seek(QUEUE_ID, 30)
97
98 play_index.assert_awaited_once_with(QUEUE_ID, 0, seek_position=30)
99 assert events == [("signal", 30), ("restart", 30)]
100 assert queue.elapsed_time == 30
101 assert queue.elapsed_time_last_updated >= before
102
103
104async def test_play_index_retry_fallback_discards_failed_items_seek_position() -> None:
105 """Falling back to another item after a load failure zeroes elapsed_time, not seek_position."""
106 ctrl, queue, signals = _controller_with_stale_queue()
107 fallback_item = QueueItem(
108 queue_id=QUEUE_ID, queue_item_id="fallback", name="fallback", duration=200
109 )
110 ctrl._queue_data[QUEUE_ID].items.append(fallback_item)
111 ctrl._load_item = AsyncMock(side_effect=[MediaNotFoundError("unplayable"), None]) # type: ignore[method-assign]
112 ctrl._get_next_index = Mock(return_value=2) # type: ignore[method-assign]
113
114 await ctrl.play_index(QUEUE_ID, 1, seek_position=45)
115
116 assert queue.current_item is not None
117 assert queue.current_item.queue_item_id == "fallback"
118 assert queue.elapsed_time == 0
119 assert all(elapsed == 0 for item_id, elapsed in signals if item_id == "fallback"), signals
120