/
/
/
1"""Tests for Flow Mode queue progress tracking."""
2
3from types import SimpleNamespace
4from typing import Any, cast
5from unittest.mock import MagicMock
6
7from music_assistant_models.enums import PlaybackState
8
9from music_assistant.controllers.player_queues.playback_tracker import PlaybackTrackerMixin
10
11QUEUE_ID = "queue-1"
12
13
14def _queue_item(item_id: str, seconds_streamed: float | None) -> SimpleNamespace:
15 """Create a minimal queue item with stream bookkeeping."""
16 return SimpleNamespace(
17 queue_item_id=item_id,
18 streamdetails=SimpleNamespace(
19 seconds_streamed=seconds_streamed,
20 seek_position=0,
21 ),
22 extra_attributes={},
23 )
24
25
26def _flow_progress(
27 elapsed: float,
28 log: list[SimpleNamespace],
29 items: list[SimpleNamespace],
30) -> tuple[int | None, float]:
31 """Calculate Flow Mode progress with a minimal playback tracker."""
32 queue = SimpleNamespace(queue_id=QUEUE_ID, current_index=0, elapsed_time=0)
33 tracker = MagicMock()
34 tracker._queue_data = {
35 QUEUE_ID: SimpleNamespace(flow_mode_stream_log=log),
36 }
37 tracker.index_by_id.side_effect = lambda _queue_id, item_id: next(
38 (index for index, item in enumerate(items) if item.queue_item_id == item_id),
39 None,
40 )
41 tracker.get_item.side_effect = lambda _queue_id, id_or_index: (
42 items[id_or_index]
43 if isinstance(id_or_index, int)
44 else next((item for item in items if item.queue_item_id == id_or_index), None)
45 )
46 player = SimpleNamespace(
47 state=SimpleNamespace(
48 corrected_elapsed_time=elapsed,
49 playback_state=PlaybackState.PLAYING,
50 )
51 )
52
53 return PlaybackTrackerMixin._get_flow_queue_stream_index(
54 cast("Any", tracker), cast("Any", queue), cast("Any", player)
55 )
56
57
58def test_flow_progress_recovers_unfinished_non_tail_probe_entry() -> None:
59 """An abandoned probe entry must not pin progress to the previous track."""
60 items = [_queue_item("track-1", 100.0), _queue_item("track-2", None)]
61 log = [
62 SimpleNamespace(queue_item_id="track-1", seconds_streamed=None),
63 SimpleNamespace(queue_item_id="track-2", seconds_streamed=None),
64 ]
65
66 index, elapsed = _flow_progress(125.0, log, items)
67
68 assert index == 1
69 assert elapsed == 25.0
70
71
72def test_flow_progress_keeps_unfinished_tail_as_active_entry() -> None:
73 """The unfinished tail entry remains the sentinel for the active track."""
74 items = [_queue_item("track-1", None)]
75 log = [SimpleNamespace(queue_item_id="track-1", seconds_streamed=None)]
76
77 index, elapsed = _flow_progress(25.0, log, items)
78
79 assert index == 0
80 assert elapsed == 25.0
81