xiaoxia-saas/tests/unit/test_original_audio_retained.py

"""回归测试：render_plan 新路径必须保留视频素材原声。

历史 Bug：mix_audio 中 `main_clips = []` 无条件丢弃源视频原声，
导致最终生成视频没有原声（与预览不一致）。本测试钉住新行为：
- 有音频流的 main/broll clip 原声必须进入最终音轨
- clip.config.volume=0 静音，volume≠1.0 应用音量滤镜
- 无音频流的素材被安全过滤
- 直通（pass-through）路径同样尊重 probe + volume
"""

from __future__ import annotations

from pathlib import Path
from unittest.mock import patch

import pytest
from video_processing.render_audio import (
    RenderContext,
    _clip_volume,
    mix_audio,
)
from video_processing.unified_render_service import (
    ResolvedClip,
    UnifiedRenderService,
)


def _ctx() -> RenderContext:
    return RenderContext(work_dir=Path("/tmp/test_render_audio_fix"), plan_id="plan_audio")


def _clip(
    cid: str,
    *,
    clip_type: str = "main",
    order: int = 0,
    duration: float = 5.0,
    config: dict | None = None,
    asset: str | None = None,
) -> ResolvedClip:
    return ResolvedClip(
        clip_id=cid,
        asset_id=asset or f"asset_{cid}.mp4",
        clip_type=clip_type,
        order=order,
        local_path=Path(f"/tmp/asset_{cid}.mp4"),
        duration=duration,
        actual_duration=duration,
        config=config or {},
    )


def _layers(svc, clips):
    # 直接把 ResolvedClip 分组为图层，跳过 _resolve_clips（后者要求原始 EditPlanClip）
    return svc._group_clips_into_layers(clips)


def _service(clips):
    paths = {c.asset_id: c.local_path for c in clips}
    return UnifiedRenderService(
        plan=type("P", (), {"id": "plan_audio", "config": {}})(),
        clips=clips,
        asset_path_map=paths,
        work_dir=Path("/tmp/test_render_audio_fix"),
        output_width=1080,
        output_height=1920,
        output_fps=25,
    )


class TestOriginalAudioRetained:
    """钉住原声不再被丢弃。"""

    def test_single_main_clip_audio_kept(self):
        svc = _service([_clip("c1")])
        with (
            patch("pathlib.Path.exists", return_value=True),
            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
            patch("video_processing.render_audio.probe_has_audio", return_value=True),
            patch("video_processing.render_audio.run_ffmpeg") as mock_run,
        ):
            result = mix_audio(_ctx(), _layers(svc, [_clip("c1")]), 5.0)

        assert result is not None
        cmd_str = " ".join(mock_run.call_args[0][0])
        assert "asset_c1.mp4" in cmd_str

    def test_multi_main_clips_concat_audio(self):
        clips = [_clip("c1", order=0), _clip("c2", order=1)]
        svc = _service(clips)
        with (
            patch("pathlib.Path.exists", return_value=True),
            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
            patch("video_processing.render_audio.probe_has_audio", return_value=True),
            patch("video_processing.render_audio.run_ffmpeg") as mock_run,
        ):
            result = mix_audio(_ctx(), _layers(svc, clips), 9.0)

        assert result is not None
        cmd_str = " ".join(mock_run.call_args[0][0])
        assert "asset_c1.mp4" in cmd_str
        assert "asset_c2.mp4" in cmd_str
        assert "concat=n=2:v=0:a=1" in cmd_str

    def test_main_audio_plus_independent_track_amix(self):
        clips = [
            _clip("c1", order=0),
            _clip("tts1", order=1, config={"role": "audio", "volume": 0.5}),
        ]
        svc = _service(clips)
        with (
            patch("pathlib.Path.exists", return_value=True),
            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
            patch("video_processing.render_audio.probe_has_audio", return_value=True),
            patch("video_processing.render_audio.run_ffmpeg") as mock_run,
        ):
            result = mix_audio(_ctx(), _layers(svc, clips), 5.0)

        assert result is not None
        cmd_str = " ".join(mock_run.call_args[0][0])
        assert "amix=inputs=2" in cmd_str
        assert "asset_c1.mp4" in cmd_str
        assert "asset_tts1.mp4" in cmd_str
        assert "volume=0.5" in cmd_str

    def test_silent_clip_volume_zero_excluded(self):
        """volume=0 的素材不送入原声 concat。"""
        clips = [_clip("mute", order=0, config={"volume": 0})]
        svc = _service(clips)
        with (
            patch("pathlib.Path.exists", return_value=True),
            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
            patch("video_processing.render_audio.probe_has_audio", return_value=True),
            patch("video_processing.render_audio.run_ffmpeg") as mock_run,
        ):
            result = mix_audio(_ctx(), _layers(svc, clips), 5.0)

        assert result is None
        mock_run.assert_not_called()

    def test_no_audio_stream_returns_none(self):
        clips = [_clip("c1")]
        svc = _service(clips)
        with (
            patch("pathlib.Path.exists", return_value=True),
            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
            patch("video_processing.render_audio.probe_has_audio", return_value=False),
            patch("video_processing.render_audio.run_ffmpeg") as mock_run,
        ):
            result = mix_audio(_ctx(), _layers(svc, clips), 5.0)

        assert result is None
        mock_run.assert_not_called()

    def test_volume_helper_default_and_override(self):
        assert _clip_volume(_clip("c1")) == 1.0
        assert _clip_volume(_clip("c2", config={"volume": 0.3})) == pytest.approx(0.3)
        assert _clip_volume(_clip("c3", config={"volume": 0})) == 0.0


class TestPassThroughAudioProbe:
    """直通路径必须先探测音频，不能无条件假设 main 有音频。"""

    def test_pass_through_probes_audio_before_encoding(self):
        clips = [_clip("c1")]
        svc = _service(clips)
        with (
            patch("pathlib.Path.exists", return_value=True),
            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
            patch(
                "video_processing.unified_render_service.probe_has_audio",
                return_value=False,
            ) as mock_probe,
            patch("video_processing.unified_render_service.run_ffmpeg") as mock_run,
        ):
            layers = svc._group_clips_into_layers(clips)
            has_audio = svc._render_pass_through(layers, Path("/tmp/out.mp4"), video_duration=5.0)

        assert has_audio is False
        mock_probe.assert_called()
        cmd_str = " ".join(mock_run.call_args[0][0])
        assert "aac" not in cmd_str

    def test_pass_through_volume_non_default_disables_stream_copy(self):
        svc = _service([_clip("c1", config={"volume": 0.5})])
        clip = _clip("c1", config={"volume": 0.5})
        can_copy, reason = svc._can_use_stream_copy(clip)
        assert can_copy is False
        assert "音量" in reason

    def test_clip_volume_static_helper(self):
        assert UnifiedRenderService._clip_volume(_clip("c1")) == 1.0
        assert UnifiedRenderService._clip_volume(_clip("c2", config={"volume": 0.7})) == pytest.approx(0.7)

    def test_pass_through_probe_exception_raises(self):
        """probe_has_audio 抛致命异常时必须向上抛出，不能静默丢音频产出无声视频。"""
        clips = [_clip("c1")]
        svc = _service(clips)
        with (
            patch("pathlib.Path.exists", return_value=True),
            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
            patch(
                "video_processing.unified_render_service.probe_has_audio",
                side_effect=RuntimeError("probe failed"),
            ),
            patch("video_processing.unified_render_service.run_ffmpeg") as mock_run,
            pytest.raises(RuntimeError, match="probe failed"),
        ):
            layers = svc._group_clips_into_layers(clips)
            svc._render_pass_through(layers, Path("/tmp/out.mp4"), video_duration=5.0)

        mock_run.assert_not_called()