saas-backend-agent
ba9837986e
- unified_render_service filter_complex Step1 trim 时长改为 effective-freeze（与直通路径同口径，tpad 补冻结） - test_duration_compensation / render_audio_utils / render_audio_pure / 两处 render_layer_utils：min(duration,actual) 断言改为目标段长为准 - estimate_total_duration 测试：cut 零重叠、非 cut 逐处扣转场（FakeClip 加 transition_effect/duration） - test_generation_tasks/preview：voice_ids kwarg/字段移除，配音改 voice_library_id
247 lines
8.4 KiB
Python
Executable File
247 lines
8.4 KiB
Python
Executable File
"""渲染音频纯逻辑测试 — 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):
|
||
"""#1749：duration > actual → 返回目标段长 duration（素材短于段长走末帧冻结/补静音，不慢放不截配音）."""
|
||
clip = _make_clip(duration=50.0, actual_duration=30.0)
|
||
assert clip_effective_duration(clip) == pytest.approx(50.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}"
|