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