diff --git a/tests/test_media_utils.py b/tests/test_media_utils.py index efe5e65f02..7e3550be67 100644 --- a/tests/test_media_utils.py +++ b/tests/test_media_utils.py @@ -871,3 +871,158 @@ 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 (#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) +_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) -> None: + self.returncode = returncode + + 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. + + 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): + called[0] = True + output_path = args[-1] + 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 called[0], "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()) + + _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) + + +@pytest.mark.asyncio +async def test_convert_audio_format_unknown_content_with_matching_ext_converts( + tmp_path, monkeypatch +): + """When magic bytes cannot be identified, proceed with conversion (safer).""" + audio_file = tmp_path / "voice.wav" + audio_file.write_bytes(b"\x00" * 64) + + 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) + + +@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) + + _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) + + +@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()) + + 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 called[0], "ffmpeg should have been invoked because content is WAV, not OGG" + assert result != str(audio_file)