/
/
/
1"""
2Tests for which buffers a queue's audio cleanup is allowed to clear.
3
4A stop tears down the audio of the session it was issued for. When playback restarts
5before that teardown gets to run - only possible once the playback lock gives up on a
6wedged holder - the replacement session's producers must survive it, while the stopped
7session's producers still have to be killed.
8"""
9
10from __future__ import annotations
11
12from typing import cast
13from unittest.mock import AsyncMock, MagicMock
14
15from music_assistant_models.enums import ContentType, MediaType, StreamType
16from music_assistant_models.media_items import AudioFormat
17from music_assistant_models.queue_item import QueueItem
18from music_assistant_models.streamdetails import StreamDetails
19
20from music_assistant.controllers.player_queues import PlayerQueuesController
21from music_assistant.controllers.player_queues.state import PlayerQueueData
22from music_assistant.controllers.streams.audio_buffer import AudioBuffer
23
24QUEUE_ID = "q1"
25
26
27def _item(item_id: str, session_id: str | None) -> QueueItem:
28 """
29 Build a queue item with a buffer attached, owned by the given playback session.
30
31 :param item_id: The queue item id.
32 :param session_id: Session to stamp on the stream details, or None to leave unstamped.
33 """
34 queue_item = QueueItem(queue_id=QUEUE_ID, queue_item_id=item_id, name=item_id, duration=180)
35 audio_buffer = MagicMock(spec=AudioBuffer)
36 audio_buffer.clear = AsyncMock()
37 queue_item.streamdetails = StreamDetails(
38 provider="local--1",
39 item_id=item_id,
40 audio_format=AudioFormat(content_type=ContentType.MP3),
41 media_type=MediaType.TRACK,
42 stream_type=StreamType.HTTP,
43 path=f"http://test.invalid/{item_id}.mp3",
44 queue_id=QUEUE_ID,
45 )
46 queue_item.streamdetails.queue_session_id = session_id
47 queue_item.streamdetails.buffer = audio_buffer
48 return queue_item
49
50
51def _controller(items: list[QueueItem], playing: str | None = None) -> PlayerQueuesController:
52 """
53 Build a bare controller holding one queue with the given items.
54
55 :param items: The queue's items.
56 :param playing: Session the queue is playing now, or None once the stop cleared it.
57 """
58 ctrl = PlayerQueuesController.__new__(PlayerQueuesController)
59 ctrl.logger = MagicMock()
60 ctrl._queue_data = {
61 QUEUE_ID: PlayerQueueData(queue=MagicMock(), items=items, session_id=playing)
62 }
63 ctrl.mass = MagicMock()
64 return ctrl
65
66
67async def test_a_stop_leaves_a_newer_sessions_buffers_alone() -> None:
68 """Playback that restarted before the teardown ran keeps the audio it prepared."""
69 stopped = _item("stopped", "sess-1")
70 replacement = _item("replacement", "sess-2")
71 ctrl = _controller([stopped, replacement], playing="sess-2")
72
73 await ctrl._cleanup_queue_audio_data(QUEUE_ID, "sess-1")
74
75 assert stopped.streamdetails is not None
76 assert stopped.streamdetails.buffer is None
77 assert replacement.streamdetails is not None
78 assert replacement.streamdetails.buffer is not None
79 replacement.streamdetails.buffer.clear.assert_not_awaited()
80
81
82async def test_a_stop_still_kills_every_buffer_of_its_own_session() -> None:
83 """The stopped session's producers are what a stop exists to release."""
84 playing = _item("playing", "sess-1")
85 preloaded = _item("preloaded", "sess-1")
86 ctrl = _controller([playing, preloaded])
87
88 await ctrl._cleanup_queue_audio_data(QUEUE_ID, "sess-1")
89
90 for item in (playing, preloaded):
91 assert item.streamdetails is not None
92 assert item.streamdetails.buffer is None
93
94
95async def test_a_stop_clears_what_a_session_that_already_ended_left_behind() -> None:
96 """
97 Audio of a session that is no longer playing is nobody's to come back for.
98
99 Sessions rotate without a stop - starting another item mints a new one - and a buffer
100 that finished filling is left attached, so a claim from an ended session would never
101 be released again if a differing stamp alone were enough to skip it.
102 """
103 leftover = _item("leftover", "sess-0")
104 own = _item("own", "sess-1")
105 ctrl = _controller([leftover, own], playing="sess-2")
106
107 await ctrl._cleanup_queue_audio_data(QUEUE_ID, "sess-1")
108
109 for item in (leftover, own):
110 assert item.streamdetails is not None
111 assert item.streamdetails.buffer is None
112
113
114async def test_a_buffer_without_a_session_is_cleared_by_a_stop() -> None:
115 """
116 Audio that cannot be proven to belong to a later session is torn down.
117
118 Leaving it would keep a producer alive - and its provider's stream slot with it -
119 which is exactly what a stop has to prevent.
120 """
121 unstamped = _item("unstamped", None)
122 ctrl = _controller([unstamped])
123
124 await ctrl._cleanup_queue_audio_data(QUEUE_ID, "sess-1")
125
126 assert unstamped.streamdetails is not None
127 assert unstamped.streamdetails.buffer is None
128
129
130async def test_without_a_session_every_buffer_is_cleared() -> None:
131 """A clear/replace drops the items themselves, so all their audio goes with them."""
132 items = [_item("a", "sess-1"), _item("b", "sess-2"), _item("c", None)]
133 ctrl = _controller(items)
134
135 await ctrl._cleanup_queue_audio_data(QUEUE_ID)
136
137 for item in items:
138 assert item.streamdetails is not None
139 assert item.streamdetails.buffer is None
140
141
142async def test_a_buffer_attached_while_it_is_released_is_kept() -> None:
143 """
144 Releasing a buffer suspends, and what a later session attaches then must survive.
145
146 Cancelling the producer waits on the producer task, so the replacement session gets to
147 run and attach its own buffer to the same stream details while that is in flight.
148 """
149 stopped = _item("stopped", "sess-1")
150 ctrl = _controller([stopped], playing="sess-2")
151 assert stopped.streamdetails is not None
152 replacement_buffer = MagicMock(spec=AudioBuffer)
153
154 async def _attach_a_replacement() -> None:
155 # stands in for the new session claiming this item while the old buffer is released
156 stopped.streamdetails.buffer = replacement_buffer # type: ignore[union-attr]
157
158 stopped.streamdetails.buffer.clear = AsyncMock(side_effect=_attach_a_replacement)
159
160 await ctrl._cleanup_queue_audio_data(QUEUE_ID, "sess-1")
161
162 assert stopped.streamdetails.buffer is replacement_buffer
163
164
165async def test_a_session_that_starts_mid_cleanup_keeps_what_it_attaches() -> None:
166 """
167 The session playing now is re-read for every buffer, not decided once up front.
168
169 Releasing a buffer suspends, so playback can start while the cleanup is part way
170 through and claim items it has not reached yet.
171 """
172 first = _item("first", "sess-1")
173 later = _item("later", "sess-1")
174 ctrl = _controller([first, later])
175
176 async def _start_a_session() -> None:
177 # the queue is idle until this runs, so nothing was protected when the loop began
178 ctrl._queue_data[QUEUE_ID].session_id = "sess-2"
179 later.streamdetails.queue_session_id = "sess-2" # type: ignore[union-attr]
180
181 first.streamdetails.buffer.clear = AsyncMock( # type: ignore[union-attr]
182 side_effect=_start_a_session
183 )
184
185 await ctrl._cleanup_queue_audio_data(QUEUE_ID, "sess-1")
186
187 assert first.streamdetails is not None
188 assert first.streamdetails.buffer is None
189 assert later.streamdetails is not None
190 assert later.streamdetails.buffer is not None
191
192
193async def test_a_pending_crossfade_handover_is_always_dropped() -> None:
194 """A restarted session starts its first track from scratch, with nothing to fade from."""
195 ctrl = _controller([_item("a", "sess-2")], playing="sess-2")
196
197 await ctrl._cleanup_queue_audio_data(QUEUE_ID, "sess-1")
198
199 clear_crossfade_handover = cast("MagicMock", ctrl.mass.streams.audio.clear_crossfade_handover)
200 clear_crossfade_handover.assert_called_once_with(QUEUE_ID)
201