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