/
/
/
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
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 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
73def _make_audio_controller() -> StreamsAudio:
74 """Build a StreamsAudio with just enough mass scaffolding to run get_media_stream."""
75 audio = StreamsAudio(MagicMock())
76 audio.mass.loop = MagicMock()
77 audio.mass.loop.time = MagicMock(return_value=0.0)
78 return audio
79
80
81def _make_pcm_format() -> AudioFormat:
82 return AudioFormat(
83 content_type=ContentType.PCM_S16LE,
84 codec_type=ContentType.PCM_S16LE,
85 sample_rate=44100,
86 bit_depth=16,
87 channels=2,
88 )
89
90
91def _make_streamdetails(
92 *,
93 audio_format: AudioFormat,
94 decoded_audio_format: AudioFormat | None = None,
95) -> StreamDetails:
96 return StreamDetails(
97 provider="test_provider",
98 item_id="main",
99 audio_format=audio_format,
100 decoded_audio_format=decoded_audio_format,
101 media_type=MediaType.AUDIO_SOURCE,
102 stream_type=StreamType.NAMED_PIPE,
103 path="/tmp/fake-fifo", # noqa: S108
104 )
105
106
107def _seekable_streamdetails() -> StreamDetails:
108 """Build seekable StreamDetails carrying provider-supplied ffmpeg input args."""
109 return StreamDetails(
110 provider="test_provider",
111 item_id="episode-1",
112 audio_format=AudioFormat(content_type=ContentType.MP3),
113 media_type=MediaType.PODCAST_EPISODE,
114 stream_type=StreamType.HTTP,
115 path="http://test.invalid/episode-1.mp3",
116 duration=3600,
117 can_seek=True,
118 allow_seek=True,
119 extra_input_args=[*_PROVIDER_INPUT_ARGS],
120 )
121
122
123async def _drain(gen: AsyncGenerator[bytes]) -> None:
124 async for _ in gen:
125 pass
126
127
128class _StallingFFMpeg(_FakeFFMpeg):
129 """FFMpeg double whose read never produces a chunk (frozen source)."""
130
131 async def iter_chunked(self, _chunk_size: int) -> AsyncGenerator[bytes]:
132 await asyncio.Event().wait() # blocks until the watchdog cancels the read
133 yield b"" # unreachable
134
135
136class _SlowConsumerFFMpeg(_FakeFFMpeg):
137 """FFMpeg double that hands over chunks instantly when asked."""
138
139 async def iter_chunked(self, _chunk_size: int) -> AsyncGenerator[bytes]:
140 for _ in range(3):
141 yield b"\x00\x01" * 256
142
143
144def _flac_streamdetails() -> StreamDetails:
145 return _make_streamdetails(
146 audio_format=AudioFormat(
147 content_type=ContentType.FLAC,
148 codec_type=ContentType.FLAC,
149 sample_rate=44100,
150 bit_depth=16,
151 channels=2,
152 )
153 )
154
155
156@pytest.mark.asyncio
157async def test_get_media_stream_raises_when_source_stalls(
158 monkeypatch: pytest.MonkeyPatch,
159) -> None:
160 """A source that stops producing audio is surfaced as an AudioError."""
161 monkeypatch.setattr(audio_mod, "FFMpeg", _StallingFFMpeg)
162 monkeypatch.setattr(audio_mod, "STREAM_START_TIMEOUT", 0.1)
163 monkeypatch.setattr(audio_mod, "STREAM_STALL_TIMEOUT", 0.1)
164
165 audio = _make_audio_controller()
166 with pytest.raises(AudioError):
167 await _drain(audio.get_media_stream(_flac_streamdetails(), _make_pcm_format()))
168
169
170@pytest.mark.asyncio
171async def test_get_media_stream_does_not_stall_on_slow_consumer(
172 monkeypatch: pytest.MonkeyPatch,
173) -> None:
174 """A consumer slower than the stall timeout must not trip the watchdog."""
175 monkeypatch.setattr(audio_mod, "FFMpeg", _SlowConsumerFFMpeg)
176 monkeypatch.setattr(audio_mod, "STREAM_START_TIMEOUT", 0.1)
177 monkeypatch.setattr(audio_mod, "STREAM_STALL_TIMEOUT", 0.1)
178
179 audio = _make_audio_controller()
180 chunks = 0
181 async for _ in audio.get_media_stream(_flac_streamdetails(), _make_pcm_format()):
182 chunks += 1
183 await asyncio.sleep(0.3) # downstream waits far longer than the stall timeout
184 assert chunks == 3
185
186
187@pytest.mark.asyncio
188async def test_get_media_stream_prefers_decoded_audio_format(
189 patch_ffmpeg: type[_FakeFFMpeg],
190) -> None:
191 """When decoded_audio_format is set, ffmpeg receives that as input_format."""
192 source_format = AudioFormat(
193 content_type=ContentType.OGG,
194 codec_type=ContentType.VORBIS,
195 sample_rate=44100,
196 bit_depth=16,
197 channels=2,
198 bit_rate=320,
199 )
200 decoded_format = AudioFormat(
201 content_type=ContentType.PCM_S16LE,
202 codec_type=ContentType.PCM_S16LE,
203 sample_rate=44100,
204 bit_depth=16,
205 channels=2,
206 )
207 streamdetails = _make_streamdetails(
208 audio_format=source_format, decoded_audio_format=decoded_format
209 )
210
211 audio = _make_audio_controller()
212 await _drain(audio.get_media_stream(streamdetails, _make_pcm_format()))
213
214 assert patch_ffmpeg.last_instance is not None
215 assert patch_ffmpeg.last_instance.input_format is decoded_format
216
217
218@pytest.mark.asyncio
219async def test_get_media_stream_falls_back_to_audio_format(
220 patch_ffmpeg: type[_FakeFFMpeg],
221) -> None:
222 """When decoded_audio_format is not set, ffmpeg receives audio_format as input_format."""
223 source_format = AudioFormat(
224 content_type=ContentType.FLAC,
225 codec_type=ContentType.FLAC,
226 sample_rate=44100,
227 bit_depth=16,
228 channels=2,
229 )
230 streamdetails = _make_streamdetails(audio_format=source_format)
231
232 audio = _make_audio_controller()
233 await _drain(audio.get_media_stream(streamdetails, _make_pcm_format()))
234
235 assert patch_ffmpeg.last_instance is not None
236 assert patch_ffmpeg.last_instance.input_format is source_format
237
238
239@pytest.mark.asyncio
240@pytest.mark.usefixtures("patch_ffmpeg")
241async def test_get_media_stream_does_not_overwrite_source_codec_when_decoded_format_set() -> None:
242 """audio_format.codec_type stays authoritative when decoded_audio_format is set."""
243 source_format = AudioFormat(
244 content_type=ContentType.OGG,
245 codec_type=ContentType.VORBIS,
246 sample_rate=44100,
247 bit_depth=16,
248 channels=2,
249 bit_rate=320,
250 )
251 decoded_format = AudioFormat(
252 content_type=ContentType.PCM_S16LE,
253 codec_type=ContentType.PCM_S16LE,
254 sample_rate=44100,
255 bit_depth=16,
256 channels=2,
257 )
258 streamdetails = _make_streamdetails(
259 audio_format=source_format, decoded_audio_format=decoded_format
260 )
261
262 audio = _make_audio_controller()
263 await _drain(audio.get_media_stream(streamdetails, _make_pcm_format()))
264
265 assert streamdetails.audio_format.codec_type is ContentType.VORBIS
266
267
268@pytest.mark.asyncio
269@pytest.mark.usefixtures("patch_ffmpeg")
270async def test_get_media_stream_writes_back_codec_when_no_decoded_format() -> None:
271 """Without decoded_audio_format, ffmpeg's probed codec_type is written back."""
272 source_format = AudioFormat(
273 content_type=ContentType.FLAC,
274 # Start with UNKNOWN so we can see the post-probe writeback take effect.
275 codec_type=ContentType.UNKNOWN,
276 sample_rate=44100,
277 bit_depth=16,
278 channels=2,
279 )
280 streamdetails = _make_streamdetails(audio_format=source_format)
281
282 audio = _make_audio_controller()
283 await _drain(audio.get_media_stream(streamdetails, _make_pcm_format()))
284
285 # _FakeFFMpeg's start() mutates input_format.codec_type to FLAC; with no
286 # decoded format that AudioFormat is the same object as streamdetails.audio_format,
287 # so the controller's writeback path is exercised end-to-end.
288 assert streamdetails.audio_format.codec_type is ContentType.FLAC
289
290
291@pytest.mark.asyncio
292async def test_get_media_stream_keeps_caller_extra_input_args_intact(
293 patch_ffmpeg: type[_FakeFFMpeg],
294) -> None:
295 """Per-call input args must not leak back onto the caller's StreamDetails."""
296 streamdetails = _seekable_streamdetails()
297 audio = _make_audio_controller()
298
299 await _drain(audio.get_media_stream(streamdetails, _make_pcm_format(), seek_position=30))
300
301 assert patch_ffmpeg.last_instance is not None
302 assert patch_ffmpeg.last_instance.extra_input_args == [*_PROVIDER_INPUT_ARGS, "-ss", "30"]
303 assert streamdetails.extra_input_args == [*_PROVIDER_INPUT_ARGS]
304
305 # StreamDetails are cached on the queue item and reach this method again on a
306 # retry, another seek or from the background analyzer: every call must build its
307 # args from the provider's list alone instead of stacking onto the previous call's.
308 await _drain(audio.get_media_stream(streamdetails, _make_pcm_format(), seek_position=600))
309
310 assert patch_ffmpeg.last_instance.extra_input_args == [*_PROVIDER_INPUT_ARGS, "-ss", "600"]
311 assert streamdetails.extra_input_args == [*_PROVIDER_INPUT_ARGS]
312