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
37
38# Standard test PCM format: 44100Hz, 16-bit, stereo
39TEST_PCM_FORMAT = AudioFormat(
40    content_type=ContentType.PCM_S16LE,
41    sample_rate=44100,
42    bit_depth=16,
43    channels=2,
44)
45
46# One second of silence in the test format
47ONE_SECOND_CHUNK = b"\x00" * TEST_PCM_FORMAT.pcm_sample_size
48
49
50def _make_stream_details(
51    media_type: MediaType,
52    *,
53    is_realtime: bool = False,
54    volume_normalization_mode: VolumeNormalizationMode | None = None,
55    queue_id: str | None = None,
56) -> StreamDetails:
57    """Build minimal stream details for AudioBuffer.get_buffer tests."""
58    return StreamDetails(
59        provider="builtin",
60        item_id="item-1",
61        audio_format=TEST_PCM_FORMAT,
62        media_type=media_type,
63        stream_type=StreamType.HTTP,
64        path="http://example.com/audio.mp3",
65        duration=180,
66        can_seek=True,
67        allow_seek=True,
68        queue_id=queue_id,
69        is_realtime=is_realtime,
70        volume_normalization_mode=volume_normalization_mode,
71    )
72
73
74async def _make_source(num_chunks: int) -> AsyncGenerator[bytes]:
75    """Create an async generator that yields one-second PCM chunks."""
76    for _ in range(num_chunks):
77        yield ONE_SECOND_CHUNK
78
79
80def _make_mass_for_get_buffer(
81    *, queue: Any | None = None
82) -> tuple[MagicMock, list[asyncio.Task[None]], list[float | None]]:
83    """Build a minimal mass stub for AudioBuffer.get_buffer tests."""
84    received_seek_positions: list[float | None] = []
85
86    def _get_media_stream(*_args: Any, **kwargs: Any) -> AsyncGenerator[bytes]:
87        received_seek_positions.append(kwargs.get("seek_position"))
88        return _make_source(1)
89
90    mass = MagicMock()
91    mass.config.get_raw_core_config_value.return_value = BufferSize.BALANCED.value
92    mass.player_queues.get.return_value = queue
93    mass.streams = SimpleNamespace(
94        audio_analysis=SimpleNamespace(start_analysis=AsyncMock(return_value=None)),
95        audio=SimpleNamespace(get_media_stream=_get_media_stream),
96    )
97    scheduled_tasks: list[asyncio.Task[None]] = []
98
99    def _create_task(coro: Any) -> asyncio.Task[None]:
100        task: asyncio.Task[None] = asyncio.ensure_future(coro)
101        scheduled_tasks.append(task)
102        return task
103
104    mass.create_task.side_effect = _create_task
105    return mass, scheduled_tasks, received_seek_positions
106
107
108def _streamdetails_for_crossfade(
109    audio_buffer: AudioBuffer | None, *, is_realtime: bool = False
110) -> StreamDetails:
111    """Build incoming track details with an optional prepared buffer."""
112    streamdetails = StreamDetails(
113        provider="test--1",
114        item_id="track-1",
115        audio_format=AudioFormat(content_type=ContentType.FLAC),
116        media_type=MediaType.TRACK,
117        stream_type=StreamType.HTTP,
118        path="http://test.invalid/track.flac",
119        duration=180,
120        is_realtime=is_realtime,
121    )
122    streamdetails.buffer = audio_buffer
123    return streamdetails
124
125
126async def _empty_mix(*_args: object, **_kwargs: object) -> AsyncGenerator[bytes]:
127    """Stand in for the mixer, producing no audio."""
128    no_audio: tuple[bytes, ...] = ()
129    for chunk in no_audio:
130        yield chunk
131
132
133def _buffer(duration_available: float, ready: bool) -> AudioBuffer:
134    """Build a valid buffer with the requested resident duration."""
135    audio_buffer = MagicMock(spec=AudioBuffer)
136    audio_buffer.has_error = False
137    audio_buffer.is_valid.return_value = True
138    audio_buffer.duration_available = duration_available
139    audio_buffer.ready = MagicMock()
140    audio_buffer.ready.is_set.return_value = ready
141    return audio_buffer
142
143
144def _stream_details_provider(streamdetails: StreamDetails) -> StreamsAudio:
145    """Build a StreamsAudio whose single provider hands back the given streamdetails."""
146    provider = MagicMock()
147    provider.instance_id = "test--1"
148    provider.domain = "test"
149    provider.available = True
150    provider.is_streaming_provider = True
151    provider.get_stream_details = AsyncMock(return_value=streamdetails)
152    mass = MagicMock()
153    mass.get_provider.side_effect = lambda instance, **_kwargs: (
154        provider if instance == "test--1" else None
155    )
156    mass.providers = []
157    mass.player_queues.queue_data_or_none.return_value = None
158    mass.streams.get_config_value.return_value = -17
159    return StreamsAudio(mass)
160
161
162def _queue_item_with_mapping(media_item_cls: type) -> QueueItem:
163    """Build a queue item whose media item carries one matching provider mapping."""
164    mapping = ProviderMapping(item_id="item-1", provider_domain="test", provider_instance="test--1")
165    media_item = media_item_cls(
166        item_id="item-1", provider="test--1", name="Item", provider_mappings={mapping}
167    )
168    return QueueItem(
169        queue_id="q1", queue_item_id="qi1", name="Item", duration=None, media_item=media_item
170    )
171
172
173# -- AudioBuffer.get_buffer: ready threshold ladder --
174
175
176@pytest.mark.parametrize(
177    (
178        "is_realtime",
179        "crossfade_enabled",
180        "normalization_mode",
181        "media_type",
182        "expected_threshold",
183    ),
184    [
185        pytest.param(True, False, None, MediaType.RADIO, 1, id="realtime_base"),
186        pytest.param(True, False, None, MediaType.AUDIO_SOURCE, 1, id="realtime_audio_source"),
187        pytest.param(True, True, None, MediaType.TRACK, 2, id="realtime_crossfade"),
188        pytest.param(
189            True,
190            False,
191            VolumeNormalizationMode.DYNAMIC,
192            MediaType.TRACK,
193            2,
194            id="realtime_dynamic_normalization",
195        ),
196        pytest.param(False, True, None, MediaType.TRACK, 8, id="non_realtime_crossfade"),
197        pytest.param(
198            False,
199            False,
200            VolumeNormalizationMode.DYNAMIC,
201            MediaType.RADIO,
202            3,
203            id="non_realtime_dynamic_radio",
204        ),
205        pytest.param(
206            False,
207            False,
208            VolumeNormalizationMode.DYNAMIC,
209            MediaType.TRACK,
210            5,
211            id="non_realtime_dynamic_track",
212        ),
213        pytest.param(False, False, None, MediaType.TRACK, 2, id="non_realtime_default"),
214    ],
215)
216async def test_ready_threshold_ladder(
217    is_realtime: bool,
218    crossfade_enabled: bool,
219    normalization_mode: VolumeNormalizationMode | None,
220    media_type: MediaType,
221    expected_threshold: int,
222) -> None:
223    """The buffered-ready threshold follows the realtime ladder, leaving the old one intact."""
224    queue = SimpleNamespace(crossfade_enabled=crossfade_enabled)
225    mass, scheduled_tasks, _seek_positions = _make_mass_for_get_buffer(queue=queue)
226    streamdetails = _make_stream_details(
227        media_type,
228        is_realtime=is_realtime,
229        volume_normalization_mode=normalization_mode,
230        queue_id="queue-1",
231    )
232
233    buffer = await AudioBuffer.get_buffer(mass, streamdetails, reason="test")
234
235    assert buffer._ready_threshold == expected_threshold
236    await asyncio.gather(*scheduled_tasks)
237    await buffer.clear()
238
239
240# -- AudioBuffer.get_buffer: seek handling --
241
242
243@pytest.mark.parametrize(
244    ("is_realtime", "seek_seconds", "expected_source_seek"),
245    [
246        pytest.param(True, 30, 30, id="realtime_short_seek_reaches_source"),
247        pytest.param(False, 30, 0, id="non_realtime_short_seek_buffers_from_start"),
248        pytest.param(False, 90, 90, id="non_realtime_long_seek_reaches_source"),
249    ],
250)
251async def test_get_buffer_seek_position_reaches_the_source(
252    is_realtime: bool, seek_seconds: int, expected_source_seek: int
253) -> None:
254    """A realtime source always seeks at the source; a non-realtime one only for a large seek."""
255    mass, scheduled_tasks, received_seek_positions = _make_mass_for_get_buffer()
256    streamdetails = _make_stream_details(MediaType.TRACK, is_realtime=is_realtime)
257
258    buffer = await AudioBuffer.get_buffer(
259        mass, streamdetails, seek_position_ms=seek_seconds * 1000, reason="test"
260    )
261
262    assert received_seek_positions == [expected_source_seek]
263    assert buffer._discarded_chunks == expected_source_seek
264    await asyncio.gather(*scheduled_tasks)
265    await buffer.clear()
266
267
268# -- AudioBuffer.eof --
269
270
271async def test_eof_reflects_producer_completion() -> None:
272    """The eof flag turns True only once the producer has delivered everything."""
273    buf = AudioBuffer(TEST_PCM_FORMAT)
274    assert not buf.eof
275    await buf._put(ONE_SECOND_CHUNK)
276    assert not buf.eof
277    await buf._set_eof()
278    assert buf.eof
279
280
281# -- StreamsAudio._crossfade_holdback_allowed --
282
283
284def test_holdback_rejected_when_crossfade_buffer_size_is_zero() -> None:
285    """A zero-size crossfade buffer has nothing to hold back."""
286    audio = StreamsAudio(MagicMock())
287    streamdetails = SimpleNamespace(
288        is_realtime=False, buffer=SimpleNamespace(eof=True, has_error=False, max_size_seconds=300)
289    )
290
291    assert audio._crossfade_holdback_allowed(cast("Any", streamdetails), 0) is False
292
293
294def test_holdback_rejected_for_realtime_source() -> None:
295    """A realtime source's tail is never held back, even once its buffer reaches EOF."""
296    audio = StreamsAudio(MagicMock())
297    streamdetails = SimpleNamespace(
298        is_realtime=True, buffer=SimpleNamespace(eof=True, has_error=False, max_size_seconds=300)
299    )
300
301    assert audio._crossfade_holdback_allowed(cast("Any", streamdetails), 10) is False
302
303
304def test_holdback_rejected_without_a_buffer() -> None:
305    """A source with no buffer yet has nothing to hold back."""
306    audio = StreamsAudio(MagicMock())
307    streamdetails = SimpleNamespace(is_realtime=False, buffer=None)
308
309    assert audio._crossfade_holdback_allowed(cast("Any", streamdetails), 10) is False
310
311
312def test_holdback_rejected_before_source_reaches_eof() -> None:
313    """A still-filling buffer keeps limiting playback, so its tail is not held back yet."""
314    audio = StreamsAudio(MagicMock())
315    streamdetails = SimpleNamespace(
316        is_realtime=False, buffer=SimpleNamespace(eof=False, has_error=False, max_size_seconds=300)
317    )
318
319    assert audio._crossfade_holdback_allowed(cast("Any", streamdetails), 10) is False
320
321
322def test_holdback_allowed_once_source_reaches_eof() -> None:
323    """A fully delivered, non-realtime source may hold back its tail for a crossfade."""
324    audio = StreamsAudio(MagicMock())
325    streamdetails = SimpleNamespace(
326        is_realtime=False, buffer=SimpleNamespace(eof=True, has_error=False, max_size_seconds=300)
327    )
328
329    assert audio._crossfade_holdback_allowed(cast("Any", streamdetails), 10) is True
330
331
332def test_holdback_rejected_for_a_failed_source() -> None:
333    """A source that failed is skipped without a fade, so its tail is played out instead."""
334    audio = StreamsAudio(MagicMock())
335    streamdetails = SimpleNamespace(
336        is_realtime=False,
337        buffer=SimpleNamespace(eof=True, has_error=True, max_size_seconds=300),
338    )
339
340    assert audio._crossfade_holdback_allowed(cast("Any", streamdetails), 10) is False
341
342
343def test_holdback_allowed_when_the_buffer_can_never_hold_the_tail() -> None:
344    """A buffer smaller than the tail collects it while the source runs, or loses the fade."""
345    audio = StreamsAudio(MagicMock())
346    streamdetails = SimpleNamespace(
347        is_realtime=False, buffer=SimpleNamespace(eof=False, has_error=False, max_size_seconds=15)
348    )
349
350    assert audio._crossfade_holdback_allowed(cast("Any", streamdetails), 45) is True
351    assert audio._crossfade_holdback_allowed(cast("Any", streamdetails), 10) is False
352
353
354def test_holdback_capacity_accounts_for_playback_speed() -> None:
355    """Buffer capacity is source time, so faster playback leaves fewer seconds to fade with."""
356    audio = StreamsAudio(MagicMock())
357    streamdetails = SimpleNamespace(
358        is_realtime=False, buffer=SimpleNamespace(eof=False, has_error=False, max_size_seconds=60)
359    )
360
361    assert audio._crossfade_holdback_allowed(cast("Any", streamdetails), 45) is False
362    assert audio._crossfade_holdback_allowed(cast("Any", streamdetails), 45, 2.0) is True
363
364
365# -- StreamsAudio._select_buffered_crossfade --
366
367
368def test_realtime_incoming_source_disables_crossfade_even_when_ready() -> None:
369    """A realtime incoming source never gets a fade-in, no matter how ready its buffer is."""
370    audio = StreamsAudio(MagicMock())
371
372    mode, duration = audio._select_buffered_crossfade(
373        _streamdetails_for_crossfade(
374            _buffer(SMART_CROSSFADE_DURATION, ready=True), is_realtime=True
375        ),
376        CrossfadeMode.SMART_CROSSFADE,
377        standard_crossfade_duration=8,
378    )
379
380    assert mode == CrossfadeMode.DISABLED
381    assert duration == 0
382
383
384# -- Path level: get_queue_item_stream_with_smartfade --
385
386
387async def test_smartfade_realtime_current_item_yields_all_audio_without_crossfade(
388    monkeypatch: pytest.MonkeyPatch,
389) -> None:
390    """A realtime current item streams straight through; nothing is held back for a fade."""
391    pcm_format = AudioFormat(
392        content_type=ContentType.PCM_S16LE,
393        sample_rate=8000,
394        bit_depth=16,
395        channels=2,
396    )
397    # the source is done delivering, so only the realtime flag can deny the holdback
398    current_details = SimpleNamespace(
399        duration=16,
400        seek_position=0,
401        seconds_streamed=0,
402        uri="test://current",
403        buffer=SimpleNamespace(eof=True, cancelled=False, has_error=False, max_size_seconds=300),
404        is_realtime=True,
405    )
406    # a fully resident, ready incoming buffer - the realtime flag on the outgoing
407    # track must be what blocks the fade, not an unprepared next item
408    next_details = SimpleNamespace(
409        audio_format=pcm_format,
410        buffer=_buffer(SMART_CROSSFADE_DURATION, ready=True),
411        duration=16,
412        seek_position=0,
413        uri="test://next",
414        is_realtime=False,
415    )
416    current_item = SimpleNamespace(
417        queue_id="queue-1",
418        queue_item_id="current",
419        name="Current",
420        streamdetails=current_details,
421        extra_attributes={},
422    )
423    next_item = SimpleNamespace(
424        queue_id="queue-1",
425        queue_item_id="next",
426        name="Next",
427        streamdetails=next_details,
428        extra_attributes={},
429        available=True,
430    )
431    queue = SimpleNamespace(
432        queue_id="queue-1",
433        display_name="Queue",
434        index_in_buffer=0,
435    )
436    player = SimpleNamespace(player_id="player-1", name="Player")
437    mass = MagicMock()
438    mass.player_queues.get.return_value = queue
439    mass.player_queues.load_next_queue_item = AsyncMock(return_value=next_item)
440    mass.player_queues.index_by_id.return_value = 1
441    audio = StreamsAudio(cast("Any", mass))
442    audio.setup()
443    build = AsyncMock()
444    monkeypatch.setattr(audio.smart_fades_mixer, "build", build)
445
446    async def _current_stream(
447        queue_item: object,
448        *_args: object,
449        **_kwargs: object,
450    ) -> AsyncGenerator[bytes]:
451        if queue_item is not current_item:
452            pytest.fail("The incoming source was opened while the current item was realtime")
453        yield bytes(pcm_format.pcm_sample_size * 8)
454        yield bytes(pcm_format.pcm_sample_size * 8)
455
456    monkeypatch.setattr(audio, "get_queue_item_stream", _current_stream)
457    stream = audio.get_queue_item_stream_with_smartfade(
458        cast("Any", player),
459        cast("Any", current_item),
460        pcm_format,
461        crossfade_mode=CrossfadeMode.STANDARD_CROSSFADE,
462        standard_crossfade_duration=8,
463    )
464
465    output = b"".join([chunk async for chunk in stream])
466
467    assert len(output) == pcm_format.pcm_sample_size * 16
468    build.assert_not_awaited()
469    assert "queue-1" not in audio._crossfade_data
470
471
472async def test_smartfade_still_filling_buffer_yields_all_audio_without_crossfade(
473    monkeypatch: pytest.MonkeyPatch,
474) -> None:
475    """A non-realtime source that has not yet reached EOF is also never held back for a fade."""
476    pcm_format = AudioFormat(
477        content_type=ContentType.PCM_S16LE,
478        sample_rate=8000,
479        bit_depth=16,
480        channels=2,
481    )
482    current_details = SimpleNamespace(
483        duration=16,
484        seek_position=0,
485        seconds_streamed=0,
486        uri="test://current",
487        buffer=SimpleNamespace(eof=False, cancelled=False, has_error=False, max_size_seconds=300),
488        is_realtime=False,
489    )
490    next_details = SimpleNamespace(
491        audio_format=pcm_format,
492        buffer=_buffer(SMART_CROSSFADE_DURATION, ready=True),
493        duration=16,
494        seek_position=0,
495        uri="test://next",
496        is_realtime=False,
497    )
498    current_item = SimpleNamespace(
499        queue_id="queue-1",
500        queue_item_id="current",
501        name="Current",
502        streamdetails=current_details,
503        extra_attributes={},
504    )
505    next_item = SimpleNamespace(
506        queue_id="queue-1",
507        queue_item_id="next",
508        name="Next",
509        streamdetails=next_details,
510        extra_attributes={},
511        available=True,
512    )
513    queue = SimpleNamespace(
514        queue_id="queue-1",
515        display_name="Queue",
516        index_in_buffer=0,
517    )
518    player = SimpleNamespace(player_id="player-1", name="Player")
519    mass = MagicMock()
520    mass.player_queues.get.return_value = queue
521    mass.player_queues.load_next_queue_item = AsyncMock(return_value=next_item)
522    mass.player_queues.index_by_id.return_value = 1
523    audio = StreamsAudio(cast("Any", mass))
524    audio.setup()
525    build = AsyncMock()
526    monkeypatch.setattr(audio.smart_fades_mixer, "build", build)
527
528    async def _current_stream(
529        queue_item: object,
530        *_args: object,
531        **_kwargs: object,
532    ) -> AsyncGenerator[bytes]:
533        if queue_item is not current_item:
534            pytest.fail("The incoming source was opened while the current buffer was still open")
535        yield bytes(pcm_format.pcm_sample_size * 8)
536        yield bytes(pcm_format.pcm_sample_size * 8)
537
538    monkeypatch.setattr(audio, "get_queue_item_stream", _current_stream)
539    stream = audio.get_queue_item_stream_with_smartfade(
540        cast("Any", player),
541        cast("Any", current_item),
542        pcm_format,
543        crossfade_mode=CrossfadeMode.STANDARD_CROSSFADE,
544        standard_crossfade_duration=8,
545    )
546
547    output = b"".join([chunk async for chunk in stream])
548
549    assert len(output) == pcm_format.pcm_sample_size * 16
550    build.assert_not_awaited()
551    assert "queue-1" not in audio._crossfade_data
552
553
554# -- Path level: get_queue_flow_stream --
555
556
557async def test_flow_realtime_item_yields_all_audio_as_plain_concatenation(
558    monkeypatch: pytest.MonkeyPatch,
559) -> None:
560    """A realtime item's flow audio is passed straight through and simply concatenated."""
561    pcm_format = AudioFormat(
562        content_type=ContentType.PCM_S16LE,
563        sample_rate=8000,
564        bit_depth=16,
565        channels=2,
566    )
567    # the source is done delivering, so only the realtime flag can deny the holdback
568    realtime_details = SimpleNamespace(
569        audio_format=pcm_format,
570        buffer=SimpleNamespace(eof=True, cancelled=False, has_error=False, max_size_seconds=300),
571        fade_in=False,
572        stream_error=False,
573        uri="test://realtime",
574        seek_position=0,
575        seconds_streamed=0,
576        duration=20,
577        is_realtime=True,
578    )
579    realtime_item = SimpleNamespace(
580        queue_id="queue-1",
581        queue_item_id="item-1",
582        name="Realtime",
583        media_type=MediaType.TRACK,
584        media_item=None,
585        streamdetails=realtime_details,
586        duration=20,
587        extra_attributes={},
588    )
589    next_details = SimpleNamespace(
590        audio_format=pcm_format,
591        buffer=None,
592        fade_in=False,
593        stream_error=False,
594        uri="test://next",
595        seek_position=0,
596        seconds_streamed=0,
597        duration=20,
598        is_realtime=False,
599    )
600    next_item = SimpleNamespace(
601        queue_id="queue-1",
602        queue_item_id="item-2",
603        name="Next",
604        media_type=MediaType.TRACK,
605        media_item=None,
606        streamdetails=next_details,
607        duration=20,
608        extra_attributes={},
609    )
610    queue = SimpleNamespace(
611        queue_id="queue-1",
612        display_name="Queue",
613        flow_mode=False,
614        overlay_enabled=False,
615        overlay_source=None,
616    )
617    queue_data = SimpleNamespace(session_id="session-1", flow_mode_stream_log=[])
618    mass = MagicMock()
619    mass.player_queues.queue_data.return_value = queue_data
620    mass.player_queues.load_next_queue_item = AsyncMock(side_effect=[next_item, QueueEmpty])
621    mass.player_queues.get.return_value = queue
622    mass.streams.get_crossfade_mode.return_value = CrossfadeMode.STANDARD_CROSSFADE
623    mass.config.get_raw_core_config_value.return_value = 8
624    mass.streams.audio_processing.update_item_context = MagicMock()
625    mass.player_queues.queue_buffer_completed = MagicMock()
626    player = MagicMock()
627    player.config.get_value.return_value = "fixed_48000"
628    player.get_supported_sample_rates.return_value = []
629    mass.players.get_player.return_value = player
630    audio = StreamsAudio(cast("Any", mass))
631    audio.setup()
632    build = AsyncMock()
633    monkeypatch.setattr(audio.smart_fades_mixer, "build", build)
634
635    realtime_chunks = [
636        bytes(pcm_format.pcm_sample_size * 8),
637        bytes(pcm_format.pcm_sample_size * 8),
638    ]
639    next_chunks = [bytes(pcm_format.pcm_sample_size * 2)]
640
641    async def _item_stream(
642        queue_item: SimpleNamespace, *_args: object, **_kwargs: object
643    ) -> AsyncGenerator[bytes]:
644        chunks = realtime_chunks if queue_item is realtime_item else next_chunks
645        for chunk in chunks:
646            yield chunk
647
648    monkeypatch.setattr(audio, "get_queue_item_stream", _item_stream)
649    select_crossfade = MagicMock(wraps=audio._select_buffered_crossfade)
650    monkeypatch.setattr(audio, "_select_buffered_crossfade", select_crossfade)
651    stream = audio.get_queue_flow_stream(
652        cast("Any", queue), cast("Any", realtime_item), pcm_format, session_id="session-1"
653    )
654
655    output = b"".join([chunk async for chunk in stream])
656
657    assert output == b"".join(realtime_chunks) + b"".join(next_chunks)
658    build.assert_not_awaited()
659    # no tail was held back, so the next item is never asked to fade into anything
660    select_crossfade.assert_not_called()
661    mass.player_queues.queue_buffer_completed.assert_called_once()
662
663
664async def test_smartfade_unaligned_chunks_still_crossfade(
665    monkeypatch: pytest.MonkeyPatch,
666) -> None:
667    """A source whose chunks are not whole seconds still collects a complete fade tail."""
668    pcm_format = AudioFormat(
669        content_type=ContentType.PCM_S16LE,
670        sample_rate=8000,
671        bit_depth=16,
672        channels=2,
673    )
674    current_details = SimpleNamespace(
675        duration=30,
676        seek_position=0,
677        seconds_streamed=0,
678        uri="test://current",
679        buffer=SimpleNamespace(eof=True, cancelled=False, has_error=False, max_size_seconds=300),
680        is_realtime=False,
681    )
682    next_details = SimpleNamespace(
683        audio_format=pcm_format,
684        buffer=_buffer(SMART_CROSSFADE_DURATION, ready=True),
685        duration=30,
686        seek_position=0,
687        uri="test://next",
688        is_realtime=False,
689        volume_normalization_mode=None,
690    )
691    current_item = SimpleNamespace(
692        queue_id="queue-1",
693        queue_item_id="current",
694        name="Current",
695        streamdetails=current_details,
696        extra_attributes={},
697    )
698    next_item = SimpleNamespace(
699        queue_id="queue-1",
700        queue_item_id="next",
701        name="Next",
702        streamdetails=next_details,
703        extra_attributes={},
704        available=True,
705    )
706    queue = SimpleNamespace(queue_id="queue-1", display_name="Queue", index_in_buffer=0)
707    player = SimpleNamespace(player_id="player-1", name="Player")
708    mass = MagicMock()
709    mass.player_queues.get.return_value = queue
710    mass.player_queues.load_next_queue_item = AsyncMock(return_value=next_item)
711    mass.player_queues.index_by_id.return_value = 1
712    audio = StreamsAudio(cast("Any", mass))
713    audio.setup()
714    audio.select_pcm_format = AsyncMock(return_value=pcm_format)  # type: ignore[method-assign]
715    audio.crossfade_allowed = MagicMock(return_value=True)  # type: ignore[method-assign]
716    build = AsyncMock(
717        return_value=SimpleNamespace(
718            timing_info=SimpleNamespace(
719                pre_crossfade_duration=2,
720                crossfade_duration=6,
721                fadein_trimmed_duration=0,
722            )
723        )
724    )
725    monkeypatch.setattr(audio.smart_fades_mixer, "build", build)
726    monkeypatch.setattr(audio.smart_fades_mixer, "mix", _empty_mix)
727
728    async def _current_stream(
729        queue_item: object, *_args: object, **_kwargs: object
730    ) -> AsyncGenerator[bytes]:
731        if queue_item is not current_item:
732            return
733        # a whole second, then chunks that never line up with a second boundary
734        yield bytes(pcm_format.pcm_sample_size * 8)
735        for _ in range(30):
736            yield bytes(pcm_format.pcm_sample_size // 3)
737
738    monkeypatch.setattr(audio, "get_queue_item_stream", _current_stream)
739    stream = audio.get_queue_item_stream_with_smartfade(
740        cast("Any", player),
741        cast("Any", current_item),
742        pcm_format,
743        crossfade_mode=CrossfadeMode.STANDARD_CROSSFADE,
744        standard_crossfade_duration=8,
745    )
746
747    async for _chunk in stream:
748        pass
749
750    build.assert_awaited_once()
751
752
753async def test_smartfade_short_remainder_still_crossfades(
754    monkeypatch: pytest.MonkeyPatch,
755) -> None:
756    """Less audio left than the configured overlap still fades with what is there."""
757    pcm_format = AudioFormat(
758        content_type=ContentType.PCM_S16LE,
759        sample_rate=8000,
760        bit_depth=16,
761        channels=2,
762    )
763    current_details = SimpleNamespace(
764        duration=180,
765        seek_position=146,
766        seconds_streamed=0,
767        uri="test://current",
768        buffer=SimpleNamespace(eof=True, cancelled=False, has_error=False, max_size_seconds=300),
769        is_realtime=False,
770    )
771    next_details = SimpleNamespace(
772        audio_format=pcm_format,
773        buffer=_buffer(SMART_CROSSFADE_DURATION, ready=True),
774        duration=180,
775        seek_position=0,
776        uri="test://next",
777        is_realtime=False,
778        volume_normalization_mode=None,
779    )
780    current_item = SimpleNamespace(
781        queue_id="queue-1",
782        queue_item_id="current",
783        name="Current",
784        streamdetails=current_details,
785        extra_attributes={},
786    )
787    next_item = SimpleNamespace(
788        queue_id="queue-1",
789        queue_item_id="next",
790        name="Next",
791        streamdetails=next_details,
792        extra_attributes={},
793        available=True,
794    )
795    queue = SimpleNamespace(queue_id="queue-1", display_name="Queue", index_in_buffer=0)
796    player = SimpleNamespace(player_id="player-1", name="Player")
797    mass = MagicMock()
798    mass.player_queues.get.return_value = queue
799    mass.player_queues.load_next_queue_item = AsyncMock(return_value=next_item)
800    mass.player_queues.index_by_id.return_value = 1
801    audio = StreamsAudio(cast("Any", mass))
802    audio.setup()
803    audio.select_pcm_format = AsyncMock(return_value=pcm_format)  # type: ignore[method-assign]
804    audio.crossfade_allowed = MagicMock(return_value=True)  # type: ignore[method-assign]
805    build = AsyncMock(
806        return_value=SimpleNamespace(
807            timing_info=SimpleNamespace(
808                pre_crossfade_duration=2,
809                crossfade_duration=6,
810                fadein_trimmed_duration=0,
811            )
812        )
813    )
814    monkeypatch.setattr(audio.smart_fades_mixer, "build", build)
815    monkeypatch.setattr(audio.smart_fades_mixer, "mix", _empty_mix)
816
817    async def _current_stream(
818        queue_item: object, *_args: object, **_kwargs: object
819    ) -> AsyncGenerator[bytes]:
820        if queue_item is not current_item:
821            return
822        # a seek near the end leaves 34s, less than the 45s smart overlap
823        yield bytes(pcm_format.pcm_sample_size * 8)
824        yield bytes(pcm_format.pcm_sample_size * 26)
825
826    monkeypatch.setattr(audio, "get_queue_item_stream", _current_stream)
827    stream = audio.get_queue_item_stream_with_smartfade(
828        cast("Any", player),
829        cast("Any", current_item),
830        pcm_format,
831        crossfade_mode=CrossfadeMode.SMART_CROSSFADE,
832        standard_crossfade_duration=8,
833    )
834
835    async for _chunk in stream:
836        pass
837
838    build.assert_awaited_once()
839    assert build.await_args is not None
840    fade_out_seconds = len(build.await_args.kwargs["fade_out_data"]) / pcm_format.pcm_sample_size
841    assert fade_out_seconds == pytest.approx(26, abs=1)
842
843
844async def test_smartfade_stub_remainder_does_not_crossfade(
845    monkeypatch: pytest.MonkeyPatch,
846) -> None:
847    """A remainder too short to overlap with is played out instead of faded."""
848    pcm_format = AudioFormat(
849        content_type=ContentType.PCM_S16LE,
850        sample_rate=8000,
851        bit_depth=16,
852        channels=2,
853    )
854    current_details = SimpleNamespace(
855        duration=180,
856        seek_position=176,
857        seconds_streamed=0,
858        uri="test://current",
859        buffer=SimpleNamespace(eof=True, cancelled=False, has_error=False, max_size_seconds=300),
860        is_realtime=False,
861    )
862    next_details = SimpleNamespace(
863        audio_format=pcm_format,
864        buffer=_buffer(SMART_CROSSFADE_DURATION, ready=True),
865        duration=180,
866        seek_position=0,
867        uri="test://next",
868        is_realtime=False,
869        volume_normalization_mode=None,
870    )
871    current_item = SimpleNamespace(
872        queue_id="queue-1",
873        queue_item_id="current",
874        name="Current",
875        streamdetails=current_details,
876        extra_attributes={},
877    )
878    next_item = SimpleNamespace(
879        queue_id="queue-1",
880        queue_item_id="next",
881        name="Next",
882        streamdetails=next_details,
883        extra_attributes={},
884        available=True,
885    )
886    queue = SimpleNamespace(queue_id="queue-1", display_name="Queue", index_in_buffer=0)
887    player = SimpleNamespace(player_id="player-1", name="Player")
888    mass = MagicMock()
889    mass.player_queues.get.return_value = queue
890    mass.player_queues.load_next_queue_item = AsyncMock(return_value=next_item)
891    mass.player_queues.index_by_id.return_value = 1
892    audio = StreamsAudio(cast("Any", mass))
893    audio.setup()
894    audio.select_pcm_format = AsyncMock(return_value=pcm_format)  # type: ignore[method-assign]
895    audio.crossfade_allowed = MagicMock(return_value=True)  # type: ignore[method-assign]
896    build = AsyncMock()
897    monkeypatch.setattr(audio.smart_fades_mixer, "build", build)
898
899    async def _current_stream(
900        queue_item: object, *_args: object, **_kwargs: object
901    ) -> AsyncGenerator[bytes]:
902        if queue_item is not current_item:
903            return
904        yield bytes(pcm_format.pcm_sample_size * 8)
905        yield bytes(pcm_format.pcm_sample_size * 2)
906
907    monkeypatch.setattr(audio, "get_queue_item_stream", _current_stream)
908    stream = audio.get_queue_item_stream_with_smartfade(
909        cast("Any", player),
910        cast("Any", current_item),
911        pcm_format,
912        crossfade_mode=CrossfadeMode.SMART_CROSSFADE,
913        standard_crossfade_duration=8,
914    )
915
916    output = b"".join([chunk async for chunk in stream])
917
918    assert len(output) == pcm_format.pcm_sample_size * 10
919    build.assert_not_awaited()
920
921
922async def test_flow_reports_no_crossfade_for_a_realtime_item(
923    monkeypatch: pytest.MonkeyPatch,
924) -> None:
925    """The audio pipeline shown for a realtime item reports that no crossfade is applied."""
926    pcm_format = AudioFormat(
927        content_type=ContentType.PCM_S16LE,
928        sample_rate=8000,
929        bit_depth=16,
930        channels=2,
931    )
932    realtime_details = SimpleNamespace(
933        audio_format=pcm_format,
934        buffer=SimpleNamespace(eof=True, cancelled=False, has_error=False, max_size_seconds=300),
935        fade_in=False,
936        stream_error=False,
937        uri="test://realtime",
938        seek_position=0,
939        seconds_streamed=0,
940        duration=20,
941        is_realtime=True,
942    )
943    realtime_item = SimpleNamespace(
944        queue_id="queue-1",
945        queue_item_id="item-1",
946        name="Realtime",
947        media_type=MediaType.TRACK,
948        media_item=None,
949        streamdetails=realtime_details,
950        duration=20,
951        extra_attributes={},
952    )
953    queue = SimpleNamespace(
954        queue_id="queue-1",
955        display_name="Queue",
956        flow_mode=False,
957        overlay_enabled=False,
958        overlay_source=None,
959    )
960    mass = MagicMock()
961    mass.player_queues.queue_data.return_value = SimpleNamespace(
962        session_id="session-1", flow_mode_stream_log=[]
963    )
964    mass.player_queues.load_next_queue_item = AsyncMock(side_effect=QueueEmpty)
965    mass.player_queues.get.return_value = queue
966    mass.streams.get_crossfade_mode.return_value = CrossfadeMode.SMART_CROSSFADE
967    mass.config.get_raw_core_config_value.return_value = 8
968    update_item_context = MagicMock()
969    mass.streams.audio_processing.update_item_context = update_item_context
970    player = MagicMock()
971    player.config.get_value.return_value = "fixed_48000"
972    player.get_supported_sample_rates.return_value = []
973    mass.players.get_player.return_value = player
974    audio = StreamsAudio(cast("Any", mass))
975    audio.setup()
976
977    async def _item_stream(*_args: object, **_kwargs: object) -> AsyncGenerator[bytes]:
978        yield bytes(pcm_format.pcm_sample_size * 4)
979
980    monkeypatch.setattr(audio, "get_queue_item_stream", _item_stream)
981    stream = audio.get_queue_flow_stream(
982        cast("Any", queue), cast("Any", realtime_item), pcm_format, session_id="session-1"
983    )
984
985    async for _chunk in stream:
986        pass
987
988    update_item_context.assert_called()
989    reported = update_item_context.call_args.kwargs["queue_processing"]
990    assert reported.crossfade_mode == CrossfadeMode.DISABLED
991
992
993async def test_flow_standard_fade_only_holds_back_its_overlap(
994    monkeypatch: pytest.MonkeyPatch,
995) -> None:
996    """A standard transition waits for its overlap, not for the whole requested window."""
997    pcm_format = AudioFormat(
998        content_type=ContentType.PCM_S16LE,
999        sample_rate=8000,
1000        bit_depth=16,
1001        channels=2,
1002    )
1003    first_details = SimpleNamespace(
1004        audio_format=pcm_format,
1005        buffer=SimpleNamespace(eof=True, cancelled=False, has_error=False, max_size_seconds=300),
1006        fade_in=False,
1007        stream_error=False,
1008        uri="test://first",
1009        seek_position=0,
1010        seconds_streamed=0,
1011        duration=300,
1012        is_realtime=False,
1013    )
1014    second_details = SimpleNamespace(
1015        audio_format=pcm_format,
1016        buffer=_buffer(SMART_CROSSFADE_DURATION, ready=True),
1017        fade_in=False,
1018        stream_error=False,
1019        uri="test://second",
1020        seek_position=0,
1021        seconds_streamed=0,
1022        duration=300,
1023        is_realtime=False,
1024        volume_normalization_mode=None,
1025    )
1026    first_item = SimpleNamespace(
1027        queue_id="queue-1",
1028        queue_item_id="item-1",
1029        name="First",
1030        media_type=MediaType.TRACK,
1031        media_item=None,
1032        streamdetails=first_details,
1033        duration=300,
1034        extra_attributes={},
1035    )
1036    second_item = SimpleNamespace(
1037        queue_id="queue-1",
1038        queue_item_id="item-2",
1039        name="Second",
1040        media_type=MediaType.TRACK,
1041        media_item=None,
1042        streamdetails=second_details,
1043        duration=300,
1044        extra_attributes={},
1045    )
1046    queue = SimpleNamespace(
1047        queue_id="queue-1",
1048        display_name="Queue",
1049        flow_mode=False,
1050        overlay_enabled=False,
1051        overlay_source=None,
1052    )
1053    mass = MagicMock()
1054    mass.player_queues.queue_data.return_value = SimpleNamespace(
1055        session_id="session-1", flow_mode_stream_log=[]
1056    )
1057    mass.player_queues.load_next_queue_item = AsyncMock(side_effect=[second_item, QueueEmpty])
1058    mass.player_queues.get.return_value = queue
1059    mass.streams.get_crossfade_mode.return_value = CrossfadeMode.SMART_CROSSFADE
1060    mass.config.get_raw_core_config_value.return_value = 8
1061    player = MagicMock()
1062    player.config.get_value.return_value = "fixed_48000"
1063    player.get_supported_sample_rates.return_value = []
1064    mass.players.get_player.return_value = player
1065    audio = StreamsAudio(cast("Any", mass))
1066    audio.setup()
1067    audio.crossfade_allowed = MagicMock(return_value=True)  # type: ignore[method-assign]
1068    # the incoming analysis is not ready, so the mixer degrades to a standard fade
1069    standard = StandardCrossFade(logger=MagicMock(), crossfade_duration=8)
1070    standard.build(
1071        pcm_format.pcm_sample_size * SMART_CROSSFADE_DURATION,
1072        pcm_format.pcm_sample_size * SMART_CROSSFADE_DURATION,
1073        pcm_format,
1074    )
1075    monkeypatch.setattr(audio.smart_fades_mixer, "build", AsyncMock(return_value=standard))
1076    monkeypatch.setattr(audio.smart_fades_mixer, "mix", _empty_mix)
1077
1078    consumed: dict[str, int] = {"second": 0}
1079
1080    async def _item_stream(
1081        queue_item: SimpleNamespace, *_args: object, **_kwargs: object
1082    ) -> AsyncGenerator[bytes]:
1083        if queue_item is first_item:
1084            for _ in range(60):
1085                yield bytes(pcm_format.pcm_sample_size)
1086            return
1087        for _ in range(SMART_CROSSFADE_DURATION + 20):
1088            consumed["second"] += 1
1089            yield bytes(pcm_format.pcm_sample_size)
1090
1091    monkeypatch.setattr(audio, "get_queue_item_stream", _item_stream)
1092    stream = audio.get_queue_flow_stream(
1093        cast("Any", queue), cast("Any", first_item), pcm_format, session_id="session-1"
1094    )
1095
1096    seconds_before_transition: int | None = None
1097    async for _chunk in stream:
1098        if seconds_before_transition is None and consumed["second"]:
1099            seconds_before_transition = consumed["second"]
1100
1101    # the overlap is 8s, so the transition must not wait for the full 45s window
1102    assert seconds_before_transition is not None
1103    assert seconds_before_transition <= SMART_CROSSFADE_DURATION / 2
1104
1105
1106# -- StreamsController.serve_queue_item_stream steering --
1107
1108
1109class _PcmFormatRequested(Exception):
1110    """Raised to stop the handler once it has decided on crossfading."""
1111
1112
1113def _single_item_handler(*, is_realtime: bool) -> tuple[Any, MagicMock, dict[str, Any]]:
1114    """Return a single-item stream handler that stops once the PCM format is picked."""
1115    streamdetails = _make_stream_details(MediaType.TRACK, is_realtime=is_realtime)
1116    queue_item = SimpleNamespace(
1117        queue_id="queue-1",
1118        queue_item_id="item-1",
1119        name="Track",
1120        duration=180,
1121        streamdetails=streamdetails,
1122        media_item=None,
1123        media_type=MediaType.TRACK,
1124        extra_attributes={},
1125        image=None,
1126    )
1127    queue = SimpleNamespace(
1128        queue_id="queue-1",
1129        display_name="Queue",
1130        current_item=queue_item,
1131        crossfade_enabled=True,
1132        overlay_enabled=False,
1133        overlay_source=None,
1134    )
1135    mass = MagicMock()
1136    mass.player_queues.get.return_value = queue
1137    mass.player_queues.queue_data.return_value = SimpleNamespace(session_id="session-1")
1138    mass.player_queues.get_item.return_value = queue_item
1139    mass.config.get_raw_core_config_value.return_value = 8
1140    player = MagicMock(player_id="player-1", protocol_parent_id=None)
1141    player.state.supported_features = {PlayerFeature.GAPLESS_PLAYBACK}
1142    player.state.name = "Player"
1143    mass.players.get_player.return_value = player
1144
1145    seen: dict[str, Any] = {}
1146
1147    async def _select_pcm_format(**kwargs: Any) -> None:
1148        seen["crossfade_enabled"] = kwargs["crossfade_enabled"]
1149        raise _PcmFormatRequested
1150
1151    audio = MagicMock()
1152    audio.select_pcm_format = _select_pcm_format
1153    controller = cast("Any", object.__new__(StreamsController))
1154    controller.mass = mass
1155    controller.audio = audio
1156    controller.logger = MagicMock()
1157    controller._log_request = MagicMock()
1158    controller.get_crossfade_mode = MagicMock(return_value=CrossfadeMode.SMART_CROSSFADE)
1159    request = MagicMock()
1160    request.method = "GET"
1161    request.match_info = {
1162        "queue_id": "queue-1",
1163        "player_id": "player-1",
1164        "session_id": "session-1",
1165        "queue_item_id": "item-1",
1166    }
1167    return controller, request, seen
1168
1169
1170async def test_single_item_handler_skips_crossfade_for_a_realtime_item() -> None:
1171    """A realtime item is never steered into the crossfaded single-item stream."""
1172    controller, request, seen = _single_item_handler(is_realtime=True)
1173
1174    with pytest.raises(_PcmFormatRequested):
1175        await controller.serve_queue_item_stream(request)
1176
1177    assert seen["crossfade_enabled"] is False
1178    controller.get_crossfade_mode.assert_not_called()
1179
1180
1181async def test_single_item_handler_keeps_crossfade_for_a_buffered_item() -> None:
1182    """A buffered item still gets the queue's configured crossfade."""
1183    controller, request, seen = _single_item_handler(is_realtime=False)
1184
1185    with pytest.raises(_PcmFormatRequested):
1186        await controller.serve_queue_item_stream(request)
1187
1188    assert seen["crossfade_enabled"] is True
1189    controller.get_crossfade_mode.assert_called_once()
1190
1191
1192# -- StreamsAudio.get_stream_details --
1193
1194
1195@pytest.mark.parametrize(
1196    ("media_item_cls", "media_type", "expected_is_realtime"),
1197    [
1198        pytest.param(Radio, MediaType.RADIO, True, id="radio"),
1199        pytest.param(AudioSource, MediaType.AUDIO_SOURCE, True, id="audio_source"),
1200        pytest.param(Track, MediaType.TRACK, False, id="track"),
1201    ],
1202)
1203async def test_get_stream_details_sets_is_realtime_by_media_type(
1204    media_item_cls: type, media_type: MediaType, expected_is_realtime: bool
1205) -> None:
1206    """RADIO and AUDIO_SOURCE streams are marked realtime; a TRACK's flag is left alone."""
1207    provider_streamdetails = StreamDetails(
1208        provider="test--1",
1209        item_id="item-1",
1210        audio_format=AudioFormat(content_type=ContentType.MP3),
1211        media_type=media_type,
1212        stream_type=StreamType.CUSTOM,
1213        duration=180 if media_type == MediaType.TRACK else None,
1214    )
1215    audio = _stream_details_provider(provider_streamdetails)
1216
1217    streamdetails = await audio.get_stream_details(
1218        queue_item=_queue_item_with_mapping(media_item_cls)
1219    )
1220
1221    assert streamdetails.is_realtime is expected_is_realtime
1222