/
/
/
1"""
2Tests for PlayerQueuesController flow-stream EOF tracking.
3
4Player providers query ``flow_stream_finished`` to recognise an end-of-stream
5condition that does not surface as a normal idle report (e.g. a Cast group that
6underruns the LIVE flow stream at EOF and keeps reporting buffering/playing).
7"""
8
9from __future__ import annotations
10
11from typing import TYPE_CHECKING, Any, cast
12from unittest.mock import MagicMock
13
14from music_assistant.controllers.player_queues import PlayerQueuesController
15from music_assistant.controllers.player_queues.state import PlayerQueueData
16
17if TYPE_CHECKING:
18 from music_assistant_models.player_queue import PlayerQueue
19
20
21def _fake_controller(queue: MagicMock) -> MagicMock:
22 """Build a MagicMock standing in for the controller, owning a single queue."""
23
24 def _close_coro(coro: Any, **_kwargs: Any) -> None:
25 # close the scheduled resume coroutine so it doesn't raise 'never awaited'
26 coro.close()
27
28 fake = MagicMock()
29 data = PlayerQueueData(queue=cast("PlayerQueue", queue))
30 # session_id is server-side state on the record now, not on the wire queue
31 data.session_id = queue.session_id
32 fake._queue_data = {queue.queue_id: data}
33 fake.get = MagicMock(
34 side_effect=lambda qid: d.queue if (d := fake._queue_data.get(qid)) else None
35 )
36 fake.queue_data_or_none = MagicMock(side_effect=lambda qid: fake._queue_data.get(qid))
37 fake.mass.create_task = MagicMock(side_effect=_close_coro)
38 return fake
39
40
41def _finished(fake: MagicMock, queue_id: str) -> bool:
42 return PlayerQueuesController.flow_stream_finished(
43 cast("PlayerQueuesController", fake), queue_id
44 )
45
46
47def test_flow_stream_finished_false_without_completion() -> None:
48 """An untouched queue reports its flow stream as not finished."""
49 queue = MagicMock(queue_id="q", session_id="sess-1")
50 fake = _fake_controller(queue)
51
52 assert _finished(fake, "q") is False
53
54
55def test_queue_buffer_completed_marks_session_finished() -> None:
56 """Buffer completion records the session so flow_stream_finished returns True."""
57 queue = MagicMock(queue_id="q", session_id="sess-1")
58 fake = _fake_controller(queue)
59
60 PlayerQueuesController.queue_buffer_completed(
61 cast("PlayerQueuesController", fake), "q", queue_exhausted=True
62 )
63
64 assert fake._queue_data["q"].flow_buffer_completed == "sess-1"
65 assert _finished(fake, "q") is True
66
67
68def test_flow_stream_finished_invalidated_by_new_session() -> None:
69 """A completion from a previous session must not count for a fresh session."""
70 queue = MagicMock(queue_id="q", session_id="sess-1")
71 fake = _fake_controller(queue)
72 PlayerQueuesController.queue_buffer_completed(
73 cast("PlayerQueuesController", fake), "q", queue_exhausted=True
74 )
75
76 # a new playback session starts (play_index assigns a fresh session_id)
77 fake._queue_data["q"].session_id = "sess-2"
78
79 assert _finished(fake, "q") is False
80
81
82def test_flow_stream_finished_false_for_unknown_queue() -> None:
83 """An unknown queue id never reports a finished flow stream."""
84 queue = MagicMock(queue_id="q", session_id="sess-1")
85 fake = _fake_controller(queue)
86
87 assert _finished(fake, "other") is False
88
89
90def _exhausted(fake: MagicMock, queue_id: str, session_id: str) -> bool:
91 return PlayerQueuesController.flow_queue_exhausted(
92 cast("PlayerQueuesController", fake), queue_id, session_id
93 )
94
95
96def test_flow_queue_exhausted_set_when_queue_ran_out() -> None:
97 """Playing the queue to its end marks the session as exhausted."""
98 queue = MagicMock(queue_id="q", session_id="sess-1")
99 fake = _fake_controller(queue)
100
101 PlayerQueuesController.queue_buffer_completed(
102 cast("PlayerQueuesController", fake), "q", queue_exhausted=True
103 )
104
105 assert _exhausted(fake, "q", "sess-1") is True
106
107
108def test_flow_queue_not_exhausted_on_early_restart() -> None:
109 """A flow that ended early to be restarted has no tail left to play out."""
110 queue = MagicMock(queue_id="q", session_id="sess-1")
111 fake = _fake_controller(queue)
112
113 PlayerQueuesController.queue_buffer_completed(
114 cast("PlayerQueuesController", fake), "q", queue_exhausted=False
115 )
116
117 assert fake._queue_data["q"].flow_buffer_completed == "sess-1"
118 assert _exhausted(fake, "q", "sess-1") is False
119
120
121def test_flow_queue_exhausted_invalidated_by_new_session() -> None:
122 """A previous session's end of queue must not count for a fresh session."""
123 queue = MagicMock(queue_id="q", session_id="sess-1")
124 fake = _fake_controller(queue)
125 PlayerQueuesController.queue_buffer_completed(
126 cast("PlayerQueuesController", fake), "q", queue_exhausted=True
127 )
128
129 fake._queue_data["q"].session_id = "sess-2"
130
131 assert _exhausted(fake, "q", "sess-1") is False
132 assert _exhausted(fake, "q", "sess-2") is False
133
134
135def test_flow_queue_exhausted_false_for_unknown_queue() -> None:
136 """An unknown queue id never reports an exhausted flow stream."""
137 queue = MagicMock(queue_id="q", session_id="sess-1")
138 fake = _fake_controller(queue)
139
140 assert _exhausted(fake, "other", "sess-1") is False
141