music-assistant-server

1"""Tests for the is_realtime gate across the buffer, holdback, and stream paths."""
2
3from __future__ import annotations
4
5import asyncio
6import struct
7from collections.abc import AsyncGenerator
8from types import SimpleNamespace
9from typing import Any, cast
10from unittest.mock import AsyncMock, MagicMock
11
12import pytest
13from music_assistant_models.enums import (
14    ContentType,
15    CrossfadeMode,
16    MediaType,
17    PlayerFeature,
18    StreamType,
19    VolumeNormalizationMode,
20)
21from music_assistant_models.errors import QueueEmpty
22from music_assistant_models.media_items import (
23    AudioFormat,
24    AudioSource,
25    ProviderMapping,
26    Radio,
27    Track,
28)
29from music_assistant_models.queue_item import QueueItem
30from music_assistant_models.streamdetails import StreamDetails
31
32from music_assistant.controllers.streams import controller as controller_mod
33from music_assistant.controllers.streams.audio import (
34    MIN_CROSSFADE_DURATION,
35    CrossfadeHandover,
36    StreamsAudio,
37    tail_hold_target,
38)
39from music_assistant.controllers.streams.audio_buffer import AudioBuffer
40from music_assistant.controllers.streams.constants import BufferSize, output_pacing_args
41from music_assistant.controllers.streams.controller import StreamsController
42from music_assistant.controllers.streams.smart_fades.fades import StandardCrossFade
43from music_assistant.controllers.streams.smart_fades.helpers import SMART_CROSSFADE_DURATION
44
45# Standard test PCM format: 44100Hz, 16-bit, stereo
46TEST_PCM_FORMAT = AudioFormat(
47    content_type=ContentType.PCM_S16LE,
48    sample_rate=44100,
49    bit_depth=16,
50    channels=2,
51)
52
53# One second of silence in the test format
54ONE_SECOND_CHUNK = b"\x00" * TEST_PCM_FORMAT.pcm_sample_size
55
56
57def _audio(pcm_format: AudioFormat, seconds: float) -> bytes:
58    """
59    Return PCM that reads as audio rather than as an item's trailing silence.
60
61    The holdback measures the silent run a buffer ends with, so a fixture filled
62    with zeroes would stand in for a track that has already finished.
63    """
64    frame = struct.pack("<2h", 9000, -9000)
65    size = int(pcm_format.pcm_sample_size * seconds)
66    return (frame * (size // len(frame) + 1))[:size]
67
68
69def _make_stream_details(
70    media_type: MediaType,
71    *,
72    is_realtime: bool = False,
73    volume_normalization_mode: VolumeNormalizationMode | None = None,
74    queue_id: str | None = None,
75) -> StreamDetails:
76    """Build minimal stream details for AudioBuffer.get_buffer tests."""
77    return StreamDetails(
78        provider="builtin",
79        item_id="item-1",
80        audio_format=TEST_PCM_FORMAT,
81        media_type=media_type,
82        stream_type=StreamType.HTTP,
83        path="http://example.com/audio.mp3",
84        duration=180,
85        can_seek=True,
86        allow_seek=True,
87        queue_id=queue_id,
88        is_realtime=is_realtime,
89        volume_normalization_mode=volume_normalization_mode,
90    )
91
92
93async def _make_source(num_chunks: int) -> AsyncGenerator[bytes]:
94    """Create an async generator that yields one-second PCM chunks."""
95    for _ in range(num_chunks):
96        yield ONE_SECOND_CHUNK
97
98
99def _make_mass_for_get_buffer(
100    *, queue: Any | None = None
101) -> tuple[MagicMock, list[asyncio.Task[None]], list[float | None]]:
102    """Build a minimal mass stub for AudioBuffer.get_buffer tests."""
103    received_seek_positions: list[float | None] = []
104
105    def _get_media_stream(*_args: Any, **kwargs: Any) -> AsyncGenerator[bytes]:
106        received_seek_positions.append(kwargs.get("seek_position"))
107        return _make_source(1)
108
109    mass = MagicMock()
110    mass.config.get_raw_core_config_value.return_value = BufferSize.BALANCED.value
111    mass.player_queues.get.return_value = queue
112    mass.streams = SimpleNamespace(
113        audio_analysis=SimpleNamespace(start_analysis=AsyncMock(return_value=None)),
114        audio=SimpleNamespace(get_media_stream=_get_media_stream),
115    )
116    scheduled_tasks: list[asyncio.Task[None]] = []
117
118    def _create_task(coro: Any) -> asyncio.Task[None]:
119        task: asyncio.Task[None] = asyncio.ensure_future(coro)
120        scheduled_tasks.append(task)
121        return task
122
123    mass.create_task.side_effect = _create_task
124    return mass, scheduled_tasks, received_seek_positions
125
126
127def _streamdetails_for_crossfade(
128    audio_buffer: AudioBuffer | None, *, is_realtime: bool = False
129) -> StreamDetails:
130    """Build incoming track details with an optional prepared buffer."""
131    streamdetails = StreamDetails(
132        provider="test--1",
133        item_id="track-1",
134        audio_format=AudioFormat(content_type=ContentType.FLAC),
135        media_type=MediaType.TRACK,
136        stream_type=StreamType.HTTP,
137        path="http://test.invalid/track.flac",
138        duration=180,
139        is_realtime=is_realtime,
140    )
141    streamdetails.buffer = audio_buffer
142    return streamdetails
143
144
145async def _empty_mix(*_args: object, **_kwargs: object) -> AsyncGenerator[bytes]:
146    """Stand in for the mixer, producing no audio."""
147    no_audio: tuple[bytes, ...] = ()
148    for chunk in no_audio:
149        yield chunk
150
151
152def _buffer(duration_available: float, ready: bool, eof: bool = False) -> AudioBuffer:
153    """Build a valid buffer with the requested resident duration."""
154    audio_buffer = MagicMock(spec=AudioBuffer)
155    audio_buffer.has_error = False
156    audio_buffer.is_valid.return_value = True
157    audio_buffer.duration_available = duration_available
158    audio_buffer.eof = eof
159    audio_buffer.ready = MagicMock()
160    audio_buffer.ready.is_set.return_value = ready
161    return audio_buffer
162
163
164def _stream_details_provider(streamdetails: StreamDetails) -> StreamsAudio:
165    """Build a StreamsAudio whose single provider hands back the given streamdetails."""
166    provider = MagicMock()
167    provider.instance_id = "test--1"
168    provider.domain = "test"
169    provider.available = True
170    provider.is_streaming_provider = True
171    provider.get_stream_details = AsyncMock(return_value=streamdetails)
172    mass = MagicMock()
173    mass.get_provider.side_effect = lambda instance, **_kwargs: (
174        provider if instance == "test--1" else None
175    )
176    mass.providers = []
177    mass.player_queues.queue_data_or_none.return_value = None
178    mass.streams.get_config_value.return_value = -17
179    return StreamsAudio(mass)
180
181
182def _queue_item_with_mapping(media_item_cls: type) -> QueueItem:
183    """Build a queue item whose media item carries one matching provider mapping."""
184    mapping = ProviderMapping(item_id="item-1", provider_domain="test", provider_instance="test--1")
185    media_item = media_item_cls(
186        item_id="item-1", provider="test--1", name="Item", provider_mappings={mapping}
187    )
188    return QueueItem(
189        queue_id="q1", queue_item_id="qi1", name="Item", duration=None, media_item=media_item
190    )
191
192
193# -- AudioBuffer.get_buffer: ready threshold ladder --
194
195
196@pytest.mark.parametrize(
197    (
198        "is_realtime",
199        "crossfade_enabled",
200        "normalization_mode",
201        "media_type",
202        "expected_threshold",
203    ),
204    [
205        pytest.param(True, False, None, MediaType.RADIO, 1, id="realtime_base"),
206        pytest.param(True, False, None, MediaType.AUDIO_SOURCE, 1, id="realtime_audio_source"),
207        # the queue's crossfade setting buys nothing for a realtime source: its fade
208        # streams in as it arrives, so a second of audio here would only be a second
209        # of extra startup delay
210        pytest.param(True, True, None, MediaType.TRACK, 1, id="realtime_crossfade"),
211        pytest.param(
212            True,
213            False,
214            VolumeNormalizationMode.DYNAMIC,
215            MediaType.TRACK,
216            2,
217            id="realtime_dynamic_normalization",
218        ),
219        pytest.param(False, True, None, MediaType.TRACK, 8, id="non_realtime_crossfade"),
220        pytest.param(
221            False,
222            False,
223            VolumeNormalizationMode.DYNAMIC,
224            MediaType.RADIO,
225            3,
226            id="non_realtime_dynamic_radio",
227        ),
228        pytest.param(
229            False,
230            False,
231            VolumeNormalizationMode.DYNAMIC,
232            MediaType.TRACK,
233            5,
234            id="non_realtime_dynamic_track",
235        ),
236        pytest.param(False, False, None, MediaType.TRACK, 2, id="non_realtime_default"),
237    ],
238)
239async def test_ready_threshold_ladder(
240    is_realtime: bool,
241    crossfade_enabled: bool,
242    normalization_mode: VolumeNormalizationMode | None,
243    media_type: MediaType,
244    expected_threshold: int,
245) -> None:
246    """The buffered-ready threshold follows the realtime ladder, leaving the old one intact."""
247    # a realtime source is only ever raised above the floor by dynamic normalization,
248    # which genuinely needs its lookahead
249    queue = SimpleNamespace(crossfade_enabled=crossfade_enabled)
250    mass, scheduled_tasks, _seek_positions = _make_mass_for_get_buffer(queue=queue)
251    streamdetails = _make_stream_details(
252        media_type,
253        is_realtime=is_realtime,
254        volume_normalization_mode=normalization_mode,
255        queue_id="queue-1",
256    )
257
258    buffer = await AudioBuffer.get_buffer(mass, streamdetails, reason="test")
259
260    assert buffer._ready_threshold == expected_threshold
261    await asyncio.gather(*scheduled_tasks)
262    await buffer.clear()
263
264
265# -- AudioBuffer.get_buffer: seek handling --
266
267
268@pytest.mark.parametrize(
269    ("is_realtime", "seek_seconds", "expected_source_seek"),
270    [
271        pytest.param(True, 30, 30, id="realtime_short_seek_reaches_source"),
272        pytest.param(False, 30, 0, id="non_realtime_short_seek_buffers_from_start"),
273        pytest.param(False, 90, 90, id="non_realtime_long_seek_reaches_source"),
274    ],
275)
276async def test_get_buffer_seek_position_reaches_the_source(
277    is_realtime: bool, seek_seconds: int, expected_source_seek: int
278) -> None:
279    """A realtime source always seeks at the source; a non-realtime one only for a large seek."""
280    mass, scheduled_tasks, received_seek_positions = _make_mass_for_get_buffer()
281    streamdetails = _make_stream_details(MediaType.TRACK, is_realtime=is_realtime)
282
283    buffer = await AudioBuffer.get_buffer(
284        mass, streamdetails, seek_position_ms=seek_seconds * 1000, reason="test"
285    )
286
287    assert received_seek_positions == [expected_source_seek]
288    assert buffer._discarded_chunks == expected_source_seek
289    await asyncio.gather(*scheduled_tasks)
290    await buffer.clear()
291
292
293# -- AudioBuffer.eof --
294
295
296async def test_eof_reflects_producer_completion() -> None:
297    """The eof flag turns True only once the producer has delivered everything."""
298    buf = AudioBuffer(TEST_PCM_FORMAT)
299    assert not buf.eof
300    await buf._put(ONE_SECOND_CHUNK)
301    assert not buf.eof
302    await buf._set_eof()
303    assert buf.eof
304
305
306def test_default_pacing_keeps_a_banked_head_start_resident() -> None:
307    """
308    The default output pacing must not flush a realtime source's banked head start.
309
310    The head start a realtime source banks into the item's buffer is the only
311    material its end-of-track crossfade can be built from. A large opening burst
312    hands it to the player at stream open and then drains above the fill rate,
313    so the buffer is empty by EOF and every boundary loses its fade.
314    """
315    default = output_pacing_args()
316    # the drain must not exceed the ~1.1x a realtime source can deliver, and the
317    # opening burst must not swallow a whole banked window
318    assert float(default[default.index("-readrate") + 1]) <= 1.1
319    assert float(default[default.index("-readrate_initial_burst") + 1]) <= 10
320
321
322# -- the holdback decision --
323
324
325def test_nothing_is_held_back_until_the_source_has_delivered_it_all() -> None:
326    """
327    The whole holdback decision: nothing before the source is done, the window after.
328
329    Anything held while the source is still delivering has to come out of audio the
330    player was waiting for, and is heard as a dropout at the boundary. Once the
331    source is finished, what is left in hand arrived after it and the player is not
332    waiting on any of it.
333    """
334    pcm_format = TEST_PCM_FORMAT
335    frame_size = (pcm_format.bit_depth // 8) * pcm_format.channels
336    window = 45 * pcm_format.pcm_sample_size
337
338    def _item(buffer: object) -> Any:
339        return cast(
340            "Any",
341            SimpleNamespace(
342                streamdetails=SimpleNamespace(buffer=buffer, duration=300, seek_position=0)
343            ),
344        )
345
346    # still delivering, however far ahead it has run: nothing may be held
347    filling = SimpleNamespace(eof=False, has_error=False, duration_available=300.0)
348    assert tail_hold_target(_item(filling), window, frame_size) == 0
349
350    # delivered in full: the whole window, aligned to a frame
351    done = SimpleNamespace(eof=True, has_error=False, duration_available=300.0)
352    target = tail_hold_target(_item(done), window, frame_size)
353    assert target == window
354    assert target % frame_size == 0
355
356    # a failed source is skipped without a fade, so its remainder plays out
357    failed = SimpleNamespace(eof=True, has_error=True, duration_available=300.0)
358    assert tail_hold_target(_item(failed), window, frame_size) == 0
359
360    # no buffer yet: opening the stream is what creates it
361    assert tail_hold_target(_item(None), window, frame_size) == 0
362    assert tail_hold_target(cast("Any", SimpleNamespace(streamdetails=None)), window, 4) == 0
363
364    # the buffer is read at decision time, so a capacity reselection that replaces
365    # the item's details is picked up rather than remembered from before
366    item = _item(filling)
367    assert tail_hold_target(item, window, frame_size) == 0
368    item.streamdetails = SimpleNamespace(buffer=done, duration=300, seek_position=0)
369    assert tail_hold_target(item, window, frame_size) == window
370
371    # a window narrower than the source has left is still the cap: the caller keeps
372    # yielding above it, so only the last part of the item is retained
373    narrow = 8 * pcm_format.pcm_sample_size
374    assert tail_hold_target(_item(done), narrow, frame_size) == narrow
375
376
377# -- StreamsAudio._select_buffered_crossfade --
378
379
380def test_the_held_tail_sizes_the_fade_the_configured_mode_picks() -> None:
381    """The mode decides which fade is applied; the held tail only sizes its window."""
382    audio = StreamsAudio(MagicMock())
383
384    # a realtime source barely delivers, yet the tail it banked carries the window:
385    # the incoming side streams in while the blend plays
386    mode, duration = audio._select_buffered_crossfade(
387        _streamdetails_for_crossfade(_buffer(2, ready=True), is_realtime=True),
388        CrossfadeMode.SMART_CROSSFADE,
389        standard_crossfade_duration=8,
390        fade_out_seconds=20,
391    )
392    assert (mode, duration) == (CrossfadeMode.SMART_CROSSFADE, 20)
393
394    # a shorter tail keeps the smart fade, on a shorter window
395    mode, duration = audio._select_buffered_crossfade(
396        _streamdetails_for_crossfade(_buffer(2, ready=True), is_realtime=True),
397        CrossfadeMode.SMART_CROSSFADE,
398        standard_crossfade_duration=8,
399        fade_out_seconds=6,
400    )
401    assert (mode, duration) == (CrossfadeMode.SMART_CROSSFADE, 6)
402
403    # a standard fade never exceeds the configured overlap
404    mode, duration = audio._select_buffered_crossfade(
405        _streamdetails_for_crossfade(_buffer(2, ready=True), is_realtime=True),
406        CrossfadeMode.STANDARD_CROSSFADE,
407        standard_crossfade_duration=8,
408        fade_out_seconds=20,
409    )
410    assert (mode, duration) == (CrossfadeMode.STANDARD_CROSSFADE, 8)
411
412
413def test_a_finished_incoming_source_caps_the_window_at_what_it_holds() -> None:
414    """A source that already ended has no more audio than what is resident."""
415    audio = StreamsAudio(MagicMock())
416
417    mode, duration = audio._select_buffered_crossfade(
418        _streamdetails_for_crossfade(_buffer(6, ready=True, eof=True), is_realtime=True),
419        CrossfadeMode.SMART_CROSSFADE,
420        standard_crossfade_duration=8,
421        fade_out_seconds=45,
422    )
423
424    assert (mode, duration) == (CrossfadeMode.SMART_CROSSFADE, 6)
425
426
427def test_a_short_incoming_track_caps_the_window() -> None:
428    """A long tail cannot claim more overlap than the next track can supply."""
429    audio = StreamsAudio(MagicMock())
430    streamdetails = _streamdetails_for_crossfade(_buffer(2, ready=True), is_realtime=True)
431    streamdetails.duration = 20
432
433    mode, duration = audio._select_buffered_crossfade(
434        streamdetails,
435        CrossfadeMode.SMART_CROSSFADE,
436        standard_crossfade_duration=8,
437        fade_out_seconds=45,
438    )
439
440    assert (mode, duration) == (CrossfadeMode.SMART_CROSSFADE, 10)
441
442
443def test_a_tail_too_short_to_blend_skips_the_fade() -> None:
444    """Below the minimum overlap the tail plays out and the boundary is a hard cut."""
445    audio = StreamsAudio(MagicMock())
446
447    mode, duration = audio._select_buffered_crossfade(
448        _streamdetails_for_crossfade(_buffer(20, ready=True), is_realtime=True),
449        CrossfadeMode.SMART_CROSSFADE,
450        standard_crossfade_duration=8,
451        fade_out_seconds=MIN_CROSSFADE_DURATION - 0.5,
452    )
453
454    assert mode == CrossfadeMode.DISABLED
455    assert duration == 0
456
457
458def test_realtime_incoming_source_not_yet_delivering_skips_the_fade() -> None:
459    """A realtime source whose buffer is not ready yet means the boundary plays clean."""
460    audio = StreamsAudio(MagicMock())
461
462    mode, duration = audio._select_buffered_crossfade(
463        _streamdetails_for_crossfade(_buffer(0, ready=False), is_realtime=True),
464        CrossfadeMode.STANDARD_CROSSFADE,
465        standard_crossfade_duration=8,
466        fade_out_seconds=8,
467    )
468
469    assert mode == CrossfadeMode.DISABLED
470    assert duration == 0
471
472
473# -- Path level: get_queue_item_stream_with_smartfade --
474
475
476async def test_smartfade_realtime_current_item_fades_once_its_source_is_done(
477    monkeypatch: pytest.MonkeyPatch,
478) -> None:
479    """A realtime item whose source finished delivering holds its tail and fades."""
480    pcm_format = AudioFormat(
481        content_type=ContentType.PCM_S16LE,
482        sample_rate=8000,
483        bit_depth=16,
484        channels=2,
485    )
486    # the source is done delivering, which is what arms the realtime holdback
487    current_details = SimpleNamespace(
488        duration=16,
489        seek_position=0,
490        seconds_streamed=0,
491        uri="test://current",
492        buffer=SimpleNamespace(
493            eof=True, cancelled=False, has_error=False, max_size_seconds=300, duration_available=0.0
494        ),
495        is_realtime=True,
496    )
497    next_details = SimpleNamespace(
498        audio_format=pcm_format,
499        buffer=_buffer(SMART_CROSSFADE_DURATION, ready=True),
500        duration=16,
501        seek_position=0,
502        uri="test://next",
503        is_realtime=False,
504        volume_normalization_mode=None,
505    )
506    current_item = SimpleNamespace(
507        queue_id="queue-1",
508        queue_item_id="current",
509        name="Current",
510        streamdetails=current_details,
511        extra_attributes={},
512    )
513    next_item = SimpleNamespace(
514        queue_id="queue-1",
515        queue_item_id="next",
516        name="Next",
517        streamdetails=next_details,
518        extra_attributes={},
519        available=True,
520    )
521    queue = SimpleNamespace(
522        queue_id="queue-1",
523        display_name="Queue",
524        index_in_buffer=0,
525    )
526    player = SimpleNamespace(player_id="player-1", name="Player")
527    mass = MagicMock()
528    mass.player_queues.get.return_value = queue
529    mass.player_queues.load_next_queue_item = AsyncMock(return_value=next_item)
530    mass.player_queues.index_by_id.return_value = 1
531    audio = StreamsAudio(cast("Any", mass))
532    audio.setup()
533    audio.select_pcm_format = AsyncMock(return_value=pcm_format)  # type: ignore[method-assign]
534    audio.crossfade_allowed = MagicMock(return_value=True)  # type: ignore[method-assign]
535    build = AsyncMock(
536        return_value=SimpleNamespace(
537            timing_info=SimpleNamespace(
538                fadein_trimmed_duration=0.0,
539                crossfade_duration=8.0,
540                pre_crossfade_duration=0.0,
541                post_crossfade_duration=0.0,
542            )
543        )
544    )
545    monkeypatch.setattr(audio.smart_fades_mixer, "build", build)
546
547    async def _concat_mix(
548        _smart_fade: object,
549        *,
550        fade_in_part: AsyncGenerator[bytes],
551        fade_out_part: bytes,
552        **_kwargs: object,
553    ) -> AsyncGenerator[bytes]:
554        yield fade_out_part
555        async for fade_in_chunk in fade_in_part:
556            yield fade_in_chunk
557
558    monkeypatch.setattr(audio.smart_fades_mixer, "mix", _concat_mix)
559
560    async def _item_stream(
561        _queue_item: object,
562        *_args: object,
563        **_kwargs: object,
564    ) -> AsyncGenerator[bytes]:
565        yield _audio(pcm_format, 8)
566        yield _audio(pcm_format, 8)
567
568    monkeypatch.setattr(audio, "get_queue_item_stream", _item_stream)
569    stream = audio.get_queue_item_stream_with_smartfade(
570        cast("Any", player),
571        cast("Any", current_item),
572        pcm_format,
573        crossfade_mode=CrossfadeMode.STANDARD_CROSSFADE,
574        standard_crossfade_duration=8,
575    )
576
577    output = b"".join([chunk async for chunk in stream])
578
579    # 8s warmup + 8s of mix output (pre+overlap); the incoming share of the mix
580    # is buffered as crossfade data for the next item's own stream
581    assert len(output) == pcm_format.pcm_sample_size * 16
582    build.assert_awaited_once()
583    crossfade_data = audio._crossfade_handover.get("queue-1")
584    assert crossfade_data is not None
585    assert crossfade_data.queue_item_id == "next"
586
587
588async def _run_smartfade_boundary(
589    monkeypatch: pytest.MonkeyPatch,
590    audio: StreamsAudio,
591    pcm_format: AudioFormat,
592) -> None:
593    """Stream one item through a boundary with the mixer and next item stubbed out."""
594    next_details = SimpleNamespace(
595        audio_format=pcm_format,
596        buffer=_buffer(SMART_CROSSFADE_DURATION, ready=True),
597        duration=16,
598        seek_position=0,
599        uri="test://next",
600        is_realtime=False,
601        volume_normalization_mode=None,
602    )
603    current_item = SimpleNamespace(
604        queue_id="queue-1",
605        queue_item_id="current",
606        name="Current",
607        streamdetails=SimpleNamespace(
608            duration=16,
609            seek_position=0,
610            seconds_streamed=0,
611            uri="test://current",
612            buffer=SimpleNamespace(
613                eof=True,
614                cancelled=False,
615                has_error=False,
616                max_size_seconds=300,
617                duration_available=0.0,
618            ),
619            is_realtime=True,
620        ),
621        extra_attributes={},
622    )
623    next_item = SimpleNamespace(
624        queue_id="queue-1",
625        queue_item_id="next",
626        name="Next",
627        streamdetails=next_details,
628        extra_attributes={},
629        available=True,
630    )
631    mass = cast("Any", audio.mass)
632    mass.player_queues.get.return_value = SimpleNamespace(
633        queue_id="queue-1", display_name="Queue", index_in_buffer=0
634    )
635    mass.player_queues.load_next_queue_item = AsyncMock(return_value=next_item)
636    mass.player_queues.index_by_id.return_value = 1
637    audio.select_pcm_format = AsyncMock(return_value=pcm_format)  # type: ignore[method-assign]
638    audio.crossfade_allowed = MagicMock(return_value=True)  # type: ignore[method-assign]
639    monkeypatch.setattr(
640        audio.smart_fades_mixer,
641        "build",
642        AsyncMock(
643            return_value=SimpleNamespace(
644                timing_info=SimpleNamespace(
645                    fadein_trimmed_duration=0.0,
646                    crossfade_duration=8.0,
647                    pre_crossfade_duration=0.0,
648                    post_crossfade_duration=0.0,
649                )
650            )
651        ),
652    )
653
654    async def _concat_mix(
655        _smart_fade: object,
656        *,
657        fade_in_part: AsyncGenerator[bytes],
658        fade_out_part: bytes,
659        **_kwargs: object,
660    ) -> AsyncGenerator[bytes]:
661        yield fade_out_part
662        async for fade_in_chunk in fade_in_part:
663            yield fade_in_chunk
664
665    monkeypatch.setattr(audio.smart_fades_mixer, "mix", _concat_mix)
666
667    async def _item_stream(
668        _queue_item: object, *_args: object, **_kwargs: object
669    ) -> AsyncGenerator[bytes]:
670        yield _audio(pcm_format, 8)
671        yield _audio(pcm_format, 8)
672
673    monkeypatch.setattr(audio, "get_queue_item_stream", _item_stream)
674    stream = audio.get_queue_item_stream_with_smartfade(
675        cast("Any", SimpleNamespace(player_id="player-1", name="Player")),
676        cast("Any", current_item),
677        pcm_format,
678        crossfade_mode=CrossfadeMode.STANDARD_CROSSFADE,
679        standard_crossfade_duration=8,
680    )
681    async for _chunk in stream:
682        pass
683
684
685async def test_the_live_post_handover_streams_into_the_next_request(
686    monkeypatch: pytest.MonkeyPatch,
687) -> None:
688    """
689    The published boundary mix plays out through the next item's own request.
690
691    The blended intro streams first, and the body continues exactly where the
692    mix stopped reading the item.
693    """
694    pcm_format = AudioFormat(
695        content_type=ContentType.PCM_S16LE, sample_rate=8000, bit_depth=16, channels=2
696    )
697
698    def _sec(value: int) -> bytes:
699        return bytes([value]) * pcm_format.pcm_sample_size
700
701    current_details = SimpleNamespace(
702        duration=16,
703        seek_position=0,
704        seconds_streamed=0,
705        uri="test://current",
706        buffer=SimpleNamespace(
707            eof=True, cancelled=False, has_error=False, max_size_seconds=300, duration_available=0.0
708        ),
709        is_realtime=True,
710    )
711    next_details = SimpleNamespace(
712        audio_format=pcm_format,
713        buffer=_buffer(16.0, ready=True),
714        duration=24,
715        seek_position=0,
716        seconds_streamed=0,
717        uri="test://next",
718        is_realtime=False,
719        volume_normalization_mode=None,
720    )
721    current_item = SimpleNamespace(
722        queue_id="queue-1",
723        queue_item_id="current",
724        name="Current",
725        streamdetails=current_details,
726        extra_attributes={},
727    )
728    next_item = SimpleNamespace(
729        queue_id="queue-1",
730        queue_item_id="next",
731        name="Next",
732        streamdetails=next_details,
733        extra_attributes={},
734        available=True,
735    )
736    queue = SimpleNamespace(queue_id="queue-1", display_name="Queue", index_in_buffer=0)
737    player = SimpleNamespace(player_id="player-1", name="Player")
738    mass = MagicMock()
739    mass.player_queues.get.return_value = queue
740    # the next item's own boundary has nothing to blend into
741    upcoming = iter([next_item])
742
743    def _load_next(*_args: object, **_kwargs: object) -> Any:
744        if (item := next(upcoming, None)) is None:
745            raise QueueEmpty("queue exhausted")
746        return item
747
748    mass.player_queues.load_next_queue_item = AsyncMock(side_effect=_load_next)
749    mass.player_queues.index_by_id.return_value = 1
750    audio = StreamsAudio(cast("Any", mass))
751    audio.setup()
752    audio.select_pcm_format = AsyncMock(return_value=pcm_format)  # type: ignore[method-assign]
753    audio.crossfade_allowed = MagicMock(return_value=True)  # type: ignore[method-assign]
754    build = AsyncMock(
755        return_value=SimpleNamespace(
756            timing_info=SimpleNamespace(
757                fadein_trimmed_duration=0.0,
758                crossfade_duration=8.0,
759                pre_crossfade_duration=0.0,
760                post_crossfade_duration=8.0,
761            )
762        )
763    )
764    monkeypatch.setattr(audio.smart_fades_mixer, "build", build)
765
766    async def _concat_mix(
767        _smart_fade: object,
768        *,
769        fade_in_part: AsyncGenerator[bytes],
770        fade_out_part: bytes,
771        **_kwargs: object,
772    ) -> AsyncGenerator[bytes]:
773        yield fade_out_part
774        async for fade_in_chunk in fade_in_part:
775            yield fade_in_chunk
776
777    monkeypatch.setattr(audio.smart_fades_mixer, "mix", _concat_mix)
778
779    async def _item_stream(
780        queue_item: Any,
781        *_args: object,
782        seek_position: float = 0.0,
783        **_kwargs: object,
784    ) -> AsyncGenerator[bytes]:
785        if queue_item.queue_item_id == "current":
786            for _ in range(16):
787                yield _sec(0x01)
788        else:
789            # a ramp: a lost, repeated or misplaced second shows in the output
790            for second in range(int(seek_position), 24):
791                yield _sec(0x10 + second)
792
793    monkeypatch.setattr(audio, "get_queue_item_stream", _item_stream)
794
795    outgoing = audio.get_queue_item_stream_with_smartfade(
796        cast("Any", player),
797        cast("Any", current_item),
798        pcm_format,
799        crossfade_mode=CrossfadeMode.STANDARD_CROSSFADE,
800        standard_crossfade_duration=8,
801    )
802    outgoing_bytes = b"".join([chunk async for chunk in outgoing])
803
804    # the outgoing request plays only its own item's audio
805    assert outgoing_bytes == _sec(0x01) * 16
806    assert audio._crossfade_handover["queue-1"].queue_item_id == "next"
807
808    incoming = audio.get_queue_item_stream_with_smartfade(
809        cast("Any", player),
810        cast("Any", next_item),
811        pcm_format,
812        crossfade_mode=CrossfadeMode.STANDARD_CROSSFADE,
813        standard_crossfade_duration=8,
814    )
815    incoming_bytes = b"".join([chunk async for chunk in incoming])
816
817    # blended intro first, then the body from exactly where the mix stopped
818    # reading: all 24 seconds of the next item, each exactly once
819    assert incoming_bytes == b"".join(_sec(0x10 + second) for second in range(24))
820    assert "queue-1" not in audio._crossfade_handover
821
822
823async def test_the_handoff_is_claimed_before_the_fade_is_even_sized(
824    monkeypatch: pytest.MonkeyPatch,
825) -> None:
826    """
827    The claim must beat the awaits that size the fade, not follow them.
828
829    Sizing a fade waits on the incoming source, up to REALTIME_FADE_SOURCE_WAIT. A
830    speaker can ask for that item's url inside that window, and it has nothing to
831    wait for unless the claim is already registered.
832    """
833    pcm_format = AudioFormat(
834        content_type=ContentType.PCM_S16LE, sample_rate=8000, bit_depth=16, channels=2
835    )
836    audio = StreamsAudio(MagicMock())
837    audio.setup()
838    claimed_during_sizing = asyncio.Event()
839
840    async def _slow_sizing(_streamdetails: object) -> None:
841        # stands in for the wait on a realtime incoming source
842        if "queue-1" in audio._crossfade_pending:
843            claimed_during_sizing.set()
844        await asyncio.sleep(0)
845
846    monkeypatch.setattr(audio, "_await_realtime_fade_source", _slow_sizing)
847    await _run_smartfade_boundary(monkeypatch, audio, pcm_format)
848
849    assert claimed_during_sizing.is_set(), (
850        "the incoming item had nothing to wait for while its fade was being sized"
851    )
852    # and the claim is gone once the boundary is done with it
853    assert "queue-1" not in audio._crossfade_pending
854
855
856async def test_the_incoming_item_waits_for_a_fade_still_being_mixed(
857    monkeypatch: pytest.MonkeyPatch,
858) -> None:
859    """A speaker asking for the next url early must not lose a nearly-ready fade."""
860    # the real bound has a speaker waiting on its first byte, so it is seconds long;
861    # this test only cares that the wait is bounded at all
862    monkeypatch.setattr("music_assistant.controllers.streams.audio.CROSSFADE_HANDOFF_WAIT", 0.2)
863    pcm_format = AudioFormat(
864        content_type=ContentType.PCM_S16LE, sample_rate=8000, bit_depth=16, channels=2
865    )
866    audio = StreamsAudio(MagicMock())
867    queue = cast("Any", SimpleNamespace(queue_id="queue-1", display_name="Queue"))
868    item = cast("Any", SimpleNamespace(queue_item_id="next", name="Next"))
869
870    # nothing being mixed: the caller is told so straight away
871    assert await audio._await_pending_crossfade(queue, item) is None
872
873    # a fade being mixed for a different item is not this item's to wait for
874    audio._crossfade_pending["queue-1"] = ("other", asyncio.Event())
875    assert await audio._await_pending_crossfade(queue, item) is None
876
877    # a fade being mixed for this item is waited for, and picked up when it lands
878    handoff = asyncio.Event()
879    audio._crossfade_pending["queue-1"] = ("next", handoff)
880    expected = CrossfadeHandover(
881        stream=None, fade_in_media_duration=0.0, pcm_format=pcm_format, queue_item_id="next"
882    )
883
884    async def _land_it() -> None:
885        await asyncio.sleep(0.05)
886        audio._crossfade_handover["queue-1"] = expected
887        handoff.set()
888
889    task = asyncio.create_task(_land_it())
890    assert await audio._await_pending_crossfade(queue, item) is expected
891    await task
892
893    # a mix that never finishes costs the fade, not the stream
894    audio._crossfade_handover.pop("queue-1", None)
895    audio._crossfade_pending["queue-1"] = ("next", asyncio.Event())
896    started = asyncio.get_event_loop().time()
897    assert await audio._await_pending_crossfade(queue, item) is None
898    assert asyncio.get_event_loop().time() - started >= 0.2
899
900
901async def test_smartfade_a_source_still_delivering_hands_over_gapless(
902    monkeypatch: pytest.MonkeyPatch,
903) -> None:
904    """
905    A source that has not finished delivering has no tail to spare for a fade.
906
907    Holding one back would take audio the player is waiting for, and the boundary
908    is heard as a dropout rather than a blend. Gapless is the honest handover.
909    """
910    pcm_format = AudioFormat(
911        content_type=ContentType.PCM_S16LE,
912        sample_rate=8000,
913        bit_depth=16,
914        channels=2,
915    )
916    current_details = SimpleNamespace(
917        duration=16,
918        seek_position=0,
919        seconds_streamed=0,
920        uri="test://current",
921        buffer=SimpleNamespace(eof=False, cancelled=False, has_error=False, max_size_seconds=300),
922        is_realtime=False,
923    )
924    next_details = SimpleNamespace(
925        audio_format=pcm_format,
926        buffer=_buffer(SMART_CROSSFADE_DURATION, ready=True),
927        duration=16,
928        seek_position=0,
929        uri="test://next",
930        volume_normalization_mode=None,
931        is_realtime=False,
932    )
933    current_item = SimpleNamespace(
934        queue_id="queue-1",
935        queue_item_id="current",
936        name="Current",
937        streamdetails=current_details,
938        extra_attributes={},
939    )
940    next_item = SimpleNamespace(
941        queue_id="queue-1",
942        queue_item_id="next",
943        name="Next",
944        streamdetails=next_details,
945        extra_attributes={},
946        available=True,
947    )
948    queue = SimpleNamespace(
949        queue_id="queue-1",
950        display_name="Queue",
951        index_in_buffer=0,
952    )
953    player = SimpleNamespace(player_id="player-1", name="Player")
954    mass = MagicMock()
955    mass.player_queues.get.return_value = queue
956    mass.player_queues.load_next_queue_item = AsyncMock(return_value=next_item)
957    mass.player_queues.index_by_id.return_value = 1
958    audio = StreamsAudio(cast("Any", mass))
959    audio.setup()
960    audio.select_pcm_format = AsyncMock(return_value=pcm_format)  # type: ignore[method-assign]
961    audio.crossfade_allowed = MagicMock(return_value=True)  # type: ignore[method-assign]
962    build = AsyncMock(
963        return_value=SimpleNamespace(
964            timing_info=SimpleNamespace(
965                fadein_trimmed_duration=0.0,
966                crossfade_duration=8.0,
967                pre_crossfade_duration=0.0,
968                post_crossfade_duration=0.0,
969            )
970        )
971    )
972    monkeypatch.setattr(audio.smart_fades_mixer, "build", build)
973
974    async def _concat_mix(
975        _smart_fade: object,
976        *,
977        fade_in_part: AsyncGenerator[bytes],
978        fade_out_part: bytes,
979        **_kwargs: object,
980    ) -> AsyncGenerator[bytes]:
981        yield fade_out_part
982        async for fade_in_chunk in fade_in_part:
983            yield fade_in_chunk
984
985    monkeypatch.setattr(audio.smart_fades_mixer, "mix", _concat_mix)
986
987    async def _item_stream(
988        _queue_item: object,
989        *_args: object,
990        **_kwargs: object,
991    ) -> AsyncGenerator[bytes]:
992        yield _audio(pcm_format, 8)
993        yield _audio(pcm_format, 8)
994
995    monkeypatch.setattr(audio, "get_queue_item_stream", _item_stream)
996    stream = audio.get_queue_item_stream_with_smartfade(
997        cast("Any", player),
998        cast("Any", current_item),
999        pcm_format,
1000        crossfade_mode=CrossfadeMode.STANDARD_CROSSFADE,
1001        standard_crossfade_duration=8,
1002    )
1003
1004    output = b"".join([chunk async for chunk in stream])
1005
1006    # every byte the source produced reaches the player, and no fade is planned
1007    assert len(output) >= pcm_format.pcm_sample_size * 16
1008    build.assert_not_awaited()
1009    assert "queue-1" not in audio._crossfade_handover
1010
1011
1012# -- Path level: get_queue_flow_stream --
1013
1014
1015async def test_flow_realtime_item_yields_all_audio_as_plain_concatenation(
1016    monkeypatch: pytest.MonkeyPatch,
1017) -> None:
1018    """A realtime item's flow audio is passed straight through and simply concatenated."""
1019    pcm_format = AudioFormat(
1020        content_type=ContentType.PCM_S16LE,
1021        sample_rate=8000,
1022        bit_depth=16,
1023        channels=2,
1024    )
1025    # the source is done delivering, so only the realtime flag can deny the holdback
1026    realtime_details = SimpleNamespace(
1027        audio_format=pcm_format,
1028        buffer=SimpleNamespace(eof=True, cancelled=False, has_error=False, max_size_seconds=300),
1029        fade_in=False,
1030        stream_error=False,
1031        uri="test://realtime",
1032        seek_position=0,
1033        seconds_streamed=0,
1034        duration=20,
1035        is_realtime=True,
1036    )
1037    realtime_item = SimpleNamespace(
1038        queue_id="queue-1",
1039        queue_item_id="item-1",
1040        name="Realtime",
1041        media_type=MediaType.TRACK,
1042        media_item=None,
1043        streamdetails=realtime_details,
1044        duration=20,
1045        extra_attributes={},
1046    )
1047    next_details = SimpleNamespace(
1048        audio_format=pcm_format,
1049        buffer=None,
1050        fade_in=False,
1051        stream_error=False,
1052        uri="test://next",
1053        seek_position=0,
1054        seconds_streamed=0,
1055        duration=20,
1056        is_realtime=False,
1057    )
1058    next_item = SimpleNamespace(
1059        queue_id="queue-1",
1060        queue_item_id="item-2",
1061        name="Next",
1062        media_type=MediaType.TRACK,
1063        media_item=None,
1064        streamdetails=next_details,
1065        duration=20,
1066        extra_attributes={},
1067    )
1068    queue = SimpleNamespace(
1069        queue_id="queue-1",
1070        display_name="Queue",
1071        flow_mode=False,
1072        overlay_enabled=False,
1073        overlay_source=None,
1074    )
1075    queue_data = SimpleNamespace(session_id="session-1", flow_mode_stream_log=[])
1076    mass = MagicMock()
1077    mass.player_queues.queue_data.return_value = queue_data
1078    mass.player_queues.load_next_queue_item = AsyncMock(side_effect=[next_item, QueueEmpty])
1079    mass.player_queues.get.return_value = queue
1080    mass.streams.get_crossfade_mode.return_value = CrossfadeMode.STANDARD_CROSSFADE
1081    mass.config.get_raw_core_config_value.return_value = 8
1082    mass.streams.audio_processing.update_item_context = MagicMock()
1083    mass.player_queues.queue_buffer_completed = MagicMock()
1084    player = MagicMock()
1085    player.config.get_value.return_value = "fixed_48000"
1086    player.get_supported_sample_rates.return_value = []
1087    mass.players.get_player.return_value = player
1088    audio = StreamsAudio(cast("Any", mass))
1089    audio.setup()
1090    build = AsyncMock()
1091    monkeypatch.setattr(audio.smart_fades_mixer, "build", build)
1092
1093    realtime_chunks = [
1094        _audio(pcm_format, 8),
1095        _audio(pcm_format, 8),
1096    ]
1097    next_chunks = [_audio(pcm_format, 2)]
1098
1099    async def _item_stream(
1100        queue_item: SimpleNamespace, *_args: object, **_kwargs: object
1101    ) -> AsyncGenerator[bytes]:
1102        chunks = realtime_chunks if queue_item is realtime_item else next_chunks
1103        for chunk in chunks:
1104            yield chunk
1105
1106    monkeypatch.setattr(audio, "get_queue_item_stream", _item_stream)
1107    select_crossfade = MagicMock(wraps=audio._select_buffered_crossfade)
1108    monkeypatch.setattr(audio, "_select_buffered_crossfade", select_crossfade)
1109    stream = audio.get_queue_flow_stream(
1110        cast("Any", queue), cast("Any", realtime_item), pcm_format, session_id="session-1"
1111    )
1112
1113    output = b"".join([chunk async for chunk in stream])
1114
1115    assert output == b"".join(realtime_chunks) + b"".join(next_chunks)
1116    build.assert_not_awaited()
1117    # no tail was held back, so the next item is never asked to fade into anything
1118    select_crossfade.assert_not_called()
1119    mass.player_queues.queue_buffer_completed.assert_called_once()
1120
1121
1122async def test_smartfade_unaligned_chunks_still_crossfade(
1123    monkeypatch: pytest.MonkeyPatch,
1124) -> None:
1125    """A source whose chunks are not whole seconds still collects a complete fade tail."""
1126    pcm_format = AudioFormat(
1127        content_type=ContentType.PCM_S16LE,
1128        sample_rate=8000,
1129        bit_depth=16,
1130        channels=2,
1131    )
1132    current_details = SimpleNamespace(
1133        duration=30,
1134        seek_position=0,
1135        seconds_streamed=0,
1136        uri="test://current",
1137        buffer=SimpleNamespace(eof=True, cancelled=False, has_error=False, max_size_seconds=300),
1138        is_realtime=False,
1139    )
1140    next_details = SimpleNamespace(
1141        audio_format=pcm_format,
1142        buffer=_buffer(SMART_CROSSFADE_DURATION, ready=True),
1143        duration=30,
1144        seek_position=0,
1145        uri="test://next",
1146        is_realtime=False,
1147        volume_normalization_mode=None,
1148    )
1149    current_item = SimpleNamespace(
1150        queue_id="queue-1",
1151        queue_item_id="current",
1152        name="Current",
1153        streamdetails=current_details,
1154        extra_attributes={},
1155    )
1156    next_item = SimpleNamespace(
1157        queue_id="queue-1",
1158        queue_item_id="next",
1159        name="Next",
1160        streamdetails=next_details,
1161        extra_attributes={},
1162        available=True,
1163    )
1164    queue = SimpleNamespace(queue_id="queue-1", display_name="Queue", index_in_buffer=0)
1165    player = SimpleNamespace(player_id="player-1", name="Player")
1166    mass = MagicMock()
1167    mass.player_queues.get.return_value = queue
1168    mass.player_queues.load_next_queue_item = AsyncMock(return_value=next_item)
1169    mass.player_queues.index_by_id.return_value = 1
1170    audio = StreamsAudio(cast("Any", mass))
1171    audio.setup()
1172    audio.select_pcm_format = AsyncMock(return_value=pcm_format)  # type: ignore[method-assign]
1173    audio.crossfade_allowed = MagicMock(return_value=True)  # type: ignore[method-assign]
1174    build = AsyncMock(
1175        return_value=SimpleNamespace(
1176            timing_info=SimpleNamespace(
1177                pre_crossfade_duration=2,
1178                crossfade_duration=6,
1179                fadein_trimmed_duration=0,
1180            )
1181        )
1182    )
1183    monkeypatch.setattr(audio.smart_fades_mixer, "build", build)
1184    monkeypatch.setattr(audio.smart_fades_mixer, "mix", _empty_mix)
1185
1186    async def _current_stream(
1187        queue_item: object, *_args: object, **_kwargs: object
1188    ) -> AsyncGenerator[bytes]:
1189        if queue_item is not current_item:
1190            return
1191        # a whole second, then chunks that never line up with a second boundary
1192        yield _audio(pcm_format, 8)
1193        for _ in range(30):
1194            yield _audio(pcm_format, 1 / 3)
1195
1196    monkeypatch.setattr(audio, "get_queue_item_stream", _current_stream)
1197    stream = audio.get_queue_item_stream_with_smartfade(
1198        cast("Any", player),
1199        cast("Any", current_item),
1200        pcm_format,
1201        crossfade_mode=CrossfadeMode.STANDARD_CROSSFADE,
1202        standard_crossfade_duration=8,
1203    )
1204
1205    async for _chunk in stream:
1206        pass
1207
1208    build.assert_awaited_once()
1209
1210
1211async def test_smartfade_short_remainder_still_crossfades(
1212    monkeypatch: pytest.MonkeyPatch,
1213) -> None:
1214    """Less audio left than the configured overlap still fades with what is there."""
1215    pcm_format = AudioFormat(
1216        content_type=ContentType.PCM_S16LE,
1217        sample_rate=8000,
1218        bit_depth=16,
1219        channels=2,
1220    )
1221    current_details = SimpleNamespace(
1222        duration=180,
1223        seek_position=146,
1224        seconds_streamed=0,
1225        uri="test://current",
1226        buffer=SimpleNamespace(eof=True, cancelled=False, has_error=False, max_size_seconds=300),
1227        is_realtime=False,
1228    )
1229    next_details = SimpleNamespace(
1230        audio_format=pcm_format,
1231        buffer=_buffer(SMART_CROSSFADE_DURATION, ready=True),
1232        duration=180,
1233        seek_position=0,
1234        uri="test://next",
1235        is_realtime=False,
1236        volume_normalization_mode=None,
1237    )
1238    current_item = SimpleNamespace(
1239        queue_id="queue-1",
1240        queue_item_id="current",
1241        name="Current",
1242        streamdetails=current_details,
1243        extra_attributes={},
1244    )
1245    next_item = SimpleNamespace(
1246        queue_id="queue-1",
1247        queue_item_id="next",
1248        name="Next",
1249        streamdetails=next_details,
1250        extra_attributes={},
1251        available=True,
1252    )
1253    queue = SimpleNamespace(queue_id="queue-1", display_name="Queue", index_in_buffer=0)
1254    player = SimpleNamespace(player_id="player-1", name="Player")
1255    mass = MagicMock()
1256    mass.player_queues.get.return_value = queue
1257    mass.player_queues.load_next_queue_item = AsyncMock(return_value=next_item)
1258    mass.player_queues.index_by_id.return_value = 1
1259    audio = StreamsAudio(cast("Any", mass))
1260    audio.setup()
1261    audio.select_pcm_format = AsyncMock(return_value=pcm_format)  # type: ignore[method-assign]
1262    audio.crossfade_allowed = MagicMock(return_value=True)  # type: ignore[method-assign]
1263    build = AsyncMock(
1264        return_value=SimpleNamespace(
1265            timing_info=SimpleNamespace(
1266                pre_crossfade_duration=2,
1267                crossfade_duration=6,
1268                fadein_trimmed_duration=0,
1269            )
1270        )
1271    )
1272    monkeypatch.setattr(audio.smart_fades_mixer, "build", build)
1273    monkeypatch.setattr(audio.smart_fades_mixer, "mix", _empty_mix)
1274
1275    async def _current_stream(
1276        queue_item: object, *_args: object, **_kwargs: object
1277    ) -> AsyncGenerator[bytes]:
1278        if queue_item is not current_item:
1279            return
1280        # a seek near the end leaves 34s, less than the 45s smart overlap
1281        yield _audio(pcm_format, 8)
1282        yield _audio(pcm_format, 26)
1283
1284    # everything left of a short remainder is fade material: nothing bypasses
1285    # the holdback anymore
1286
1287    monkeypatch.setattr(audio, "get_queue_item_stream", _current_stream)
1288    stream = audio.get_queue_item_stream_with_smartfade(
1289        cast("Any", player),
1290        cast("Any", current_item),
1291        pcm_format,
1292        crossfade_mode=CrossfadeMode.SMART_CROSSFADE,
1293        standard_crossfade_duration=8,
1294    )
1295
1296    async for _chunk in stream:
1297        pass
1298
1299    build.assert_awaited_once()
1300    assert build.await_args is not None
1301    fade_out_seconds = len(build.await_args.kwargs["fade_out_data"]) / pcm_format.pcm_sample_size
1302    assert fade_out_seconds == pytest.approx(34, abs=1)
1303
1304
1305async def test_smartfade_stub_remainder_does_not_crossfade(
1306    monkeypatch: pytest.MonkeyPatch,
1307) -> None:
1308    """A remainder too short to overlap with is played out instead of faded."""
1309    pcm_format = AudioFormat(
1310        content_type=ContentType.PCM_S16LE,
1311        sample_rate=8000,
1312        bit_depth=16,
1313        channels=2,
1314    )
1315    current_details = SimpleNamespace(
1316        duration=180,
1317        seek_position=176,
1318        seconds_streamed=0,
1319        uri="test://current",
1320        buffer=SimpleNamespace(eof=True, cancelled=False, has_error=False, max_size_seconds=300),
1321        is_realtime=False,
1322    )
1323    next_details = SimpleNamespace(
1324        audio_format=pcm_format,
1325        buffer=_buffer(SMART_CROSSFADE_DURATION, ready=True),
1326        duration=180,
1327        seek_position=0,
1328        uri="test://next",
1329        is_realtime=False,
1330        volume_normalization_mode=None,
1331    )
1332    current_item = SimpleNamespace(
1333        queue_id="queue-1",
1334        queue_item_id="current",
1335        name="Current",
1336        streamdetails=current_details,
1337        extra_attributes={},
1338    )
1339    next_item = SimpleNamespace(
1340        queue_id="queue-1",
1341        queue_item_id="next",
1342        name="Next",
1343        streamdetails=next_details,
1344        extra_attributes={},
1345        available=True,
1346    )
1347    queue = SimpleNamespace(queue_id="queue-1", display_name="Queue", index_in_buffer=0)
1348    player = SimpleNamespace(player_id="player-1", name="Player")
1349    mass = MagicMock()
1350    mass.player_queues.get.return_value = queue
1351    mass.player_queues.load_next_queue_item = AsyncMock(return_value=next_item)
1352    mass.player_queues.index_by_id.return_value = 1
1353    audio = StreamsAudio(cast("Any", mass))
1354    audio.setup()
1355    audio.select_pcm_format = AsyncMock(return_value=pcm_format)  # type: ignore[method-assign]
1356    audio.crossfade_allowed = MagicMock(return_value=True)  # type: ignore[method-assign]
1357    build = AsyncMock()
1358    monkeypatch.setattr(audio.smart_fades_mixer, "build", build)
1359
1360    async def _current_stream(
1361        queue_item: object, *_args: object, **_kwargs: object
1362    ) -> AsyncGenerator[bytes]:
1363        if queue_item is not current_item:
1364            return
1365        # 2s remainder: under MIN_CROSSFADE_DURATION, so nothing to blend with
1366        yield _audio(pcm_format, 2)
1367
1368    monkeypatch.setattr(audio, "get_queue_item_stream", _current_stream)
1369    stream = audio.get_queue_item_stream_with_smartfade(
1370        cast("Any", player),
1371        cast("Any", current_item),
1372        pcm_format,
1373        crossfade_mode=CrossfadeMode.SMART_CROSSFADE,
1374        standard_crossfade_duration=8,
1375    )
1376
1377    output = b"".join([chunk async for chunk in stream])
1378
1379    assert len(output) == pcm_format.pcm_sample_size * 2
1380    build.assert_not_awaited()
1381
1382
1383async def test_flow_reports_no_fade_for_a_realtime_item_until_one_renders(
1384    monkeypatch: pytest.MonkeyPatch,
1385) -> None:
1386    """
1387    A realtime item is not credited with any fade up front.
1388
1389    A fade is only reported once one is really rendered at its boundary; the
1390    source-delegation reporting is gone along with the delegation itself.
1391    """
1392    pcm_format = AudioFormat(
1393        content_type=ContentType.PCM_S16LE,
1394        sample_rate=8000,
1395        bit_depth=16,
1396        channels=2,
1397    )
1398    realtime_details = SimpleNamespace(
1399        audio_format=pcm_format,
1400        buffer=SimpleNamespace(eof=True, cancelled=False, has_error=False, max_size_seconds=300),
1401        fade_in=False,
1402        stream_error=False,
1403        uri="test://realtime",
1404        seek_position=0,
1405        seconds_streamed=0,
1406        duration=20,
1407        is_realtime=True,
1408    )
1409    realtime_item = SimpleNamespace(
1410        queue_id="queue-1",
1411        queue_item_id="item-1",
1412        name="Realtime",
1413        media_type=MediaType.TRACK,
1414        media_item=None,
1415        streamdetails=realtime_details,
1416        duration=20,
1417        extra_attributes={},
1418    )
1419    queue = SimpleNamespace(
1420        queue_id="queue-1",
1421        display_name="Queue",
1422        flow_mode=False,
1423        overlay_enabled=False,
1424        overlay_source=None,
1425    )
1426    mass = MagicMock()
1427    mass.player_queues.queue_data.return_value = SimpleNamespace(
1428        session_id="session-1", flow_mode_stream_log=[]
1429    )
1430    mass.player_queues.load_next_queue_item = AsyncMock(side_effect=QueueEmpty)
1431    mass.player_queues.get.return_value = queue
1432    mass.streams.get_crossfade_mode.return_value = CrossfadeMode.SMART_CROSSFADE
1433    mass.config.get_raw_core_config_value.return_value = 8
1434    update_item_context = MagicMock()
1435    mass.streams.audio_processing.update_item_context = update_item_context
1436    player = MagicMock()
1437    player.config.get_value.return_value = "fixed_48000"
1438    player.get_supported_sample_rates.return_value = []
1439    mass.players.get_player.return_value = player
1440    audio = StreamsAudio(cast("Any", mass))
1441    audio.setup()
1442
1443    async def _item_stream(*_args: object, **_kwargs: object) -> AsyncGenerator[bytes]:
1444        yield _audio(pcm_format, 4)
1445
1446    monkeypatch.setattr(audio, "get_queue_item_stream", _item_stream)
1447    stream = audio.get_queue_flow_stream(
1448        cast("Any", queue), cast("Any", realtime_item), pcm_format, session_id="session-1"
1449    )
1450
1451    async for _chunk in stream:
1452        pass
1453
1454    update_item_context.assert_called()
1455    reported = update_item_context.call_args.kwargs["queue_processing"]
1456    assert reported.crossfade_mode == CrossfadeMode.DISABLED
1457
1458
1459async def test_flow_standard_fade_only_holds_back_its_overlap(
1460    monkeypatch: pytest.MonkeyPatch,
1461) -> None:
1462    """A standard transition waits for its overlap, not for the whole requested window."""
1463    pcm_format = AudioFormat(
1464        content_type=ContentType.PCM_S16LE,
1465        sample_rate=8000,
1466        bit_depth=16,
1467        channels=2,
1468    )
1469    first_details = SimpleNamespace(
1470        audio_format=pcm_format,
1471        buffer=SimpleNamespace(eof=True, cancelled=False, has_error=False, max_size_seconds=300),
1472        fade_in=False,
1473        stream_error=False,
1474        uri="test://first",
1475        seek_position=0,
1476        seconds_streamed=0,
1477        duration=300,
1478        is_realtime=False,
1479    )
1480    second_details = SimpleNamespace(
1481        audio_format=pcm_format,
1482        buffer=_buffer(SMART_CROSSFADE_DURATION, ready=True),
1483        fade_in=False,
1484        stream_error=False,
1485        uri="test://second",
1486        seek_position=0,
1487        seconds_streamed=0,
1488        duration=300,
1489        is_realtime=False,
1490        volume_normalization_mode=None,
1491    )
1492    first_item = SimpleNamespace(
1493        queue_id="queue-1",
1494        queue_item_id="item-1",
1495        name="First",
1496        media_type=MediaType.TRACK,
1497        media_item=None,
1498        streamdetails=first_details,
1499        duration=300,
1500        extra_attributes={},
1501    )
1502    second_item = SimpleNamespace(
1503        queue_id="queue-1",
1504        queue_item_id="item-2",
1505        name="Second",
1506        media_type=MediaType.TRACK,
1507        media_item=None,
1508        streamdetails=second_details,
1509        duration=300,
1510        extra_attributes={},
1511    )
1512    queue = SimpleNamespace(
1513        queue_id="queue-1",
1514        display_name="Queue",
1515        flow_mode=False,
1516        overlay_enabled=False,
1517        overlay_source=None,
1518    )
1519    mass = MagicMock()
1520    mass.player_queues.queue_data.return_value = SimpleNamespace(
1521        session_id="session-1", flow_mode_stream_log=[]
1522    )
1523    mass.player_queues.load_next_queue_item = AsyncMock(side_effect=[second_item, QueueEmpty])
1524    mass.player_queues.get.return_value = queue
1525    mass.streams.get_crossfade_mode.return_value = CrossfadeMode.SMART_CROSSFADE
1526    mass.config.get_raw_core_config_value.return_value = 8
1527    player = MagicMock()
1528    player.config.get_value.return_value = "fixed_48000"
1529    player.get_supported_sample_rates.return_value = []
1530    mass.players.get_player.return_value = player
1531    audio = StreamsAudio(cast("Any", mass))
1532    audio.setup()
1533    audio.crossfade_allowed = MagicMock(return_value=True)  # type: ignore[method-assign]
1534    # the incoming analysis is not ready, so the mixer degrades to a standard fade
1535    standard = StandardCrossFade(logger=MagicMock(), crossfade_duration=8)
1536    standard.build(
1537        pcm_format.pcm_sample_size * SMART_CROSSFADE_DURATION,
1538        pcm_format.pcm_sample_size * SMART_CROSSFADE_DURATION,
1539        pcm_format,
1540    )
1541    monkeypatch.setattr(audio.smart_fades_mixer, "build", AsyncMock(return_value=standard))
1542    monkeypatch.setattr(audio.smart_fades_mixer, "mix", _empty_mix)
1543
1544    consumed: dict[str, int] = {"second": 0}
1545
1546    async def _item_stream(
1547        queue_item: SimpleNamespace, *_args: object, **_kwargs: object
1548    ) -> AsyncGenerator[bytes]:
1549        if queue_item is first_item:
1550            for _ in range(60):
1551                yield bytes(pcm_format.pcm_sample_size)
1552            return
1553        for _ in range(SMART_CROSSFADE_DURATION + 20):
1554            consumed["second"] += 1
1555            yield bytes(pcm_format.pcm_sample_size)
1556
1557    monkeypatch.setattr(audio, "get_queue_item_stream", _item_stream)
1558    stream = audio.get_queue_flow_stream(
1559        cast("Any", queue), cast("Any", first_item), pcm_format, session_id="session-1"
1560    )
1561
1562    seconds_before_transition: int | None = None
1563    async for _chunk in stream:
1564        if seconds_before_transition is None and consumed["second"]:
1565            seconds_before_transition = consumed["second"]
1566
1567    # the overlap is 8s, so the transition must not wait for the full 45s window
1568    assert seconds_before_transition is not None
1569    assert seconds_before_transition <= SMART_CROSSFADE_DURATION / 2
1570
1571
1572# -- StreamsController.serve_queue_item_stream steering --
1573
1574
1575class _PcmFormatRequested(Exception):
1576    """Raised to stop the handler once it has decided on crossfading."""
1577
1578
1579class _FfmpegArgsCaptured(Exception):
1580    """Raised to stop the handler once the encode ffmpeg would start."""
1581
1582
1583class _FakeStreamResponse:
1584    """Accept the handler's response plumbing without a real HTTP transport."""
1585
1586    def __init__(self, **_kwargs: Any) -> None:
1587        self.content_type: str | None = None
1588        self.content_length: int | None = None
1589
1590    def enable_chunked_encoding(self) -> None:
1591        """Accept the chunked-profile branch."""
1592
1593    async def prepare(self, request: Any) -> None:
1594        """Accept the response start."""
1595
1596
1597def _single_item_handler(
1598    *,
1599    is_realtime: bool,
1600    capture_ffmpeg: pytest.MonkeyPatch | None = None,
1601    player_provider_domain: str = "test",
1602) -> tuple[Any, MagicMock, dict[str, Any]]:
1603    """
1604    Return a single-item stream handler rigged to stop early.
1605
1606    Without ``capture_ffmpeg`` it stops once the PCM format is picked; with it, the
1607    handler runs on to the encode ffmpeg call and records its keyword arguments.
1608    """
1609    streamdetails = _make_stream_details(MediaType.TRACK, is_realtime=is_realtime)
1610    queue_item = SimpleNamespace(
1611        queue_id="queue-1",
1612        queue_item_id="item-1",
1613        name="Track",
1614        duration=180,
1615        streamdetails=streamdetails,
1616        media_item=None,
1617        media_type=MediaType.TRACK,
1618        extra_attributes={},
1619        image=None,
1620    )
1621    queue = SimpleNamespace(
1622        queue_id="queue-1",
1623        display_name="Queue",
1624        current_item=queue_item,
1625        crossfade_enabled=True,
1626        overlay_enabled=False,
1627        overlay_source=None,
1628    )
1629    mass = MagicMock()
1630    mass.player_queues.get.return_value = queue
1631    mass.player_queues.queue_data.return_value = SimpleNamespace(session_id="session-1")
1632    mass.player_queues.get_item.return_value = queue_item
1633    mass.config.get_raw_core_config_value.return_value = 8
1634    player = MagicMock(player_id="player-1", protocol_parent_id=None)
1635    player.provider.domain = player_provider_domain
1636    player.state.supported_features = {PlayerFeature.GAPLESS_PLAYBACK}
1637    player.state.name = "Player"
1638    mass.players.get_player.return_value = player
1639
1640    seen: dict[str, Any] = {}
1641
1642    async def _select_pcm_format(**kwargs: Any) -> Any:
1643        seen["crossfade_enabled"] = kwargs["crossfade_enabled"]
1644        if capture_ffmpeg is None:
1645            raise _PcmFormatRequested
1646        return TEST_PCM_FORMAT
1647
1648    audio = MagicMock()
1649    audio.select_pcm_format = _select_pcm_format
1650    controller = cast("Any", object.__new__(StreamsController))
1651    controller.mass = mass
1652    controller.audio = audio
1653    controller.logger = MagicMock()
1654    controller._log_request = MagicMock()
1655    controller.get_crossfade_mode = MagicMock(return_value=CrossfadeMode.SMART_CROSSFADE)
1656    request = MagicMock()
1657    request.method = "GET"
1658    request.match_info = {
1659        "queue_id": "queue-1",
1660        "player_id": "player-1",
1661        "session_id": "session-1",
1662        "queue_item_id": "item-1",
1663        "fmt": "flac",
1664    }
1665    if capture_ffmpeg is not None:
1666        audio.get_output_format = AsyncMock(
1667            return_value=AudioFormat(
1668                content_type=ContentType.FLAC, sample_rate=44100, bit_depth=16, channels=2
1669            )
1670        )
1671        player.get_config_value = MagicMock(return_value="default")
1672        controller._update_audio_processing_context = MagicMock()
1673
1674        def _capture_ffmpeg_args(**kwargs: Any) -> None:
1675            seen["extra_input_args"] = kwargs["extra_input_args"]
1676            raise _FfmpegArgsCaptured
1677
1678        capture_ffmpeg.setattr(controller_mod, "get_ffmpeg_stream", _capture_ffmpeg_args)
1679        capture_ffmpeg.setattr(
1680            controller_mod, "web", SimpleNamespace(StreamResponse=_FakeStreamResponse)
1681        )
1682    return controller, request, seen
1683
1684
1685async def test_single_item_handler_keeps_crossfade_for_a_realtime_item() -> None:
1686    """A realtime item whose source does not fade keeps the queue's crossfade."""
1687    controller, request, seen = _single_item_handler(is_realtime=True)
1688
1689    with pytest.raises(_PcmFormatRequested):
1690        await controller.serve_queue_item_stream(request)
1691
1692    assert seen["crossfade_enabled"] is True
1693    controller.get_crossfade_mode.assert_called_once()
1694
1695
1696async def test_single_item_handler_keeps_crossfade_for_a_buffered_item() -> None:
1697    """A buffered item still gets the queue's configured crossfade."""
1698    controller, request, seen = _single_item_handler(is_realtime=False)
1699
1700    with pytest.raises(_PcmFormatRequested):
1701        await controller.serve_queue_item_stream(request)
1702
1703    assert seen["crossfade_enabled"] is True
1704    controller.get_crossfade_mode.assert_called_once()
1705
1706
1707@pytest.mark.parametrize(
1708    ("player_provider_domain", "profile"),
1709    [("sonos", "default"), ("musiccast", "gapless_burst")],
1710    ids=["default", "musiccast"],
1711)
1712async def test_single_item_handler_paces_by_player(
1713    monkeypatch: pytest.MonkeyPatch, player_provider_domain: str, profile: str
1714) -> None:
1715    """Every player gets the gentle default; MusicCast gets its gapless opening burst."""
1716    controller, request, seen = _single_item_handler(
1717        is_realtime=True,
1718        capture_ffmpeg=monkeypatch,
1719        player_provider_domain=player_provider_domain,
1720    )
1721
1722    with pytest.raises(_FfmpegArgsCaptured):
1723        await controller.serve_queue_item_stream(request)
1724
1725    assert seen["extra_input_args"] == output_pacing_args(profile)  # type: ignore[arg-type]
1726
1727
1728# -- StreamsAudio.get_stream_details --
1729
1730
1731@pytest.mark.parametrize(
1732    ("media_item_cls", "media_type", "expected_is_realtime"),
1733    [
1734        pytest.param(Radio, MediaType.RADIO, True, id="radio"),
1735        pytest.param(AudioSource, MediaType.AUDIO_SOURCE, True, id="audio_source"),
1736        pytest.param(Track, MediaType.TRACK, False, id="track"),
1737    ],
1738)
1739async def test_get_stream_details_sets_is_realtime_by_media_type(
1740    media_item_cls: type, media_type: MediaType, expected_is_realtime: bool
1741) -> None:
1742    """RADIO and AUDIO_SOURCE streams are marked realtime; a TRACK's flag is left alone."""
1743    provider_streamdetails = StreamDetails(
1744        provider="test--1",
1745        item_id="item-1",
1746        audio_format=AudioFormat(content_type=ContentType.MP3),
1747        media_type=media_type,
1748        stream_type=StreamType.CUSTOM,
1749        duration=180 if media_type == MediaType.TRACK else None,
1750    )
1751    audio = _stream_details_provider(provider_streamdetails)
1752
1753    streamdetails = await audio.get_stream_details(
1754        queue_item=_queue_item_with_mapping(media_item_cls)
1755    )
1756
1757    assert streamdetails.is_realtime is expected_is_realtime
1758