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) -> StreamDetails:
107    return StreamDetails(
108        provider="test_provider",
109        item_id="main",
110        audio_format=audio_format,
111        decoded_audio_format=decoded_audio_format,
112        media_type=MediaType.AUDIO_SOURCE,
113        stream_type=StreamType.NAMED_PIPE,
114        path="/tmp/fake-fifo",  # noqa: S108
115    )
116
117
118def _seekable_streamdetails() -> StreamDetails:
119    """Build seekable StreamDetails carrying provider-supplied ffmpeg input args."""
120    return StreamDetails(
121        provider="test_provider",
122        item_id="episode-1",
123        audio_format=AudioFormat(content_type=ContentType.MP3),
124        media_type=MediaType.PODCAST_EPISODE,
125        stream_type=StreamType.HTTP,
126        path="http://test.invalid/episode-1.mp3",
127        duration=3600,
128        can_seek=True,
129        allow_seek=True,
130        extra_input_args=[*_PROVIDER_INPUT_ARGS],
131    )
132
133
134async def _drain(gen: AsyncGenerator[bytes]) -> None:
135    async for _ in gen:
136        pass
137
138
139def _recording_multi_file_stream() -> tuple[Any, list[int]]:
140    """
141    Build a stand-in for the concat stream plus the list of seek positions it received.
142
143    Avoids a real ffmpeg process and temp file while still proving the seek was
144    handed off to the source rather than applied through the -ss argument.
145    """
146    received_seeks: list[int] = []
147
148    async def _empty_stream() -> AsyncGenerator[bytes]:
149        yield b""
150
151    # record on call rather than on first iteration: the FFMpeg double never
152    # consumes the generator it is handed, so its body would never run
153    def _fake_stream(
154        _streamdetails: StreamDetails, seek_position: int = 0
155    ) -> AsyncGenerator[bytes]:
156        received_seeks.append(seek_position)
157        return _empty_stream()
158
159    return _fake_stream, received_seeks
160
161
162class _StallingFFMpeg(_FakeFFMpeg):
163    """FFMpeg double whose read never produces a chunk (frozen source)."""
164
165    async def iter_chunked(self, _chunk_size: int) -> AsyncGenerator[bytes]:
166        await asyncio.Event().wait()  # blocks until the watchdog cancels the read
167        yield b""  # unreachable
168
169
170class _SlowConsumerFFMpeg(_FakeFFMpeg):
171    """FFMpeg double that hands over chunks instantly when asked."""
172
173    async def iter_chunked(self, _chunk_size: int) -> AsyncGenerator[bytes]:
174        for _ in range(3):
175            yield b"\x00\x01" * 256
176
177
178class _TwoMinuteFFMpeg(_FakeFFMpeg):
179    """FFMpeg double that emits exactly two minutes of PCM at the format below."""
180
181    seconds_emitted = 120
182
183    async def iter_chunked(self, _chunk_size: int) -> AsyncGenerator[bytes]:
184        for _ in range(self.seconds_emitted):
185            yield b"\x00" * _PCM_SAMPLE_SIZE
186
187
188def _multi_part_streamdetails() -> StreamDetails:
189    """Build StreamDetails for a multi-file audiobook of two 30 minute parts."""
190    return StreamDetails(
191        provider="test_provider",
192        item_id="audiobook-1",
193        audio_format=AudioFormat(content_type=ContentType.MP3),
194        media_type=MediaType.AUDIOBOOK,
195        stream_type=StreamType.HTTP,
196        path=[
197            MultiPartPath(path="http://test.invalid/part-1.mp3", duration=1800),
198            MultiPartPath(path="http://test.invalid/part-2.mp3", duration=1800),
199        ],
200        duration=3600,
201        can_seek=True,
202        allow_seek=True,
203    )
204
205
206def _flac_streamdetails() -> StreamDetails:
207    return _make_streamdetails(
208        audio_format=AudioFormat(
209            content_type=ContentType.FLAC,
210            codec_type=ContentType.FLAC,
211            sample_rate=44100,
212            bit_depth=16,
213            channels=2,
214        )
215    )
216
217
218@pytest.mark.asyncio
219async def test_get_media_stream_raises_when_source_stalls(
220    monkeypatch: pytest.MonkeyPatch,
221) -> None:
222    """A source that stops producing audio is surfaced as an AudioError."""
223    monkeypatch.setattr(audio_mod, "FFMpeg", _StallingFFMpeg)
224    monkeypatch.setattr(audio_mod, "STREAM_START_TIMEOUT", 0.1)
225    monkeypatch.setattr(audio_mod, "STREAM_STALL_TIMEOUT", 0.1)
226
227    audio = _make_audio_controller()
228    with pytest.raises(AudioError):
229        await _drain(audio.get_media_stream(_flac_streamdetails(), _make_pcm_format()))
230
231
232@pytest.mark.asyncio
233async def test_get_media_stream_does_not_stall_on_slow_consumer(
234    monkeypatch: pytest.MonkeyPatch,
235) -> None:
236    """A consumer slower than the stall timeout must not trip the watchdog."""
237    monkeypatch.setattr(audio_mod, "FFMpeg", _SlowConsumerFFMpeg)
238    monkeypatch.setattr(audio_mod, "STREAM_START_TIMEOUT", 0.1)
239    monkeypatch.setattr(audio_mod, "STREAM_STALL_TIMEOUT", 0.1)
240
241    audio = _make_audio_controller()
242    chunks = 0
243    async for _ in audio.get_media_stream(_flac_streamdetails(), _make_pcm_format()):
244        chunks += 1
245        await asyncio.sleep(0.3)  # downstream waits far longer than the stall timeout
246    assert chunks == 3
247
248
249@pytest.mark.asyncio
250async def test_get_media_stream_prefers_decoded_audio_format(
251    patch_ffmpeg: type[_FakeFFMpeg],
252) -> None:
253    """When decoded_audio_format is set, ffmpeg receives that as input_format."""
254    source_format = AudioFormat(
255        content_type=ContentType.OGG,
256        codec_type=ContentType.VORBIS,
257        sample_rate=44100,
258        bit_depth=16,
259        channels=2,
260        bit_rate=320,
261    )
262    decoded_format = AudioFormat(
263        content_type=ContentType.PCM_S16LE,
264        codec_type=ContentType.PCM_S16LE,
265        sample_rate=44100,
266        bit_depth=16,
267        channels=2,
268    )
269    streamdetails = _make_streamdetails(
270        audio_format=source_format, decoded_audio_format=decoded_format
271    )
272
273    audio = _make_audio_controller()
274    await _drain(audio.get_media_stream(streamdetails, _make_pcm_format()))
275
276    assert patch_ffmpeg.last_instance is not None
277    assert patch_ffmpeg.last_instance.input_format is decoded_format
278
279
280@pytest.mark.asyncio
281async def test_get_media_stream_falls_back_to_audio_format(
282    patch_ffmpeg: type[_FakeFFMpeg],
283) -> None:
284    """When decoded_audio_format is not set, ffmpeg receives audio_format as input_format."""
285    source_format = AudioFormat(
286        content_type=ContentType.FLAC,
287        codec_type=ContentType.FLAC,
288        sample_rate=44100,
289        bit_depth=16,
290        channels=2,
291    )
292    streamdetails = _make_streamdetails(audio_format=source_format)
293
294    audio = _make_audio_controller()
295    await _drain(audio.get_media_stream(streamdetails, _make_pcm_format()))
296
297    assert patch_ffmpeg.last_instance is not None
298    assert patch_ffmpeg.last_instance.input_format is source_format
299
300
301@pytest.mark.asyncio
302@pytest.mark.usefixtures("patch_ffmpeg")
303async def test_get_media_stream_does_not_overwrite_source_codec_when_decoded_format_set() -> None:
304    """audio_format.codec_type stays authoritative when decoded_audio_format is set."""
305    source_format = AudioFormat(
306        content_type=ContentType.OGG,
307        codec_type=ContentType.VORBIS,
308        sample_rate=44100,
309        bit_depth=16,
310        channels=2,
311        bit_rate=320,
312    )
313    decoded_format = AudioFormat(
314        content_type=ContentType.PCM_S16LE,
315        codec_type=ContentType.PCM_S16LE,
316        sample_rate=44100,
317        bit_depth=16,
318        channels=2,
319    )
320    streamdetails = _make_streamdetails(
321        audio_format=source_format, decoded_audio_format=decoded_format
322    )
323
324    audio = _make_audio_controller()
325    await _drain(audio.get_media_stream(streamdetails, _make_pcm_format()))
326
327    assert streamdetails.audio_format.codec_type is ContentType.VORBIS
328
329
330@pytest.mark.asyncio
331@pytest.mark.usefixtures("patch_ffmpeg")
332async def test_get_media_stream_writes_back_codec_when_no_decoded_format() -> None:
333    """Without decoded_audio_format, ffmpeg's probed codec_type is written back."""
334    source_format = AudioFormat(
335        content_type=ContentType.FLAC,
336        # Start with UNKNOWN so we can see the post-probe writeback take effect.
337        codec_type=ContentType.UNKNOWN,
338        sample_rate=44100,
339        bit_depth=16,
340        channels=2,
341    )
342    streamdetails = _make_streamdetails(audio_format=source_format)
343
344    audio = _make_audio_controller()
345    await _drain(audio.get_media_stream(streamdetails, _make_pcm_format()))
346
347    # _FakeFFMpeg's start() mutates input_format.codec_type to FLAC; with no
348    # decoded format that AudioFormat is the same object as streamdetails.audio_format,
349    # so the controller's writeback path is exercised end-to-end.
350    assert streamdetails.audio_format.codec_type is ContentType.FLAC
351
352
353@pytest.mark.asyncio
354async def test_get_media_stream_stores_measured_duration_for_full_playthrough(
355    monkeypatch: pytest.MonkeyPatch,
356    patch_two_minute_ffmpeg: type[_TwoMinuteFFMpeg],
357) -> None:
358    """A multi-file item streamed from the start gets its measured duration stored."""
359    streamdetails = _multi_part_streamdetails()
360    audio = _make_audio_controller()
361    fake_stream, _ = _recording_multi_file_stream()
362    monkeypatch.setattr(audio, "get_multi_file_stream", fake_stream)
363
364    await _drain(audio.get_media_stream(streamdetails, _make_pcm_format()))
365
366    assert streamdetails.duration == patch_two_minute_ffmpeg.seconds_emitted
367
368
369@pytest.mark.asyncio
370async def test_get_media_stream_keeps_duration_when_multi_file_seek_is_delegated(
371    monkeypatch: pytest.MonkeyPatch,
372    patch_two_minute_ffmpeg: type[_TwoMinuteFFMpeg],
373) -> None:
374    """Resuming a multi-file audiobook must not shrink its duration to the remainder."""
375    streamdetails = _multi_part_streamdetails()
376    audio = _make_audio_controller()
377    fake_stream, received_seeks = _recording_multi_file_stream()
378    monkeypatch.setattr(audio, "get_multi_file_stream", fake_stream)
379
380    await _drain(audio.get_media_stream(streamdetails, _make_pcm_format(), seek_position=1800))
381
382    # the concat stream consumes the seek itself, which clears the local seek
383    # position before the duration writeback runs at the end of the stream
384    assert received_seeks == [1800]
385    assert patch_two_minute_ffmpeg.last_instance is not None
386    assert "-ss" not in (patch_two_minute_ffmpeg.last_instance.extra_input_args or [])
387    assert streamdetails.duration == 3600
388
389
390@pytest.mark.asyncio
391async def test_get_media_stream_keeps_duration_when_provider_seek_is_delegated(
392    patch_two_minute_ffmpeg: type[_TwoMinuteFFMpeg],
393) -> None:
394    """A seekable provider stream must not shrink its duration to the remainder either."""
395    streamdetails = StreamDetails(
396        provider="test_provider",
397        item_id="track-1",
398        audio_format=AudioFormat(content_type=ContentType.OGG),
399        media_type=MediaType.TRACK,
400        stream_type=StreamType.CUSTOM,
401        duration=240,
402        can_seek=True,
403        allow_seek=True,
404    )
405    audio = _make_audio_controller()
406    provider = cast("MagicMock", audio.mass).get_provider.return_value
407
408    await _drain(audio.get_media_stream(streamdetails, _make_pcm_format(), seek_position=90))
409
410    # a provider that can seek receives the position and the local one is cleared,
411    # so only the remaining audio reaches ffmpeg
412    provider.get_audio_stream.assert_called_once_with(streamdetails, seek_position=90)
413    assert patch_two_minute_ffmpeg.last_instance is not None
414    assert "-ss" not in (patch_two_minute_ffmpeg.last_instance.extra_input_args or [])
415    assert streamdetails.duration == 240
416
417
418@pytest.mark.asyncio
419async def test_get_media_stream_keeps_caller_extra_input_args_intact(
420    patch_ffmpeg: type[_FakeFFMpeg],
421) -> None:
422    """Per-call input args must not leak back onto the caller's StreamDetails."""
423    streamdetails = _seekable_streamdetails()
424    audio = _make_audio_controller()
425
426    await _drain(audio.get_media_stream(streamdetails, _make_pcm_format(), seek_position=30))
427
428    assert patch_ffmpeg.last_instance is not None
429    assert patch_ffmpeg.last_instance.extra_input_args == [*_PROVIDER_INPUT_ARGS, "-ss", "30"]
430    assert streamdetails.extra_input_args == [*_PROVIDER_INPUT_ARGS]
431
432    # StreamDetails are cached on the queue item and reach this method again on a
433    # retry, another seek or from the background analyzer: every call must build its
434    # args from the provider's list alone instead of stacking onto the previous call's.
435    await _drain(audio.get_media_stream(streamdetails, _make_pcm_format(), seek_position=600))
436
437    assert patch_ffmpeg.last_instance.extra_input_args == [*_PROVIDER_INPUT_ARGS, "-ss", "600"]
438    assert streamdetails.extra_input_args == [*_PROVIDER_INPUT_ARGS]
439