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