Test additive video audio mixing

tests/test_mix_audio.py

@@ -0,0 +1,89 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from app.core.exceptions import InputError
+from app.models.media import InputMedia, MediaSource
+from app.operations.mix_audio import add_audio
+
+
+def _media(path: Path, filename: str, mime: str) -> InputMedia:
+    path.write_bytes(b"input")
+    return InputMedia(
+        source=MediaSource.MULTIPART,
+        filename=filename,
+        mime_type=mime,
+        temp_path=path,
+        size=path.stat().st_size,
+    )
+
+
+class FakeFFmpeg:
+    def __init__(self) -> None:
+        self.args: list[str | Path] = []
+
+    async def run(self, args, operation: str) -> None:
+        self.args = list(args)
+        output = Path(args[-1])
+        output.parent.mkdir(parents=True, exist_ok=True)
+        output.write_bytes(b"output")
+
+
+@pytest.mark.asyncio
+async def test_add_audio_preserves_video_and_mixes_sfx(tmp_path: Path) -> None:
+    video = _media(tmp_path / "video.mp4", "video.mp4", "video/mp4")
+    sfx = _media(tmp_path / "impact.wav", "impact.wav", "audio/wav")
+    ffmpeg = FakeFFmpeg()
+
+    result = await add_audio(
+        ffmpeg,
+        [video, sfx],
+        {"start": 2.5, "volume": 0.4},
+        tmp_path / "out",
+    )
+
+    assert result.path == tmp_path / "out" / "audio-mix.mp4"
+    command = [str(value) for value in ffmpeg.args]
+    assert "-map" in command
+    assert "0:v:0" in command
+    assert "[mixed]" in command
+    assert "-c:v" in command
+    assert command[command.index("-c:v") + 1] == "copy"
+    filter_complex = command[command.index("-filter_complex") + 1]
+    assert "[1:a]volume=0.4,adelay=2500:all=1[a1]" in filter_complex
+    assert "amix=inputs=2:duration=first:dropout_transition=0:normalize=0[mixed]" in filter_complex
+
+
+@pytest.mark.asyncio
+async def test_add_audio_supports_multiple_sfx_tracks(tmp_path: Path) -> None:
+    video = _media(tmp_path / "video.mp4", "video.mp4", "video/mp4")
+    sfx1 = _media(tmp_path / "one.wav", "one.wav", "audio/wav")
+    sfx2 = _media(tmp_path / "two.wav", "two.wav", "audio/wav")
+    ffmpeg = FakeFFmpeg()
+
+    await add_audio(
+        ffmpeg,
+        [video, sfx1, sfx2],
+        {"audio_tracks": '[{"start": 1, "volume": 0.5}, {"start": 4.25, "volume": 0.25}]'},
+        tmp_path / "out",
+    )
+
+    filter_complex = str(ffmpeg.args[ffmpeg.args.index("-filter_complex") + 1])
+    assert "[1:a]volume=0.5,adelay=1000:all=1[a1]" in filter_complex
+    assert "[2:a]volume=0.25,adelay=4250:all=1[a2]" in filter_complex
+    assert "amix=inputs=3:duration=first:dropout_transition=0:normalize=0[mixed]" in filter_complex
+
+
+@pytest.mark.asyncio
+async def test_add_audio_rejects_track_count_mismatch(tmp_path: Path) -> None:
+    video = _media(tmp_path / "video.mp4", "video.mp4", "video/mp4")
+    sfx = _media(tmp_path / "one.wav", "one.wav", "audio/wav")
+    with pytest.raises(InputError, match="one object for each"):
+        await add_audio(
+            FakeFFmpeg(),
+            [video, sfx],
+            {"audio_tracks": '[{"start": 1}, {"start": 2}]'},
+            tmp_path / "out",
+        )