xiaoxia-saas/tests/unit/test_render_audio_pure.py

"""渲染音频纯逻辑测试 — clip_effective_duration + clip_has_audio缓存."""

from __future__ import annotations

from dataclasses import field
from pathlib import Path
from unittest.mock import patch

import pytest
from video_processing.render_audio import (
    RenderContext,
    clip_effective_duration,
    clip_has_audio,
    mix_with_independent_audio,
)
from video_processing.unified_render_service import ResolvedClip


def _make_ctx(work_dir: str = "/tmp") -> RenderContext:
    """创建测试用RenderContext."""
    return RenderContext(work_dir=Path(work_dir), plan_id="test_plan")


def _make_clip(
    *,
    duration: float = 0.0,
    actual_duration: float = 0.0,
    local_path: str = "/tmp/test.mp4",
    clip_type: str = "main",
    clip_id: str = "c1",
    asset_id: str = "a1",
    order: int = 0,
) -> ResolvedClip:
    """快速创建测试用ResolvedClip."""
    return ResolvedClip(
        clip_id=clip_id,
        asset_id=asset_id,
        local_path=Path(local_path),
        clip_type=clip_type,
        order=order,
        duration=duration,
        actual_duration=actual_duration,
    )


class TestClipEffectiveDuration:
    """clip_effective_duration 有效时长计算测试."""

    def test_both_zero_returns_zero(self):
        """duration和actual_duration都是0 → 0."""
        clip = _make_clip(duration=0, actual_duration=0)
        assert clip_effective_duration(clip) == 0.0

    def test_only_actual_duration(self):
        """只有actual_duration → 返回actual_duration."""
        clip = _make_clip(duration=0, actual_duration=30.0)
        assert clip_effective_duration(clip) == pytest.approx(30.0)

    def test_duration_less_than_actual(self):
        """duration < actual → 返回duration（剪辑后的时长）."""
        clip = _make_clip(duration=10.0, actual_duration=30.0)
        assert clip_effective_duration(clip) == pytest.approx(10.0)

    def test_duration_greater_than_actual(self):
        """duration > actual → 返回actual（不能超过素材时长）."""
        clip = _make_clip(duration=50.0, actual_duration=30.0)
        assert clip_effective_duration(clip) == pytest.approx(30.0)

    def test_duration_equals_actual(self):
        """duration == actual → 返回该值."""
        clip = _make_clip(duration=20.0, actual_duration=20.0)
        assert clip_effective_duration(clip) == pytest.approx(20.0)

    def test_no_actual_with_positive_duration(self):
        """actual_duration=0但duration>0 → 返回duration（还没probe时）."""
        clip = _make_clip(duration=15.0, actual_duration=0.0)
        assert clip_effective_duration(clip) == pytest.approx(15.0)


class TestClipHasAudio:
    """clip_has_audio 音频探测+缓存测试."""

    def test_has_audio_true(self):
        """有音频时返回True."""
        ctx = _make_ctx()
        clip = _make_clip(local_path="/tmp/video1.mp4")

        with patch("video_processing.render_audio.probe_has_audio", return_value=True):
            result = clip_has_audio(ctx, clip)
            assert result is True

    def test_has_audio_false(self):
        """无音频时返回False."""
        ctx = _make_ctx()
        clip = _make_clip(local_path="/tmp/video2.mp4")

        with patch("video_processing.render_audio.probe_has_audio", return_value=False):
            result = clip_has_audio(ctx, clip)
            assert result is False

    def test_cache_avoids_reprobe(self):
        """同一个clip多次调用只probe一次（缓存生效）."""
        ctx = _make_ctx()
        clip = _make_clip(local_path="/tmp/cached.mp4")

        call_count = 0

        def fake_probe(path):
            nonlocal call_count
            call_count += 1
            return True

        with patch("video_processing.render_audio.probe_has_audio", side_effect=fake_probe):
            result1 = clip_has_audio(ctx, clip)
            result2 = clip_has_audio(ctx, clip)
            result3 = clip_has_audio(ctx, clip)

        assert result1 is True
        assert result2 is True
        assert result3 is True
        assert call_count == 1  # 只调用了一次

    def test_different_clips_both_probed(self):
        """不同clip各自probe一次."""
        ctx = _make_ctx()
        clip1 = _make_clip(clip_id="c1", local_path="/tmp/v1.mp4")
        clip2 = _make_clip(clip_id="c2", local_path="/tmp/v2.mp4")

        probe_call_count = 0

        def fake_probe(path):
            nonlocal probe_call_count
            probe_call_count += 1
            return "v1" in str(path)

        with patch("video_processing.render_audio.probe_has_audio", side_effect=fake_probe):
            r1 = clip_has_audio(ctx, clip1)
            r2 = clip_has_audio(ctx, clip2)

        assert r1 is True
        assert r2 is False
        assert probe_call_count == 2


class TestRenderContext:
    """RenderContext 渲染上下文测试."""

    def test_default_noise_reduction_none(self):
        """默认无降噪配置."""
        ctx = _make_ctx()
        assert ctx.noise_reduction_config is None

    def test_cache_starts_empty(self):
        """音频缓存初始为空."""
        ctx = _make_ctx()
        assert ctx._audio_cache == {}


class TestMixWithIndependentAudioVolume:
    """测试 mix_with_independent_audio 中主视频 clip 音量滤镜是否生效。"""

    def _make_clip(self, volume=None, clip_id="c1", duration=5.0):
        """创建测试用 ResolvedClip。"""
        config = {}
        if volume is not None:
            config["volume"] = volume
        return ResolvedClip(
            clip_id=clip_id,
            asset_id=f"a_{clip_id}",
            local_path=Path(f"/tmp/{clip_id}.mp4"),
            clip_type="main",
            order=0,
            duration=duration,
            actual_duration=duration,
            config=config,
        )

    def test_main_clip_volume_zero_applied_in_filter(self):
        """volume=0 的主视频 clip 应在 filter_complex 中包含 volume=0.0000。"""
        from unittest.mock import patch

        ctx = _make_ctx()
        main_clip = self._make_clip(volume=0, clip_id="m1")
        captured_cmd = {}

        def fake_run_ffmpeg(cmd):
            captured_cmd["cmd"] = cmd

        with patch("video_processing.render_audio.run_ffmpeg", side_effect=fake_run_ffmpeg):
            mix_with_independent_audio(
                ctx=ctx,
                main_clips=[main_clip],
                audio_clips=[],
                output_path=Path("/tmp/out.m4a"),
                video_duration=5.0,
            )

        filter_complex = captured_cmd["cmd"][captured_cmd["cmd"].index("-filter_complex") + 1]
        assert "volume=0.0000" in filter_complex, f"volume filter missing in: {filter_complex}"

    def test_main_clip_default_volume_no_filter(self):
        """默认 volume=1.0 时不应添加 volume 滤镜。"""
        from unittest.mock import patch

        ctx = _make_ctx()
        main_clip = self._make_clip(clip_id="m1")  # no volume set
        captured_cmd = {}

        def fake_run_ffmpeg(cmd):
            captured_cmd["cmd"] = cmd

        with patch("video_processing.render_audio.run_ffmpeg", side_effect=fake_run_ffmpeg):
            mix_with_independent_audio(
                ctx=ctx,
                main_clips=[main_clip],
                audio_clips=[],
                output_path=Path("/tmp/out.m4a"),
                video_duration=5.0,
            )

        filter_complex = captured_cmd["cmd"][captured_cmd["cmd"].index("-filter_complex") + 1]
        # 默认音量不应出现 volume= 滤镜
        assert "volume=" not in filter_complex, f"unexpected volume filter in: {filter_complex}"

    def test_main_clip_partial_volume_applied(self):
        """volume=0.5 的主视频 clip 应包含 volume=0.5000。"""
        from unittest.mock import patch

        ctx = _make_ctx()
        main_clip = self._make_clip(volume=0.5, clip_id="m1")
        captured_cmd = {}

        def fake_run_ffmpeg(cmd):
            captured_cmd["cmd"] = cmd

        with patch("video_processing.render_audio.run_ffmpeg", side_effect=fake_run_ffmpeg):
            mix_with_independent_audio(
                ctx=ctx,
                main_clips=[main_clip],
                audio_clips=[],
                output_path=Path("/tmp/out.m4a"),
                video_duration=5.0,
            )

        filter_complex = captured_cmd["cmd"][captured_cmd["cmd"].index("-filter_complex") + 1]
        assert "volume=0.5000" in filter_complex, f"volume filter missing in: {filter_complex}"