/
/
/
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 yield bytes(size)
175 await asyncio.sleep(0)
176 exhausted_at[item_id] = dict(consumed)
177
178 monkeypatch.setattr(audio, "get_queue_item_stream", _item_stream)
179 return opened, consumed, exhausted_at
180
181
182def _reported(mass: MagicMock) -> list[tuple[str, CrossfadeMode]]:
183 """Return the crossfade modes published for each queue item, in order."""
184 return [
185 (call.kwargs["queue_item_id"], call.kwargs["queue_processing"].crossfade_mode)
186 for call in mass.streams.audio_processing.update_item_context.call_args_list
187 ]
188
189
190async def _drain(stream: AsyncGenerator[bytes]) -> int:
191 """Consume a flow stream, yielding to the loop like a real consumer does."""
192 total = 0
193 async for chunk in stream:
194 total += len(chunk)
195 await asyncio.sleep(0)
196 return total
197
198
199async def test_flow_prefetches_the_incoming_fade_in_during_the_holdback(
200 monkeypatch: pytest.MonkeyPatch,
201) -> None:
202 """The incoming overlap is gathered while the outgoing tail is still being held back."""
203 first_item = _queue_item("item-1", "First")
204 second_item = _queue_item("item-2", "Second")
205 audio, queue, _mass = _flow_audio(
206 monkeypatch, next_item=second_item, load_next=[second_item, QueueEmpty]
207 )
208 opened, _consumed, exhausted_at = _install_item_streams(
209 monkeypatch, audio, {"item-1": 40, "item-2": 20}
210 )
211
212 stream = audio.get_queue_flow_stream(
213 cast("Any", queue), cast("Any", first_item), TEST_PCM_FORMAT, session_id="session-1"
214 )
215 emitted = await _drain(stream)
216
217 # the whole overlap was already in hand when the outgoing track ran out
218 overlap_size = TEST_PCM_FORMAT.pcm_sample_size * STANDARD_CROSSFADE_DURATION
219 assert exhausted_at["item-1"]["item-2"] >= overlap_size
220 # the prefetched stream is adopted, so the incoming track is only ever opened once
221 assert opened == ["item-1", "item-2"]
222 assert emitted == TEST_PCM_FORMAT.pcm_sample_size * 60
223
224
225async def test_flow_falls_back_when_the_next_item_changed(
226 monkeypatch: pytest.MonkeyPatch,
227) -> None:
228 """A prefetch for another item is dropped and the real next item is streamed."""
229 first_item = _queue_item("item-1", "First")
230 second_item = _queue_item("item-2", "Second")
231 other_item = _queue_item("item-3", "Other")
232 audio, queue, _mass = _flow_audio(
233 monkeypatch, next_item=other_item, load_next=[second_item, QueueEmpty]
234 )
235 opened, consumed, _exhausted_at = _install_item_streams(
236 monkeypatch, audio, {"item-1": 40, "item-2": 20, "item-3": 20}
237 )
238
239 stream = audio.get_queue_flow_stream(
240 cast("Any", queue), cast("Any", first_item), TEST_PCM_FORMAT, session_id="session-1"
241 )
242 emitted = await _drain(stream)
243
244 # the stale prefetch is dropped and the real next item is opened once
245 assert opened[:3] == ["item-1", "item-3", "item-2"]
246 assert opened.count("item-2") == 1
247 # the discarded prefetch never reaches the listener
248 assert emitted == TEST_PCM_FORMAT.pcm_sample_size * 60
249 assert consumed["item-2"] == TEST_PCM_FORMAT.pcm_sample_size * 20
250
251
252async def test_flow_reports_the_crossfade_that_actually_happens(
253 monkeypatch: pytest.MonkeyPatch,
254) -> None:
255 """A fade is reported on both of its sides, once the boundary has decided."""
256 first_item = _queue_item("item-1", "First")
257 second_item = _queue_item("item-2", "Second")
258 audio, queue, mass = _flow_audio(
259 monkeypatch, next_item=second_item, load_next=[second_item, QueueEmpty]
260 )
261 _install_item_streams(monkeypatch, audio, {"item-1": 40, "item-2": 20})
262
263 stream = audio.get_queue_flow_stream(
264 cast("Any", queue), cast("Any", first_item), TEST_PCM_FORMAT, session_id="session-1"
265 )
266 await _drain(stream)
267
268 assert _reported(mass) == [
269 # each track starts out crediting no fade
270 ("item-1", CrossfadeMode.DISABLED),
271 ("item-2", CrossfadeMode.DISABLED),
272 # both sides are only credited once the blend has really been rendered
273 ("item-2", CrossfadeMode.STANDARD_CROSSFADE),
274 ("item-1", CrossfadeMode.STANDARD_CROSSFADE),
275 ]
276
277
278async def test_flow_reports_no_crossfade_when_the_transition_is_denied(
279 monkeypatch: pytest.MonkeyPatch,
280) -> None:
281 """A transition that never happens is not reported as a crossfade."""
282 first_item = _queue_item("item-1", "First")
283 second_item = _queue_item("item-2", "Second")
284 audio, queue, mass = _flow_audio(
285 monkeypatch,
286 next_item=second_item,
287 load_next=[second_item, QueueEmpty],
288 crossfade_mode=CrossfadeMode.SMART_CROSSFADE,
289 crossfade_allowed=False,
290 )
291 _install_item_streams(monkeypatch, audio, {"item-1": 40, "item-2": 20})
292
293 stream = audio.get_queue_flow_stream(
294 cast("Any", queue), cast("Any", first_item), TEST_PCM_FORMAT, session_id="session-1"
295 )
296 await _drain(stream)
297
298 assert _reported(mass) == [
299 ("item-1", CrossfadeMode.DISABLED),
300 ("item-2", CrossfadeMode.DISABLED),
301 ]
302
303
304async def test_flow_reports_a_smart_fade_that_degraded_to_standard(
305 monkeypatch: pytest.MonkeyPatch,
306) -> None:
307 """A smart fade the mixer could not plan is reported as the standard one it became."""
308 first_item = _queue_item("item-1", "First")
309 second_item = _queue_item("item-2", "Second")
310 degraded = StandardCrossFade(logger=MagicMock(), crossfade_duration=STANDARD_CROSSFADE_DURATION)
311 overlap_size = TEST_PCM_FORMAT.pcm_sample_size * STANDARD_CROSSFADE_DURATION
312 degraded.build(overlap_size, overlap_size, TEST_PCM_FORMAT)
313 audio, queue, mass = _flow_audio(
314 monkeypatch,
315 next_item=second_item,
316 load_next=[second_item, QueueEmpty],
317 crossfade_mode=CrossfadeMode.SMART_CROSSFADE,
318 build_result=degraded,
319 )
320 _install_item_streams(monkeypatch, audio, {"item-1": 60, "item-2": 60})
321
322 stream = audio.get_queue_flow_stream(
323 cast("Any", queue), cast("Any", first_item), TEST_PCM_FORMAT, session_id="session-1"
324 )
325 await _drain(stream)
326
327 assert _reported(mass) == [
328 ("item-1", CrossfadeMode.DISABLED),
329 ("item-2", CrossfadeMode.DISABLED),
330 ("item-2", CrossfadeMode.STANDARD_CROSSFADE),
331 ("item-1", CrossfadeMode.STANDARD_CROSSFADE),
332 ]
333
334
335async def test_flow_reopens_the_incoming_track_when_the_prefetch_broke(
336 monkeypatch: pytest.MonkeyPatch,
337) -> None:
338 """A prefetch whose source failed is dropped so the track gets a fresh attempt."""
339 first_item = _queue_item("item-1", "First")
340 second_item = _queue_item("item-2", "Second")
341 audio, queue, _mass = _flow_audio(
342 monkeypatch, next_item=second_item, load_next=[second_item, QueueEmpty]
343 )
344 opened: list[str] = []
345
346 async def _item_stream(
347 queue_item: SimpleNamespace, *_args: object, **_kwargs: object
348 ) -> AsyncGenerator[bytes]:
349 queue_item.streamdetails.stream_error = False
350 opened.append(queue_item.queue_item_id)
351 if queue_item is second_item and opened.count("item-2") == 1:
352 # the source dies before handing over any audio
353 queue_item.streamdetails.stream_error = True
354 return
355 total = TEST_PCM_FORMAT.pcm_sample_size * 40
356 sent = 0
357 while sent < total:
358 size = min(CHUNK_SIZE, total - sent)
359 sent += size
360 yield bytes(size)
361 await asyncio.sleep(0)
362
363 monkeypatch.setattr(audio, "get_queue_item_stream", _item_stream)
364 stream = audio.get_queue_flow_stream(
365 cast("Any", queue), cast("Any", first_item), TEST_PCM_FORMAT, session_id="session-1"
366 )
367 emitted = await _drain(stream)
368
369 assert opened == ["item-1", "item-2", "item-2"]
370 # the retry serves the whole track, so nothing is lost to the failed prefetch
371 assert emitted == TEST_PCM_FORMAT.pcm_sample_size * 80
372
373
374async def test_flow_drops_a_prefetch_opened_at_another_position(
375 monkeypatch: pytest.MonkeyPatch,
376) -> None:
377 """A prefetch started at a stale seek position is not adopted."""
378 first_item = _queue_item("item-1", "First")
379 second_item = _queue_item("item-2", "Second")
380 # a leftover from an earlier crossfade into this track
381 second_item.streamdetails.seek_position = 8
382
383 loads = {"count": 0}
384
385 async def _load_next(*_args: object, **_kwargs: object) -> SimpleNamespace:
386 loads["count"] += 1
387 if loads["count"] > 1:
388 raise QueueEmpty
389 # loading the item resolves its stream details again, back to the track start
390 second_item.streamdetails.seek_position = 0
391 return second_item
392
393 audio, queue, _mass = _flow_audio(monkeypatch, next_item=second_item, load_next=_load_next)
394 opened, _consumed, _exhausted_at = _install_item_streams(
395 monkeypatch, audio, {"item-1": 40, "item-2": 20}
396 )
397
398 stream = audio.get_queue_flow_stream(
399 cast("Any", queue), cast("Any", first_item), TEST_PCM_FORMAT, session_id="session-1"
400 )
401 emitted = await _drain(stream)
402
403 assert opened == ["item-1", "item-2", "item-2"]
404 assert emitted == TEST_PCM_FORMAT.pcm_sample_size * 60
405
406
407async def test_flow_never_prefetches_a_short_track_to_its_end(
408 monkeypatch: pytest.MonkeyPatch,
409) -> None:
410 """The prefetch stops short of the end, so the track is not reported as streamed."""
411 first_item = _queue_item("item-1", "First")
412 second_item = _queue_item("item-2", "Second", duration=20)
413 audio, queue, _mass = _flow_audio(
414 monkeypatch,
415 next_item=second_item,
416 load_next=[second_item, QueueEmpty],
417 crossfade_mode=CrossfadeMode.SMART_CROSSFADE,
418 )
419 _opened, _consumed, exhausted_at = _install_item_streams(
420 monkeypatch, audio, {"item-1": 40, "item-2": 20}
421 )
422
423 stream = audio.get_queue_flow_stream(
424 cast("Any", queue), cast("Any", first_item), TEST_PCM_FORMAT, session_id="session-1"
425 )
426 await _drain(stream)
427
428 # the requested 45s window is clamped to half the incoming track
429 assert exhausted_at["item-1"]["item-2"] <= TEST_PCM_FORMAT.pcm_sample_size * 10 + CHUNK_SIZE
430
431
432async def test_flow_skips_the_prefetch_without_a_known_duration(
433 monkeypatch: pytest.MonkeyPatch,
434) -> None:
435 """A track of unknown length is not prefetched, since its end cannot be avoided."""
436 first_item = _queue_item("item-1", "First")
437 second_item = _queue_item("item-2", "Second")
438 second_item.streamdetails.duration = None
439 audio, queue, _mass = _flow_audio(
440 monkeypatch, next_item=second_item, load_next=[second_item, QueueEmpty]
441 )
442 opened, _consumed, exhausted_at = _install_item_streams(
443 monkeypatch, audio, {"item-1": 40, "item-2": 20}
444 )
445
446 stream = audio.get_queue_flow_stream(
447 cast("Any", queue), cast("Any", first_item), TEST_PCM_FORMAT, session_id="session-1"
448 )
449 await _drain(stream)
450
451 # opened once, at the transition, with nothing read while the tail was held back
452 assert opened == ["item-1", "item-2"]
453 assert exhausted_at["item-1"]["item-2"] == 0
454
455
456async def test_flow_reopens_a_track_whose_prefetch_ran_out_early(
457 monkeypatch: pytest.MonkeyPatch,
458) -> None:
459 """A source that stops short of the clamp is not trusted to serve the track."""
460 first_item = _queue_item("item-1", "First")
461 second_item = _queue_item("item-2", "Second")
462 audio, queue, _mass = _flow_audio(
463 monkeypatch, next_item=second_item, load_next=[second_item, QueueEmpty]
464 )
465 opened: list[str] = []
466
467 async def _item_stream(
468 queue_item: SimpleNamespace, *_args: object, **_kwargs: object
469 ) -> AsyncGenerator[bytes]:
470 opened.append(queue_item.queue_item_id)
471 # the incoming source ends cleanly long before the prefetch target
472 seconds = 2 if queue_item is second_item and opened.count("item-2") == 1 else 40
473 total = TEST_PCM_FORMAT.pcm_sample_size * seconds
474 sent = 0
475 while sent < total:
476 size = min(CHUNK_SIZE, total - sent)
477 sent += size
478 yield bytes(size)
479 await asyncio.sleep(0)
480
481 monkeypatch.setattr(audio, "get_queue_item_stream", _item_stream)
482 stream = audio.get_queue_flow_stream(
483 cast("Any", queue), cast("Any", first_item), TEST_PCM_FORMAT, session_id="session-1"
484 )
485 emitted = await _drain(stream)
486
487 assert opened == ["item-1", "item-2", "item-2"]
488 # the truncated prefetch is discarded rather than played as the whole track
489 assert emitted == TEST_PCM_FORMAT.pcm_sample_size * 80
490
491
492async def test_flow_keeps_the_prefetch_clear_of_the_end_after_a_seek(
493 monkeypatch: pytest.MonkeyPatch,
494) -> None:
495 """The clamp follows the seek position, so a near-the-end start is not read to EOF."""
496 first_item = _queue_item("item-1", "First")
497 second_item = _queue_item("item-2", "Second")
498 # resuming with only 10 seconds of the track left
499 second_item.streamdetails.seek_position = 290
500 audio, queue, _mass = _flow_audio(
501 monkeypatch, next_item=second_item, load_next=[second_item, QueueEmpty]
502 )
503 opened, _consumed, exhausted_at = _install_item_streams(
504 monkeypatch, audio, {"item-1": 40, "item-2": 10}
505 )
506
507 stream = audio.get_queue_flow_stream(
508 cast("Any", queue), cast("Any", first_item), TEST_PCM_FORMAT, session_id="session-1"
509 )
510 await _drain(stream)
511
512 # half of the 10s that remain, so the source is never read to its end in the background
513 assert exhausted_at["item-1"]["item-2"] <= TEST_PCM_FORMAT.pcm_sample_size * 5 + CHUNK_SIZE
514 assert opened.count("item-2") == 1
515
516
517async def test_prefetch_handover_gives_up_on_a_stalled_source(
518 monkeypatch: pytest.MonkeyPatch,
519) -> None:
520 """A source that stopped delivering must not hold the handover open."""
521 monkeypatch.setattr(audio_module, "PREFETCH_HANDOVER_TIMEOUT", 0.1)
522 prefetcher = audio_module._IncomingFadePrefetcher(
523 cast("Any", MagicMock()), TEST_PCM_FORMAT, "session-1"
524 )
525 stalled = asyncio.Event()
526
527 async def _stalled_stream() -> AsyncGenerator[bytes]:
528 # the source went quiet: the collector blocks here, never seeing a new target
529 await stalled.wait()
530 yield b""
531
532 streamdetails = SimpleNamespace(stream_error=False)
533 queue_item = SimpleNamespace(queue_item_id="item-2", streamdetails=streamdetails)
534 stream = _stalled_stream()
535 prefetcher._queue_item_id = "item-2"
536 prefetcher._streamdetails = cast("Any", streamdetails)
537 prefetcher._seek_position = 0
538 prefetcher._stream = stream
539 prefetcher._chunks = deque()
540 prefetcher._target = TEST_PCM_FORMAT.pcm_sample_size * 45
541 prefetcher._task = asyncio.create_task(prefetcher._collect(stream, prefetcher._chunks))
542
543 # the flow stream opens the track itself rather than waiting on a dead prefetch;
544 # the timeout keeps a regression here a failure instead of a hung test run
545 async with asyncio.timeout(10):
546 assert await prefetcher.take(cast("Any", queue_item), 0) is None
547