/
/
1"""Tests for the flow stream's transition: incoming prefetch and crossfade reporting."""
2
3from __future__ import annotations
4
5import asyncio
6from collections.abc import AsyncGenerator
7from types import SimpleNamespace
8from typing import TYPE_CHECKING, Any, cast
9from unittest.mock import AsyncMock, MagicMock
10
11from music_assistant_models.enums import ContentType, CrossfadeMode, MediaType
12from music_assistant_models.errors import QueueEmpty
13from music_assistant_models.media_items import AudioFormat
14
15from music_assistant.controllers.streams.audio import StreamsAudio
16from music_assistant.controllers.streams.audio_buffer import AudioBuffer
17from music_assistant.controllers.streams.smart_fades.fades import StandardCrossFade
18
19if TYPE_CHECKING:
20 import pytest
21
22TEST_PCM_FORMAT = AudioFormat(
23 content_type=ContentType.PCM_S16LE,
24 sample_rate=8000,
25 bit_depth=16,
26 channels=2,
27)
28# deliberately not a whole second and not frame-aligned, so any assumption about
29# chunk boundaries in the transition path shows up as wrong audio
30CHUNK_SIZE = TEST_PCM_FORMAT.pcm_sample_size // 3 + 2
31STANDARD_CROSSFADE_DURATION = 8
32
33
34def _buffer(*, duration_available: float = 45.0, eof: bool = True) -> AudioBuffer:
35 """Build a valid, fully resident buffer."""
36 audio_buffer = MagicMock(spec=AudioBuffer)
37 audio_buffer.has_error = False
38 audio_buffer.cancelled = False
39 audio_buffer.eof = eof
40 audio_buffer.max_size_seconds = 300
41 audio_buffer.is_valid.return_value = True
42 audio_buffer.duration_available = duration_available
43 audio_buffer.ready = MagicMock()
44 audio_buffer.ready.is_set.return_value = True
45 return audio_buffer
46
47
48def _queue_item(item_id: str, name: str, duration: int = 300) -> SimpleNamespace:
49 """Build a flow-streamable track with a prepared buffer."""
50 streamdetails = SimpleNamespace(
51 audio_format=TEST_PCM_FORMAT,
52 buffer=_buffer(),
53 fade_in=False,
54 stream_error=False,
55 uri=f"test://{item_id}",
56 seek_position=0,
57 seconds_streamed=0,
58 duration=300,
59 is_realtime=False,
60 volume_normalization_mode=None,
61 )
62 streamdetails.duration = duration
63 return SimpleNamespace(
64 queue_id="queue-1",
65 queue_item_id=item_id,
66 name=name,
67 media_type=MediaType.TRACK,
68 media_item=None,
69 streamdetails=streamdetails,
70 duration=duration,
71 extra_attributes={},
72 )
73
74
75def _flow_audio(
76 monkeypatch: pytest.MonkeyPatch,
77 *,
78 next_item: SimpleNamespace | None,
79 load_next: Any,
80 crossfade_mode: CrossfadeMode = CrossfadeMode.STANDARD_CROSSFADE,
81 crossfade_allowed: bool = True,
82 build_result: object | None = None,
83) -> tuple[StreamsAudio, SimpleNamespace, MagicMock]:
84 """Build a StreamsAudio wired for a two-track flow stream."""
85 queue = SimpleNamespace(
86 queue_id="queue-1",
87 display_name="Queue",
88 flow_mode=False,
89 overlay_enabled=False,
90 overlay_source=None,
91 )
92 mass = MagicMock()
93 mass.player_queues.queue_data.return_value = SimpleNamespace(
94 session_id="session-1", flow_mode_stream_log=[]
95 )
96 mass.player_queues.load_next_queue_item = AsyncMock(side_effect=load_next)
97 mass.player_queues.get.return_value = queue
98 mass.player_queues.get_next_item.return_value = next_item
99 mass.streams.get_crossfade_mode.return_value = crossfade_mode
100 mass.config.get_raw_core_config_value.return_value = STANDARD_CROSSFADE_DURATION
101 mass.streams.audio_processing.update_item_context = MagicMock()
102 player = MagicMock()
103 player.config.get_value.return_value = "fixed_48000"
104 player.get_supported_sample_rates.return_value = []
105 mass.players.get_player.return_value = player
106
107 audio = StreamsAudio(cast("Any", mass))
108 audio.setup()
109 audio.crossfade_allowed = MagicMock(return_value=crossfade_allowed) # type: ignore[method-assign]
110 monkeypatch.setattr(
111 audio.smart_fades_mixer,
112 "build",
113 AsyncMock(
114 return_value=build_result
115 or SimpleNamespace(
116 timing_info=SimpleNamespace(
117 fadein_trimmed_duration=0.0,
118 crossfade_duration=float(STANDARD_CROSSFADE_DURATION),
119 pre_crossfade_duration=0.0,
120 )
121 )
122 ),
123 )
124
125 async def _concat_mix(
126 _smart_fade: object,
127 *,
128 fade_in_part: bytes,
129 fade_out_part: bytes,
130 **_kwargs: object,
131 ) -> AsyncGenerator[bytes]:
132 # a lossless stand-in for the mixer, so the emitted total stays checkable
133 yield fade_out_part
134 yield fade_in_part
135
136 monkeypatch.setattr(audio.smart_fades_mixer, "mix", _concat_mix)
137 return audio, queue, mass
138
139
140def _install_item_streams(
141 monkeypatch: pytest.MonkeyPatch,
142 audio: StreamsAudio,
143 seconds_per_item: dict[str, int],
144) -> tuple[list[str], dict[str, int], dict[str, dict[str, int]]]:
145 """
146 Serve each queue item unaligned chunks.
147
148 Returns the order in which streams were opened, how much of each item was read, and
149 a snapshot of that reading taken the moment each item's stream ran out.
150 """
151 opened: list[str] = []
152 consumed: dict[str, int] = dict.fromkeys(seconds_per_item, 0)
153 exhausted_at: dict[str, dict[str, int]] = {}
154
155 async def _item_stream(
156 queue_item: SimpleNamespace, *_args: object, **_kwargs: object
157 ) -> AsyncGenerator[bytes]:
158 item_id = queue_item.queue_item_id
159 opened.append(item_id)
160 total = TEST_PCM_FORMAT.pcm_sample_size * seconds_per_item[item_id]
161 sent = 0
162 while sent < total:
163 size = min(CHUNK_SIZE, total - sent)
164 sent += size
165 consumed[item_id] += size
166 yield bytes(size)
167 await asyncio.sleep(0)
168 exhausted_at[item_id] = dict(consumed)
169
170 monkeypatch.setattr(audio, "get_queue_item_stream", _item_stream)
171 return opened, consumed, exhausted_at
172
173
174def _reported(mass: MagicMock) -> list[tuple[str, CrossfadeMode]]:
175 """Return the crossfade modes published for each queue item, in order."""
176 return [
177 (call.kwargs["queue_item_id"], call.kwargs["queue_processing"].crossfade_mode)
178 for call in mass.streams.audio_processing.update_item_context.call_args_list
179 ]
180
181
182async def _drain(stream: AsyncGenerator[bytes]) -> int:
183 """Consume a flow stream, yielding to the loop like a real consumer does."""
184 total = 0
185 async for chunk in stream:
186 total += len(chunk)
187 await asyncio.sleep(0)
188 return total
189
190
191async def test_flow_prefetches_the_incoming_fade_in_during_the_holdback(
192 monkeypatch: pytest.MonkeyPatch,
193) -> None:
194 """The incoming overlap is gathered while the outgoing tail is still being held back."""
195 first_item = _queue_item("item-1", "First")
196 second_item = _queue_item("item-2", "Second")
197 audio, queue, _mass = _flow_audio(
198 monkeypatch, next_item=second_item, load_next=[second_item, QueueEmpty]
199 )
200 opened, _consumed, exhausted_at = _install_item_streams(
201 monkeypatch, audio, {"item-1": 40, "item-2": 20}
202 )
203
204 stream = audio.get_queue_flow_stream(
205 cast("Any", queue), cast("Any", first_item), TEST_PCM_FORMAT, session_id="session-1"
206 )
207 emitted = await _drain(stream)
208
209 # the whole overlap was already in hand when the outgoing track ran out
210 overlap_size = TEST_PCM_FORMAT.pcm_sample_size * STANDARD_CROSSFADE_DURATION
211 assert exhausted_at["item-1"]["item-2"] >= overlap_size
212 # the prefetched stream is adopted, so the incoming track is only ever opened once
213 assert opened == ["item-1", "item-2"]
214 assert emitted == TEST_PCM_FORMAT.pcm_sample_size * 60
215
216
217async def test_flow_falls_back_when_the_next_item_changed(
218 monkeypatch: pytest.MonkeyPatch,
219) -> None:
220 """A prefetch for another item is dropped and the real next item is streamed."""
221 first_item = _queue_item("item-1", "First")
222 second_item = _queue_item("item-2", "Second")
223 other_item = _queue_item("item-3", "Other")
224 audio, queue, _mass = _flow_audio(
225 monkeypatch, next_item=other_item, load_next=[second_item, QueueEmpty]
226 )
227 opened, consumed, _exhausted_at = _install_item_streams(
228 monkeypatch, audio, {"item-1": 40, "item-2": 20, "item-3": 20}
229 )
230
231 stream = audio.get_queue_flow_stream(
232 cast("Any", queue), cast("Any", first_item), TEST_PCM_FORMAT, session_id="session-1"
233 )
234 emitted = await _drain(stream)
235
236 # the stale prefetch is dropped and the real next item is opened once
237 assert opened[:3] == ["item-1", "item-3", "item-2"]
238 assert opened.count("item-2") == 1
239 # the discarded prefetch never reaches the listener
240 assert emitted == TEST_PCM_FORMAT.pcm_sample_size * 60
241 assert consumed["item-2"] == TEST_PCM_FORMAT.pcm_sample_size * 20
242
243
244async def test_flow_reports_the_crossfade_that_actually_happens(
245 monkeypatch: pytest.MonkeyPatch,
246) -> None:
247 """A fade is reported on both of its sides, once the boundary has decided."""
248 first_item = _queue_item("item-1", "First")
249 second_item = _queue_item("item-2", "Second")
250 audio, queue, mass = _flow_audio(
251 monkeypatch, next_item=second_item, load_next=[second_item, QueueEmpty]
252 )
253 _install_item_streams(monkeypatch, audio, {"item-1": 40, "item-2": 20})
254
255 stream = audio.get_queue_flow_stream(
256 cast("Any", queue), cast("Any", first_item), TEST_PCM_FORMAT, session_id="session-1"
257 )
258 await _drain(stream)
259
260 assert _reported(mass) == [
261 # each track starts out crediting no fade
262 ("item-1", CrossfadeMode.DISABLED),
263 ("item-2", CrossfadeMode.DISABLED),
264 # both sides are only credited once the blend has really been rendered
265 ("item-2", CrossfadeMode.STANDARD_CROSSFADE),
266 ("item-1", CrossfadeMode.STANDARD_CROSSFADE),
267 ]
268
269
270async def test_flow_reports_no_crossfade_when_the_transition_is_denied(
271 monkeypatch: pytest.MonkeyPatch,
272) -> None:
273 """A transition that never happens is not reported as a crossfade."""
274 first_item = _queue_item("item-1", "First")
275 second_item = _queue_item("item-2", "Second")
276 audio, queue, mass = _flow_audio(
277 monkeypatch,
278 next_item=second_item,
279 load_next=[second_item, QueueEmpty],
280 crossfade_mode=CrossfadeMode.SMART_CROSSFADE,
281 crossfade_allowed=False,
282 )
283 _install_item_streams(monkeypatch, audio, {"item-1": 40, "item-2": 20})
284
285 stream = audio.get_queue_flow_stream(
286 cast("Any", queue), cast("Any", first_item), TEST_PCM_FORMAT, session_id="session-1"
287 )
288 await _drain(stream)
289
290 assert _reported(mass) == [
291 ("item-1", CrossfadeMode.DISABLED),
292 ("item-2", CrossfadeMode.DISABLED),
293 ]
294
295
296async def test_flow_reports_a_smart_fade_that_degraded_to_standard(
297 monkeypatch: pytest.MonkeyPatch,
298) -> None:
299 """A smart fade the mixer could not plan is reported as the standard one it became."""
300 first_item = _queue_item("item-1", "First")
301 second_item = _queue_item("item-2", "Second")
302 degraded = StandardCrossFade(logger=MagicMock(), crossfade_duration=STANDARD_CROSSFADE_DURATION)
303 overlap_size = TEST_PCM_FORMAT.pcm_sample_size * STANDARD_CROSSFADE_DURATION
304 degraded.build(overlap_size, overlap_size, TEST_PCM_FORMAT)
305 audio, queue, mass = _flow_audio(
306 monkeypatch,
307 next_item=second_item,
308 load_next=[second_item, QueueEmpty],
309 crossfade_mode=CrossfadeMode.SMART_CROSSFADE,
310 build_result=degraded,
311 )
312 _install_item_streams(monkeypatch, audio, {"item-1": 60, "item-2": 60})
313
314 stream = audio.get_queue_flow_stream(
315 cast("Any", queue), cast("Any", first_item), TEST_PCM_FORMAT, session_id="session-1"
316 )
317 await _drain(stream)
318
319 assert _reported(mass) == [
320 ("item-1", CrossfadeMode.DISABLED),
321 ("item-2", CrossfadeMode.DISABLED),
322 ("item-2", CrossfadeMode.STANDARD_CROSSFADE),
323 ("item-1", CrossfadeMode.STANDARD_CROSSFADE),
324 ]
325
326
327async def test_flow_reopens_the_incoming_track_when_the_prefetch_broke(
328 monkeypatch: pytest.MonkeyPatch,
329) -> None:
330 """A prefetch whose source failed is dropped so the track gets a fresh attempt."""
331 first_item = _queue_item("item-1", "First")
332 second_item = _queue_item("item-2", "Second")
333 audio, queue, _mass = _flow_audio(
334 monkeypatch, next_item=second_item, load_next=[second_item, QueueEmpty]
335 )
336 opened: list[str] = []
337
338 async def _item_stream(
339 queue_item: SimpleNamespace, *_args: object, **_kwargs: object
340 ) -> AsyncGenerator[bytes]:
341 queue_item.streamdetails.stream_error = False
342 opened.append(queue_item.queue_item_id)
343 if queue_item is second_item and opened.count("item-2") == 1:
344 # the source dies before handing over any audio
345 queue_item.streamdetails.stream_error = True
346 return
347 total = TEST_PCM_FORMAT.pcm_sample_size * 40
348 sent = 0
349 while sent < total:
350 size = min(CHUNK_SIZE, total - sent)
351 sent += size
352 yield bytes(size)
353 await asyncio.sleep(0)
354
355 monkeypatch.setattr(audio, "get_queue_item_stream", _item_stream)
356 stream = audio.get_queue_flow_stream(
357 cast("Any", queue), cast("Any", first_item), TEST_PCM_FORMAT, session_id="session-1"
358 )
359 emitted = await _drain(stream)
360
361 assert opened == ["item-1", "item-2", "item-2"]
362 # the retry serves the whole track, so nothing is lost to the failed prefetch
363 assert emitted == TEST_PCM_FORMAT.pcm_sample_size * 80
364
365
366async def test_flow_drops_a_prefetch_opened_at_another_position(
367 monkeypatch: pytest.MonkeyPatch,
368) -> None:
369 """A prefetch started at a stale seek position is not adopted."""
370 first_item = _queue_item("item-1", "First")
371 second_item = _queue_item("item-2", "Second")
372 # a leftover from an earlier crossfade into this track
373 second_item.streamdetails.seek_position = 8
374
375 loads = {"count": 0}
376
377 async def _load_next(*_args: object, **_kwargs: object) -> SimpleNamespace:
378 loads["count"] += 1
379 if loads["count"] > 1:
380 raise QueueEmpty
381 # loading the item resolves its stream details again, back to the track start
382 second_item.streamdetails.seek_position = 0
383 return second_item
384
385 audio, queue, _mass = _flow_audio(monkeypatch, next_item=second_item, load_next=_load_next)
386 opened, _consumed, _exhausted_at = _install_item_streams(
387 monkeypatch, audio, {"item-1": 40, "item-2": 20}
388 )
389
390 stream = audio.get_queue_flow_stream(
391 cast("Any", queue), cast("Any", first_item), TEST_PCM_FORMAT, session_id="session-1"
392 )
393 emitted = await _drain(stream)
394
395 assert opened == ["item-1", "item-2", "item-2"]
396 assert emitted == TEST_PCM_FORMAT.pcm_sample_size * 60
397
398
399async def test_flow_never_prefetches_a_short_track_to_its_end(
400 monkeypatch: pytest.MonkeyPatch,
401) -> None:
402 """The prefetch stops short of the end, so the track is not reported as streamed."""
403 first_item = _queue_item("item-1", "First")
404 second_item = _queue_item("item-2", "Second", duration=20)
405 audio, queue, _mass = _flow_audio(
406 monkeypatch,
407 next_item=second_item,
408 load_next=[second_item, QueueEmpty],
409 crossfade_mode=CrossfadeMode.SMART_CROSSFADE,
410 )
411 _opened, _consumed, exhausted_at = _install_item_streams(
412 monkeypatch, audio, {"item-1": 40, "item-2": 20}
413 )
414
415 stream = audio.get_queue_flow_stream(
416 cast("Any", queue), cast("Any", first_item), TEST_PCM_FORMAT, session_id="session-1"
417 )
418 await _drain(stream)
419
420 # the requested 45s window is clamped to half the incoming track
421 assert exhausted_at["item-1"]["item-2"] <= TEST_PCM_FORMAT.pcm_sample_size * 10 + CHUNK_SIZE
422
423
424async def test_flow_skips_the_prefetch_without_a_known_duration(
425 monkeypatch: pytest.MonkeyPatch,
426) -> None:
427 """A track of unknown length is not prefetched, since its end cannot be avoided."""
428 first_item = _queue_item("item-1", "First")
429 second_item = _queue_item("item-2", "Second")
430 second_item.streamdetails.duration = None
431 audio, queue, _mass = _flow_audio(
432 monkeypatch, next_item=second_item, load_next=[second_item, QueueEmpty]
433 )
434 opened, _consumed, exhausted_at = _install_item_streams(
435 monkeypatch, audio, {"item-1": 40, "item-2": 20}
436 )
437
438 stream = audio.get_queue_flow_stream(
439 cast("Any", queue), cast("Any", first_item), TEST_PCM_FORMAT, session_id="session-1"
440 )
441 await _drain(stream)
442
443 # opened once, at the transition, with nothing read while the tail was held back
444 assert opened == ["item-1", "item-2"]
445 assert exhausted_at["item-1"]["item-2"] == 0
446
447
448async def test_flow_reopens_a_track_whose_prefetch_ran_out_early(
449 monkeypatch: pytest.MonkeyPatch,
450) -> None:
451 """A source that stops short of the clamp is not trusted to serve the track."""
452 first_item = _queue_item("item-1", "First")
453 second_item = _queue_item("item-2", "Second")
454 audio, queue, _mass = _flow_audio(
455 monkeypatch, next_item=second_item, load_next=[second_item, QueueEmpty]
456 )
457 opened: list[str] = []
458
459 async def _item_stream(
460 queue_item: SimpleNamespace, *_args: object, **_kwargs: object
461 ) -> AsyncGenerator[bytes]:
462 opened.append(queue_item.queue_item_id)
463 # the incoming source ends cleanly long before the prefetch target
464 seconds = 2 if queue_item is second_item and opened.count("item-2") == 1 else 40
465 total = TEST_PCM_FORMAT.pcm_sample_size * seconds
466 sent = 0
467 while sent < total:
468 size = min(CHUNK_SIZE, total - sent)
469 sent += size
470 yield bytes(size)
471 await asyncio.sleep(0)
472
473 monkeypatch.setattr(audio, "get_queue_item_stream", _item_stream)
474 stream = audio.get_queue_flow_stream(
475 cast("Any", queue), cast("Any", first_item), TEST_PCM_FORMAT, session_id="session-1"
476 )
477 emitted = await _drain(stream)
478
479 assert opened == ["item-1", "item-2", "item-2"]
480 # the truncated prefetch is discarded rather than played as the whole track
481 assert emitted == TEST_PCM_FORMAT.pcm_sample_size * 80
482
483
484async def test_flow_keeps_the_prefetch_clear_of_the_end_after_a_seek(
485 monkeypatch: pytest.MonkeyPatch,
486) -> None:
487 """The clamp follows the seek position, so a near-the-end start is not read to EOF."""
488 first_item = _queue_item("item-1", "First")
489 second_item = _queue_item("item-2", "Second")
490 # resuming with only 10 seconds of the track left
491 second_item.streamdetails.seek_position = 290
492 audio, queue, _mass = _flow_audio(
493 monkeypatch, next_item=second_item, load_next=[second_item, QueueEmpty]
494 )
495 opened, _consumed, exhausted_at = _install_item_streams(
496 monkeypatch, audio, {"item-1": 40, "item-2": 10}
497 )
498
499 stream = audio.get_queue_flow_stream(
500 cast("Any", queue), cast("Any", first_item), TEST_PCM_FORMAT, session_id="session-1"
501 )
502 await _drain(stream)
503
504 # half of the 10s that remain, so the source is never read to its end in the background
505 assert exhausted_at["item-1"]["item-2"] <= TEST_PCM_FORMAT.pcm_sample_size * 5 + CHUNK_SIZE
506 assert opened.count("item-2") == 1
507