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