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