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