/
/
/
1"""Tests for the ffmpeg helper module."""
2
3from __future__ import annotations
4
5import asyncio
6import subprocess
7from array import array
8from collections.abc import AsyncGenerator, Sequence
9from math import sqrt
10from pathlib import Path
11
12import pytest
13from music_assistant_models.enums import ContentType
14from music_assistant_models.errors import AudioError
15from music_assistant_models.helpers import get_global_cache_value, set_global_cache_values
16from music_assistant_models.media_items import AudioFormat
17
18from music_assistant.helpers.dsp import ComplexFilter, ComplexFilterInput
19from music_assistant.helpers.ffmpeg import (
20 _INPUT_READ_ARGS,
21 CACHE_ATTR_HLS_CMAF_BLOCKED,
22 FFMpeg,
23 FFMpegStreamInfo,
24 _build_filtergraph_args,
25 _build_overlay_mixer,
26 _get_overlay_volume_filter,
27 check_ffmpeg_version,
28 get_ffmpeg_args,
29 get_ffmpeg_hls_cmaf_input_args,
30 get_ffmpeg_overlay_stream,
31 get_ffmpeg_stream,
32 parse_ffmpeg_duration,
33 parse_ffmpeg_stream_info,
34)
35
36
37def test_get_ffmpeg_args_does_not_mutate_filters() -> None:
38 """Automatic resampling must not alter a caller-owned filter plan."""
39 input_format = AudioFormat(
40 content_type=ContentType.PCM_F32LE,
41 sample_rate=96000,
42 bit_depth=32,
43 channels=2,
44 )
45 output_format = AudioFormat(
46 content_type=ContentType.FLAC,
47 sample_rate=48000,
48 bit_depth=16,
49 channels=2,
50 )
51 filter_params = ["volume=-1dB"]
52
53 get_ffmpeg_args(input_format, output_format, filter_params)
54
55 assert filter_params == ["volume=-1dB"]
56
57
58def test_get_ffmpeg_args_downmixes_multichannel_for_single_channel_output() -> None:
59 """A surround source is folded to stereo before the output is narrowed to one channel."""
60 input_format = AudioFormat(
61 content_type=ContentType.PCM_F32LE,
62 sample_rate=48000,
63 bit_depth=32,
64 channels=6,
65 )
66 output_format = AudioFormat(
67 content_type=ContentType.FLAC,
68 sample_rate=48000,
69 bit_depth=16,
70 channels=1,
71 )
72
73 args = get_ffmpeg_args(input_format, output_format, ["pan=mono|c0=0.5*FL+0.5*FR"])
74
75 filter_graph = args[args.index("-af") + 1]
76 assert filter_graph.index("aformat=channel_layouts=stereo") < filter_graph.index(
77 "pan=mono|c0=0.5*FL+0.5*FR"
78 )
79
80
81def _split_at_input(args: list[str]) -> tuple[list[str], list[str]]:
82 """Split generated ffmpeg args into the part describing the input and the output."""
83 idx = args.index("-i")
84 return args[:idx], args[idx + 2 :]
85
86
87@pytest.mark.parametrize(
88 ("channels", "expected_layout"),
89 [(1, "mono"), (2, "stereo")],
90)
91def test_get_ffmpeg_args_names_layout_up_to_stereo(channels: int, expected_layout: str) -> None:
92 """Mono and stereo PCM are described by both their channel count and their layout."""
93 fmt = AudioFormat(
94 content_type=ContentType.PCM_S24LE,
95 sample_rate=48000,
96 bit_depth=24,
97 channels=channels,
98 )
99
100 input_args, output_args = _split_at_input(get_ffmpeg_args(fmt, fmt, []))
101
102 for part in (input_args, output_args):
103 assert part[part.index("-ac") + 1] == str(channels)
104 assert part[part.index("-channel_layout") + 1] == expected_layout
105
106
107def test_get_ffmpeg_args_omits_layout_above_stereo() -> None:
108 """Surround PCM is described by its channel count alone, never as a stereo layout."""
109 fmt = AudioFormat(
110 content_type=ContentType.PCM_S24LE,
111 sample_rate=48000,
112 bit_depth=24,
113 channels=6,
114 )
115
116 input_args, output_args = _split_at_input(get_ffmpeg_args(fmt, fmt, []))
117
118 for part in (input_args, output_args):
119 assert part[part.index("-ac") + 1] == "6"
120 assert "-channel_layout" not in part
121
122
123def test_multichannel_pcm_folds_down_without_stretching(tmp_path: Path) -> None:
124 """Raw surround PCM keeps its real width and length, and its rear channels survive."""
125 source = tmp_path / "surround.pcm"
126 out = tmp_path / "out.flac"
127 # only the rear channels carry a tone: a downmix that drops them yields silence
128 subprocess.run( # noqa: S603
129 [ # noqa: S607
130 "ffmpeg",
131 "-y",
132 "-f",
133 "lavfi",
134 "-i",
135 "sine=frequency=1000:duration=1:sample_rate=48000",
136 "-af",
137 "pan=5.1|BL=c0|BR=c0",
138 "-f",
139 "s16le",
140 str(source),
141 ],
142 check=True,
143 capture_output=True,
144 )
145 pcm_format = AudioFormat(
146 content_type=ContentType.PCM_S16LE,
147 sample_rate=48000,
148 bit_depth=16,
149 channels=6,
150 )
151 output_format = AudioFormat(
152 content_type=ContentType.FLAC,
153 sample_rate=48000,
154 bit_depth=16,
155 channels=2,
156 )
157 args = get_ffmpeg_args(
158 pcm_format, output_format, [], input_path=str(source), output_path=str(out)
159 )
160
161 result = subprocess.run([*args, "-y"], capture_output=True, text=True, check=False) # noqa: S603
162 assert result.returncode == 0, result.stderr
163
164 duration = subprocess.run( # noqa: S603
165 [ # noqa: S607
166 "ffprobe",
167 "-v",
168 "error",
169 "-show_entries",
170 "format=duration",
171 "-of",
172 "csv=p=0",
173 str(out),
174 ],
175 capture_output=True,
176 text=True,
177 check=True,
178 ).stdout
179 assert float(duration) == pytest.approx(1.0, abs=0.05)
180 assert _rms_db(out) > -40
181
182
183def _output_args(args: list[str]) -> list[str]:
184 """Return the output section of an ffmpeg command line (everything past the input path)."""
185 return args[args.index("-i") + 2 :]
186
187
188@pytest.mark.parametrize(
189 ("content_type", "encoder_args"),
190 [
191 (ContentType.AAC, ["-f", "adts", "-c:a", "aac", "-b:a", "256k"]),
192 (ContentType.MP3, ["-f", "mp3", "-b:a", "320k"]),
193 (ContentType.WAV, ["-ar", "44100", "-acodec", "pcm_s16le", "-f", "wav"]),
194 (
195 ContentType.FLAC,
196 ["-sample_fmt", "s16", "-ar", "44100", "-f", "flac", "-compression_level", "0"],
197 ),
198 ],
199)
200def test_get_ffmpeg_args_encoded_output_declares_channels(
201 content_type: ContentType, encoder_args: list[str]
202) -> None:
203 """Every encoded output format is handed the requested channel count."""
204 input_format = AudioFormat(
205 content_type=ContentType.PCM_S16LE, sample_rate=44100, bit_depth=16, channels=2
206 )
207 output_format = AudioFormat(content_type=content_type, sample_rate=44100, bit_depth=16)
208
209 args = get_ffmpeg_args(input_format, output_format, [])
210
211 assert _output_args(args) == ["-ac", "2", "-channel_layout", "stereo", *encoder_args, "-"]
212
213
214def test_get_ffmpeg_args_single_channel_output_declares_mono() -> None:
215 """A one channel target is declared as mono rather than stereo."""
216 fmt = AudioFormat(
217 content_type=ContentType.PCM_S16LE, sample_rate=44100, bit_depth=16, channels=1
218 )
219 output_format = AudioFormat(
220 content_type=ContentType.WAV, sample_rate=44100, bit_depth=16, channels=1
221 )
222
223 args = get_ffmpeg_args(fmt, output_format, [])
224
225 assert _output_args(args) == [
226 "-ac",
227 "1",
228 "-channel_layout",
229 "mono",
230 "-ar",
231 "44100",
232 "-acodec",
233 "pcm_s16le",
234 "-f",
235 "wav",
236 "-",
237 ]
238
239
240def test_get_ffmpeg_args_pcm_input_ignores_stale_source_codec() -> None:
241 """A PCM input keeps its raw decoder even when a source codec is still attached."""
242 # background analysis builds its PCM format from the track's own AudioFormat,
243 # which swaps content_type but carries the encoded codec_type over
244 input_format = AudioFormat(
245 content_type=ContentType.PCM_S16LE,
246 codec_type=ContentType.FLAC,
247 sample_rate=96000,
248 bit_depth=16,
249 channels=2,
250 )
251
252 args = get_ffmpeg_args(input_format, input_format, [], input_path="-", output_path="NULL")
253 input_args = args[: args.index("-i")]
254
255 assert input_args.count("-acodec") == 1
256 assert input_args[input_args.index("-acodec") + 1] == "pcm_s16le"
257 assert "flac" not in input_args
258
259
260def test_get_ffmpeg_args_encoded_input_still_declares_source_codec() -> None:
261 """A non-PCM input keeps declaring the codec ffmpeg should decode with."""
262 input_format = AudioFormat(
263 content_type=ContentType.FLAC,
264 codec_type=ContentType.FLAC,
265 sample_rate=96000,
266 bit_depth=16,
267 channels=2,
268 )
269 output_format = AudioFormat(
270 content_type=ContentType.PCM_S16LE, sample_rate=96000, bit_depth=16, channels=2
271 )
272
273 args = get_ffmpeg_args(input_format, output_format, [], input_path="-")
274 input_args = args[: args.index("-i")]
275
276 assert input_args.count("-acodec") == 1
277 assert input_args[input_args.index("-acodec") + 1] == "flac"
278
279
280@pytest.mark.parametrize(
281 ("output_path", "output_content_type", "expected"),
282 [
283 ("NULL", ContentType.FLAC, ["-f", "null", "-"]),
284 ("-", ContentType.NUT, ["-vn", "-dn", "-sn", "-acodec", "copy", "-f", "nut", "-"]),
285 ],
286)
287def test_get_ffmpeg_args_passthrough_sinks_omit_channels(
288 output_path: str, output_content_type: ContentType, expected: list[str]
289) -> None:
290 """The analysis sink and the cache passthrough declare no channel count of their own."""
291 input_format = AudioFormat(
292 content_type=ContentType.PCM_S16LE, sample_rate=44100, bit_depth=16, channels=1
293 )
294 output_format = AudioFormat(
295 content_type=output_content_type, sample_rate=44100, bit_depth=16, channels=1
296 )
297
298 args = get_ffmpeg_args(input_format, output_format, [], output_path=output_path)
299
300 assert _output_args(args) == expected
301
302
303def test_get_ffmpeg_args_duplicates_mono_source_for_stereo_output() -> None:
304 """A mono source is widened by duplication, ahead of the caller's own filters."""
305 input_format = AudioFormat(
306 content_type=ContentType.PCM_S16LE, sample_rate=44100, bit_depth=16, channels=1
307 )
308 output_format = AudioFormat(
309 content_type=ContentType.FLAC, sample_rate=44100, bit_depth=16, channels=2
310 )
311
312 args = get_ffmpeg_args(input_format, output_format, ["volume=-1dB"])
313
314 assert args[args.index("-af") + 1] == "pan=stereo|c0=c0|c1=c0,volume=-1dB"
315
316
317def test_get_ffmpeg_args_mono_source_to_mono_output_is_not_widened() -> None:
318 """A mono source kept at one channel needs no channel filter at all."""
319 fmt = AudioFormat(
320 content_type=ContentType.PCM_S16LE, sample_rate=44100, bit_depth=16, channels=1
321 )
322 output_format = AudioFormat(
323 content_type=ContentType.FLAC, sample_rate=44100, bit_depth=16, channels=1
324 )
325
326 args = get_ffmpeg_args(fmt, output_format, [])
327
328 assert "-af" not in args
329
330
331# -- parse_ffmpeg_stream_info --
332
333
334def test_parse_stream_info_mp3() -> None:
335 """Lossy MP3 line yields codec/sample rate/bit rate, but no bit depth."""
336 line = "Stream #0:0: Audio: mp3, 44100 Hz, stereo, fltp, 320 kb/s"
337 info = parse_ffmpeg_stream_info(line)
338 assert info == FFMpegStreamInfo(
339 codec=ContentType.MP3,
340 sample_rate=44100,
341 bit_depth=None,
342 bit_rate=320,
343 )
344
345
346def test_parse_stream_info_aac_with_profile_and_language() -> None:
347 """AAC line with profile annotation and language tag is parsed correctly."""
348 line = "Stream #0:0(eng): Audio: aac (LC) (mp4a / 0x6134706D), 44100 Hz, stereo, fltp, 254 kb/s"
349 info = parse_ffmpeg_stream_info(line)
350 assert info is not None
351 assert info.codec == ContentType.AAC
352 assert info.sample_rate == 44100
353 assert info.bit_rate == 254
354 assert info.bit_depth is None
355
356
357def test_parse_stream_info_flac_16bit() -> None:
358 """16-bit FLAC: bit depth is inferred from the s16 sample format token."""
359 line = "Stream #0:0: Audio: flac, 44100 Hz, stereo, s16, 1024 kb/s"
360 info = parse_ffmpeg_stream_info(line)
361 assert info == FFMpegStreamInfo(
362 codec=ContentType.FLAC,
363 sample_rate=44100,
364 bit_depth=16,
365 bit_rate=1024,
366 )
367
368
369def test_parse_stream_info_flac_24bit_in_s32() -> None:
370 """24-bit FLAC is stored in s32; the explicit "(24 bit)" annotation wins."""
371 line = "Stream #0:0: Audio: flac, 96000 Hz, stereo, s32 (24 bit)"
372 info = parse_ffmpeg_stream_info(line)
373 assert info == FFMpegStreamInfo(
374 codec=ContentType.FLAC,
375 sample_rate=96000,
376 bit_depth=24,
377 bit_rate=None,
378 )
379
380
381def test_parse_stream_info_flac_24bit_hires_with_bitrate() -> None:
382 """High-resolution 24-bit FLAC at 192k with reported bit rate."""
383 line = "Stream #0:0: Audio: flac, 192000 Hz, stereo, s32 (24 bit), 5644 kb/s"
384 info = parse_ffmpeg_stream_info(line)
385 assert info == FFMpegStreamInfo(
386 codec=ContentType.FLAC,
387 sample_rate=192000,
388 bit_depth=24,
389 bit_rate=5644,
390 )
391
392
393def test_parse_stream_info_pcm_s16le() -> None:
394 """PCM stream reports codec via try_parse, sample format gives bit depth."""
395 line = "Stream #0:0: Audio: pcm_s16le, 44100 Hz, stereo, s16, 1411 kb/s"
396 info = parse_ffmpeg_stream_info(line)
397 assert info == FFMpegStreamInfo(
398 codec=ContentType.PCM_S16LE,
399 sample_rate=44100,
400 bit_depth=16,
401 bit_rate=1411,
402 )
403
404
405def test_parse_stream_info_opus_without_bitrate() -> None:
406 """Opus often omits bit rate; we still get codec and sample rate."""
407 line = "Stream #0:0: Audio: opus, 48000 Hz, stereo, fltp"
408 info = parse_ffmpeg_stream_info(line)
409 assert info == FFMpegStreamInfo(
410 codec=ContentType.OPUS,
411 sample_rate=48000,
412 bit_depth=None,
413 bit_rate=None,
414 )
415
416
417def test_parse_stream_info_alac_planar() -> None:
418 """ALAC reported with s16p (planar) sample format still yields 16-bit depth."""
419 line = "Stream #0:0: Audio: alac (alac / 0x63616C61), 44100 Hz, stereo, s16p"
420 info = parse_ffmpeg_stream_info(line)
421 assert info is not None
422 assert info.codec == ContentType.ALAC
423 assert info.sample_rate == 44100
424 assert info.bit_depth == 16
425
426
427def test_parse_stream_info_returns_none_for_non_stream_line() -> None:
428 """Non-stream log lines must return None."""
429 assert parse_ffmpeg_stream_info("Duration: 00:03:25.78, start: 0.000000") is None
430 assert parse_ffmpeg_stream_info("[error] Invalid data found") is None
431 assert parse_ffmpeg_stream_info("") is None
432
433
434def test_parse_stream_info_ignores_video_stream() -> None:
435 """Video stream lines must not be misparsed as audio."""
436 line = "Stream #0:0: Video: h264 (High), yuv420p, 1920x1080, 5000 kb/s, 25 fps"
437 assert parse_ffmpeg_stream_info(line) is None
438
439
440def test_parse_stream_info_unknown_codec_still_yields_other_fields() -> None:
441 """Unrecognised codec token returns UNKNOWN but sample rate / bit rate are still parsed."""
442 line = "Stream #0:0: Audio: somenewcodec, 48000 Hz, stereo, 192 kb/s"
443 info = parse_ffmpeg_stream_info(line)
444 assert info is not None
445 assert info.codec == ContentType.UNKNOWN
446 assert info.sample_rate == 48000
447 assert info.bit_rate == 192
448 assert info.bit_depth is None
449
450
451# -- parse_ffmpeg_duration --
452
453
454def test_parse_duration_typical() -> None:
455 """Typical 'Duration: HH:MM:SS.ms' line yields total seconds (floor)."""
456 line = "Duration: 00:03:25.78, start: 0.000000, bitrate: 320 kb/s"
457 assert parse_ffmpeg_duration(line) == 3 * 60 + 25
458
459
460def test_parse_duration_one_hour() -> None:
461 """Hours component is honoured."""
462 assert parse_ffmpeg_duration("Duration: 01:00:00.00, bitrate: 128 kb/s") == 3600
463
464
465def test_parse_duration_under_one_second() -> None:
466 """Sub-second durations round down to 0."""
467 assert parse_ffmpeg_duration("Duration: 00:00:00.50, bitrate: 128 kb/s") == 0
468
469
470def test_parse_duration_na_returns_none() -> None:
471 """Live streams report 'Duration: N/A' — must not match."""
472 assert parse_ffmpeg_duration("Duration: N/A, start: 0.000000, bitrate: N/A") is None
473
474
475def test_parse_duration_unrelated_line_returns_none() -> None:
476 """Random log lines must return None."""
477 assert parse_ffmpeg_duration("Stream #0:0: Audio: mp3, 44100 Hz") is None
478 assert parse_ffmpeg_duration("") is None
479
480
481# -- get_ffmpeg_overlay_stream (end-to-end with a real ffmpeg process) --
482
483_PCM_FORMAT = AudioFormat(
484 content_type=ContentType.PCM_S16LE, sample_rate=44100, bit_depth=16, channels=2
485)
486_BYTES_PER_SECOND = _PCM_FORMAT.pcm_sample_size # 1 second of PCM audio
487
488
489@pytest.fixture
490def overlay_file(tmp_path: Path) -> Path:
491 """Generate a 1 second mono sine-tone wav file to use as overlay source."""
492 overlay_path = tmp_path / "overlay.wav"
493 subprocess.run( # noqa: S603
494 ["ffmpeg", "-f", "lavfi", "-i", "sine=frequency=440:duration=1", str(overlay_path)], # noqa: S607
495 check=True,
496 capture_output=True,
497 )
498 return overlay_path
499
500
501@pytest.fixture
502def overlay_file_stereo(tmp_path: Path) -> Path:
503 """Generate a stereo overlay wav carrying the same tone as ``overlay_file`` on both channels."""
504 overlay_path = tmp_path / "overlay_stereo.wav"
505 subprocess.run( # noqa: S603
506 [ # noqa: S607
507 "ffmpeg",
508 "-f",
509 "lavfi",
510 "-i",
511 "sine=frequency=440:duration=1",
512 "-af",
513 "pan=stereo|c0=c0|c1=c0",
514 str(overlay_path),
515 ],
516 check=True,
517 capture_output=True,
518 )
519 return overlay_path
520
521
522@pytest.fixture
523def overlay_file_wide_stereo(tmp_path: Path) -> Path:
524 """Generate a 1 second stereo overlay wav with the tone on the left channel only."""
525 overlay_path = tmp_path / "overlay_wide.wav"
526 subprocess.run( # noqa: S603
527 [ # noqa: S607
528 "ffmpeg",
529 "-f",
530 "lavfi",
531 "-i",
532 "sine=frequency=440:duration=1",
533 "-af",
534 "pan=stereo|c0=c0",
535 str(overlay_path),
536 ],
537 check=True,
538 capture_output=True,
539 )
540 return overlay_path
541
542
543@pytest.fixture
544def overlay_file_with_silent_intro(tmp_path: Path) -> Path:
545 """Generate a 2 second overlay wav that starts with 1s of silence then a 1s tone."""
546 overlay_path = tmp_path / "overlay_silent_intro.wav"
547 subprocess.run( # noqa: S603
548 [ # noqa: S607
549 "ffmpeg",
550 "-f",
551 "lavfi",
552 "-i",
553 "sine=frequency=440:duration=1",
554 "-af",
555 "adelay=1000:all=1",
556 str(overlay_path),
557 ],
558 check=True,
559 capture_output=True,
560 )
561 return overlay_path
562
563
564async def _silence(seconds: int) -> AsyncGenerator[bytes]:
565 """Yield the given amount of seconds of PCM silence in 1-second chunks."""
566 for _ in range(seconds):
567 yield b"\x00" * _BYTES_PER_SECOND
568
569
570async def _collect_chunks(stream: AsyncGenerator[bytes]) -> list[bytes]:
571 return [chunk async for chunk in stream]
572
573
574@pytest.mark.parametrize("source_error", [RuntimeError("source failed"), BrokenPipeError()])
575async def test_ffmpeg_stream_surfaces_stdin_feeder_error(source_error: Exception) -> None:
576 """An input generator failure is surfaced after FFmpeg emits its buffered output."""
577
578 async def failing_input() -> AsyncGenerator[bytes]:
579 yield b"\x00" * _BYTES_PER_SECOND
580 raise source_error
581
582 with pytest.raises(AudioError, match="Error while feeding audio to FFmpeg") as err:
583 await _collect_chunks(
584 get_ffmpeg_stream(
585 audio_input=failing_input(),
586 input_format=AudioFormat(
587 content_type=ContentType.PCM_S16LE,
588 sample_rate=44100,
589 bit_depth=16,
590 channels=2,
591 ),
592 output_format=_PCM_FORMAT,
593 )
594 )
595
596 assert err.value.__cause__ is source_error
597
598
599async def test_ffmpeg_stream_ignores_cancelled_stdin_feeder() -> None:
600 """A cancelled input generator ends the FFmpeg stream without an error."""
601
602 async def cancelled_input() -> AsyncGenerator[bytes]:
603 yield b"\x00" * _BYTES_PER_SECOND
604 raise asyncio.CancelledError
605
606 chunks = await _collect_chunks(
607 get_ffmpeg_stream(
608 audio_input=cancelled_input(),
609 input_format=AudioFormat(
610 content_type=ContentType.PCM_S16LE,
611 sample_rate=44100,
612 bit_depth=16,
613 channels=2,
614 ),
615 output_format=_PCM_FORMAT,
616 )
617 )
618
619 assert b"".join(chunks) == b"\x00" * _BYTES_PER_SECOND
620
621
622async def test_ffmpeg_stream_ignores_early_stdin_close() -> None:
623 """FFmpeg ending its input early does not report a source failure."""
624 chunks = await _collect_chunks(
625 get_ffmpeg_stream(
626 audio_input=_silence(30),
627 input_format=_PCM_FORMAT,
628 output_format=_PCM_FORMAT,
629 extra_input_args=["-t", "0.1"],
630 )
631 )
632
633 assert chunks
634
635
636async def test_overlay_stream_surfaces_stdin_feeder_error(overlay_file: Path) -> None:
637 """A failure in the main input is surfaced after the mixed output is emitted."""
638
639 async def failing_input() -> AsyncGenerator[bytes]:
640 yield b"\x00" * _BYTES_PER_SECOND
641 raise RuntimeError("source failed")
642
643 with pytest.raises(AudioError, match="Error while feeding audio to FFmpeg") as err:
644 await _collect_chunks(
645 get_ffmpeg_overlay_stream(
646 audio_input=failing_input(),
647 overlay_input=str(overlay_file),
648 pcm_format=_PCM_FORMAT,
649 )
650 )
651
652 assert isinstance(err.value.__cause__, RuntimeError)
653
654
655def _samples(pcm: bytes, channel: int | None = None) -> Sequence[int]:
656 """Return the samples of the given PCM audio, optionally for one channel only."""
657 samples = array("h")
658 samples.frombytes(pcm)
659 return samples if channel is None else samples[channel :: _PCM_FORMAT.channels]
660
661
662def _rms(samples: Sequence[int]) -> float:
663 """Return the RMS level of the given PCM samples."""
664 return sqrt(sum(sample * sample for sample in samples) / len(samples))
665
666
667async def _mix_overlay(overlay_input: Path) -> bytes:
668 """Mix the given overlay source into 1 second of silence and return the result."""
669 return b"".join(
670 await _collect_chunks(
671 get_ffmpeg_overlay_stream(
672 audio_input=_silence(1),
673 overlay_input=str(overlay_input),
674 pcm_format=_PCM_FORMAT,
675 )
676 )
677 )
678
679
680async def test_overlay_stream_mixes_loops_and_preserves_length(overlay_file: Path) -> None:
681 """The overlay is looped and mixed in while length, format and chunking stay intact."""
682 chunks = await _collect_chunks(
683 get_ffmpeg_overlay_stream(
684 audio_input=_silence(3),
685 overlay_input=str(overlay_file),
686 pcm_format=_PCM_FORMAT,
687 chunk_size=_BYTES_PER_SECOND,
688 )
689 )
690 output = b"".join(chunks)
691 # duration=first: output length exactly matches the 3s main input
692 assert len(output) == 3 * _BYTES_PER_SECOND
693 # all chunks except the last are exactly chunk_size
694 assert all(len(chunk) == _BYTES_PER_SECOND for chunk in chunks[:-1])
695 # the main input was pure silence, so any signal proves the overlay was mixed in;
696 # signal in the third second proves the 1s overlay file was looped
697 assert any(output[:_BYTES_PER_SECOND])
698 assert any(output[2 * _BYTES_PER_SECOND :])
699
700
701async def test_overlay_stream_does_not_mutate_pcm_format(overlay_file: Path) -> None:
702 """Mixing an overlay leaves the caller's PCM format unchanged."""
703 pcm_format = AudioFormat(
704 content_type=ContentType.PCM_F32LE,
705 sample_rate=48000,
706 bit_depth=32,
707 channels=2,
708 )
709 original_format = pcm_format.to_dict()
710
711 async def silence() -> AsyncGenerator[bytes]:
712 yield b"\x00" * pcm_format.pcm_sample_size
713
714 await _collect_chunks(
715 get_ffmpeg_overlay_stream(
716 audio_input=silence(),
717 overlay_input=str(overlay_file),
718 pcm_format=pcm_format,
719 )
720 )
721
722 assert pcm_format.to_dict() == original_format
723
724
725async def test_overlay_stream_applies_volume(overlay_file: Path) -> None:
726 """Overlay volume 0% silences the overlay entirely (gain is applied)."""
727 output = b"".join(
728 await _collect_chunks(
729 get_ffmpeg_overlay_stream(
730 audio_input=_silence(1),
731 overlay_input=str(overlay_file),
732 pcm_format=_PCM_FORMAT,
733 overlay_volume=0,
734 )
735 )
736 )
737 assert len(output) == _BYTES_PER_SECOND
738 assert not any(output)
739
740
741async def test_overlay_stream_trims_leading_silence(
742 overlay_file_with_silent_intro: Path,
743) -> None:
744 """A near-silent intro on the overlay source is trimmed so it plays immediately."""
745 output = b"".join(
746 await _collect_chunks(
747 get_ffmpeg_overlay_stream(
748 audio_input=_silence(1),
749 overlay_input=str(overlay_file_with_silent_intro),
750 pcm_format=_PCM_FORMAT,
751 )
752 )
753 )
754 assert len(output) == _BYTES_PER_SECOND
755 # without trimming, the first second would be the overlay's silent intro;
756 # the trim makes the tone play from the start, so the first second has signal
757 assert any(output)
758
759
760async def test_overlay_stream_level_is_independent_of_source_channel_count(
761 overlay_file: Path, overlay_file_stereo: Path
762) -> None:
763 """A mono overlay source mixes in at the same level as an identical stereo one."""
764 mono_level = _rms(_samples(await _mix_overlay(overlay_file)))
765 stereo_level = _rms(_samples(await _mix_overlay(overlay_file_stereo)))
766 assert mono_level > 0
767 # left to FFmpeg, the mono source would be spread at 1/sqrt(2) per channel
768 # and land a factor sqrt(2) (3 dB) below the stereo one
769 assert mono_level == pytest.approx(stereo_level, rel=0.02)
770
771
772async def test_overlay_stream_preserves_stereo_image(overlay_file_wide_stereo: Path) -> None:
773 """A stereo overlay keeps its channels apart instead of being folded to dual mono."""
774 output = await _mix_overlay(overlay_file_wide_stereo)
775 # the source carries the tone on the left only, so a fold would leak it into the right
776 assert _rms(_samples(output, channel=0)) > 0
777 assert not any(_samples(output, channel=1))
778
779
780# -- overlay volume filter --
781
782
783def test_overlay_volume_filter_compensates_only_for_a_stereo_output() -> None:
784 """Only the widening to stereo costs a mono source level, so only there is it scaled up."""
785 stereo_output = _get_overlay_volume_filter(100, 2)
786 assert "nb_channels" in stereo_output
787 # a comma would read as the end of this filter in the graph
788 assert "," not in stereo_output
789 # a mono source keeps its level when it is widened further, or not at all
790 assert _get_overlay_volume_filter(100, 1) == "volume=1.0"
791 assert _get_overlay_volume_filter(60, 6) == "volume=0.6"
792
793
794def test_overlay_mixer_loops_its_source() -> None:
795 """The overlay source is looped for as long as the main stream runs."""
796 (overlay,) = _build_overlay_mixer("/sound.wav", _PCM_FORMAT, 100).inputs
797 # a local file has nothing to reconnect to
798 assert overlay.input_args == ["-stream_loop", "-1"]
799
800
801def test_overlay_mixer_reconnects_for_http_sources() -> None:
802 """An http overlay source additionally gets the reconnect options."""
803 (overlay,) = _build_overlay_mixer("http://host/sound.mp3", _PCM_FORMAT, 100).inputs
804 assert "-reconnect" in overlay.input_args
805 assert overlay.input_args[-2:] == ["-stream_loop", "-1"]
806
807
808def test_overlay_args_probe_the_main_input_and_add_no_filters() -> None:
809 """Both inputs are read under our own limits and no filter is injected on top."""
810 args = get_ffmpeg_args(
811 _PCM_FORMAT, _PCM_FORMAT, [_build_overlay_mixer("/sound.wav", _PCM_FORMAT, 100)]
812 )
813 main_input = args.index("-i")
814 # the limits must reach the main input, which is the first one, and the overlay
815 assert args.index("-probesize") < main_input
816 assert args.count("-probesize") == 2
817 assert args.index("-stream_loop") > main_input
818 # the overlay format matches the output, so nothing gets resampled or reconformed
819 assert "-af" not in args
820 assert args.count("-filter_complex") == 1
821 assert not any(arg.startswith(("pan=", "aresample=resampler=")) for arg in args)
822
823
824@pytest.mark.parametrize(
825 "extra_input_args",
826 [
827 # the concat demuxer brings its own -f, and needs the whitelist for the listed files
828 ["-safe", "0", "-f", "concat", "-i", "/list.txt"],
829 # a caller raising the probe limits relies on its own values landing last
830 ["-probesize", "65536", "-analyzeduration", "5000000"],
831 ],
832 ids=["caller-supplied-input-format", "caller-raised-probe-limits"],
833)
834def test_read_args_lead_the_main_input(extra_input_args: list[str]) -> None:
835 """Every main input is opened under our read limits, which the caller may override."""
836 args = get_ffmpeg_args(_PCM_FORMAT, _PCM_FORMAT, [], extra_input_args=extra_input_args)
837 start = args.index("-protocol_whitelist")
838 end = start + len(_INPUT_READ_ARGS)
839 assert args[start:end] == _INPUT_READ_ARGS
840 # the caller's args follow ours within the same input group, so theirs win on a conflict
841 assert args[end : end + len(extra_input_args)] == extra_input_args
842 # exactly one input either way: we add ours only when the caller brings none
843 assert args.count("-i") == 1
844
845
846# -- _log_reader_task (decode-error flood guard) --
847
848
849class _FakeStream:
850 """Stand-in for a StreamReader/StreamWriter: already closed/at EOF, nothing to drain."""
851
852 def is_closing(self) -> bool:
853 return True
854
855 def at_eof(self) -> bool:
856 return True
857
858
859class _FakeProc:
860 """Minimal stand-in for asyncio.subprocess.Process — just enough for close() to run."""
861
862 def __init__(self) -> None:
863 self.pid = 12345
864 self.returncode: int | None = None
865 self.stdin: _FakeStream | None = _FakeStream()
866 self.stdout = _FakeStream()
867
868 async def communicate(self) -> tuple[bytes, bytes]:
869 self.returncode = 0
870 return b"", b""
871
872 def send_signal(self, _sig: int) -> None:
873 pass
874
875
876class _FakeProcRacingExit(_FakeProc):
877 """A no-stdin process that has already exited, so send_signal raises ProcessLookupError."""
878
879 def __init__(self) -> None:
880 super().__init__()
881 # no stdin routes close() down the send_signal(SIGINT) branch
882 self.stdin = None
883
884 def send_signal(self, _sig: int) -> None:
885 raise ProcessLookupError("no such process")
886
887
888async def test_log_reader_reports_decode_errors_once_and_aborts() -> None:
889 """
890 Crossing the decode-error threshold logs a single line and aborts the stream.
891
892 Regression test: previously every stderr line was re-promoted to ERROR for the
893 rest of the process once 50 "Invalid data" lines were seen, flooding the log
894 with thousands of lines for a single corrupted file.
895 """
896 ffmpeg = FFMpeg(audio_input="-", input_format=_PCM_FORMAT, output_format=_PCM_FORMAT)
897
898 async def fake_stderr() -> AsyncGenerator[str]:
899 for _ in range(60):
900 yield "Invalid data found when processing input"
901 # noise that a genuinely corrupted stream keeps emitting after the threshold;
902 # none of this should reach ERROR level under the fix
903 for _ in range(20):
904 yield "Reserved bit set."
905
906 ffmpeg.iter_stderr = fake_stderr # type: ignore[method-assign]
907
908 error_lines: list[str] = []
909 ffmpeg.logger.error = lambda msg, *args: error_lines.append(msg % args if args else msg) # type: ignore[method-assign]
910
911 await ffmpeg._log_reader_task()
912 assert ffmpeg._abort_task is not None
913 await ffmpeg._abort_task
914
915 assert error_lines == ["Excessive decode errors (50+) for this stream; aborting"]
916 assert ffmpeg.closed
917
918
919async def test_log_reader_below_threshold_does_not_abort() -> None:
920 """A handful of decode errors, well under the threshold, triggers no report or abort."""
921 ffmpeg = FFMpeg(audio_input="-", input_format=_PCM_FORMAT, output_format=_PCM_FORMAT)
922
923 async def fake_stderr() -> AsyncGenerator[str]:
924 for _ in range(10):
925 yield "Invalid data found when processing input"
926 yield "Reserved bit set."
927
928 ffmpeg.iter_stderr = fake_stderr # type: ignore[method-assign]
929
930 error_lines: list[str] = []
931 ffmpeg.logger.error = lambda msg, *args: error_lines.append(msg % args if args else msg) # type: ignore[method-assign]
932
933 await ffmpeg._log_reader_task()
934
935 assert error_lines == []
936 assert ffmpeg._abort_task is None
937 assert not ffmpeg.closed
938
939
940async def test_log_reader_abort_does_not_self_deadlock() -> None:
941 """
942 The detached abort task can close() the reader's own process without a self-await.
943
944 Regression test for the deadlock this PR reintroduces abort-on-close around:
945 close() does ``await asyncio.wait_for(self._stderr_reader_task, 5)``, and
946 ``_stderr_reader_task`` here is wired to the very task running
947 ``_log_reader_task`` — the same setup ``start()`` uses in production. If the
948 abort were awaited inline from within ``_log_reader_task`` instead of via a
949 detached task, that task would be awaiting itself, which asyncio turns into
950 ``RuntimeError: Task cannot await on itself`` rather than a hang.
951 """
952 ffmpeg = FFMpeg(audio_input="-", input_format=_PCM_FORMAT, output_format=_PCM_FORMAT)
953 ffmpeg.proc = _FakeProc() # type: ignore[assignment]
954
955 async def fake_stderr() -> AsyncGenerator[str]:
956 for _ in range(50):
957 yield "Invalid data found when processing input"
958
959 ffmpeg.iter_stderr = fake_stderr # type: ignore[method-assign]
960
961 reader_task = asyncio.create_task(ffmpeg._log_reader_task())
962 ffmpeg._stderr_reader_task = reader_task
963
964 await asyncio.wait_for(reader_task, timeout=2)
965 assert ffmpeg._abort_task is not None
966 await asyncio.wait_for(ffmpeg._abort_task, timeout=2)
967
968 assert ffmpeg.closed
969
970
971async def test_abort_survives_send_signal_racing_process_exit() -> None:
972 """
973 The fire-and-forget abort must not crash if the process exits before it is signalled.
974
975 close() sends SIGINT to no-stdin processes, which raises ProcessLookupError when the
976 process already exited between the returncode check and the signal. The abort task is
977 never awaited by a caller, so that race must be swallowed inside close() rather than
978 escaping as an untracked "Task exception was never retrieved".
979 """
980 ffmpeg = FFMpeg(audio_input="-", input_format=_PCM_FORMAT, output_format=_PCM_FORMAT)
981 ffmpeg.proc = _FakeProcRacingExit() # type: ignore[assignment]
982
983 async def fake_stderr() -> AsyncGenerator[str]:
984 for _ in range(50):
985 yield "Invalid data found when processing input"
986
987 ffmpeg.iter_stderr = fake_stderr # type: ignore[method-assign]
988
989 reader_task = asyncio.create_task(ffmpeg._log_reader_task())
990 ffmpeg._stderr_reader_task = reader_task
991
992 await asyncio.wait_for(reader_task, timeout=2)
993 assert ffmpeg._abort_task is not None
994 # must complete without propagating ProcessLookupError
995 await asyncio.wait_for(ffmpeg._abort_task, timeout=2)
996
997 assert ffmpeg.closed
998
999
1000# -- _build_filtergraph_args (DSP chain assembly) --
1001
1002
1003def test_build_filtergraph_all_simple_uses_af() -> None:
1004 """A chain of plain filters renders to a single -af comma chain."""
1005 assert _build_filtergraph_args(["equalizer=x", "volume=3dB"]) == (
1006 [],
1007 ["-af", "equalizer=x,volume=3dB"],
1008 )
1009
1010
1011def test_build_filtergraph_empty_returns_no_args() -> None:
1012 """An empty chain produces no ffmpeg arguments."""
1013 assert _build_filtergraph_args([]) == ([], [])
1014
1015
1016def test_build_filtergraph_single_complex_fragment() -> None:
1017 """A complex fragment renders a labelled -filter_complex graph with -map."""
1018 result = _build_filtergraph_args(
1019 [ComplexFilter("afir=irnorm=1", [ComplexFilterInput("/ir.wav", "aresample=48000")])]
1020 )
1021 assert result == (
1022 [*_INPUT_READ_ARGS, "-i", "/ir.wav"],
1023 [
1024 "-filter_complex",
1025 "[1:a]aresample=48000[dsp1];[0:a][dsp1]afir=irnorm=1[dsp2]",
1026 "-map",
1027 "[dsp2]",
1028 ],
1029 )
1030
1031
1032def test_build_filtergraph_complex_between_simple_runs() -> None:
1033 """Simple runs on either side of a complex fragment weave into labelled pads."""
1034 result = _build_filtergraph_args(
1035 [
1036 "equalizer=x",
1037 ComplexFilter("afir=irnorm=1", [ComplexFilterInput("/ir.wav", "aresample=48000")]),
1038 "volume=2dB",
1039 ]
1040 )
1041 assert result == (
1042 [*_INPUT_READ_ARGS, "-i", "/ir.wav"],
1043 [
1044 "-filter_complex",
1045 "[0:a]equalizer=x[dsp1];[1:a]aresample=48000[dsp2];"
1046 "[dsp1][dsp2]afir=irnorm=1[dsp3];[dsp3]volume=2dB[dsp4]",
1047 "-map",
1048 "[dsp4]",
1049 ],
1050 )
1051
1052
1053def test_build_filtergraph_multiple_inputs() -> None:
1054 """A fragment with several inputs numbers them in order and feeds them to the body."""
1055 result = _build_filtergraph_args(
1056 [ComplexFilter("amerge", [ComplexFilterInput("/a.wav"), ComplexFilterInput("/b.wav")])]
1057 )
1058 assert result == (
1059 [*_INPUT_READ_ARGS, "-i", "/a.wav", *_INPUT_READ_ARGS, "-i", "/b.wav"],
1060 ["-filter_complex", "[0:a][1:a][2:a]amerge[dsp1]", "-map", "[dsp1]"],
1061 )
1062
1063
1064def test_build_filtergraph_input_args_precede_the_input() -> None:
1065 """An input's own ffmpeg options are emitted directly before its -i."""
1066 result = _build_filtergraph_args(
1067 [
1068 ComplexFilter(
1069 "amix=inputs=2",
1070 [ComplexFilterInput("/loop.wav", input_args=["-stream_loop", "-1"])],
1071 )
1072 ]
1073 )
1074 assert result == (
1075 [*_INPUT_READ_ARGS, "-stream_loop", "-1", "-i", "/loop.wav"],
1076 ["-filter_complex", "[0:a][1:a]amix=inputs=2[dsp1]", "-map", "[dsp1]"],
1077 )
1078
1079
1080def test_get_ffmpeg_args_uses_af_without_complex_filter() -> None:
1081 """Plain filter chains keep the -af path (no -filter_complex/-map)."""
1082 fmt = AudioFormat(
1083 content_type=ContentType.PCM_S16LE, sample_rate=48000, bit_depth=16, channels=2
1084 )
1085 args = get_ffmpeg_args(fmt, fmt, ["volume=-1dB"])
1086 assert "-af" in args
1087 assert "-filter_complex" not in args
1088
1089
1090def test_get_ffmpeg_args_uses_filter_complex_with_complex_filter() -> None:
1091 """A complex fragment switches the whole chain to -filter_complex with -map."""
1092 fmt = AudioFormat(
1093 content_type=ContentType.PCM_S16LE, sample_rate=48000, bit_depth=16, channels=2
1094 )
1095 args = get_ffmpeg_args(
1096 fmt, fmt, [ComplexFilter("afir=irnorm=1", [ComplexFilterInput("/ir.wav")])]
1097 )
1098 assert "-filter_complex" in args
1099 assert "-map" in args
1100 assert "-af" not in args
1101 # the impulse response is a real input, so it never passes through graph quoting
1102 assert args.count("-i") == 2
1103 assert args.index("/ir.wav") > args.index("-i")
1104
1105
1106def _rms_db(path: Path) -> float:
1107 """Return the overall RMS level of an audio file in dB via ffmpeg astats."""
1108 output = subprocess.run( # noqa: S603
1109 [ # noqa: S607
1110 "ffmpeg",
1111 "-hide_banner",
1112 "-nostats",
1113 "-i",
1114 str(path),
1115 "-af",
1116 "astats=measure_perchannel=none",
1117 "-f",
1118 "null",
1119 "-",
1120 ],
1121 capture_output=True,
1122 text=True,
1123 check=True,
1124 ).stderr
1125 for line in output.splitlines():
1126 if "RMS level dB" in line:
1127 return float(line.split("RMS level dB:")[-1])
1128 raise AssertionError("no RMS level in astats output")
1129
1130
1131def test_filtergraph_complex_runs_in_ffmpeg(tmp_path: Path) -> None:
1132 """The generated -filter_complex graph is valid and an identity IR passes audio through."""
1133 main = tmp_path / "main.wav"
1134 ir = tmp_path / "ir.wav"
1135 out = tmp_path / "out.wav"
1136 subprocess.run( # noqa: S603
1137 [ # noqa: S607
1138 "ffmpeg",
1139 "-y",
1140 "-f",
1141 "lavfi",
1142 "-i",
1143 "sine=frequency=1000:duration=1:sample_rate=48000",
1144 "-ac",
1145 "2",
1146 str(main),
1147 ],
1148 check=True,
1149 capture_output=True,
1150 )
1151 # a single-sample impulse is the identity IR: convolving with it returns the input
1152 subprocess.run( # noqa: S603
1153 [ # noqa: S607
1154 "ffmpeg",
1155 "-y",
1156 "-f",
1157 "lavfi",
1158 "-i",
1159 "aevalsrc=eq(n\\,0):d=0.01:s=48000:c=stereo",
1160 str(ir),
1161 ],
1162 check=True,
1163 capture_output=True,
1164 )
1165 input_args, filter_args = _build_filtergraph_args(
1166 [ComplexFilter("afir=irnorm=1", [ComplexFilterInput(str(ir), "aresample=48000")])]
1167 )
1168 result = subprocess.run( # noqa: S603
1169 [ # noqa: S607
1170 "ffmpeg",
1171 "-hide_banner",
1172 "-loglevel",
1173 "error",
1174 "-y",
1175 "-i",
1176 str(main),
1177 *input_args,
1178 *filter_args,
1179 str(out),
1180 ],
1181 capture_output=True,
1182 text=True,
1183 check=False,
1184 )
1185 assert result.returncode == 0, result.stderr
1186 assert out.exists()
1187 # identity IR => output level matches input level
1188 assert abs(_rms_db(out) - _rms_db(main)) < 0.5
1189
1190
1191def test_mono_source_keeps_its_level_when_widened_to_stereo(tmp_path: Path) -> None:
1192 """Widening a mono source to stereo duplicates it, where a rematrix would cost 3 dB."""
1193 main = tmp_path / "mono.wav"
1194 out = tmp_path / "stereo.wav"
1195 subprocess.run( # noqa: S603
1196 [ # noqa: S607
1197 "ffmpeg",
1198 "-y",
1199 "-f",
1200 "lavfi",
1201 "-i",
1202 "sine=frequency=1000:duration=1:sample_rate=44100",
1203 "-ac",
1204 "1",
1205 str(main),
1206 ],
1207 check=True,
1208 capture_output=True,
1209 )
1210 args = get_ffmpeg_args(
1211 AudioFormat(content_type=ContentType.WAV, sample_rate=44100, bit_depth=16, channels=1),
1212 AudioFormat(content_type=ContentType.WAV, sample_rate=44100, bit_depth=16, channels=2),
1213 [],
1214 input_path=str(main),
1215 output_path=str(out),
1216 )
1217 result = subprocess.run(args, capture_output=True, text=True, check=False) # noqa: S603
1218
1219 assert result.returncode == 0, result.stderr
1220 assert abs(_rms_db(out) - _rms_db(main)) < 0.5
1221
1222
1223async def test_get_ffmpeg_hls_cmaf_input_args_relax_a_build_that_blocks_cmaf() -> None:
1224 """A build whose HLS demuxer rejects CMAF gets the extension check turned off."""
1225 await set_global_cache_values({CACHE_ATTR_HLS_CMAF_BLOCKED: True})
1226
1227 assert get_ffmpeg_hls_cmaf_input_args() == ["-extension_picky", "0"]
1228
1229
1230async def test_get_ffmpeg_hls_cmaf_input_args_leave_a_capable_build_alone() -> None:
1231 """A build that accepts CMAF keeps its extension check, which guards hostile playlists."""
1232 await set_global_cache_values({CACHE_ATTR_HLS_CMAF_BLOCKED: False})
1233
1234 assert get_ffmpeg_hls_cmaf_input_args() == []
1235
1236
1237# Trimmed `ffmpeg -h demuxer=hls` output for the three generations of the segment extension
1238# check: absent before 7.1.1, present without CMAF in 7.1.1, present with CMAF from 7.1.2 on.
1239_HLS_OPTIONS_WITHOUT_CHECK = b"""Demuxer hls [Apple HTTP Live Streaming]:
1240hls demuxer AVOptions:
1241 -allowed_extensions <string> .D......... List of file extensions that hls is allowed to access (default "3gp,aac,m3u8,m4a,m4s,mp4,mpegts,ts,wav")
1242 -max_reload <int> .D......... Maximum number of times a insufficient list is attempted to be reloaded (from 0 to INT_MAX) (default 100)
1243"""
1244_HLS_OPTIONS_BLOCKING_CMAF = b"""Demuxer hls [Apple HTTP Live Streaming]:
1245hls demuxer AVOptions:
1246 -allowed_extensions <string> .D......... List of file extensions that hls is allowed to access (default "3gp,aac,m3u8,m4a,m4s,mp4,mpegts,ts,wav")
1247 -extension_picky <boolean> .D......... Be picky with all extensions matching (default true)
1248 -max_reload <int> .D......... Maximum number of times a insufficient list is attempted to be reloaded (from 0 to INT_MAX) (default 100)
1249"""
1250_HLS_OPTIONS_ALLOWING_CMAF = b"""Demuxer hls [Apple HTTP Live Streaming]:
1251hls demuxer AVOptions:
1252 -allowed_extensions <string> .D......... List of file extensions that hls is allowed to access (default "3gp,aac,m3u8,m4a,m4s,mp4,mpegts,ts,wav,cmfv,cmfa,ec3,fmp4")
1253 -allowed_segment_extensions <string> .D......... List of file extensions that hls is allowed to access (default "3gp,aac,m3u8,m4a,m4s,mp4,mpegts,ts,wav,cmfv,cmfa,ec3,fmp4,html")
1254 -extension_picky <boolean> .D......... Be picky with all extensions matching (default true)
1255 -max_reload <int> .D......... Maximum number of times a insufficient list is attempted to be reloaded (from 0 to INT_MAX) (default 100)
1256"""
1257_FFMPEG_VERSION_OUTPUT = (
1258 b"ffmpeg version 7.1.1 Copyright (c) 2000-2025 the FFmpeg developers\n"
1259 b"configuration: --enable-libsoxr\n"
1260)
1261
1262
1263def _fake_ffmpeg_probes(
1264 monkeypatch: pytest.MonkeyPatch, hls_options: bytes, hls_returncode: int = 0
1265) -> None:
1266 """Answer the version and HLS demuxer probes with canned output."""
1267
1268 async def _check_output(
1269 *args: str, _env: dict[str, str] | None = None, _timeout: float | None = None
1270 ) -> tuple[int, bytes]:
1271 if "-h" in args:
1272 return (hls_returncode, hls_options)
1273 return (0, _FFMPEG_VERSION_OUTPUT)
1274
1275 monkeypatch.setattr("music_assistant.helpers.ffmpeg.check_output", _check_output)
1276
1277
1278async def test_check_ffmpeg_version_finds_a_demuxer_that_blocks_cmaf(
1279 monkeypatch: pytest.MonkeyPatch,
1280) -> None:
1281 """A demuxer that is picky about extensions but does not know CMAF blocks those segments."""
1282 _fake_ffmpeg_probes(monkeypatch, _HLS_OPTIONS_BLOCKING_CMAF)
1283
1284 await check_ffmpeg_version()
1285
1286 assert get_global_cache_value(CACHE_ATTR_HLS_CMAF_BLOCKED) is True
1287
1288
1289async def test_check_ffmpeg_version_leaves_a_demuxer_that_whitelists_cmaf_alone(
1290 monkeypatch: pytest.MonkeyPatch,
1291) -> None:
1292 """A demuxer that lists CMAF among the extensions it accepts needs no relaxation."""
1293 _fake_ffmpeg_probes(monkeypatch, _HLS_OPTIONS_ALLOWING_CMAF)
1294
1295 await check_ffmpeg_version()
1296
1297 assert get_global_cache_value(CACHE_ATTR_HLS_CMAF_BLOCKED) is False
1298
1299
1300async def test_check_ffmpeg_version_leaves_a_demuxer_without_the_check_alone(
1301 monkeypatch: pytest.MonkeyPatch,
1302) -> None:
1303 """A demuxer that never gained the extension check accepts CMAF as it is."""
1304 _fake_ffmpeg_probes(monkeypatch, _HLS_OPTIONS_WITHOUT_CHECK)
1305
1306 await check_ffmpeg_version()
1307
1308 assert get_global_cache_value(CACHE_ATTR_HLS_CMAF_BLOCKED) is False
1309
1310
1311async def test_check_ffmpeg_version_keeps_the_check_when_the_probe_fails(
1312 monkeypatch: pytest.MonkeyPatch,
1313) -> None:
1314 """An unreadable probe must not relax a check that may well be doing its job."""
1315 _fake_ffmpeg_probes(monkeypatch, b"Unknown demuxer 'hls'.\n", hls_returncode=1)
1316
1317 await check_ffmpeg_version()
1318
1319 assert get_global_cache_value(CACHE_ATTR_HLS_CMAF_BLOCKED) is False
1320
1321
1322async def test_check_ffmpeg_version_keeps_the_check_when_the_probe_exits_nonzero(
1323 monkeypatch: pytest.MonkeyPatch,
1324) -> None:
1325 """Output that reads as blocking is worthless once the probe itself reported failure."""
1326 _fake_ffmpeg_probes(monkeypatch, _HLS_OPTIONS_BLOCKING_CMAF, hls_returncode=1)
1327
1328 await check_ffmpeg_version()
1329
1330 assert get_global_cache_value(CACHE_ATTR_HLS_CMAF_BLOCKED) is False
1331