music-assistant-server

1"""Tests for crossfade degradation when incoming source capacity is unavailable."""
2
3from __future__ import annotations
4
5from collections.abc import AsyncGenerator
6from types import SimpleNamespace
7from typing import Any, cast
8from unittest.mock import AsyncMock, MagicMock
9
10import pytest
11from music_assistant_models.enums import ContentType, CrossfadeMode, MediaType, StreamType
12from music_assistant_models.media_items import AudioFormat
13from music_assistant_models.streamdetails import StreamDetails
14
15from music_assistant.controllers.streams.audio import (
16    MIN_CROSSFADE_FALLBACK_DURATION,
17    StreamsAudio,
18)
19from music_assistant.controllers.streams.audio_buffer import AudioBuffer
20from music_assistant.controllers.streams.smart_fades.helpers import SMART_CROSSFADE_DURATION
21
22
23def _streamdetails(audio_buffer: AudioBuffer | None) -> StreamDetails:
24    """Build incoming track details with an optional prepared buffer."""
25    streamdetails = StreamDetails(
26        provider="test--1",
27        item_id="track-1",
28        audio_format=AudioFormat(content_type=ContentType.FLAC),
29        media_type=MediaType.TRACK,
30        stream_type=StreamType.HTTP,
31        path="http://test.invalid/track.flac",
32        duration=180,
33    )
34    streamdetails.buffer = audio_buffer
35    return streamdetails
36
37
38def _buffer(duration_available: float, ready: bool) -> AudioBuffer:
39    """Build a valid buffer with the requested resident duration."""
40    audio_buffer = MagicMock(spec=AudioBuffer)
41    audio_buffer.has_error = False
42    audio_buffer.is_valid.return_value = True
43    audio_buffer.duration_available = duration_available
44    audio_buffer.ready = MagicMock()
45    audio_buffer.ready.is_set.return_value = ready
46    return audio_buffer
47
48
49def test_ready_incoming_buffer_keeps_smart_crossfade() -> None:
50    """A fully resident incoming buffer keeps the requested Smart Fade."""
51    audio = StreamsAudio(MagicMock())
52
53    mode, duration = audio._select_buffered_crossfade(
54        _streamdetails(_buffer(SMART_CROSSFADE_DURATION, ready=True)),
55        CrossfadeMode.SMART_CROSSFADE,
56        standard_crossfade_duration=8,
57    )
58
59    assert mode == CrossfadeMode.SMART_CROSSFADE
60    assert duration == SMART_CROSSFADE_DURATION
61
62
63def test_partial_incoming_buffer_degrades_to_short_standard_crossfade() -> None:
64    """Five resident seconds are crossfaded without waiting for more source PCM."""
65    audio = StreamsAudio(MagicMock())
66    available_seconds = MIN_CROSSFADE_FALLBACK_DURATION + 2
67
68    mode, duration = audio._select_buffered_crossfade(
69        _streamdetails(_buffer(available_seconds, ready=False)),
70        CrossfadeMode.SMART_CROSSFADE,
71        standard_crossfade_duration=8,
72    )
73
74    assert mode == CrossfadeMode.STANDARD_CROSSFADE
75    assert duration == available_seconds
76
77
78def test_crossfade_resident_duration_accounts_for_playback_speed() -> None:
79    """Fast playback cannot claim more post-filter overlap than resident PCM can produce."""
80    audio = StreamsAudio(MagicMock())
81
82    mode, duration = audio._select_buffered_crossfade(
83        _streamdetails(_buffer(8, ready=False)),
84        CrossfadeMode.SMART_CROSSFADE,
85        standard_crossfade_duration=8,
86        playback_speed=2.0,
87    )
88
89    assert mode == CrossfadeMode.DISABLED
90    assert duration == 0
91
92
93@pytest.mark.parametrize(
94    "audio_buffer",
95    [
96        None,
97        _buffer(MIN_CROSSFADE_FALLBACK_DURATION - 0.5, ready=False),
98    ],
99)
100def test_unprepared_incoming_buffer_disables_crossfade(
101    audio_buffer: AudioBuffer | None,
102) -> None:
103    """An unusable incoming buffer falls back to gapless/no-crossfade playback."""
104    audio = StreamsAudio(MagicMock())
105
106    mode, duration = audio._select_buffered_crossfade(
107        _streamdetails(audio_buffer),
108        CrossfadeMode.SMART_CROSSFADE,
109        standard_crossfade_duration=8,
110    )
111
112    assert mode == CrossfadeMode.DISABLED
113    assert duration == 0
114
115
116async def test_unprepared_next_track_flushes_outgoing_tail_without_opening_source(
117    monkeypatch: pytest.MonkeyPatch,
118) -> None:
119    """Missing incoming PCM emits the complete outgoing track without a blocking fade fetch."""
120    pcm_format = AudioFormat(
121        content_type=ContentType.PCM_S16LE,
122        sample_rate=8000,
123        bit_depth=16,
124        channels=2,
125    )
126    current_details = SimpleNamespace(
127        duration=16,
128        seek_position=0,
129        seconds_streamed=0,
130        uri="test://current",
131        buffer=None,
132    )
133    next_details = SimpleNamespace(
134        audio_format=pcm_format,
135        buffer=None,
136        duration=16,
137        seek_position=0,
138        uri="test://next",
139    )
140    current_item = SimpleNamespace(
141        queue_id="queue-1",
142        queue_item_id="current",
143        name="Current",
144        streamdetails=current_details,
145        extra_attributes={},
146    )
147    next_item = SimpleNamespace(
148        queue_id="queue-1",
149        queue_item_id="next",
150        name="Next",
151        streamdetails=next_details,
152        extra_attributes={},
153        available=True,
154    )
155    queue = SimpleNamespace(
156        queue_id="queue-1",
157        display_name="Queue",
158        index_in_buffer=0,
159    )
160    player = SimpleNamespace(player_id="player-1", name="Player")
161    mass = MagicMock()
162    mass.player_queues.get.return_value = queue
163    mass.player_queues.load_next_queue_item = AsyncMock(return_value=next_item)
164    mass.player_queues.index_by_id.return_value = 1
165    audio = StreamsAudio(cast("Any", mass))
166    audio.setup()
167    audio.select_pcm_format = AsyncMock(return_value=pcm_format)  # type: ignore[method-assign]
168    audio.crossfade_allowed = MagicMock(return_value=True)  # type: ignore[method-assign]
169    build = AsyncMock()
170    monkeypatch.setattr(audio.smart_fades_mixer, "build", build)
171
172    async def _current_stream(
173        queue_item: object,
174        *_args: object,
175        **_kwargs: object,
176    ) -> AsyncGenerator[bytes]:
177        if queue_item is not current_item:
178            pytest.fail("The incoming source was opened during crossfade fallback")
179        yield bytes(pcm_format.pcm_sample_size * 8)
180        yield bytes(pcm_format.pcm_sample_size * 8)
181
182    monkeypatch.setattr(audio, "get_queue_item_stream", _current_stream)
183    stream = audio.get_queue_item_stream_with_smartfade(
184        cast("Any", player),
185        cast("Any", current_item),
186        pcm_format,
187        crossfade_mode=CrossfadeMode.STANDARD_CROSSFADE,
188        standard_crossfade_duration=8,
189    )
190
191    output = b"".join([chunk async for chunk in stream])
192
193    assert len(output) == pcm_format.pcm_sample_size * 16
194    assert next_item.available
195    build.assert_not_awaited()
196
197
198@pytest.mark.parametrize(
199    ("playback_speed", "resident_media_duration"),
200    [(0.5, 2.5), (2.0, 10.0)],
201)
202async def test_partial_crossfade_resumes_at_consumed_media_time(
203    monkeypatch: pytest.MonkeyPatch,
204    playback_speed: float,
205    resident_media_duration: float,
206) -> None:
207    """Crossfade output bytes resume the raw source at the matching media-time position."""
208    pcm_format = AudioFormat(
209        content_type=ContentType.PCM_S16LE,
210        sample_rate=8000,
211        bit_depth=16,
212        channels=2,
213    )
214    current_details = SimpleNamespace(
215        duration=16,
216        seek_position=0,
217        seconds_streamed=0,
218        uri="test://current",
219        buffer=None,
220    )
221    next_details = SimpleNamespace(
222        audio_format=pcm_format,
223        buffer=_buffer(resident_media_duration, ready=False),
224        duration=16,
225        seek_position=0,
226        uri="test://next",
227        volume_normalization_mode=None,
228    )
229    current_item = SimpleNamespace(
230        queue_id="queue-1",
231        queue_item_id="current",
232        name="Current",
233        streamdetails=current_details,
234        extra_attributes={},
235    )
236    next_item = SimpleNamespace(
237        queue_id="queue-1",
238        queue_item_id="next",
239        name="Next",
240        streamdetails=next_details,
241        extra_attributes={"playback_speed": playback_speed},
242        available=True,
243    )
244    queue = SimpleNamespace(
245        queue_id="queue-1",
246        display_name="Queue",
247        index_in_buffer=0,
248    )
249    player = SimpleNamespace(player_id="player-1", name="Player")
250    mass = MagicMock()
251    mass.player_queues.get.return_value = queue
252    mass.player_queues.load_next_queue_item = AsyncMock(return_value=next_item)
253    mass.player_queues.index_by_id.return_value = 1
254    audio = StreamsAudio(cast("Any", mass))
255    audio.setup()
256    audio.select_pcm_format = AsyncMock(return_value=pcm_format)  # type: ignore[method-assign]
257    audio.crossfade_allowed = MagicMock(return_value=True)  # type: ignore[method-assign]
258    smart_fade = SimpleNamespace(
259        timing_info=SimpleNamespace(
260            pre_crossfade_duration=3,
261            crossfade_duration=5,
262            fadein_trimmed_duration=0,
263        )
264    )
265    monkeypatch.setattr(
266        audio.smart_fades_mixer,
267        "build",
268        AsyncMock(return_value=smart_fade),
269    )
270
271    async def _mix(
272        _smart_fade: object,
273        *,
274        fade_in_part: AsyncGenerator[bytes],
275        **_kwargs: object,
276    ) -> AsyncGenerator[bytes]:
277        async for chunk in fade_in_part:
278            yield chunk
279
280    monkeypatch.setattr(audio.smart_fades_mixer, "mix", _mix)
281    requested_beyond_resident = False
282    seek_positions: list[float | None] = []
283
284    async def _item_stream(
285        queue_item: object,
286        *_args: object,
287        **kwargs: object,
288    ) -> AsyncGenerator[bytes]:
289        nonlocal requested_beyond_resident
290        if queue_item is current_item:
291            yield bytes(pcm_format.pcm_sample_size * 8)
292            yield bytes(pcm_format.pcm_sample_size * 8)
293            return
294        seek_positions.append(cast("float | None", kwargs.get("seek_position")))
295        yield bytes(pcm_format.pcm_sample_size * MIN_CROSSFADE_FALLBACK_DURATION)
296        requested_beyond_resident = True
297        pytest.fail("Crossfade requested audio beyond the resident buffer")
298
299    monkeypatch.setattr(audio, "get_queue_item_stream", _item_stream)
300    stream = audio.get_queue_item_stream_with_smartfade(
301        cast("Any", player),
302        cast("Any", current_item),
303        pcm_format,
304        crossfade_mode=CrossfadeMode.STANDARD_CROSSFADE,
305        standard_crossfade_duration=8,
306    )
307
308    _ = [chunk async for chunk in stream]
309
310    assert not requested_beyond_resident
311    crossfade_data = audio._crossfade_data["queue-1"]
312    assert crossfade_data.fade_in_media_duration == resident_media_duration
313    assert crossfade_data.elapsed_time_offset == 5 * playback_speed
314
315    next_stream = audio.get_queue_item_stream_with_smartfade(
316        cast("Any", player),
317        cast("Any", next_item),
318        pcm_format,
319        crossfade_mode=CrossfadeMode.STANDARD_CROSSFADE,
320        standard_crossfade_duration=8,
321    )
322    await anext(next_stream)
323    await next_stream.aclose()
324
325    assert seek_positions == [None, resident_media_duration]
326    assert next_details.seek_position == 5 * playback_speed
327