music-assistant-server

1"""Tests for the ffmpeg input arguments StreamsAudio.get_media_stream builds."""
2
3from __future__ import annotations
4
5import asyncio
6from collections.abc import AsyncGenerator
7from typing import Any, cast
8from unittest.mock import MagicMock
9
10import pytest
11from music_assistant_models.enums import ContentType, MediaType, StreamType
12from music_assistant_models.errors import AudioError
13from music_assistant_models.media_items import AudioFormat
14from music_assistant_models.streamdetails import MultiPartPath, StreamDetails
15
16import music_assistant.controllers.streams.audio as audio_mod
17from music_assistant.controllers.streams.audio import StreamsAudio
18
19# input args a provider may attach to its StreamDetails (podcastfeed does exactly this).
20# Kept as a tuple so the tests below can never assert against a mutated expectation.
21_PROVIDER_INPUT_ARGS = ("-user_agent", "Test/1.0")
22
23
24class _FakeFFMpeg:
25    """FFMpeg test double that records the arguments it was constructed with."""
26
27    last_instance: _FakeFFMpeg | None = None
28
29    def __init__(
30        self,
31        *,
32        audio_input: object,
33        input_format: AudioFormat,
34        extra_input_args: list[str] | None = None,
35        **_kwargs: Any,
36    ) -> None:
37        self.audio_input = audio_input
38        self.extra_input_args = extra_input_args
39        # Mirror the real FFMpeg, which mutates this object's codec_type after probe.
40        # Tests inspect the original `input_format` AudioFormat passed in to confirm
41        # which one the controller picked.
42        self.input_format = input_format
43        self._probed_codec_type = ContentType.FLAC  # arbitrary, distinct from PCM/OGG
44        self.parsed_duration: int | None = None
45        self.returncode: int | None = 0
46        self.log_history: list[str] = []
47        self.proc = MagicMock(pid=1234)
48        type(self).last_instance = self
49
50    async def start(self) -> None:
51        # Simulate ffmpeg's post-probe codec detection: real FFMpeg mutates
52        # self.input_format.codec_type once it reads the input header.
53        self.input_format.codec_type = self._probed_codec_type
54
55    async def iter_chunked(self, _chunk_size: int) -> AsyncGenerator[bytes]:
56        yield b"\x00\x01" * 256
57
58    async def wait_with_timeout(self, _timeout: float) -> None:
59        return None
60
61    async def close(self) -> None:
62        return None
63
64
65@pytest.fixture
66def patch_ffmpeg(monkeypatch: pytest.MonkeyPatch) -> type[_FakeFFMpeg]:
67    """Swap the real FFMpeg in the streams.audio module for the fake."""
68    _FakeFFMpeg.last_instance = None
69    monkeypatch.setattr(audio_mod, "FFMpeg", _FakeFFMpeg)
70    return _FakeFFMpeg
71
72
73@pytest.fixture
74def patch_two_minute_ffmpeg(monkeypatch: pytest.MonkeyPatch) -> type[_TwoMinuteFFMpeg]:
75    """Swap the real FFMpeg for the fake that emits a fixed amount of audio."""
76    _TwoMinuteFFMpeg.last_instance = None
77    monkeypatch.setattr(audio_mod, "FFMpeg", _TwoMinuteFFMpeg)
78    return _TwoMinuteFFMpeg
79
80
81def _make_audio_controller() -> StreamsAudio:
82    """Build a StreamsAudio with just enough mass scaffolding to run get_media_stream."""
83    audio = StreamsAudio(MagicMock())
84    audio.mass.loop = MagicMock()
85    audio.mass.loop.time = MagicMock(return_value=0.0)
86    return audio
87
88
89def _make_pcm_format() -> AudioFormat:
90    return AudioFormat(
91        content_type=ContentType.PCM_S16LE,
92        codec_type=ContentType.PCM_S16LE,
93        sample_rate=44100,
94        bit_depth=16,
95        channels=2,
96    )
97
98
99_PCM_SAMPLE_SIZE = _make_pcm_format().pcm_sample_size
100
101
102def _make_streamdetails(
103    *,
104    audio_format: AudioFormat,
105    decoded_audio_format: AudioFormat | None = None,
106    extra_input_args: list[str] | None = None,
107) -> StreamDetails:
108    return StreamDetails(
109        provider="test_provider",
110        item_id="main",
111        audio_format=audio_format,
112        decoded_audio_format=decoded_audio_format,
113        media_type=MediaType.AUDIO_SOURCE,
114        stream_type=StreamType.NAMED_PIPE,
115        path="/tmp/fake-fifo",  # noqa: S108
116        extra_input_args=extra_input_args or [],
117    )
118
119
120def _seekable_streamdetails() -> StreamDetails:
121    """Build seekable StreamDetails carrying provider-supplied ffmpeg input args."""
122    return StreamDetails(
123        provider="test_provider",
124        item_id="episode-1",
125        audio_format=AudioFormat(content_type=ContentType.MP3),
126        media_type=MediaType.PODCAST_EPISODE,
127        stream_type=StreamType.HTTP,
128        path="http://test.invalid/episode-1.mp3",
129        duration=3600,
130        can_seek=True,
131        allow_seek=True,
132        extra_input_args=[*_PROVIDER_INPUT_ARGS],
133    )
134
135
136async def _drain(gen: AsyncGenerator[bytes]) -> None:
137    async for _ in gen:
138        pass
139
140
141def _recording_multi_file_stream() -> tuple[Any, list[int]]:
142    """
143    Build a stand-in for the concat stream plus the list of seek positions it received.
144
145    Avoids a real ffmpeg process and temp file while still proving the seek was
146    handed off to the source rather than applied through the -ss argument.
147    """
148    received_seeks: list[int] = []
149
150    async def _empty_stream() -> AsyncGenerator[bytes]:
151        yield b""
152
153    # record on call rather than on first iteration: the FFMpeg double never
154    # consumes the generator it is handed, so its body would never run
155    def _fake_stream(
156        _streamdetails: StreamDetails, seek_position: int = 0
157    ) -> AsyncGenerator[bytes]:
158        received_seeks.append(seek_position)
159        return _empty_stream()
160
161    return _fake_stream, received_seeks
162
163
164class _StallingFFMpeg(_FakeFFMpeg):
165    """FFMpeg double whose read never produces a chunk (frozen source)."""
166
167    async def iter_chunked(self, _chunk_size: int) -> AsyncGenerator[bytes]:
168        await asyncio.Event().wait()  # blocks until the watchdog cancels the read
169        yield b""  # unreachable
170
171
172class _SlowConsumerFFMpeg(_FakeFFMpeg):
173    """FFMpeg double that hands over chunks instantly when asked."""
174
175    async def iter_chunked(self, _chunk_size: int) -> AsyncGenerator[bytes]:
176        for _ in range(3):
177            yield b"\x00\x01" * 256
178
179
180class _TwoMinuteFFMpeg(_FakeFFMpeg):
181    """FFMpeg double that emits exactly two minutes of PCM at the format below."""
182
183    seconds_emitted = 120
184
185    async def iter_chunked(self, _chunk_size: int) -> AsyncGenerator[bytes]:
186        for _ in range(self.seconds_emitted):
187            yield b"\x00" * _PCM_SAMPLE_SIZE
188
189
190def _multi_part_streamdetails() -> StreamDetails:
191    """Build StreamDetails for a multi-file audiobook of two 30 minute parts."""
192    return StreamDetails(
193        provider="test_provider",
194        item_id="audiobook-1",
195        audio_format=AudioFormat(content_type=ContentType.MP3),
196        media_type=MediaType.AUDIOBOOK,
197        stream_type=StreamType.HTTP,
198        path=[
199            MultiPartPath(path="http://test.invalid/part-1.mp3", duration=1800),
200            MultiPartPath(path="http://test.invalid/part-2.mp3", duration=1800),
201        ],
202        duration=3600,
203        can_seek=True,
204        allow_seek=True,
205    )
206
207
208def _flac_streamdetails(extra_input_args: list[str] | None = None) -> StreamDetails:
209    return _make_streamdetails(
210        audio_format=AudioFormat(
211            content_type=ContentType.FLAC,
212            codec_type=ContentType.FLAC,
213            sample_rate=44100,
214            bit_depth=16,
215            channels=2,
216        ),
217        extra_input_args=extra_input_args,
218    )
219
220
221@pytest.mark.asyncio
222async def test_get_media_stream_raises_when_source_stalls(
223    monkeypatch: pytest.MonkeyPatch,
224) -> None:
225    """A source that stops producing audio is surfaced as an AudioError."""
226    monkeypatch.setattr(audio_mod, "FFMpeg", _StallingFFMpeg)
227    monkeypatch.setattr(audio_mod, "STREAM_START_TIMEOUT", 0.1)
228    monkeypatch.setattr(audio_mod, "STREAM_STALL_TIMEOUT", 0.1)
229
230    audio = _make_audio_controller()
231    with pytest.raises(AudioError):
232        await _drain(audio.get_media_stream(_flac_streamdetails(), _make_pcm_format()))
233
234
235@pytest.mark.asyncio
236async def test_get_media_stream_does_not_stall_on_slow_consumer(
237    monkeypatch: pytest.MonkeyPatch,
238) -> None:
239    """A consumer slower than the stall timeout must not trip the watchdog."""
240    monkeypatch.setattr(audio_mod, "FFMpeg", _SlowConsumerFFMpeg)
241    monkeypatch.setattr(audio_mod, "STREAM_START_TIMEOUT", 0.1)
242    monkeypatch.setattr(audio_mod, "STREAM_STALL_TIMEOUT", 0.1)
243
244    audio = _make_audio_controller()
245    chunks = 0
246    async for _ in audio.get_media_stream(_flac_streamdetails(), _make_pcm_format()):
247        chunks += 1
248        await asyncio.sleep(0.3)  # downstream waits far longer than the stall timeout
249    assert chunks == 3
250
251
252@pytest.mark.asyncio
253async def test_get_media_stream_prefers_decoded_audio_format(
254    patch_ffmpeg: type[_FakeFFMpeg],
255) -> None:
256    """When decoded_audio_format is set, ffmpeg receives that as input_format."""
257    source_format = AudioFormat(
258        content_type=ContentType.OGG,
259        codec_type=ContentType.VORBIS,
260        sample_rate=44100,
261        bit_depth=16,
262        channels=2,
263        bit_rate=320,
264    )
265    decoded_format = AudioFormat(
266        content_type=ContentType.PCM_S16LE,
267        codec_type=ContentType.PCM_S16LE,
268        sample_rate=44100,
269        bit_depth=16,
270        channels=2,
271    )
272    streamdetails = _make_streamdetails(
273        audio_format=source_format, decoded_audio_format=decoded_format
274    )
275
276    audio = _make_audio_controller()
277    await _drain(audio.get_media_stream(streamdetails, _make_pcm_format()))
278
279    assert patch_ffmpeg.last_instance is not None
280    assert patch_ffmpeg.last_instance.input_format is decoded_format
281
282
283@pytest.mark.asyncio
284async def test_get_media_stream_falls_back_to_audio_format(
285    patch_ffmpeg: type[_FakeFFMpeg],
286) -> None:
287    """When decoded_audio_format is not set, ffmpeg receives audio_format as input_format."""
288    source_format = AudioFormat(
289        content_type=ContentType.FLAC,
290        codec_type=ContentType.FLAC,
291        sample_rate=44100,
292        bit_depth=16,
293        channels=2,
294    )
295    streamdetails = _make_streamdetails(audio_format=source_format)
296
297    audio = _make_audio_controller()
298    await _drain(audio.get_media_stream(streamdetails, _make_pcm_format()))
299
300    assert patch_ffmpeg.last_instance is not None
301    assert patch_ffmpeg.last_instance.input_format is source_format
302
303
304@pytest.mark.asyncio
305@pytest.mark.usefixtures("patch_ffmpeg")
306async def test_get_media_stream_does_not_overwrite_source_codec_when_decoded_format_set() -> None:
307    """audio_format.codec_type stays authoritative when decoded_audio_format is set."""
308    source_format = AudioFormat(
309        content_type=ContentType.OGG,
310        codec_type=ContentType.VORBIS,
311        sample_rate=44100,
312        bit_depth=16,
313        channels=2,
314        bit_rate=320,
315    )
316    decoded_format = AudioFormat(
317        content_type=ContentType.PCM_S16LE,
318        codec_type=ContentType.PCM_S16LE,
319        sample_rate=44100,
320        bit_depth=16,
321        channels=2,
322    )
323    streamdetails = _make_streamdetails(
324        audio_format=source_format, decoded_audio_format=decoded_format
325    )
326
327    audio = _make_audio_controller()
328    await _drain(audio.get_media_stream(streamdetails, _make_pcm_format()))
329
330    assert streamdetails.audio_format.codec_type is ContentType.VORBIS
331
332
333@pytest.mark.asyncio
334@pytest.mark.usefixtures("patch_ffmpeg")
335async def test_get_media_stream_writes_back_codec_when_no_decoded_format() -> None:
336    """Without decoded_audio_format, ffmpeg's probed codec_type is written back."""
337    source_format = AudioFormat(
338        content_type=ContentType.FLAC,
339        # Start with UNKNOWN so we can see the post-probe writeback take effect.
340        codec_type=ContentType.UNKNOWN,
341        sample_rate=44100,
342        bit_depth=16,
343        channels=2,
344    )
345    streamdetails = _make_streamdetails(audio_format=source_format)
346
347    audio = _make_audio_controller()
348    await _drain(audio.get_media_stream(streamdetails, _make_pcm_format()))
349
350    # _FakeFFMpeg's start() mutates input_format.codec_type to FLAC; with no
351    # decoded format that AudioFormat is the same object as streamdetails.audio_format,
352    # so the controller's writeback path is exercised end-to-end.
353    assert streamdetails.audio_format.codec_type is ContentType.FLAC
354
355
356@pytest.mark.asyncio
357async def test_get_media_stream_stores_measured_duration_for_full_playthrough(
358    monkeypatch: pytest.MonkeyPatch,
359    patch_two_minute_ffmpeg: type[_TwoMinuteFFMpeg],
360) -> None:
361    """A multi-file item streamed from the start gets its measured duration stored."""
362    streamdetails = _multi_part_streamdetails()
363    audio = _make_audio_controller()
364    fake_stream, _ = _recording_multi_file_stream()
365    monkeypatch.setattr(audio, "get_multi_file_stream", fake_stream)
366
367    await _drain(audio.get_media_stream(streamdetails, _make_pcm_format()))
368
369    assert streamdetails.duration == patch_two_minute_ffmpeg.seconds_emitted
370
371
372@pytest.mark.asyncio
373async def test_get_media_stream_keeps_duration_when_multi_file_seek_is_delegated(
374    monkeypatch: pytest.MonkeyPatch,
375    patch_two_minute_ffmpeg: type[_TwoMinuteFFMpeg],
376) -> None:
377    """Resuming a multi-file audiobook must not shrink its duration to the remainder."""
378    streamdetails = _multi_part_streamdetails()
379    audio = _make_audio_controller()
380    fake_stream, received_seeks = _recording_multi_file_stream()
381    monkeypatch.setattr(audio, "get_multi_file_stream", fake_stream)
382
383    await _drain(audio.get_media_stream(streamdetails, _make_pcm_format(), seek_position=1800))
384
385    # the concat stream consumes the seek itself, which clears the local seek
386    # position before the duration writeback runs at the end of the stream
387    assert received_seeks == [1800]
388    assert patch_two_minute_ffmpeg.last_instance is not None
389    assert "-ss" not in (patch_two_minute_ffmpeg.last_instance.extra_input_args or [])
390    assert streamdetails.duration == 3600
391
392
393@pytest.mark.asyncio
394async def test_get_media_stream_keeps_duration_when_provider_seek_is_delegated(
395    patch_two_minute_ffmpeg: type[_TwoMinuteFFMpeg],
396) -> None:
397    """A seekable provider stream must not shrink its duration to the remainder either."""
398    streamdetails = StreamDetails(
399        provider="test_provider",
400        item_id="track-1",
401        audio_format=AudioFormat(content_type=ContentType.OGG),
402        media_type=MediaType.TRACK,
403        stream_type=StreamType.CUSTOM,
404        duration=240,
405        can_seek=True,
406        allow_seek=True,
407    )
408    audio = _make_audio_controller()
409    provider = cast("MagicMock", audio.mass).get_provider.return_value
410
411    await _drain(audio.get_media_stream(streamdetails, _make_pcm_format(), seek_position=90))
412
413    # a provider that can seek receives the position and the local one is cleared,
414    # so only the remaining audio reaches ffmpeg
415    provider.get_audio_stream.assert_called_once_with(streamdetails, seek_position=90)
416    assert patch_two_minute_ffmpeg.last_instance is not None
417    assert "-ss" not in (patch_two_minute_ffmpeg.last_instance.extra_input_args or [])
418    assert streamdetails.duration == 240
419
420
421@pytest.mark.asyncio
422async def test_get_media_stream_keeps_caller_extra_input_args_intact(
423    patch_ffmpeg: type[_FakeFFMpeg],
424) -> None:
425    """Per-call input args must not leak back onto the caller's StreamDetails."""
426    streamdetails = _seekable_streamdetails()
427    audio = _make_audio_controller()
428
429    await _drain(audio.get_media_stream(streamdetails, _make_pcm_format(), seek_position=30))
430
431    assert patch_ffmpeg.last_instance is not None
432    assert patch_ffmpeg.last_instance.extra_input_args == [*_PROVIDER_INPUT_ARGS, "-ss", "30"]
433    assert streamdetails.extra_input_args == [*_PROVIDER_INPUT_ARGS]
434
435    # StreamDetails are cached on the queue item and reach this method again on a
436    # retry, another seek or from the background analyzer: every call must build its
437    # args from the provider's list alone instead of stacking onto the previous call's.
438    await _drain(audio.get_media_stream(streamdetails, _make_pcm_format(), seek_position=600))
439
440    assert patch_ffmpeg.last_instance.extra_input_args == [*_PROVIDER_INPUT_ARGS, "-ss", "600"]
441    assert streamdetails.extra_input_args == [*_PROVIDER_INPUT_ARGS]
442
443
444@pytest.mark.asyncio
445async def test_get_media_stream_adds_realtime_pacing_for_audio_source(
446    patch_ffmpeg: type[_FakeFFMpeg],
447) -> None:
448    """A live AudioSource gets realtime pacing with a small initial burst of headroom."""
449    audio = _make_audio_controller()
450    await _drain(audio.get_media_stream(_flac_streamdetails(), _make_pcm_format()))
451
452    assert patch_ffmpeg.last_instance is not None
453    assert patch_ffmpeg.last_instance.extra_input_args == [
454        "-readrate",
455        "1",
456        "-readrate_initial_burst",
457        "0.5",
458    ]
459
460
461@pytest.mark.asyncio
462@pytest.mark.parametrize(
463    "provider_pacing_args",
464    [["-readrate", "1.0", "-readrate_initial_burst", "2"], ["-re"]],
465    ids=["readrate", "re"],
466)
467async def test_get_media_stream_respects_provider_pacing_args(
468    patch_ffmpeg: type[_FakeFFMpeg],
469    provider_pacing_args: list[str],
470) -> None:
471    """Provider-supplied -re/-readrate args suppress the automatic AudioSource pacing."""
472    streamdetails = _flac_streamdetails(extra_input_args=list(provider_pacing_args))
473    audio = _make_audio_controller()
474    await _drain(audio.get_media_stream(streamdetails, _make_pcm_format()))
475
476    assert patch_ffmpeg.last_instance is not None
477    assert patch_ffmpeg.last_instance.extra_input_args == provider_pacing_args
478