e83a0b86f8
CI/CD Pipeline / Check if frontend-only change (push) Has been skipped
CI/CD Pipeline / Dedup Check - skip PR tests when covered by push pipeline (push) Successful in 2s
CI/CD Pipeline / Check push changed paths (pull_request) Has been skipped
CI/CD Pipeline / PR Build Worker Image (push) Has been skipped
CI/CD Pipeline / PR Build Web Image (push) Has been skipped
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 2s
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Dedup Check - skip PR tests when covered by push pipeline (pull_request) Successful in 3s
CI/CD Pipeline / PR Build API Image (push) Has been skipped
CI/CD Pipeline / Frontend Lint (push) Has been skipped
CI/CD Pipeline / Validate - Style (pull_request) Has been skipped
CI/CD Pipeline / Validate - Security (pull_request) Has been skipped
CI/CD Pipeline / Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / Validate - Python (mypy + alembic) (pull_request) Has been skipped
CI/CD Pipeline / Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / Frontend Lint (pull_request) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / Check push changed paths (push) Successful in 9s
CI/CD Pipeline / Retag skipped Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Retag skipped Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been skipped
CI/CD Pipeline / PR Build API Image (pull_request) Successful in 35s
CI/CD Pipeline / PR Build Worker Image (pull_request) Successful in 38s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been skipped
CI/CD Pipeline / Canary Release to Production (pull_request) Has been skipped
CI/CD Pipeline / Build Staging API Image (push) Successful in 58s
CI/CD Pipeline / Build Staging Worker Image (push) Successful in 55s
CI/CD Pipeline / Validate - Python (mypy + alembic) (push) Successful in 2m5s
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 2m18s
CI/CD Pipeline / Validate - Style (push) Successful in 2m31s
CI/CD Pipeline / PR Build Web Image (pull_request) Successful in 2m30s
CI/CD Pipeline / Integration Tests (push) Successful in 2m35s
CI/CD Pipeline / CI Gate (pull_request) Successful in 1s
CI/CD Pipeline / Build Staging Web Image (push) Successful in 2m7s
CI/CD Pipeline / Retag skipped Staging API Image (push) Has been skipped
CI/CD Pipeline / Retag skipped Staging Worker Image (push) Has been skipped
CI/CD Pipeline / Retag skipped Staging Web Image (push) Has been skipped
PR Automation / Auto Approve on CI Green (pull_request) Successful in 3m15s
PR Automation / Auto Merge on CI Green + Approved (pull_request) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Successful in 1m30s
CI/CD Pipeline / Validate - Security (push) Successful in 4m34s
CI/CD Pipeline / ACR Image Cleanup (push) Successful in 1m37s
CI/CD Pipeline / Frontend Unit Tests (push) Successful in 5m54s
AI Code Review / AI Code Review (pull_request) Successful in 6m19s
CI/CD Pipeline / Staging API Integration Tests (push) Successful in 3m57s
CI/CD Pipeline / Unit Tests (push) Successful in 8m33s
CI/CD Pipeline / Build Production API Image (push) Has been skipped
CI/CD Pipeline / Build Production Web Image (push) Has been skipped
CI/CD Pipeline / Build Production Worker Image (push) Has been skipped
CI/CD Pipeline / CI Gate (push) Has been skipped
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / Canary Release to Production (push) Has been skipped
CI/CD Pipeline / Staging E2E Tests (push) Failing after 4m48s
Co-authored-by: xiaoxia <dev@xiaoxiajianji.com> Co-committed-by: xiaoxia <dev@xiaoxiajianji.com>
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}"
|