/
/
/
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
200def _install_counting_mix(monkeypatch: pytest.MonkeyPatch, audio: StreamsAudio) -> list[None]:
201 """Replace the mixer with the lossless concat stand-in, recording each invocation."""
202 calls: list[None] = []
203
204 async def _counting_mix(
205 _smart_fade: object,
206 *,
207 fade_in_part: bytes | AsyncGenerator[bytes],
208 fade_out_part: bytes,
209 **_kwargs: object,
210 ) -> AsyncGenerator[bytes]:
211 calls.append(None)
212 yield fade_out_part
213 if isinstance(fade_in_part, bytes):
214 yield fade_in_part
215 else:
216 async for fade_in_chunk in fade_in_part:
217 yield fade_in_chunk
218
219 monkeypatch.setattr(audio.smart_fades_mixer, "mix", _counting_mix)
220 return calls
221
222
223async def test_flow_prefetches_the_incoming_fade_in_during_the_holdback(
224 monkeypatch: pytest.MonkeyPatch,
225) -> None:
226 """The incoming overlap is gathered while the outgoing tail is still being held back."""
227 first_item = _queue_item("item-1", "First")
228 second_item = _queue_item("item-2", "Second")
229 audio, queue, _mass = _flow_audio(
230 monkeypatch, next_item=second_item, load_next=[second_item, QueueEmpty]
231 )
232 opened, _consumed, exhausted_at = _install_item_streams(
233 monkeypatch, audio, {"item-1": 40, "item-2": 20}
234 )
235
236 stream = audio.get_queue_flow_stream(
237 cast("Any", queue), cast("Any", first_item), TEST_PCM_FORMAT, session_id="session-1"
238 )
239 emitted = await _drain(stream)
240
241 # the whole overlap was already in hand when the outgoing track ran out
242 overlap_size = TEST_PCM_FORMAT.pcm_sample_size * STANDARD_CROSSFADE_DURATION
243 assert exhausted_at["item-1"]["item-2"] >= overlap_size
244 # the prefetched stream is adopted, so the incoming track is only ever opened once
245 assert opened == ["item-1", "item-2"]
246 assert emitted == TEST_PCM_FORMAT.pcm_sample_size * 60
247
248
249async def test_flow_falls_back_when_the_next_item_changed(
250 monkeypatch: pytest.MonkeyPatch,
251) -> None:
252 """A prefetch for another item is dropped and the real next item is streamed."""
253 first_item = _queue_item("item-1", "First")
254 second_item = _queue_item("item-2", "Second")
255 other_item = _queue_item("item-3", "Other")
256 audio, queue, _mass = _flow_audio(
257 monkeypatch, next_item=other_item, load_next=[second_item, QueueEmpty]
258 )
259 opened, consumed, _exhausted_at = _install_item_streams(
260 monkeypatch, audio, {"item-1": 40, "item-2": 20, "item-3": 20}
261 )
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 emitted = await _drain(stream)
267
268 # the stale prefetch is dropped and the real next item is opened once
269 assert opened[:3] == ["item-1", "item-3", "item-2"]
270 assert opened.count("item-2") == 1
271 # the discarded prefetch never reaches the listener
272 assert emitted == TEST_PCM_FORMAT.pcm_sample_size * 60
273 assert consumed["item-2"] == TEST_PCM_FORMAT.pcm_sample_size * 20
274
275
276async def test_flow_reports_the_crossfade_that_actually_happens(
277 monkeypatch: pytest.MonkeyPatch,
278) -> None:
279 """A fade is reported on both of its sides, once the boundary has decided."""
280 first_item = _queue_item("item-1", "First")
281 second_item = _queue_item("item-2", "Second")
282 audio, queue, mass = _flow_audio(
283 monkeypatch, next_item=second_item, load_next=[second_item, QueueEmpty]
284 )
285 _install_item_streams(monkeypatch, audio, {"item-1": 40, "item-2": 20})
286
287 stream = audio.get_queue_flow_stream(
288 cast("Any", queue), cast("Any", first_item), TEST_PCM_FORMAT, session_id="session-1"
289 )
290 await _drain(stream)
291
292 assert _reported(mass) == [
293 # each track starts out crediting no fade
294 ("item-1", CrossfadeMode.DISABLED),
295 ("item-2", CrossfadeMode.DISABLED),
296 # both sides are only credited once the blend has really been rendered
297 ("item-2", CrossfadeMode.STANDARD_CROSSFADE),
298 ("item-1", CrossfadeMode.STANDARD_CROSSFADE),
299 ]
300
301
302async def test_flow_reports_no_crossfade_when_the_transition_is_denied(
303 monkeypatch: pytest.MonkeyPatch,
304) -> None:
305 """A transition that never happens is not reported as a crossfade."""
306 first_item = _queue_item("item-1", "First")
307 second_item = _queue_item("item-2", "Second")
308 audio, queue, mass = _flow_audio(
309 monkeypatch,
310 next_item=second_item,
311 load_next=[second_item, QueueEmpty],
312 crossfade_mode=CrossfadeMode.SMART_CROSSFADE,
313 crossfade_allowed=False,
314 )
315 _install_item_streams(monkeypatch, audio, {"item-1": 40, "item-2": 20})
316
317 stream = audio.get_queue_flow_stream(
318 cast("Any", queue), cast("Any", first_item), TEST_PCM_FORMAT, session_id="session-1"
319 )
320 await _drain(stream)
321
322 assert _reported(mass) == [
323 ("item-1", CrossfadeMode.DISABLED),
324 ("item-2", CrossfadeMode.DISABLED),
325 ]
326
327
328async def test_flow_reports_a_smart_fade_that_degraded_to_standard(
329 monkeypatch: pytest.MonkeyPatch,
330) -> None:
331 """A smart fade the mixer could not plan is reported as the standard one it became."""
332 first_item = _queue_item("item-1", "First")
333 second_item = _queue_item("item-2", "Second")
334 degraded = StandardCrossFade(logger=MagicMock(), crossfade_duration=STANDARD_CROSSFADE_DURATION)
335 overlap_size = TEST_PCM_FORMAT.pcm_sample_size * STANDARD_CROSSFADE_DURATION
336 degraded.build(overlap_size, overlap_size, TEST_PCM_FORMAT)
337 audio, queue, mass = _flow_audio(
338 monkeypatch,
339 next_item=second_item,
340 load_next=[second_item, QueueEmpty],
341 crossfade_mode=CrossfadeMode.SMART_CROSSFADE,
342 build_result=degraded,
343 )
344 _install_item_streams(monkeypatch, audio, {"item-1": 60, "item-2": 60})
345
346 stream = audio.get_queue_flow_stream(
347 cast("Any", queue), cast("Any", first_item), TEST_PCM_FORMAT, session_id="session-1"
348 )
349 await _drain(stream)
350
351 assert _reported(mass) == [
352 ("item-1", CrossfadeMode.DISABLED),
353 ("item-2", CrossfadeMode.DISABLED),
354 ("item-2", CrossfadeMode.STANDARD_CROSSFADE),
355 ("item-1", CrossfadeMode.STANDARD_CROSSFADE),
356 ]
357
358
359async def test_flow_reopens_the_incoming_track_when_the_prefetch_broke(
360 monkeypatch: pytest.MonkeyPatch,
361) -> None:
362 """A prefetch whose source failed is dropped so the track gets a fresh attempt."""
363 first_item = _queue_item("item-1", "First")
364 second_item = _queue_item("item-2", "Second")
365 audio, queue, _mass = _flow_audio(
366 monkeypatch, next_item=second_item, load_next=[second_item, QueueEmpty]
367 )
368 opened: list[str] = []
369
370 async def _item_stream(
371 queue_item: SimpleNamespace, *_args: object, **_kwargs: object
372 ) -> AsyncGenerator[bytes]:
373 queue_item.streamdetails.stream_error = False
374 opened.append(queue_item.queue_item_id)
375 if queue_item is second_item and opened.count("item-2") == 1:
376 # the source dies before handing over any audio
377 queue_item.streamdetails.stream_error = True
378 return
379 total = TEST_PCM_FORMAT.pcm_sample_size * 40
380 sent = 0
381 while sent < total:
382 size = min(CHUNK_SIZE, total - sent)
383 sent += size
384 yield (b"\x10\x20" * (size // 2 + 1))[:size]
385 await asyncio.sleep(0)
386
387 monkeypatch.setattr(audio, "get_queue_item_stream", _item_stream)
388 stream = audio.get_queue_flow_stream(
389 cast("Any", queue), cast("Any", first_item), TEST_PCM_FORMAT, session_id="session-1"
390 )
391 emitted = await _drain(stream)
392
393 assert opened == ["item-1", "item-2", "item-2"]
394 # the retry serves the whole track, so nothing is lost to the failed prefetch
395 assert emitted == TEST_PCM_FORMAT.pcm_sample_size * 80
396
397
398async def test_flow_drops_a_prefetch_opened_at_another_position(
399 monkeypatch: pytest.MonkeyPatch,
400) -> None:
401 """A prefetch started at a stale seek position is not adopted."""
402 first_item = _queue_item("item-1", "First")
403 second_item = _queue_item("item-2", "Second")
404 # a leftover from an earlier crossfade into this track
405 second_item.streamdetails.seek_position = 8
406
407 loads = {"count": 0}
408
409 async def _load_next(*_args: object, **_kwargs: object) -> SimpleNamespace:
410 loads["count"] += 1
411 if loads["count"] > 1:
412 raise QueueEmpty
413 # loading the item resolves its stream details again, back to the track start
414 second_item.streamdetails.seek_position = 0
415 return second_item
416
417 audio, queue, _mass = _flow_audio(monkeypatch, next_item=second_item, load_next=_load_next)
418 opened, _consumed, _exhausted_at = _install_item_streams(
419 monkeypatch, audio, {"item-1": 40, "item-2": 20}
420 )
421
422 stream = audio.get_queue_flow_stream(
423 cast("Any", queue), cast("Any", first_item), TEST_PCM_FORMAT, session_id="session-1"
424 )
425 emitted = await _drain(stream)
426
427 assert opened == ["item-1", "item-2", "item-2"]
428 assert emitted == TEST_PCM_FORMAT.pcm_sample_size * 60
429
430
431async def test_flow_never_prefetches_a_short_track_to_its_end(
432 monkeypatch: pytest.MonkeyPatch,
433) -> None:
434 """The prefetch stops short of the end, so the track is not reported as streamed."""
435 first_item = _queue_item("item-1", "First")
436 second_item = _queue_item("item-2", "Second", duration=20)
437 audio, queue, _mass = _flow_audio(
438 monkeypatch,
439 next_item=second_item,
440 load_next=[second_item, QueueEmpty],
441 crossfade_mode=CrossfadeMode.SMART_CROSSFADE,
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 # the requested 45s window is clamped to half the incoming track
453 assert exhausted_at["item-1"]["item-2"] <= TEST_PCM_FORMAT.pcm_sample_size * 10 + CHUNK_SIZE
454
455
456async def test_flow_skips_the_prefetch_without_a_known_duration(
457 monkeypatch: pytest.MonkeyPatch,
458) -> None:
459 """A track of unknown length is not prefetched, since its end cannot be avoided."""
460 first_item = _queue_item("item-1", "First")
461 second_item = _queue_item("item-2", "Second")
462 second_item.streamdetails.duration = None
463 audio, queue, _mass = _flow_audio(
464 monkeypatch, next_item=second_item, load_next=[second_item, QueueEmpty]
465 )
466 opened, _consumed, exhausted_at = _install_item_streams(
467 monkeypatch, audio, {"item-1": 40, "item-2": 20}
468 )
469
470 stream = audio.get_queue_flow_stream(
471 cast("Any", queue), cast("Any", first_item), TEST_PCM_FORMAT, session_id="session-1"
472 )
473 await _drain(stream)
474
475 # opened once, at the transition, with nothing read while the tail was held back
476 assert opened == ["item-1", "item-2"]
477 assert exhausted_at["item-1"]["item-2"] == 0
478
479
480async def test_flow_reopens_a_track_whose_prefetch_ran_out_early(
481 monkeypatch: pytest.MonkeyPatch,
482) -> None:
483 """A source that stops short of the clamp is not trusted to serve the track."""
484 first_item = _queue_item("item-1", "First")
485 second_item = _queue_item("item-2", "Second")
486 audio, queue, _mass = _flow_audio(
487 monkeypatch, next_item=second_item, load_next=[second_item, QueueEmpty]
488 )
489 opened: list[str] = []
490
491 async def _item_stream(
492 queue_item: SimpleNamespace, *_args: object, **_kwargs: object
493 ) -> AsyncGenerator[bytes]:
494 opened.append(queue_item.queue_item_id)
495 # the incoming source ends cleanly long before the prefetch target
496 seconds = 2 if queue_item is second_item and opened.count("item-2") == 1 else 40
497 total = TEST_PCM_FORMAT.pcm_sample_size * seconds
498 sent = 0
499 while sent < total:
500 size = min(CHUNK_SIZE, total - sent)
501 sent += size
502 yield (b"\x10\x20" * (size // 2 + 1))[:size]
503 await asyncio.sleep(0)
504
505 monkeypatch.setattr(audio, "get_queue_item_stream", _item_stream)
506 stream = audio.get_queue_flow_stream(
507 cast("Any", queue), cast("Any", first_item), TEST_PCM_FORMAT, session_id="session-1"
508 )
509 emitted = await _drain(stream)
510
511 assert opened == ["item-1", "item-2", "item-2"]
512 # the truncated prefetch is discarded rather than played as the whole track
513 assert emitted == TEST_PCM_FORMAT.pcm_sample_size * 80
514
515
516async def test_flow_keeps_the_prefetch_clear_of_the_end_after_a_seek(
517 monkeypatch: pytest.MonkeyPatch,
518) -> None:
519 """The clamp follows the seek position, so a near-the-end start is not read to EOF."""
520 first_item = _queue_item("item-1", "First")
521 second_item = _queue_item("item-2", "Second")
522 # resuming with only 10 seconds of the track left
523 second_item.streamdetails.seek_position = 290
524 audio, queue, _mass = _flow_audio(
525 monkeypatch, next_item=second_item, load_next=[second_item, QueueEmpty]
526 )
527 opened, _consumed, exhausted_at = _install_item_streams(
528 monkeypatch, audio, {"item-1": 40, "item-2": 10}
529 )
530
531 stream = audio.get_queue_flow_stream(
532 cast("Any", queue), cast("Any", first_item), TEST_PCM_FORMAT, session_id="session-1"
533 )
534 await _drain(stream)
535
536 # half of the 10s that remain, so the source is never read to its end in the background
537 assert exhausted_at["item-1"]["item-2"] <= TEST_PCM_FORMAT.pcm_sample_size * 5 + CHUNK_SIZE
538 assert opened.count("item-2") == 1
539
540
541async def test_prefetch_handover_gives_up_on_a_stalled_source(
542 monkeypatch: pytest.MonkeyPatch,
543) -> None:
544 """A source that stopped delivering must not hold the handover open."""
545 monkeypatch.setattr(audio_module, "PREFETCH_HANDOVER_TIMEOUT", 0.1)
546 prefetcher = audio_module._IncomingFadePrefetcher(
547 cast("Any", MagicMock()), TEST_PCM_FORMAT, "session-1"
548 )
549 stalled = asyncio.Event()
550
551 async def _stalled_stream() -> AsyncGenerator[bytes]:
552 # the source went quiet: the collector blocks here, never seeing a new target
553 await stalled.wait()
554 yield b""
555
556 streamdetails = SimpleNamespace(stream_error=False)
557 queue_item = SimpleNamespace(queue_item_id="item-2", streamdetails=streamdetails)
558 stream = _stalled_stream()
559 prefetcher._queue_item_id = "item-2"
560 prefetcher._streamdetails = cast("Any", streamdetails)
561 prefetcher._seek_position = 0
562 prefetcher._stream = stream
563 prefetcher._chunks = deque()
564 prefetcher._target = TEST_PCM_FORMAT.pcm_sample_size * 45
565 prefetcher._task = asyncio.create_task(prefetcher._collect(stream, prefetcher._chunks))
566
567 # the flow stream opens the track itself rather than waiting on a dead prefetch;
568 # the timeout keeps a regression here a failure instead of a hung test run
569 async with asyncio.timeout(10):
570 assert await prefetcher.take(cast("Any", queue_item), 0) is None
571
572
573async def test_enable_crossfade_mid_session_applies_after_current_track(
574 monkeypatch: pytest.MonkeyPatch,
575) -> None:
576 """Turning crossfade on mid-session skips the current track and fades the next one."""
577 first_item = _queue_item("item-1", "First")
578 second_item = _queue_item("item-2", "Second")
579 third_item = _queue_item("item-3", "Third")
580 audio, queue, mass = _flow_audio(
581 monkeypatch,
582 next_item=third_item,
583 load_next=[second_item, third_item, QueueEmpty],
584 )
585 # snapshot at session start and item-1's own iteration still see the old setting;
586 # item-2 and item-3 see it enabled (padded so the sequence can't run dry)
587 mass.streams.get_crossfade_mode.side_effect = [
588 CrossfadeMode.DISABLED,
589 CrossfadeMode.DISABLED,
590 CrossfadeMode.STANDARD_CROSSFADE,
591 CrossfadeMode.STANDARD_CROSSFADE,
592 CrossfadeMode.STANDARD_CROSSFADE,
593 CrossfadeMode.STANDARD_CROSSFADE,
594 ]
595 _install_item_streams(monkeypatch, audio, {"item-1": 40, "item-2": 40, "item-3": 40})
596 mix_calls = _install_counting_mix(monkeypatch, audio)
597
598 stream = audio.get_queue_flow_stream(
599 cast("Any", queue), cast("Any", first_item), TEST_PCM_FORMAT, session_id="session-1"
600 )
601 await _drain(stream)
602
603 # item-1's tail was never held back, so 1->2 cannot fade; only 2->3 does
604 assert len(mix_calls) == 1
605 assert _reported(mass) == [
606 ("item-1", CrossfadeMode.DISABLED),
607 ("item-2", CrossfadeMode.DISABLED),
608 ("item-3", CrossfadeMode.DISABLED),
609 ("item-3", CrossfadeMode.STANDARD_CROSSFADE),
610 ("item-2", CrossfadeMode.STANDARD_CROSSFADE),
611 ]
612
613
614async def test_disable_crossfade_mid_session_applies_at_next_transition(
615 monkeypatch: pytest.MonkeyPatch,
616) -> None:
617 """Turning crossfade off mid-session flushes the held-back tail instead of fading it."""
618 first_item = _queue_item("item-1", "First")
619 second_item = _queue_item("item-2", "Second")
620 audio, queue, mass = _flow_audio(
621 monkeypatch, next_item=second_item, load_next=[second_item, QueueEmpty]
622 )
623 # snapshot and item-1's own iteration still see the old setting; item-2 sees it disabled
624 mass.streams.get_crossfade_mode.side_effect = [
625 CrossfadeMode.STANDARD_CROSSFADE,
626 CrossfadeMode.STANDARD_CROSSFADE,
627 CrossfadeMode.DISABLED,
628 CrossfadeMode.DISABLED,
629 CrossfadeMode.DISABLED,
630 ]
631 _install_item_streams(monkeypatch, audio, {"item-1": 40, "item-2": 20})
632 mix_calls = _install_counting_mix(monkeypatch, audio)
633
634 stream = audio.get_queue_flow_stream(
635 cast("Any", queue), cast("Any", first_item), TEST_PCM_FORMAT, session_id="session-1"
636 )
637 emitted = await _drain(stream)
638
639 # item-1's held-back tail is flushed unfaded rather than blended or dropped
640 assert len(mix_calls) == 0
641 assert _reported(mass) == [
642 ("item-1", CrossfadeMode.DISABLED),
643 ("item-2", CrossfadeMode.DISABLED),
644 ]
645 assert emitted == TEST_PCM_FORMAT.pcm_sample_size * 60
646