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 StreamsAudio
32from music_assistant.controllers.streams.audio_buffer import AudioBuffer
33from music_assistant.controllers.streams.constants import BufferSize
34from music_assistant.controllers.streams.controller import StreamsController
35from music_assistant.controllers.streams.smart_fades.fades import StandardCrossFade
36from music_assistant.controllers.streams.smart_fades.helpers import SMART_CROSSFADE_DURATION
37from music_assistant.models.music_provider import MusicProvider
38
39# Standard test PCM format: 44100Hz, 16-bit, stereo
40TEST_PCM_FORMAT = AudioFormat(
41    content_type=ContentType.PCM_S16LE,
42    sample_rate=44100,
43    bit_depth=16,
44    channels=2,
45)
46
47# One second of silence in the test format
48ONE_SECOND_CHUNK = b"\x00" * TEST_PCM_FORMAT.pcm_sample_size
49
50
51def _make_stream_details(
52    media_type: MediaType,
53    *,
54    is_realtime: bool = False,
55    volume_normalization_mode: VolumeNormalizationMode | None = None,
56    queue_id: str | None = None,
57) -> StreamDetails:
58    """Build minimal stream details for AudioBuffer.get_buffer tests."""
59    return StreamDetails(
60        provider="builtin",
61        item_id="item-1",
62        audio_format=TEST_PCM_FORMAT,
63        media_type=media_type,
64        stream_type=StreamType.HTTP,
65        path="http://example.com/audio.mp3",
66        duration=180,
67        can_seek=True,
68        allow_seek=True,
69        queue_id=queue_id,
70        is_realtime=is_realtime,
71        volume_normalization_mode=volume_normalization_mode,
72    )
73
74
75async def _make_source(num_chunks: int) -> AsyncGenerator[bytes]:
76    """Create an async generator that yields one-second PCM chunks."""
77    for _ in range(num_chunks):
78        yield ONE_SECOND_CHUNK
79
80
81def _make_mass_for_get_buffer(
82    *, queue: Any | None = None
83) -> tuple[MagicMock, list[asyncio.Task[None]], list[float | None]]:
84    """Build a minimal mass stub for AudioBuffer.get_buffer tests."""
85    received_seek_positions: list[float | None] = []
86
87    def _get_media_stream(*_args: Any, **kwargs: Any) -> AsyncGenerator[bytes]:
88        received_seek_positions.append(kwargs.get("seek_position"))
89        return _make_source(1)
90
91    mass = MagicMock()
92    mass.config.get_raw_core_config_value.return_value = BufferSize.BALANCED.value
93    mass.player_queues.get.return_value = queue
94    mass.streams = SimpleNamespace(
95        audio_analysis=SimpleNamespace(start_analysis=AsyncMock(return_value=None)),
96        audio=SimpleNamespace(get_media_stream=_get_media_stream),
97    )
98    scheduled_tasks: list[asyncio.Task[None]] = []
99
100    def _create_task(coro: Any) -> asyncio.Task[None]:
101        task: asyncio.Task[None] = asyncio.ensure_future(coro)
102        scheduled_tasks.append(task)
103        return task
104
105    mass.create_task.side_effect = _create_task
106    return mass, scheduled_tasks, received_seek_positions
107
108
109def _streamdetails_for_crossfade(
110    audio_buffer: AudioBuffer | None, *, is_realtime: bool = False
111) -> StreamDetails:
112    """Build incoming track details with an optional prepared buffer."""
113    streamdetails = StreamDetails(
114        provider="test--1",
115        item_id="track-1",
116        audio_format=AudioFormat(content_type=ContentType.FLAC),
117        media_type=MediaType.TRACK,
118        stream_type=StreamType.HTTP,
119        path="http://test.invalid/track.flac",
120        duration=180,
121        is_realtime=is_realtime,
122    )
123    streamdetails.buffer = audio_buffer
124    return streamdetails
125
126
127async def _empty_mix(*_args: object, **_kwargs: object) -> AsyncGenerator[bytes]:
128    """Stand in for the mixer, producing no audio."""
129    no_audio: tuple[bytes, ...] = ()
130    for chunk in no_audio:
131        yield chunk
132
133
134def _buffer(duration_available: float, ready: bool) -> AudioBuffer:
135    """Build a valid buffer with the requested resident duration."""
136    audio_buffer = MagicMock(spec=AudioBuffer)
137    audio_buffer.has_error = False
138    audio_buffer.is_valid.return_value = True
139    audio_buffer.duration_available = duration_available
140    audio_buffer.ready = MagicMock()
141    audio_buffer.ready.is_set.return_value = ready
142    return audio_buffer
143
144
145def _stream_details_provider(streamdetails: StreamDetails) -> StreamsAudio:
146    """Build a StreamsAudio whose single provider hands back the given streamdetails."""
147    provider = MagicMock()
148    provider.instance_id = "test--1"
149    provider.domain = "test"
150    provider.available = True
151    provider.is_streaming_provider = True
152    provider.get_stream_details = AsyncMock(return_value=streamdetails)
153    mass = MagicMock()
154    mass.get_provider.side_effect = lambda instance, **_kwargs: (
155        provider if instance == "test--1" else None
156    )
157    mass.providers = []
158    mass.player_queues.queue_data_or_none.return_value = None
159    mass.streams.get_config_value.return_value = -17
160    return StreamsAudio(mass)
161
162
163def _queue_item_with_mapping(media_item_cls: type) -> QueueItem:
164    """Build a queue item whose media item carries one matching provider mapping."""
165    mapping = ProviderMapping(item_id="item-1", provider_domain="test", provider_instance="test--1")
166    media_item = media_item_cls(
167        item_id="item-1", provider="test--1", name="Item", provider_mappings={mapping}
168    )
169    return QueueItem(
170        queue_id="q1", queue_item_id="qi1", name="Item", duration=None, media_item=media_item
171    )
172
173
174# -- AudioBuffer.get_buffer: ready threshold ladder --
175
176
177@pytest.mark.parametrize(
178    (
179        "is_realtime",
180        "crossfade_enabled",
181        "normalization_mode",
182        "media_type",
183        "expected_threshold",
184    ),
185    [
186        pytest.param(True, False, None, MediaType.RADIO, 1, id="realtime_base"),
187        pytest.param(True, False, None, MediaType.AUDIO_SOURCE, 1, id="realtime_audio_source"),
188        # the queue's crossfade setting buys nothing for a realtime source: MA's own
189        # crossfade is force-disabled for one, so a second of audio here would only
190        # be a second of extra startup delay
191        pytest.param(True, True, None, MediaType.TRACK, 1, id="realtime_crossfade"),
192        pytest.param(
193            True,
194            False,
195            VolumeNormalizationMode.DYNAMIC,
196            MediaType.TRACK,
197            2,
198            id="realtime_dynamic_normalization",
199        ),
200        pytest.param(False, True, None, MediaType.TRACK, 8, id="non_realtime_crossfade"),
201        pytest.param(
202            False,
203            False,
204            VolumeNormalizationMode.DYNAMIC,
205            MediaType.RADIO,
206            3,
207            id="non_realtime_dynamic_radio",
208        ),
209        pytest.param(
210            False,
211            False,
212            VolumeNormalizationMode.DYNAMIC,
213            MediaType.TRACK,
214            5,
215            id="non_realtime_dynamic_track",
216        ),
217        pytest.param(False, False, None, MediaType.TRACK, 2, id="non_realtime_default"),
218    ],
219)
220async def test_ready_threshold_ladder(
221    is_realtime: bool,
222    crossfade_enabled: bool,
223    normalization_mode: VolumeNormalizationMode | None,
224    media_type: MediaType,
225    expected_threshold: int,
226) -> None:
227    """The buffered-ready threshold follows the realtime ladder, leaving the old one intact."""
228    # a realtime source is only ever raised above the floor by dynamic normalization,
229    # which genuinely needs its lookahead
230    queue = SimpleNamespace(crossfade_enabled=crossfade_enabled)
231    mass, scheduled_tasks, _seek_positions = _make_mass_for_get_buffer(queue=queue)
232    streamdetails = _make_stream_details(
233        media_type,
234        is_realtime=is_realtime,
235        volume_normalization_mode=normalization_mode,
236        queue_id="queue-1",
237    )
238
239    buffer = await AudioBuffer.get_buffer(mass, streamdetails, reason="test")
240
241    assert buffer._ready_threshold == expected_threshold
242    await asyncio.gather(*scheduled_tasks)
243    await buffer.clear()
244
245
246# -- AudioBuffer.get_buffer: seek handling --
247
248
249@pytest.mark.parametrize(
250    ("is_realtime", "seek_seconds", "expected_source_seek"),
251    [
252        pytest.param(True, 30, 30, id="realtime_short_seek_reaches_source"),
253        pytest.param(False, 30, 0, id="non_realtime_short_seek_buffers_from_start"),
254        pytest.param(False, 90, 90, id="non_realtime_long_seek_reaches_source"),
255    ],
256)
257async def test_get_buffer_seek_position_reaches_the_source(
258    is_realtime: bool, seek_seconds: int, expected_source_seek: int
259) -> None:
260    """A realtime source always seeks at the source; a non-realtime one only for a large seek."""
261    mass, scheduled_tasks, received_seek_positions = _make_mass_for_get_buffer()
262    streamdetails = _make_stream_details(MediaType.TRACK, is_realtime=is_realtime)
263
264    buffer = await AudioBuffer.get_buffer(
265        mass, streamdetails, seek_position_ms=seek_seconds * 1000, reason="test"
266    )
267
268    assert received_seek_positions == [expected_source_seek]
269    assert buffer._discarded_chunks == expected_source_seek
270    await asyncio.gather(*scheduled_tasks)
271    await buffer.clear()
272
273
274# -- AudioBuffer.eof --
275
276
277async def test_eof_reflects_producer_completion() -> None:
278    """The eof flag turns True only once the producer has delivered everything."""
279    buf = AudioBuffer(TEST_PCM_FORMAT)
280    assert not buf.eof
281    await buf._put(ONE_SECOND_CHUNK)
282    assert not buf.eof
283    await buf._set_eof()
284    assert buf.eof
285
286
287# -- StreamsAudio._crossfade_holdback_allowed --
288
289
290def test_holdback_rejected_when_crossfade_buffer_size_is_zero() -> None:
291    """A zero-size crossfade buffer has nothing to hold back."""
292    audio = StreamsAudio(MagicMock())
293    streamdetails = SimpleNamespace(
294        is_realtime=False, buffer=SimpleNamespace(eof=True, has_error=False, max_size_seconds=300)
295    )
296
297    assert audio._crossfade_holdback_allowed(cast("Any", streamdetails), 0) is False
298
299
300def test_holdback_rejected_for_realtime_source() -> None:
301    """A realtime source's tail is never held back, even once its buffer reaches EOF."""
302    audio = StreamsAudio(MagicMock())
303    streamdetails = SimpleNamespace(
304        is_realtime=True, buffer=SimpleNamespace(eof=True, has_error=False, max_size_seconds=300)
305    )
306
307    assert audio._crossfade_holdback_allowed(cast("Any", streamdetails), 10) is False
308
309
310def test_holdback_rejected_without_a_buffer() -> None:
311    """A source with no buffer yet has nothing to hold back."""
312    audio = StreamsAudio(MagicMock())
313    streamdetails = SimpleNamespace(is_realtime=False, buffer=None)
314
315    assert audio._crossfade_holdback_allowed(cast("Any", streamdetails), 10) is False
316
317
318def test_holdback_rejected_before_source_reaches_eof() -> None:
319    """A still-filling buffer keeps limiting playback, so its tail is not held back yet."""
320    audio = StreamsAudio(MagicMock())
321    streamdetails = SimpleNamespace(
322        is_realtime=False, buffer=SimpleNamespace(eof=False, has_error=False, max_size_seconds=300)
323    )
324
325    assert audio._crossfade_holdback_allowed(cast("Any", streamdetails), 10) is False
326
327
328def test_holdback_allowed_once_source_reaches_eof() -> None:
329    """A fully delivered, non-realtime source may hold back its tail for a crossfade."""
330    audio = StreamsAudio(MagicMock())
331    streamdetails = SimpleNamespace(
332        is_realtime=False, buffer=SimpleNamespace(eof=True, has_error=False, max_size_seconds=300)
333    )
334
335    assert audio._crossfade_holdback_allowed(cast("Any", streamdetails), 10) is True
336
337
338def test_holdback_rejected_for_a_failed_source() -> None:
339    """A source that failed is skipped without a fade, so its tail is played out instead."""
340    audio = StreamsAudio(MagicMock())
341    streamdetails = SimpleNamespace(
342        is_realtime=False,
343        buffer=SimpleNamespace(eof=True, has_error=True, max_size_seconds=300),
344    )
345
346    assert audio._crossfade_holdback_allowed(cast("Any", streamdetails), 10) is False
347
348
349def test_holdback_allowed_when_the_buffer_can_never_hold_the_tail() -> None:
350    """A buffer smaller than the tail collects it while the source runs, or loses the fade."""
351    audio = StreamsAudio(MagicMock())
352    streamdetails = SimpleNamespace(
353        is_realtime=False, buffer=SimpleNamespace(eof=False, has_error=False, max_size_seconds=15)
354    )
355
356    assert audio._crossfade_holdback_allowed(cast("Any", streamdetails), 45) is True
357    assert audio._crossfade_holdback_allowed(cast("Any", streamdetails), 10) is False
358
359
360def test_holdback_capacity_accounts_for_playback_speed() -> None:
361    """Buffer capacity is source time, so faster playback leaves fewer seconds to fade with."""
362    audio = StreamsAudio(MagicMock())
363    streamdetails = SimpleNamespace(
364        is_realtime=False, buffer=SimpleNamespace(eof=False, has_error=False, max_size_seconds=60)
365    )
366
367    assert audio._crossfade_holdback_allowed(cast("Any", streamdetails), 45) is False
368    assert audio._crossfade_holdback_allowed(cast("Any", streamdetails), 45, 2.0) is True
369
370
371# -- StreamsAudio._select_buffered_crossfade --
372
373
374def test_realtime_incoming_source_disables_crossfade_even_when_ready() -> None:
375    """A realtime incoming source never gets a fade-in, no matter how ready its buffer is."""
376    audio = StreamsAudio(MagicMock())
377
378    mode, duration = audio._select_buffered_crossfade(
379        _streamdetails_for_crossfade(
380            _buffer(SMART_CROSSFADE_DURATION, ready=True), is_realtime=True
381        ),
382        CrossfadeMode.SMART_CROSSFADE,
383        standard_crossfade_duration=8,
384    )
385
386    assert mode == CrossfadeMode.DISABLED
387    assert duration == 0
388
389
390# -- Path level: get_queue_item_stream_with_smartfade --
391
392
393async def test_smartfade_realtime_current_item_yields_all_audio_without_crossfade(
394    monkeypatch: pytest.MonkeyPatch,
395) -> None:
396    """A realtime current item streams straight through; nothing is held back for a fade."""
397    pcm_format = AudioFormat(
398        content_type=ContentType.PCM_S16LE,
399        sample_rate=8000,
400        bit_depth=16,
401        channels=2,
402    )
403    # the source is done delivering, so only the realtime flag can deny the holdback
404    current_details = SimpleNamespace(
405        duration=16,
406        seek_position=0,
407        seconds_streamed=0,
408        uri="test://current",
409        buffer=SimpleNamespace(eof=True, cancelled=False, has_error=False, max_size_seconds=300),
410        is_realtime=True,
411    )
412    # a fully resident, ready incoming buffer - the realtime flag on the outgoing
413    # track must be what blocks the fade, not an unprepared next item
414    next_details = SimpleNamespace(
415        audio_format=pcm_format,
416        buffer=_buffer(SMART_CROSSFADE_DURATION, ready=True),
417        duration=16,
418        seek_position=0,
419        uri="test://next",
420        is_realtime=False,
421    )
422    current_item = SimpleNamespace(
423        queue_id="queue-1",
424        queue_item_id="current",
425        name="Current",
426        streamdetails=current_details,
427        extra_attributes={},
428    )
429    next_item = SimpleNamespace(
430        queue_id="queue-1",
431        queue_item_id="next",
432        name="Next",
433        streamdetails=next_details,
434        extra_attributes={},
435        available=True,
436    )
437    queue = SimpleNamespace(
438        queue_id="queue-1",
439        display_name="Queue",
440        index_in_buffer=0,
441    )
442    player = SimpleNamespace(player_id="player-1", name="Player")
443    mass = MagicMock()
444    mass.player_queues.get.return_value = queue
445    mass.player_queues.load_next_queue_item = AsyncMock(return_value=next_item)
446    mass.player_queues.index_by_id.return_value = 1
447    audio = StreamsAudio(cast("Any", mass))
448    audio.setup()
449    build = AsyncMock()
450    monkeypatch.setattr(audio.smart_fades_mixer, "build", build)
451
452    async def _current_stream(
453        queue_item: object,
454        *_args: object,
455        **_kwargs: object,
456    ) -> AsyncGenerator[bytes]:
457        if queue_item is not current_item:
458            pytest.fail("The incoming source was opened while the current item was realtime")
459        yield bytes(pcm_format.pcm_sample_size * 8)
460        yield bytes(pcm_format.pcm_sample_size * 8)
461
462    monkeypatch.setattr(audio, "get_queue_item_stream", _current_stream)
463    stream = audio.get_queue_item_stream_with_smartfade(
464        cast("Any", player),
465        cast("Any", current_item),
466        pcm_format,
467        crossfade_mode=CrossfadeMode.STANDARD_CROSSFADE,
468        standard_crossfade_duration=8,
469    )
470
471    output = b"".join([chunk async for chunk in stream])
472
473    assert len(output) == pcm_format.pcm_sample_size * 16
474    build.assert_not_awaited()
475    assert "queue-1" not in audio._crossfade_data
476
477
478async def test_smartfade_still_filling_buffer_yields_all_audio_without_crossfade(
479    monkeypatch: pytest.MonkeyPatch,
480) -> None:
481    """A non-realtime source that has not yet reached EOF is also never held back for a fade."""
482    pcm_format = AudioFormat(
483        content_type=ContentType.PCM_S16LE,
484        sample_rate=8000,
485        bit_depth=16,
486        channels=2,
487    )
488    current_details = SimpleNamespace(
489        duration=16,
490        seek_position=0,
491        seconds_streamed=0,
492        uri="test://current",
493        buffer=SimpleNamespace(eof=False, cancelled=False, has_error=False, max_size_seconds=300),
494        is_realtime=False,
495    )
496    next_details = SimpleNamespace(
497        audio_format=pcm_format,
498        buffer=_buffer(SMART_CROSSFADE_DURATION, ready=True),
499        duration=16,
500        seek_position=0,
501        uri="test://next",
502        is_realtime=False,
503    )
504    current_item = SimpleNamespace(
505        queue_id="queue-1",
506        queue_item_id="current",
507        name="Current",
508        streamdetails=current_details,
509        extra_attributes={},
510    )
511    next_item = SimpleNamespace(
512        queue_id="queue-1",
513        queue_item_id="next",
514        name="Next",
515        streamdetails=next_details,
516        extra_attributes={},
517        available=True,
518    )
519    queue = SimpleNamespace(
520        queue_id="queue-1",
521        display_name="Queue",
522        index_in_buffer=0,
523    )
524    player = SimpleNamespace(player_id="player-1", name="Player")
525    mass = MagicMock()
526    mass.player_queues.get.return_value = queue
527    mass.player_queues.load_next_queue_item = AsyncMock(return_value=next_item)
528    mass.player_queues.index_by_id.return_value = 1
529    audio = StreamsAudio(cast("Any", mass))
530    audio.setup()
531    build = AsyncMock()
532    monkeypatch.setattr(audio.smart_fades_mixer, "build", build)
533
534    async def _current_stream(
535        queue_item: object,
536        *_args: object,
537        **_kwargs: object,
538    ) -> AsyncGenerator[bytes]:
539        if queue_item is not current_item:
540            pytest.fail("The incoming source was opened while the current buffer was still open")
541        yield bytes(pcm_format.pcm_sample_size * 8)
542        yield bytes(pcm_format.pcm_sample_size * 8)
543
544    monkeypatch.setattr(audio, "get_queue_item_stream", _current_stream)
545    stream = audio.get_queue_item_stream_with_smartfade(
546        cast("Any", player),
547        cast("Any", current_item),
548        pcm_format,
549        crossfade_mode=CrossfadeMode.STANDARD_CROSSFADE,
550        standard_crossfade_duration=8,
551    )
552
553    output = b"".join([chunk async for chunk in stream])
554
555    assert len(output) == pcm_format.pcm_sample_size * 16
556    build.assert_not_awaited()
557    assert "queue-1" not in audio._crossfade_data
558
559
560# -- Path level: get_queue_flow_stream --
561
562
563async def test_flow_realtime_item_yields_all_audio_as_plain_concatenation(
564    monkeypatch: pytest.MonkeyPatch,
565) -> None:
566    """A realtime item's flow audio is passed straight through and simply concatenated."""
567    pcm_format = AudioFormat(
568        content_type=ContentType.PCM_S16LE,
569        sample_rate=8000,
570        bit_depth=16,
571        channels=2,
572    )
573    # the source is done delivering, so only the realtime flag can deny the holdback
574    realtime_details = SimpleNamespace(
575        audio_format=pcm_format,
576        buffer=SimpleNamespace(eof=True, cancelled=False, has_error=False, max_size_seconds=300),
577        fade_in=False,
578        stream_error=False,
579        uri="test://realtime",
580        seek_position=0,
581        seconds_streamed=0,
582        duration=20,
583        is_realtime=True,
584    )
585    realtime_item = SimpleNamespace(
586        queue_id="queue-1",
587        queue_item_id="item-1",
588        name="Realtime",
589        media_type=MediaType.TRACK,
590        media_item=None,
591        streamdetails=realtime_details,
592        duration=20,
593        extra_attributes={},
594    )
595    next_details = SimpleNamespace(
596        audio_format=pcm_format,
597        buffer=None,
598        fade_in=False,
599        stream_error=False,
600        uri="test://next",
601        seek_position=0,
602        seconds_streamed=0,
603        duration=20,
604        is_realtime=False,
605    )
606    next_item = SimpleNamespace(
607        queue_id="queue-1",
608        queue_item_id="item-2",
609        name="Next",
610        media_type=MediaType.TRACK,
611        media_item=None,
612        streamdetails=next_details,
613        duration=20,
614        extra_attributes={},
615    )
616    queue = SimpleNamespace(
617        queue_id="queue-1",
618        display_name="Queue",
619        flow_mode=False,
620        overlay_enabled=False,
621        overlay_source=None,
622    )
623    queue_data = SimpleNamespace(session_id="session-1", flow_mode_stream_log=[])
624    mass = MagicMock()
625    mass.player_queues.queue_data.return_value = queue_data
626    mass.player_queues.load_next_queue_item = AsyncMock(side_effect=[next_item, QueueEmpty])
627    mass.player_queues.get.return_value = queue
628    mass.streams.get_crossfade_mode.return_value = CrossfadeMode.STANDARD_CROSSFADE
629    mass.config.get_raw_core_config_value.return_value = 8
630    mass.streams.audio_processing.update_item_context = MagicMock()
631    mass.player_queues.queue_buffer_completed = MagicMock()
632    player = MagicMock()
633    player.config.get_value.return_value = "fixed_48000"
634    player.get_supported_sample_rates.return_value = []
635    mass.players.get_player.return_value = player
636    audio = StreamsAudio(cast("Any", mass))
637    audio.setup()
638    build = AsyncMock()
639    monkeypatch.setattr(audio.smart_fades_mixer, "build", build)
640
641    realtime_chunks = [
642        bytes(pcm_format.pcm_sample_size * 8),
643        bytes(pcm_format.pcm_sample_size * 8),
644    ]
645    next_chunks = [bytes(pcm_format.pcm_sample_size * 2)]
646
647    async def _item_stream(
648        queue_item: SimpleNamespace, *_args: object, **_kwargs: object
649    ) -> AsyncGenerator[bytes]:
650        chunks = realtime_chunks if queue_item is realtime_item else next_chunks
651        for chunk in chunks:
652            yield chunk
653
654    monkeypatch.setattr(audio, "get_queue_item_stream", _item_stream)
655    select_crossfade = MagicMock(wraps=audio._select_buffered_crossfade)
656    monkeypatch.setattr(audio, "_select_buffered_crossfade", select_crossfade)
657    stream = audio.get_queue_flow_stream(
658        cast("Any", queue), cast("Any", realtime_item), pcm_format, session_id="session-1"
659    )
660
661    output = b"".join([chunk async for chunk in stream])
662
663    assert output == b"".join(realtime_chunks) + b"".join(next_chunks)
664    build.assert_not_awaited()
665    # no tail was held back, so the next item is never asked to fade into anything
666    select_crossfade.assert_not_called()
667    mass.player_queues.queue_buffer_completed.assert_called_once()
668
669
670async def test_smartfade_unaligned_chunks_still_crossfade(
671    monkeypatch: pytest.MonkeyPatch,
672) -> None:
673    """A source whose chunks are not whole seconds still collects a complete fade tail."""
674    pcm_format = AudioFormat(
675        content_type=ContentType.PCM_S16LE,
676        sample_rate=8000,
677        bit_depth=16,
678        channels=2,
679    )
680    current_details = SimpleNamespace(
681        duration=30,
682        seek_position=0,
683        seconds_streamed=0,
684        uri="test://current",
685        buffer=SimpleNamespace(eof=True, cancelled=False, has_error=False, max_size_seconds=300),
686        is_realtime=False,
687    )
688    next_details = SimpleNamespace(
689        audio_format=pcm_format,
690        buffer=_buffer(SMART_CROSSFADE_DURATION, ready=True),
691        duration=30,
692        seek_position=0,
693        uri="test://next",
694        is_realtime=False,
695        volume_normalization_mode=None,
696    )
697    current_item = SimpleNamespace(
698        queue_id="queue-1",
699        queue_item_id="current",
700        name="Current",
701        streamdetails=current_details,
702        extra_attributes={},
703    )
704    next_item = SimpleNamespace(
705        queue_id="queue-1",
706        queue_item_id="next",
707        name="Next",
708        streamdetails=next_details,
709        extra_attributes={},
710        available=True,
711    )
712    queue = SimpleNamespace(queue_id="queue-1", display_name="Queue", index_in_buffer=0)
713    player = SimpleNamespace(player_id="player-1", name="Player")
714    mass = MagicMock()
715    mass.player_queues.get.return_value = queue
716    mass.player_queues.load_next_queue_item = AsyncMock(return_value=next_item)
717    mass.player_queues.index_by_id.return_value = 1
718    audio = StreamsAudio(cast("Any", mass))
719    audio.setup()
720    audio.select_pcm_format = AsyncMock(return_value=pcm_format)  # type: ignore[method-assign]
721    audio.crossfade_allowed = MagicMock(return_value=True)  # type: ignore[method-assign]
722    build = AsyncMock(
723        return_value=SimpleNamespace(
724            timing_info=SimpleNamespace(
725                pre_crossfade_duration=2,
726                crossfade_duration=6,
727                fadein_trimmed_duration=0,
728            )
729        )
730    )
731    monkeypatch.setattr(audio.smart_fades_mixer, "build", build)
732    monkeypatch.setattr(audio.smart_fades_mixer, "mix", _empty_mix)
733
734    async def _current_stream(
735        queue_item: object, *_args: object, **_kwargs: object
736    ) -> AsyncGenerator[bytes]:
737        if queue_item is not current_item:
738            return
739        # a whole second, then chunks that never line up with a second boundary
740        yield bytes(pcm_format.pcm_sample_size * 8)
741        for _ in range(30):
742            yield bytes(pcm_format.pcm_sample_size // 3)
743
744    monkeypatch.setattr(audio, "get_queue_item_stream", _current_stream)
745    stream = audio.get_queue_item_stream_with_smartfade(
746        cast("Any", player),
747        cast("Any", current_item),
748        pcm_format,
749        crossfade_mode=CrossfadeMode.STANDARD_CROSSFADE,
750        standard_crossfade_duration=8,
751    )
752
753    async for _chunk in stream:
754        pass
755
756    build.assert_awaited_once()
757
758
759async def test_smartfade_short_remainder_still_crossfades(
760    monkeypatch: pytest.MonkeyPatch,
761) -> None:
762    """Less audio left than the configured overlap still fades with what is there."""
763    pcm_format = AudioFormat(
764        content_type=ContentType.PCM_S16LE,
765        sample_rate=8000,
766        bit_depth=16,
767        channels=2,
768    )
769    current_details = SimpleNamespace(
770        duration=180,
771        seek_position=146,
772        seconds_streamed=0,
773        uri="test://current",
774        buffer=SimpleNamespace(eof=True, cancelled=False, has_error=False, max_size_seconds=300),
775        is_realtime=False,
776    )
777    next_details = SimpleNamespace(
778        audio_format=pcm_format,
779        buffer=_buffer(SMART_CROSSFADE_DURATION, ready=True),
780        duration=180,
781        seek_position=0,
782        uri="test://next",
783        is_realtime=False,
784        volume_normalization_mode=None,
785    )
786    current_item = SimpleNamespace(
787        queue_id="queue-1",
788        queue_item_id="current",
789        name="Current",
790        streamdetails=current_details,
791        extra_attributes={},
792    )
793    next_item = SimpleNamespace(
794        queue_id="queue-1",
795        queue_item_id="next",
796        name="Next",
797        streamdetails=next_details,
798        extra_attributes={},
799        available=True,
800    )
801    queue = SimpleNamespace(queue_id="queue-1", display_name="Queue", index_in_buffer=0)
802    player = SimpleNamespace(player_id="player-1", name="Player")
803    mass = MagicMock()
804    mass.player_queues.get.return_value = queue
805    mass.player_queues.load_next_queue_item = AsyncMock(return_value=next_item)
806    mass.player_queues.index_by_id.return_value = 1
807    audio = StreamsAudio(cast("Any", mass))
808    audio.setup()
809    audio.select_pcm_format = AsyncMock(return_value=pcm_format)  # type: ignore[method-assign]
810    audio.crossfade_allowed = MagicMock(return_value=True)  # type: ignore[method-assign]
811    build = AsyncMock(
812        return_value=SimpleNamespace(
813            timing_info=SimpleNamespace(
814                pre_crossfade_duration=2,
815                crossfade_duration=6,
816                fadein_trimmed_duration=0,
817            )
818        )
819    )
820    monkeypatch.setattr(audio.smart_fades_mixer, "build", build)
821    monkeypatch.setattr(audio.smart_fades_mixer, "mix", _empty_mix)
822
823    async def _current_stream(
824        queue_item: object, *_args: object, **_kwargs: object
825    ) -> AsyncGenerator[bytes]:
826        if queue_item is not current_item:
827            return
828        # a seek near the end leaves 34s, less than the 45s smart overlap
829        yield bytes(pcm_format.pcm_sample_size * 8)
830        yield bytes(pcm_format.pcm_sample_size * 26)
831
832    monkeypatch.setattr(audio, "get_queue_item_stream", _current_stream)
833    stream = audio.get_queue_item_stream_with_smartfade(
834        cast("Any", player),
835        cast("Any", current_item),
836        pcm_format,
837        crossfade_mode=CrossfadeMode.SMART_CROSSFADE,
838        standard_crossfade_duration=8,
839    )
840
841    async for _chunk in stream:
842        pass
843
844    build.assert_awaited_once()
845    assert build.await_args is not None
846    fade_out_seconds = len(build.await_args.kwargs["fade_out_data"]) / pcm_format.pcm_sample_size
847    assert fade_out_seconds == pytest.approx(26, abs=1)
848
849
850async def test_smartfade_stub_remainder_does_not_crossfade(
851    monkeypatch: pytest.MonkeyPatch,
852) -> None:
853    """A remainder too short to overlap with is played out instead of faded."""
854    pcm_format = AudioFormat(
855        content_type=ContentType.PCM_S16LE,
856        sample_rate=8000,
857        bit_depth=16,
858        channels=2,
859    )
860    current_details = SimpleNamespace(
861        duration=180,
862        seek_position=176,
863        seconds_streamed=0,
864        uri="test://current",
865        buffer=SimpleNamespace(eof=True, cancelled=False, has_error=False, max_size_seconds=300),
866        is_realtime=False,
867    )
868    next_details = SimpleNamespace(
869        audio_format=pcm_format,
870        buffer=_buffer(SMART_CROSSFADE_DURATION, ready=True),
871        duration=180,
872        seek_position=0,
873        uri="test://next",
874        is_realtime=False,
875        volume_normalization_mode=None,
876    )
877    current_item = SimpleNamespace(
878        queue_id="queue-1",
879        queue_item_id="current",
880        name="Current",
881        streamdetails=current_details,
882        extra_attributes={},
883    )
884    next_item = SimpleNamespace(
885        queue_id="queue-1",
886        queue_item_id="next",
887        name="Next",
888        streamdetails=next_details,
889        extra_attributes={},
890        available=True,
891    )
892    queue = SimpleNamespace(queue_id="queue-1", display_name="Queue", index_in_buffer=0)
893    player = SimpleNamespace(player_id="player-1", name="Player")
894    mass = MagicMock()
895    mass.player_queues.get.return_value = queue
896    mass.player_queues.load_next_queue_item = AsyncMock(return_value=next_item)
897    mass.player_queues.index_by_id.return_value = 1
898    audio = StreamsAudio(cast("Any", mass))
899    audio.setup()
900    audio.select_pcm_format = AsyncMock(return_value=pcm_format)  # type: ignore[method-assign]
901    audio.crossfade_allowed = MagicMock(return_value=True)  # type: ignore[method-assign]
902    build = AsyncMock()
903    monkeypatch.setattr(audio.smart_fades_mixer, "build", build)
904
905    async def _current_stream(
906        queue_item: object, *_args: object, **_kwargs: object
907    ) -> AsyncGenerator[bytes]:
908        if queue_item is not current_item:
909            return
910        yield bytes(pcm_format.pcm_sample_size * 8)
911        yield bytes(pcm_format.pcm_sample_size * 2)
912
913    monkeypatch.setattr(audio, "get_queue_item_stream", _current_stream)
914    stream = audio.get_queue_item_stream_with_smartfade(
915        cast("Any", player),
916        cast("Any", current_item),
917        pcm_format,
918        crossfade_mode=CrossfadeMode.SMART_CROSSFADE,
919        standard_crossfade_duration=8,
920    )
921
922    output = b"".join([chunk async for chunk in stream])
923
924    assert len(output) == pcm_format.pcm_sample_size * 10
925    build.assert_not_awaited()
926
927
928@pytest.mark.parametrize("source_crossfade_mode", [CrossfadeMode.DISABLED, CrossfadeMode.SOURCE])
929async def test_flow_reports_the_crossfade_a_realtime_item_really_gets(
930    monkeypatch: pytest.MonkeyPatch,
931    source_crossfade_mode: CrossfadeMode,
932) -> None:
933    """
934    A realtime item is credited with its source's fade, never one of ours.
935
936    Music Assistant has no audio to spare for an overlap on either side of such an
937    item, so the only fade it can report is the one the source applies itself.
938    """
939    pcm_format = AudioFormat(
940        content_type=ContentType.PCM_S16LE,
941        sample_rate=8000,
942        bit_depth=16,
943        channels=2,
944    )
945    realtime_details = SimpleNamespace(
946        audio_format=pcm_format,
947        buffer=SimpleNamespace(eof=True, cancelled=False, has_error=False, max_size_seconds=300),
948        fade_in=False,
949        stream_error=False,
950        uri="test://realtime",
951        seek_position=0,
952        seconds_streamed=0,
953        duration=20,
954        is_realtime=True,
955    )
956    realtime_item = SimpleNamespace(
957        queue_id="queue-1",
958        queue_item_id="item-1",
959        name="Realtime",
960        media_type=MediaType.TRACK,
961        media_item=None,
962        streamdetails=realtime_details,
963        duration=20,
964        extra_attributes={},
965    )
966    queue = SimpleNamespace(
967        queue_id="queue-1",
968        display_name="Queue",
969        flow_mode=False,
970        overlay_enabled=False,
971        overlay_source=None,
972    )
973    mass = MagicMock()
974    mass.player_queues.queue_data.return_value = SimpleNamespace(
975        session_id="session-1", flow_mode_stream_log=[]
976    )
977    mass.player_queues.load_next_queue_item = AsyncMock(side_effect=QueueEmpty)
978    mass.player_queues.get.return_value = queue
979    mass.streams.get_crossfade_mode.return_value = CrossfadeMode.SMART_CROSSFADE
980    mass.streams.get_source_crossfade_mode.return_value = source_crossfade_mode
981    mass.config.get_raw_core_config_value.return_value = 8
982    update_item_context = MagicMock()
983    mass.streams.audio_processing.update_item_context = update_item_context
984    player = MagicMock()
985    player.config.get_value.return_value = "fixed_48000"
986    player.get_supported_sample_rates.return_value = []
987    mass.players.get_player.return_value = player
988    audio = StreamsAudio(cast("Any", mass))
989    audio.setup()
990
991    async def _item_stream(*_args: object, **_kwargs: object) -> AsyncGenerator[bytes]:
992        yield bytes(pcm_format.pcm_sample_size * 4)
993
994    monkeypatch.setattr(audio, "get_queue_item_stream", _item_stream)
995    stream = audio.get_queue_flow_stream(
996        cast("Any", queue), cast("Any", realtime_item), pcm_format, session_id="session-1"
997    )
998
999    async for _chunk in stream:
1000        pass
1001
1002    update_item_context.assert_called()
1003    reported = update_item_context.call_args.kwargs["queue_processing"]
1004    assert reported.crossfade_mode == source_crossfade_mode
1005
1006
1007async def test_flow_standard_fade_only_holds_back_its_overlap(
1008    monkeypatch: pytest.MonkeyPatch,
1009) -> None:
1010    """A standard transition waits for its overlap, not for the whole requested window."""
1011    pcm_format = AudioFormat(
1012        content_type=ContentType.PCM_S16LE,
1013        sample_rate=8000,
1014        bit_depth=16,
1015        channels=2,
1016    )
1017    first_details = SimpleNamespace(
1018        audio_format=pcm_format,
1019        buffer=SimpleNamespace(eof=True, cancelled=False, has_error=False, max_size_seconds=300),
1020        fade_in=False,
1021        stream_error=False,
1022        uri="test://first",
1023        seek_position=0,
1024        seconds_streamed=0,
1025        duration=300,
1026        is_realtime=False,
1027    )
1028    second_details = SimpleNamespace(
1029        audio_format=pcm_format,
1030        buffer=_buffer(SMART_CROSSFADE_DURATION, ready=True),
1031        fade_in=False,
1032        stream_error=False,
1033        uri="test://second",
1034        seek_position=0,
1035        seconds_streamed=0,
1036        duration=300,
1037        is_realtime=False,
1038        volume_normalization_mode=None,
1039    )
1040    first_item = SimpleNamespace(
1041        queue_id="queue-1",
1042        queue_item_id="item-1",
1043        name="First",
1044        media_type=MediaType.TRACK,
1045        media_item=None,
1046        streamdetails=first_details,
1047        duration=300,
1048        extra_attributes={},
1049    )
1050    second_item = SimpleNamespace(
1051        queue_id="queue-1",
1052        queue_item_id="item-2",
1053        name="Second",
1054        media_type=MediaType.TRACK,
1055        media_item=None,
1056        streamdetails=second_details,
1057        duration=300,
1058        extra_attributes={},
1059    )
1060    queue = SimpleNamespace(
1061        queue_id="queue-1",
1062        display_name="Queue",
1063        flow_mode=False,
1064        overlay_enabled=False,
1065        overlay_source=None,
1066    )
1067    mass = MagicMock()
1068    mass.player_queues.queue_data.return_value = SimpleNamespace(
1069        session_id="session-1", flow_mode_stream_log=[]
1070    )
1071    mass.player_queues.load_next_queue_item = AsyncMock(side_effect=[second_item, QueueEmpty])
1072    mass.player_queues.get.return_value = queue
1073    mass.streams.get_crossfade_mode.return_value = CrossfadeMode.SMART_CROSSFADE
1074    mass.config.get_raw_core_config_value.return_value = 8
1075    player = MagicMock()
1076    player.config.get_value.return_value = "fixed_48000"
1077    player.get_supported_sample_rates.return_value = []
1078    mass.players.get_player.return_value = player
1079    audio = StreamsAudio(cast("Any", mass))
1080    audio.setup()
1081    audio.crossfade_allowed = MagicMock(return_value=True)  # type: ignore[method-assign]
1082    # the incoming analysis is not ready, so the mixer degrades to a standard fade
1083    standard = StandardCrossFade(logger=MagicMock(), crossfade_duration=8)
1084    standard.build(
1085        pcm_format.pcm_sample_size * SMART_CROSSFADE_DURATION,
1086        pcm_format.pcm_sample_size * SMART_CROSSFADE_DURATION,
1087        pcm_format,
1088    )
1089    monkeypatch.setattr(audio.smart_fades_mixer, "build", AsyncMock(return_value=standard))
1090    monkeypatch.setattr(audio.smart_fades_mixer, "mix", _empty_mix)
1091
1092    consumed: dict[str, int] = {"second": 0}
1093
1094    async def _item_stream(
1095        queue_item: SimpleNamespace, *_args: object, **_kwargs: object
1096    ) -> AsyncGenerator[bytes]:
1097        if queue_item is first_item:
1098            for _ in range(60):
1099                yield bytes(pcm_format.pcm_sample_size)
1100            return
1101        for _ in range(SMART_CROSSFADE_DURATION + 20):
1102            consumed["second"] += 1
1103            yield bytes(pcm_format.pcm_sample_size)
1104
1105    monkeypatch.setattr(audio, "get_queue_item_stream", _item_stream)
1106    stream = audio.get_queue_flow_stream(
1107        cast("Any", queue), cast("Any", first_item), pcm_format, session_id="session-1"
1108    )
1109
1110    seconds_before_transition: int | None = None
1111    async for _chunk in stream:
1112        if seconds_before_transition is None and consumed["second"]:
1113            seconds_before_transition = consumed["second"]
1114
1115    # the overlap is 8s, so the transition must not wait for the full 45s window
1116    assert seconds_before_transition is not None
1117    assert seconds_before_transition <= SMART_CROSSFADE_DURATION / 2
1118
1119
1120# -- StreamsController.serve_queue_item_stream steering --
1121
1122
1123class _PcmFormatRequested(Exception):
1124    """Raised to stop the handler once it has decided on crossfading."""
1125
1126
1127def _single_item_handler(*, is_realtime: bool) -> tuple[Any, MagicMock, dict[str, Any]]:
1128    """Return a single-item stream handler that stops once the PCM format is picked."""
1129    streamdetails = _make_stream_details(MediaType.TRACK, is_realtime=is_realtime)
1130    queue_item = SimpleNamespace(
1131        queue_id="queue-1",
1132        queue_item_id="item-1",
1133        name="Track",
1134        duration=180,
1135        streamdetails=streamdetails,
1136        media_item=None,
1137        media_type=MediaType.TRACK,
1138        extra_attributes={},
1139        image=None,
1140    )
1141    queue = SimpleNamespace(
1142        queue_id="queue-1",
1143        display_name="Queue",
1144        current_item=queue_item,
1145        crossfade_enabled=True,
1146        overlay_enabled=False,
1147        overlay_source=None,
1148    )
1149    mass = MagicMock()
1150    mass.player_queues.get.return_value = queue
1151    mass.player_queues.queue_data.return_value = SimpleNamespace(session_id="session-1")
1152    mass.player_queues.get_item.return_value = queue_item
1153    mass.config.get_raw_core_config_value.return_value = 8
1154    player = MagicMock(player_id="player-1", protocol_parent_id=None)
1155    player.state.supported_features = {PlayerFeature.GAPLESS_PLAYBACK}
1156    player.state.name = "Player"
1157    mass.players.get_player.return_value = player
1158
1159    seen: dict[str, Any] = {}
1160
1161    async def _select_pcm_format(**kwargs: Any) -> None:
1162        seen["crossfade_enabled"] = kwargs["crossfade_enabled"]
1163        raise _PcmFormatRequested
1164
1165    audio = MagicMock()
1166    audio.select_pcm_format = _select_pcm_format
1167    controller = cast("Any", object.__new__(StreamsController))
1168    controller.mass = mass
1169    controller.audio = audio
1170    controller.logger = MagicMock()
1171    controller._log_request = MagicMock()
1172    controller.get_crossfade_mode = MagicMock(return_value=CrossfadeMode.SMART_CROSSFADE)
1173    request = MagicMock()
1174    request.method = "GET"
1175    request.match_info = {
1176        "queue_id": "queue-1",
1177        "player_id": "player-1",
1178        "session_id": "session-1",
1179        "queue_item_id": "item-1",
1180    }
1181    return controller, request, seen
1182
1183
1184async def test_single_item_handler_skips_crossfade_for_a_realtime_item() -> None:
1185    """A realtime item is never steered into the crossfaded single-item stream."""
1186    controller, request, seen = _single_item_handler(is_realtime=True)
1187
1188    with pytest.raises(_PcmFormatRequested):
1189        await controller.serve_queue_item_stream(request)
1190
1191    assert seen["crossfade_enabled"] is False
1192    controller.get_crossfade_mode.assert_not_called()
1193
1194
1195async def test_single_item_handler_keeps_crossfade_for_a_buffered_item() -> None:
1196    """A buffered item still gets the queue's configured crossfade."""
1197    controller, request, seen = _single_item_handler(is_realtime=False)
1198
1199    with pytest.raises(_PcmFormatRequested):
1200        await controller.serve_queue_item_stream(request)
1201
1202    assert seen["crossfade_enabled"] is True
1203    controller.get_crossfade_mode.assert_called_once()
1204
1205
1206# -- StreamsController.get_source_crossfade_mode --
1207
1208
1209class _CrossfadingProvider(MusicProvider):
1210    """A music provider whose running source crossfades this item's boundary."""
1211
1212    def delivers_crossfaded_audio(self, streamdetails: StreamDetails) -> bool | None:
1213        """Declare the source fade."""
1214        return True
1215
1216
1217class _NonCrossfadingServingProvider(MusicProvider):
1218    """A music provider serving this item, whose session fades neither of its boundaries."""
1219
1220    def delivers_crossfaded_audio(self, streamdetails: StreamDetails) -> bool | None:
1221        """Deny the source fade for this item."""
1222        return False
1223
1224
1225class _UndecidedCrossfadingProvider(MusicProvider):
1226    """A music provider that can crossfade its own playback but serves nothing yet."""
1227
1228    def delivers_crossfaded_audio(self, streamdetails: StreamDetails) -> bool | None:
1229        """Leave the answer to the queue's own setting."""
1230        return None
1231
1232
1233@pytest.mark.parametrize(
1234    ("queue_crossfade_mode", "provider", "expected"),
1235    [
1236        # with no session running the queue preference is all there is to go on
1237        (
1238            CrossfadeMode.DISABLED,
1239            object.__new__(_UndecidedCrossfadingProvider),
1240            CrossfadeMode.DISABLED,
1241        ),
1242        (
1243            CrossfadeMode.SMART_CROSSFADE,
1244            object.__new__(_UndecidedCrossfadingProvider),
1245            CrossfadeMode.SOURCE,
1246        ),
1247        # a running source answers for itself, whatever the setting says now
1248        (
1249            CrossfadeMode.DISABLED,
1250            object.__new__(_CrossfadingProvider),
1251            CrossfadeMode.SOURCE,
1252        ),
1253        # a provider that does not fade its own playback, and a plugin-served item
1254        (CrossfadeMode.SMART_CROSSFADE, object.__new__(MusicProvider), CrossfadeMode.DISABLED),
1255        (CrossfadeMode.SMART_CROSSFADE, None, CrossfadeMode.DISABLED),
1256    ],
1257)
1258def test_only_a_declared_source_fade_is_reported_as_such(
1259    queue_crossfade_mode: CrossfadeMode,
1260    provider: Any,
1261    expected: CrossfadeMode,
1262) -> None:
1263    """Only a provider that says it crossfades its own playback is credited with one."""
1264    controller = _source_crossfade_controller(provider, queue_crossfade_mode)
1265
1266    assert controller.get_source_crossfade_mode(MagicMock(), _realtime_track()) == expected
1267
1268
1269@pytest.mark.parametrize(
1270    ("media_type", "is_realtime"),
1271    [
1272        # our own mixer owns both of these, so the source's fade is not in play
1273        (MediaType.TRACK, False),
1274        (MediaType.AUDIOBOOK, True),
1275    ],
1276)
1277def test_no_source_fade_for_audio_we_mix_ourselves(
1278    media_type: MediaType, is_realtime: bool
1279) -> None:
1280    """A source fade is only reported for audio Music Assistant does not mix."""
1281    controller = _source_crossfade_controller(
1282        object.__new__(_CrossfadingProvider), CrossfadeMode.SMART_CROSSFADE
1283    )
1284    controller.mass.player_queues.get_next_item.return_value = _follower("same-provider")
1285    queue_item = SimpleNamespace(
1286        queue_item_id="item-1",
1287        media_type=media_type,
1288        streamdetails=_make_stream_details(media_type, is_realtime=is_realtime),
1289    )
1290
1291    assert controller.get_source_crossfade_mode(MagicMock(), queue_item) == CrossfadeMode.DISABLED
1292
1293
1294@pytest.mark.parametrize(
1295    ("neighbour_kind", "expected"),
1296    [
1297        # a boundary the source owns both sides of, as a provider item and as a
1298        # library item that carries the source in its mappings instead
1299        ("same-provider", CrossfadeMode.SOURCE),
1300        ("library-mapped", CrossfadeMode.SOURCE),
1301        # nothing next to it, so there is no boundary at all
1302        ("none", CrossfadeMode.DISABLED),
1303        # the source plays the current item out at any of these, so the cut is hard
1304        ("other-provider", CrossfadeMode.DISABLED),
1305        ("not-a-track", CrossfadeMode.DISABLED),
1306        ("unresolvable", CrossfadeMode.DISABLED),
1307    ],
1308)
1309@pytest.mark.parametrize("side", ["follower", "predecessor"])
1310def test_an_undecided_source_needs_a_boundary_it_would_own(
1311    neighbour_kind: str, side: str, expected: CrossfadeMode
1312) -> None:
1313    """
1314    A source that is not serving this queue yet is credited only where it could fade.
1315
1316    It cannot answer for the item, so the boundary stands in: either side counts,
1317    the same way one of our own fades credits both of its sides. Reporting a fade
1318    anywhere else would describe an overlap nobody rendered - we do not mix a
1319    realtime item's boundaries either, so it is a hard cut.
1320    """
1321    controller = _source_crossfade_controller(
1322        object.__new__(_UndecidedCrossfadingProvider), CrossfadeMode.SMART_CROSSFADE
1323    )
1324    queues = controller.mass.player_queues
1325    neighbour = _follower(neighbour_kind)
1326    if side == "follower":
1327        queues.get_next_item.return_value = neighbour
1328    else:
1329        queues.get_next_item.return_value = None
1330        queues.index_by_id.return_value = 1
1331        queues.get_item.return_value = neighbour
1332    queue_item = SimpleNamespace(
1333        queue_item_id="item-1",
1334        media_type=MediaType.TRACK,
1335        streamdetails=_make_stream_details(MediaType.TRACK, is_realtime=True),
1336    )
1337
1338    assert controller.get_source_crossfade_mode(MagicMock(), queue_item) == expected
1339
1340
1341@pytest.mark.parametrize(
1342    ("provider", "neighbour_kind", "expected"),
1343    [
1344        # the source says it fades this item, so its own boundary answer stands even
1345        # where the queue offers nothing to fade against
1346        (_CrossfadingProvider, "none", CrossfadeMode.SOURCE),
1347        # and it says it does not, so the neighbour it happens to own is not credited:
1348        # a list predecessor that never played is exactly this case
1349        (_NonCrossfadingServingProvider, "same-provider", CrossfadeMode.DISABLED),
1350    ],
1351)
1352def test_a_serving_source_is_not_second_guessed_by_the_queue(
1353    provider: type[MusicProvider], neighbour_kind: str, expected: CrossfadeMode
1354) -> None:
1355    """
1356    A source already serving the item answers for its boundaries, and that answer stands.
1357
1358    It knows what it fed and what it played across; the neighbour check is only a
1359    necessary condition, so applying it on top would both deny fades that happened
1360    and credit ones that did not.
1361    """
1362    controller = _source_crossfade_controller(
1363        object.__new__(provider), CrossfadeMode.SMART_CROSSFADE
1364    )
1365    controller.mass.player_queues.get_next_item.return_value = _follower(neighbour_kind)
1366
1367    assert controller.get_source_crossfade_mode(MagicMock(), _realtime_track()) == expected
1368
1369
1370def test_the_raw_pcm_path_reports_a_source_fade_too() -> None:
1371    """
1372    A source fade reaches the processing details on the raw-PCM entry point as well.
1373
1374    Gapless players are served per item straight from ``get_stream`` rather than over
1375    the http route, so leaving it out there would report nothing for most players.
1376    """
1377    queue_item = SimpleNamespace(
1378        queue_id="queue-1",
1379        queue_item_id="item-1",
1380        name="Track",
1381        media_type=MediaType.TRACK,
1382        media_item=None,
1383        streamdetails=_make_stream_details(MediaType.TRACK, is_realtime=True),
1384        extra_attributes={},
1385        image=None,
1386    )
1387    queue = SimpleNamespace(
1388        queue_id="queue-1",
1389        display_name="Queue",
1390        crossfade_enabled=True,
1391        overlay_enabled=False,
1392        overlay_source=None,
1393    )
1394    controller = cast("Any", object.__new__(StreamsController))
1395    controller.mass = MagicMock()
1396    controller.mass.player_queues.get.return_value = queue
1397    controller.mass.player_queues.get_item.return_value = queue_item
1398    controller.mass.players.get_player.return_value = None
1399    controller.logger = MagicMock()
1400    controller.audio = MagicMock()
1401    controller._update_audio_processing_context = MagicMock()
1402    controller.get_source_crossfade_mode = MagicMock(return_value=CrossfadeMode.SOURCE)
1403    media = SimpleNamespace(
1404        media_type=MediaType.TRACK,
1405        source_id="queue-1",
1406        queue_item_id="item-1",
1407        queue_session_id="session-1",
1408        uri="",
1409    )
1410
1411    controller.get_stream(cast("Any", media), TEST_PCM_FORMAT)
1412
1413    assert (
1414        controller._update_audio_processing_context.call_args.kwargs["source_crossfade_mode"]
1415        == CrossfadeMode.SOURCE
1416    )
1417
1418
1419def test_a_music_provider_declares_no_source_fade_by_default() -> None:
1420    """The declaration is opt-in: nothing downstream verifies it."""
1421    assert (
1422        object.__new__(MusicProvider).delivers_crossfaded_audio(
1423            _make_stream_details(MediaType.TRACK, is_realtime=True)
1424        )
1425        is False
1426    )
1427
1428
1429def _source_crossfade_controller(provider: Any, queue_crossfade_mode: CrossfadeMode) -> Any:
1430    """
1431    Return a controller resolving source fades against one provider and setting.
1432
1433    The queue follows on with another item of the same source, so only the provider
1434    and the setting decide; the boundary cases are covered on their own.
1435    """
1436    controller = cast("Any", object.__new__(StreamsController))
1437    controller.mass = MagicMock()
1438    controller.mass.get_provider.return_value = provider
1439    controller.mass.player_queues.get_next_item.return_value = _follower("same-provider")
1440    # first in the queue, so nothing precedes it and only the follower decides
1441    controller.mass.player_queues.index_by_id.return_value = 0
1442    controller.get_crossfade_mode = MagicMock(return_value=queue_crossfade_mode)
1443    return controller
1444
1445
1446def _follower(kind: str) -> Any:
1447    """Return the queue item that follows, in each shape the resolver has to handle."""
1448    if kind == "none":
1449        return None
1450    if kind == "not-a-track":
1451        return SimpleNamespace(media_type=MediaType.AUDIOBOOK, streamdetails=None, media_item=None)
1452    if kind == "same-provider":
1453        # already resolved to the provider that will serve it
1454        return SimpleNamespace(
1455            media_type=MediaType.TRACK,
1456            streamdetails=_make_stream_details(MediaType.TRACK, is_realtime=True),
1457            media_item=None,
1458        )
1459    if kind == "other-provider":
1460        other = _make_stream_details(MediaType.TRACK, is_realtime=True)
1461        other.provider = "other--1"
1462        return SimpleNamespace(media_type=MediaType.TRACK, streamdetails=other, media_item=None)
1463    if kind == "library-mapped":
1464        return SimpleNamespace(
1465            media_type=MediaType.TRACK,
1466            streamdetails=None,
1467            media_item=SimpleNamespace(
1468                provider="library",
1469                provider_mappings=[SimpleNamespace(provider_instance="builtin")],
1470            ),
1471        )
1472    # nothing to resolve it with at all
1473    return SimpleNamespace(media_type=MediaType.TRACK, streamdetails=None, media_item=None)
1474
1475
1476def _realtime_track() -> Any:
1477    """Return a queue item whose track arrives at playback pace."""
1478    return SimpleNamespace(
1479        queue_item_id="item-1",
1480        media_type=MediaType.TRACK,
1481        streamdetails=_make_stream_details(MediaType.TRACK, is_realtime=True),
1482    )
1483
1484
1485def test_the_context_update_carries_a_source_fade_the_audio_layer_never_sees() -> None:
1486    """
1487    A crossfade performed by the source is published with the item's context.
1488
1489    Our own mixer reports the fades it renders, but it renders none here, so this is
1490    the only place the source's fade can reach the processing details.
1491    """
1492    controller = cast("Any", object.__new__(StreamsController))
1493    controller.mass = MagicMock()
1494    controller.mass.player_queues.queue_data_or_none.return_value = SimpleNamespace(
1495        session_id="session-1"
1496    )
1497    controller.audio_processing = MagicMock()
1498    queue_item = SimpleNamespace(
1499        queue_item_id="item-1",
1500        streamdetails=_make_stream_details(MediaType.TRACK, is_realtime=True),
1501        extra_attributes={},
1502    )
1503
1504    controller._update_audio_processing_context(
1505        queue=SimpleNamespace(queue_id="queue-1"),
1506        queue_item=queue_item,
1507        pcm_format=TEST_PCM_FORMAT,
1508        overlay_enabled=False,
1509        session_id="session-1",
1510        source_crossfade_mode=CrossfadeMode.SOURCE,
1511    )
1512
1513    reported = controller.audio_processing.update_item_context.call_args.kwargs["queue_processing"]
1514    assert reported.crossfade_mode == CrossfadeMode.SOURCE
1515
1516
1517# -- StreamsAudio.get_stream_details --
1518
1519
1520@pytest.mark.parametrize(
1521    ("media_item_cls", "media_type", "expected_is_realtime"),
1522    [
1523        pytest.param(Radio, MediaType.RADIO, True, id="radio"),
1524        pytest.param(AudioSource, MediaType.AUDIO_SOURCE, True, id="audio_source"),
1525        pytest.param(Track, MediaType.TRACK, False, id="track"),
1526    ],
1527)
1528async def test_get_stream_details_sets_is_realtime_by_media_type(
1529    media_item_cls: type, media_type: MediaType, expected_is_realtime: bool
1530) -> None:
1531    """RADIO and AUDIO_SOURCE streams are marked realtime; a TRACK's flag is left alone."""
1532    provider_streamdetails = StreamDetails(
1533        provider="test--1",
1534        item_id="item-1",
1535        audio_format=AudioFormat(content_type=ContentType.MP3),
1536        media_type=media_type,
1537        stream_type=StreamType.CUSTOM,
1538        duration=180 if media_type == MediaType.TRACK else None,
1539    )
1540    audio = _stream_details_provider(provider_streamdetails)
1541
1542    streamdetails = await audio.get_stream_details(
1543        queue_item=_queue_item_with_mapping(media_item_cls)
1544    )
1545
1546    assert streamdetails.is_realtime is expected_is_realtime
1547