From ba73781ae762f7c3d545e2670d5767705be70ba6 Mon Sep 17 00:00:00 2001 From: xiaoyuyu6420 <93528429+xiaoyuyu6420@users.noreply.github.com> Date: Thu, 13 Aug 2026 03:06:29 +0800 Subject: [PATCH 1/4] fix: verify audio magic bytes before extension-based short-circuit in convert_audio_format When a platform (e.g. NapCat) saves AMR-encoded audio with a .wav extension, the extension-only short-circuit returned the raw AMR file unchanged. Downstream STT providers then received malformed WAV data and returned HTTP 400. The fix reuses the existing _get_audio_magic_type() helper to verify the file content actually matches the target format before skipping ffmpeg conversion. When the detected format differs from the output format, conversion proceeds normally. Unrecognised content still falls back to extension matching to preserve existing behaviour. Closes #9594. --- tests/test_media_utils.py | 133 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 133 insertions(+) diff --git a/tests/test_media_utils.py b/tests/test_media_utils.py index efe5e65f02..1d231d24cf 100644 --- a/tests/test_media_utils.py +++ b/tests/test_media_utils.py @@ -871,3 +871,136 @@ async def test_wav_to_tencent_silk_skips_resample_for_supported_rate( assert len(fake.calls) == 1 assert fake.calls[0]["sample_rate"] == 24000 + + +# --------------------------------------------------------------------------- +# convert_audio_format: extension vs magic-byte mismatch (issue #9594) +# --------------------------------------------------------------------------- + +# Minimal AMR header (magic bytes: #!AMR) +_AMR_HEADER = b"#!AMR\n" + + +def _make_wav_bytes() -> bytes: + """Build a minimal valid WAV header (RIFF/WAVE).""" + return b"RIFF\x24\x00\x00\x00WAVEfmt " + b"\x00" * 16 + + +class _FakeFFmpegProcess: + """Minimal async subprocess mock for ffmpeg.""" + + def __init__(self, returncode: int = 0, stderr: bytes = b"") -> None: + self.returncode = returncode + + async def communicate(self) -> tuple[bytes, bytes]: + return (b"", b"") + + +@pytest.mark.asyncio +async def test_convert_audio_format_amr_with_wav_extension_does_not_short_circuit( + tmp_path, monkeypatch +): + """AMR content saved with a .wav extension must still be converted (#9594).""" + # NapCat saves AMR-encoded QQ voice with a .wav extension. + audio_file = tmp_path / "voice.wav" + audio_file.write_bytes(_AMR_HEADER + b"\x00" * 50) + + # Intercept ffmpeg subprocess so the test never depends on a real binary. + ffmpeg_called = False + + async def fake_exec(*args, **kwargs): + nonlocal ffmpeg_called + ffmpeg_called = True + # Write a dummy output file so convert_audio_format can return it. + output_path = args[-1] + Path(output_path).write_bytes(_make_wav_bytes()) + return _FakeFFmpegProcess() + + monkeypatch.setattr(media_utils.asyncio, "create_subprocess_exec", fake_exec) + monkeypatch.setattr(media_utils, "get_astrbot_temp_path", lambda: str(tmp_path)) + + result = await media_utils.convert_audio_format(str(audio_file), "wav") + + assert ffmpeg_called, ( + "ffmpeg should have been invoked because content is AMR, not WAV" + ) + assert result != str(audio_file) + + +@pytest.mark.asyncio +async def test_convert_audio_format_real_wav_with_wav_extension_short_circuits( + tmp_path, monkeypatch +): + """Genuine WAV content with a .wav extension should skip conversion.""" + audio_file = tmp_path / "voice.wav" + audio_file.write_bytes(_make_wav_bytes()) + + async def fake_exec(*args, **kwargs): + raise AssertionError("ffmpeg should not be called for a real WAV file") + + monkeypatch.setattr(media_utils.asyncio, "create_subprocess_exec", fake_exec) + + result = await media_utils.convert_audio_format(str(audio_file), "wav") + assert result == str(audio_file) + + +@pytest.mark.asyncio +async def test_convert_audio_format_unknown_content_with_matching_ext_short_circuits( + tmp_path, monkeypatch +): + """When magic bytes cannot be identified, fall back to extension matching.""" + audio_file = tmp_path / "voice.wav" + audio_file.write_bytes(b"\x00" * 64) + + async def fake_exec(*args, **kwargs): + raise AssertionError("ffmpeg should not be called for unrecognised content") + + monkeypatch.setattr(media_utils.asyncio, "create_subprocess_exec", fake_exec) + + result = await media_utils.convert_audio_format(str(audio_file), "wav") + assert result == str(audio_file) + + +@pytest.mark.asyncio +async def test_convert_audio_format_real_amr_with_amr_extension_short_circuits( + tmp_path, monkeypatch +): + """AMR content with a matching .amr extension should skip conversion.""" + audio_file = tmp_path / "voice.amr" + audio_file.write_bytes(_AMR_HEADER + b"\x00" * 50) + + async def fake_exec(*args, **kwargs): + raise AssertionError("ffmpeg should not be called for a real AMR file") + + monkeypatch.setattr(media_utils.asyncio, "create_subprocess_exec", fake_exec) + + result = await media_utils.convert_audio_format(str(audio_file), "amr") + assert result == str(audio_file) + + +@pytest.mark.asyncio +async def test_convert_audio_format_wav_with_ogg_extension_does_not_short_circuit( + tmp_path, monkeypatch +): + """WAV content saved with a .ogg extension must still be converted.""" + audio_file = tmp_path / "voice.ogg" + audio_file.write_bytes(_make_wav_bytes()) + + ffmpeg_called = False + + async def fake_exec(*args, **kwargs): + nonlocal ffmpeg_called + ffmpeg_called = True + output_path = args[-1] + Path(output_path).write_bytes(b"OggS" + b"\x00" * 60) + return _FakeFFmpegProcess() + + monkeypatch.setattr(media_utils.asyncio, "create_subprocess_exec", fake_exec) + monkeypatch.setattr(media_utils, "get_astrbot_temp_path", lambda: str(tmp_path)) + + result = await media_utils.convert_audio_format(str(audio_file), "ogg") + + assert ffmpeg_called, ( + "ffmpeg should have been invoked because content is WAV, not OGG" + ) + assert result != str(audio_file) From 750f133ee3d4d37d0ac64e349d269d9dd5507177 Mon Sep 17 00:00:00 2001 From: xiaoyuyu6420 <93528429+xiaoyuyu6420@users.noreply.github.com> Date: Fri, 14 Aug 2026 03:02:07 +0800 Subject: [PATCH 2/4] refactor: extract ffmpeg mock helper and remove unused stderr param Address Sourcery review feedback on #9651: - Extract _patch_ffmpeg() helper to deduplicate the fake_exec setup across the three conversion-expected tests. - Remove the unused stderr parameter from _FakeFFmpegProcess. --- tests/test_media_utils.py | 62 +++++++++++++++++++-------------------- 1 file changed, 30 insertions(+), 32 deletions(-) diff --git a/tests/test_media_utils.py b/tests/test_media_utils.py index 1d231d24cf..2842e78b75 100644 --- a/tests/test_media_utils.py +++ b/tests/test_media_utils.py @@ -889,41 +889,51 @@ def _make_wav_bytes() -> bytes: class _FakeFFmpegProcess: """Minimal async subprocess mock for ffmpeg.""" - def __init__(self, returncode: int = 0, stderr: bytes = b"") -> None: + def __init__(self, returncode: int = 0) -> None: self.returncode = returncode async def communicate(self) -> tuple[bytes, bytes]: return (b"", b"") -@pytest.mark.asyncio -async def test_convert_audio_format_amr_with_wav_extension_does_not_short_circuit( - tmp_path, monkeypatch -): - """AMR content saved with a .wav extension must still be converted (#9594).""" - # NapCat saves AMR-encoded QQ voice with a .wav extension. - audio_file = tmp_path / "voice.wav" - audio_file.write_bytes(_AMR_HEADER + b"\x00" * 50) +def _patch_ffmpeg(monkeypatch, tmp_path, *, output_content: bytes = b""): + """Monkeypatch ``asyncio.create_subprocess_exec`` to simulate ffmpeg. - # Intercept ffmpeg subprocess so the test never depends on a real binary. - ffmpeg_called = False + Args: + monkeypatch: pytest monkeypatch fixture. + tmp_path: pytest tmp_path fixture (used for temp dir patching). + output_content: Bytes written to the output path so + ``convert_audio_format`` can return it. + + Returns: + A mutable list whose first element tracks whether ffmpeg was called. + """ + called = [False] async def fake_exec(*args, **kwargs): - nonlocal ffmpeg_called - ffmpeg_called = True - # Write a dummy output file so convert_audio_format can return it. + called[0] = True output_path = args[-1] - Path(output_path).write_bytes(_make_wav_bytes()) + Path(output_path).write_bytes(output_content) return _FakeFFmpegProcess() monkeypatch.setattr(media_utils.asyncio, "create_subprocess_exec", fake_exec) monkeypatch.setattr(media_utils, "get_astrbot_temp_path", lambda: str(tmp_path)) + return called + + +@pytest.mark.asyncio +async def test_convert_audio_format_amr_with_wav_extension_does_not_short_circuit( + tmp_path, monkeypatch +): + """AMR content saved with a .wav extension must still be converted (#9594).""" + audio_file = tmp_path / "voice.wav" + audio_file.write_bytes(_AMR_HEADER + b"\x00" * 50) + + called = _patch_ffmpeg(monkeypatch, tmp_path, output_content=_make_wav_bytes()) result = await media_utils.convert_audio_format(str(audio_file), "wav") - assert ffmpeg_called, ( - "ffmpeg should have been invoked because content is AMR, not WAV" - ) + assert called[0], "ffmpeg should have been invoked because content is AMR, not WAV" assert result != str(audio_file) @@ -986,21 +996,9 @@ async def test_convert_audio_format_wav_with_ogg_extension_does_not_short_circui audio_file = tmp_path / "voice.ogg" audio_file.write_bytes(_make_wav_bytes()) - ffmpeg_called = False - - async def fake_exec(*args, **kwargs): - nonlocal ffmpeg_called - ffmpeg_called = True - output_path = args[-1] - Path(output_path).write_bytes(b"OggS" + b"\x00" * 60) - return _FakeFFmpegProcess() - - monkeypatch.setattr(media_utils.asyncio, "create_subprocess_exec", fake_exec) - monkeypatch.setattr(media_utils, "get_astrbot_temp_path", lambda: str(tmp_path)) + called = _patch_ffmpeg(monkeypatch, tmp_path, output_content=b"OggS" + b"\x00" * 60) result = await media_utils.convert_audio_format(str(audio_file), "ogg") - assert ffmpeg_called, ( - "ffmpeg should have been invoked because content is WAV, not OGG" - ) + assert called[0], "ffmpeg should have been invoked because content is WAV, not OGG" assert result != str(audio_file) From 1cb8961ca172ee1371e8b22270a969476e8ab310 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9E=AD=E9=92=B0=E9=92=B0?= <672178818@qq.com> Date: Sat, 15 Aug 2026 09:37:10 +0800 Subject: [PATCH 3/4] refactor: extract shared never-called ffmpeg mock helper Collapse the three duplicated fake_exec stubs in the short-circuit tests into _patch_ffmpeg_not_called(monkeypatch, reason), finishing the Sourcery review note on duplicated ffmpeg mocking. --- tests/test_media_utils.py | 28 ++++++++++++++++------------ 1 file changed, 16 insertions(+), 12 deletions(-) diff --git a/tests/test_media_utils.py b/tests/test_media_utils.py index 2842e78b75..f2327754bc 100644 --- a/tests/test_media_utils.py +++ b/tests/test_media_utils.py @@ -896,6 +896,19 @@ async def communicate(self) -> tuple[bytes, bytes]: return (b"", b"") +def _patch_ffmpeg_not_called(monkeypatch, reason: str) -> None: + """Monkeypatch ``asyncio.create_subprocess_exec`` to fail if ffmpeg runs. + + Args: + monkeypatch: pytest monkeypatch fixture. + reason: scenario description included in the failure message. + """ + async def fake_exec(*args, **kwargs): + raise AssertionError(f"ffmpeg should not be called {reason}") + + monkeypatch.setattr(media_utils.asyncio, "create_subprocess_exec", fake_exec) + + def _patch_ffmpeg(monkeypatch, tmp_path, *, output_content: bytes = b""): """Monkeypatch ``asyncio.create_subprocess_exec`` to simulate ffmpeg. @@ -945,10 +958,7 @@ async def test_convert_audio_format_real_wav_with_wav_extension_short_circuits( audio_file = tmp_path / "voice.wav" audio_file.write_bytes(_make_wav_bytes()) - async def fake_exec(*args, **kwargs): - raise AssertionError("ffmpeg should not be called for a real WAV file") - - monkeypatch.setattr(media_utils.asyncio, "create_subprocess_exec", fake_exec) + _patch_ffmpeg_not_called(monkeypatch, "for a real WAV file") result = await media_utils.convert_audio_format(str(audio_file), "wav") assert result == str(audio_file) @@ -962,10 +972,7 @@ async def test_convert_audio_format_unknown_content_with_matching_ext_short_circ audio_file = tmp_path / "voice.wav" audio_file.write_bytes(b"\x00" * 64) - async def fake_exec(*args, **kwargs): - raise AssertionError("ffmpeg should not be called for unrecognised content") - - monkeypatch.setattr(media_utils.asyncio, "create_subprocess_exec", fake_exec) + _patch_ffmpeg_not_called(monkeypatch, "for unrecognised content") result = await media_utils.convert_audio_format(str(audio_file), "wav") assert result == str(audio_file) @@ -979,10 +986,7 @@ async def test_convert_audio_format_real_amr_with_amr_extension_short_circuits( audio_file = tmp_path / "voice.amr" audio_file.write_bytes(_AMR_HEADER + b"\x00" * 50) - async def fake_exec(*args, **kwargs): - raise AssertionError("ffmpeg should not be called for a real AMR file") - - monkeypatch.setattr(media_utils.asyncio, "create_subprocess_exec", fake_exec) + _patch_ffmpeg_not_called(monkeypatch, "for a real AMR file") result = await media_utils.convert_audio_format(str(audio_file), "amr") assert result == str(audio_file) From 35b8c4740102d09f33aadfd90f81a290041344bc Mon Sep 17 00:00:00 2001 From: xiaoyuyu6420 <93528429+xiaoyuyu6420@users.noreply.github.com> Date: Fri, 28 Aug 2026 19:19:34 +0800 Subject: [PATCH 4/4] test: align convert_audio_format tests with master's magic-byte verification The core fix for #9594 was already merged in #9612 (detect audio format from file content). This commit adapts the test suite to match that implementation: - Update unknown-content test: master converts unrecognised formats rather than short-circuiting (safer behavior). - Add missing-file test: master returns the path as-is when the file does not exist yet (NapCat race condition handling). - Update section header to reference both #9594 and #9612. --- tests/test_media_utils.py | 28 ++++++++++++++++++++++++---- 1 file changed, 24 insertions(+), 4 deletions(-) diff --git a/tests/test_media_utils.py b/tests/test_media_utils.py index f2327754bc..7e3550be67 100644 --- a/tests/test_media_utils.py +++ b/tests/test_media_utils.py @@ -874,7 +874,9 @@ async def test_wav_to_tencent_silk_skips_resample_for_supported_rate( # --------------------------------------------------------------------------- -# convert_audio_format: extension vs magic-byte mismatch (issue #9594) +# convert_audio_format: extension vs magic-byte mismatch (#9594, fixed in #9612) +# These tests guard the magic-byte verification logic that prevents platforms +# like NapCat from passing AMR content through with a .wav extension. # --------------------------------------------------------------------------- # Minimal AMR header (magic bytes: #!AMR) @@ -903,6 +905,7 @@ def _patch_ffmpeg_not_called(monkeypatch, reason: str) -> None: monkeypatch: pytest monkeypatch fixture. reason: scenario description included in the failure message. """ + async def fake_exec(*args, **kwargs): raise AssertionError(f"ffmpeg should not be called {reason}") @@ -965,14 +968,31 @@ async def test_convert_audio_format_real_wav_with_wav_extension_short_circuits( @pytest.mark.asyncio -async def test_convert_audio_format_unknown_content_with_matching_ext_short_circuits( +async def test_convert_audio_format_unknown_content_with_matching_ext_converts( tmp_path, monkeypatch ): - """When magic bytes cannot be identified, fall back to extension matching.""" + """When magic bytes cannot be identified, proceed with conversion (safer).""" audio_file = tmp_path / "voice.wav" audio_file.write_bytes(b"\x00" * 64) - _patch_ffmpeg_not_called(monkeypatch, "for unrecognised content") + called = _patch_ffmpeg(monkeypatch, tmp_path, output_content=_make_wav_bytes()) + + result = await media_utils.convert_audio_format(str(audio_file), "wav") + + assert called[0], ( + "ffmpeg should be invoked for unrecognised content (safer to convert)" + ) + assert result != str(audio_file) + + +@pytest.mark.asyncio +async def test_convert_audio_format_missing_file_with_matching_ext_returns_path( + tmp_path, monkeypatch +): + """When the file does not exist yet (e.g. NapCat race), return the path as-is.""" + audio_file = tmp_path / "voice.wav" # never created + + _patch_ffmpeg_not_called(monkeypatch, "for a missing file") result = await media_utils.convert_audio_format(str(audio_file), "wav") assert result == str(audio_file)