xiaoxia
4328854f58
CI/CD Pipeline / Validate Code Quality And Tests (push) Successful in 1m7s
CI/CD Pipeline / Frontend Lint (push) Successful in 2m31s
CI/CD Pipeline / Build Production Runtime Images (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 / Build & Push Staging (Watchtower auto-deploy) (push) Successful in 2m57s
CI/CD Pipeline / Staging API Integration Tests (push) Successful in 5m5s
CI/CD Pipeline / Staging E2E Tests (push) Successful in 5m29s
405 lines
14 KiB
Python
405 lines
14 KiB
Python
"""UnifiedRenderService 单元测试.
|
||
|
||
测试图层分组算法、filter_complex 构建、以及渲染流程。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from dataclasses import dataclass, field
|
||
from pathlib import Path
|
||
from typing import Any
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
import pytest
|
||
from video_processing.unified_render_service import (
|
||
RenderLayer,
|
||
RenderResult,
|
||
ResolvedClip,
|
||
UnifiedRenderService,
|
||
_resolve_layer_role,
|
||
)
|
||
|
||
# ── Fixtures ──────────────────────────────────────────────────────────────────
|
||
|
||
|
||
@dataclass
|
||
class FakeClip:
|
||
"""模拟 EditPlanClip。"""
|
||
|
||
id: str
|
||
plan_id: str = "plan_001"
|
||
clip_type: str = "main"
|
||
order: int = 0
|
||
asset_id: str = ""
|
||
text_content: str = ""
|
||
start_time: float = 0.0
|
||
duration: float = 0.0
|
||
transition_effect: str = "cut"
|
||
status: str = "ready"
|
||
config: dict[str, Any] = field(default_factory=dict)
|
||
|
||
|
||
@dataclass
|
||
class FakePlan:
|
||
"""模拟 EditPlan。"""
|
||
|
||
id: str = "plan_001"
|
||
name: str = "测试计划"
|
||
|
||
|
||
def _make_clip(
|
||
clip_id: str,
|
||
clip_type: str = "main",
|
||
order: int = 0,
|
||
asset_id: str = "",
|
||
duration: float = 0.0,
|
||
transition_effect: str = "cut",
|
||
config: dict[str, Any] | None = None,
|
||
) -> FakeClip:
|
||
return FakeClip(
|
||
id=clip_id,
|
||
clip_type=clip_type,
|
||
order=order,
|
||
asset_id=asset_id or f"asset_{clip_id}.mp4",
|
||
duration=duration,
|
||
transition_effect=transition_effect,
|
||
config=config or {},
|
||
)
|
||
|
||
|
||
def _make_service(
|
||
clips: list[FakeClip] | None = None,
|
||
asset_paths: dict[str, Path] | None = None,
|
||
work_dir: Path | None = None,
|
||
) -> UnifiedRenderService:
|
||
"""创建测试用的 UnifiedRenderService 实例。
|
||
|
||
如果未提供 asset_paths，自动从 clips 生成默认映射
|
||
（asset_id → /tmp/asset_{clip_id}.mp4）。
|
||
"""
|
||
plan = FakePlan()
|
||
clips = clips or []
|
||
work_dir = work_dir or Path("/tmp/test_render")
|
||
if asset_paths is None:
|
||
asset_paths = {}
|
||
for c in clips:
|
||
if c.asset_id:
|
||
asset_paths[c.asset_id] = Path(f"/tmp/{c.asset_id}")
|
||
return UnifiedRenderService(
|
||
plan=plan,
|
||
clips=clips,
|
||
asset_path_map=asset_paths,
|
||
work_dir=work_dir,
|
||
)
|
||
|
||
|
||
def _patch_path_exists():
|
||
"""Patch Path.exists() 让测试路径返回 True。"""
|
||
return patch("pathlib.Path.exists", return_value=True)
|
||
|
||
|
||
# ── 测试 _resolve_layer_role ─────────────────────────────────────────────────
|
||
|
||
|
||
class TestResolveLayerRole:
|
||
"""测试 clip_type → layer role 映射。"""
|
||
|
||
def test_main_default(self):
|
||
assert _resolve_layer_role("main", {}) == "main"
|
||
|
||
def test_main_with_b_roll_role(self):
|
||
assert _resolve_layer_role("main", {"role": "b_roll"}) == "broll"
|
||
|
||
def test_overlay(self):
|
||
assert _resolve_layer_role("overlay", {}) == "overlay"
|
||
|
||
def test_background(self):
|
||
assert _resolve_layer_role("background", {}) == "background"
|
||
|
||
def test_corner_voice(self):
|
||
assert _resolve_layer_role("corner_voice", {}) == "corner_voice"
|
||
|
||
def test_b_roll(self):
|
||
assert _resolve_layer_role("b_roll", {}) == "broll"
|
||
|
||
def test_intro(self):
|
||
assert _resolve_layer_role("intro", {}) == "main"
|
||
|
||
def test_outro(self):
|
||
assert _resolve_layer_role("outro", {}) == "main"
|
||
|
||
|
||
# ── 测试图层分组 ──────────────────────────────────────────────────────────────
|
||
|
||
|
||
class TestGroupClipsIntoLayers:
|
||
"""测试 _group_clips_into_layers 方法。"""
|
||
|
||
def test_group_clips_one_take(self):
|
||
"""4 个 main clips → 1 个 main_layer。"""
|
||
clips = [
|
||
_make_clip("c1", "main", order=0),
|
||
_make_clip("c2", "main", order=1),
|
||
_make_clip("c3", "main", order=2),
|
||
_make_clip("c4", "main", order=3),
|
||
]
|
||
svc = _make_service(clips)
|
||
with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
|
||
resolved = svc._resolve_clips()
|
||
layers = svc._group_clips_into_layers(resolved)
|
||
|
||
assert len(layers) == 1
|
||
assert layers[0].role == "main"
|
||
assert len(layers[0].clips) == 4
|
||
assert layers[0].z_index == 0
|
||
|
||
def test_group_clips_pip(self):
|
||
"""1 main + 2 overlay → main_layer + overlay_layer。"""
|
||
clips = [
|
||
_make_clip("c1", "main", order=0),
|
||
_make_clip("c2", "overlay", order=1),
|
||
_make_clip("c3", "overlay", order=2),
|
||
]
|
||
svc = _make_service(clips)
|
||
with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
|
||
resolved = svc._resolve_clips()
|
||
layers = svc._group_clips_into_layers(resolved)
|
||
|
||
roles = {lyr.role for lyr in layers}
|
||
assert "main" in roles
|
||
assert "overlay" in roles
|
||
|
||
main_layer = next(lyr for lyr in layers if lyr.role == "main")
|
||
overlay_layer = next(lyr for lyr in layers if lyr.role == "overlay")
|
||
assert len(main_layer.clips) == 1
|
||
assert len(overlay_layer.clips) == 2
|
||
assert overlay_layer.z_index > main_layer.z_index
|
||
|
||
def test_group_clips_voice_over(self):
|
||
"""3 个 main(b_roll) clips → 1 个 broll_layer。"""
|
||
clips = [
|
||
_make_clip("c1", "main", order=0, config={"role": "b_roll"}),
|
||
_make_clip("c2", "main", order=1, config={"role": "b_roll"}),
|
||
_make_clip("c3", "main", order=2, config={"role": "b_roll"}),
|
||
]
|
||
svc = _make_service(clips)
|
||
with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
|
||
resolved = svc._resolve_clips()
|
||
layers = svc._group_clips_into_layers(resolved)
|
||
|
||
assert len(layers) == 1
|
||
assert layers[0].role == "broll"
|
||
assert len(layers[0].clips) == 3
|
||
|
||
def test_group_clips_voice_pip(self):
|
||
"""1 background + 1 corner_voice + 2 b_roll → 3 layers。"""
|
||
clips = [
|
||
_make_clip("c1", "background", order=0),
|
||
_make_clip("c2", "corner_voice", order=1),
|
||
_make_clip("c3", "b_roll", order=2),
|
||
_make_clip("c4", "b_roll", order=3),
|
||
]
|
||
svc = _make_service(clips)
|
||
with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
|
||
resolved = svc._resolve_clips()
|
||
layers = svc._group_clips_into_layers(resolved)
|
||
|
||
roles = {lyr.role for lyr in layers}
|
||
assert roles == {"background", "corner_voice", "broll"}
|
||
assert len(layers) == 3
|
||
|
||
# z_index 排序
|
||
assert layers[0].z_index <= layers[1].z_index <= layers[2].z_index
|
||
|
||
def test_group_clips_intro_outro(self):
|
||
"""intro + 2 main + outro → 1 main_layer（4 clips，按 order 排序）。"""
|
||
clips = [
|
||
_make_clip("intro", "intro", order=0),
|
||
_make_clip("c1", "main", order=1),
|
||
_make_clip("c2", "main", order=2),
|
||
_make_clip("outro", "outro", order=3),
|
||
]
|
||
svc = _make_service(clips)
|
||
with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
|
||
resolved = svc._resolve_clips()
|
||
layers = svc._group_clips_into_layers(resolved)
|
||
|
||
assert len(layers) == 1
|
||
assert layers[0].role == "main"
|
||
assert len(layers[0].clips) == 4
|
||
# 按 order 排序
|
||
orders = [c.order for c in layers[0].clips]
|
||
assert orders == [0, 1, 2, 3]
|
||
|
||
|
||
# ── 测试 _resolve_clips ──────────────────────────────────────────────────────
|
||
|
||
|
||
class TestResolveClips:
|
||
"""测试 _resolve_clips 方法。"""
|
||
|
||
def test_skip_missing_asset(self):
|
||
"""跳过 asset_id 在 asset_path_map 中找不到的 clip。"""
|
||
clips = [
|
||
_make_clip("c1", "main", order=0, asset_id="asset_1.mp4"),
|
||
_make_clip("c2", "main", order=1, asset_id="missing.mp4"),
|
||
]
|
||
# 只有 asset_1.mp4 存在
|
||
asset_paths = {"asset_1.mp4": Path("/tmp/asset_1.mp4")}
|
||
|
||
svc = _make_service(clips, asset_paths)
|
||
|
||
with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
|
||
resolved = svc._resolve_clips()
|
||
|
||
assert len(resolved) == 1
|
||
assert resolved[0].clip_id == "c1"
|
||
|
||
def test_skip_empty_asset_id(self):
|
||
"""跳过 asset_id 为空的 clip。"""
|
||
clips = [
|
||
_make_clip("c1", "main", order=0, asset_id=""),
|
||
_make_clip("c2", "main", order=1, asset_id="asset_2.mp4"),
|
||
]
|
||
asset_paths = {"asset_2.mp4": Path("/tmp/asset_2.mp4")}
|
||
|
||
svc = _make_service(clips, asset_paths)
|
||
|
||
with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
|
||
resolved = svc._resolve_clips()
|
||
|
||
assert len(resolved) == 1
|
||
assert resolved[0].clip_id == "c2"
|
||
|
||
def test_sort_by_order(self):
|
||
"""解析后的 clips 按 order 排序。"""
|
||
clips = [
|
||
_make_clip("c3", "main", order=3, asset_id="a3.mp4"),
|
||
_make_clip("c1", "main", order=1, asset_id="a1.mp4"),
|
||
_make_clip("c2", "main", order=2, asset_id="a2.mp4"),
|
||
]
|
||
asset_paths = {
|
||
"a1.mp4": Path("/tmp/a1.mp4"),
|
||
"a2.mp4": Path("/tmp/a2.mp4"),
|
||
"a3.mp4": Path("/tmp/a3.mp4"),
|
||
}
|
||
|
||
svc = _make_service(clips, asset_paths)
|
||
|
||
with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
|
||
resolved = svc._resolve_clips()
|
||
|
||
orders = [c.order for c in resolved]
|
||
assert orders == [1, 2, 3]
|
||
|
||
|
||
# ── 测试 _build_filter_complex ───────────────────────────────────────────────
|
||
|
||
|
||
class TestBuildFilterComplex:
|
||
"""测试 _build_filter_complex 方法。"""
|
||
|
||
def test_single_layer_single_clip(self):
|
||
"""只有 1 个 main clip → 简单 scale + setpts。"""
|
||
clips = [_make_clip("c1", "main", order=0)]
|
||
asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
|
||
svc = _make_service(clips, asset_paths)
|
||
|
||
with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
|
||
resolved = svc._resolve_clips()
|
||
layers = svc._group_clips_into_layers(resolved)
|
||
fc, input_args = svc._build_filter_complex(layers)
|
||
|
||
assert "-i" in input_args
|
||
assert "/tmp/asset_c1.mp4" in input_args
|
||
assert "scale=" in fc
|
||
assert "[final_video]" in fc
|
||
|
||
def test_single_layer_multi_clips(self):
|
||
"""多个 main clips → xfade 串联。"""
|
||
clips = [
|
||
_make_clip("c1", "main", order=0, duration=3.0),
|
||
_make_clip("c2", "main", order=1, duration=3.0),
|
||
]
|
||
asset_paths = {
|
||
"asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
|
||
"asset_c2.mp4": Path("/tmp/asset_c2.mp4"),
|
||
}
|
||
svc = _make_service(clips, asset_paths)
|
||
|
||
with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
|
||
resolved = svc._resolve_clips()
|
||
layers = svc._group_clips_into_layers(resolved)
|
||
fc, input_args = svc._build_filter_complex(layers)
|
||
|
||
assert input_args.count("-i") == 2
|
||
assert "xfade=" in fc
|
||
assert "[final_video]" in fc
|
||
|
||
def test_with_overlay(self):
|
||
"""main + overlay → overlay 滤镜。"""
|
||
clips = [
|
||
_make_clip("c1", "main", order=0),
|
||
_make_clip("c2", "overlay", order=1),
|
||
]
|
||
asset_paths = {
|
||
"asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
|
||
"asset_c2.mp4": Path("/tmp/asset_c2.mp4"),
|
||
}
|
||
svc = _make_service(clips, asset_paths)
|
||
|
||
with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
|
||
resolved = svc._resolve_clips()
|
||
layers = svc._group_clips_into_layers(resolved)
|
||
fc, input_args = svc._build_filter_complex(layers)
|
||
|
||
assert "overlay=" in fc
|
||
assert "[final_video]" in fc
|
||
|
||
def test_empty_layers_raises(self):
|
||
"""空图层列表抛出 ValueError。"""
|
||
svc = _make_service()
|
||
with pytest.raises(ValueError, match="没有可渲染的图层"):
|
||
svc._build_filter_complex([])
|
||
|
||
|
||
# ── 测试 render 方法 ─────────────────────────────────────────────────────────
|
||
|
||
|
||
class TestRender:
|
||
"""测试 render 方法。"""
|
||
|
||
def test_render_empty_clips_raises(self):
|
||
"""没有 clips 时抛出 ValueError。"""
|
||
svc = _make_service(clips=[], asset_paths={})
|
||
with pytest.raises(ValueError, match="没有可渲染的片段"):
|
||
svc.render()
|
||
|
||
def test_render_with_missing_assets_raises(self):
|
||
"""所有 clips 素材缺失时抛出 ValueError。"""
|
||
clips = [_make_clip("c1", "main", order=0, asset_id="missing.mp4")]
|
||
svc = _make_service(clips, asset_paths={})
|
||
with pytest.raises(ValueError, match="没有可渲染的片段"):
|
||
svc.render()
|
||
|
||
def test_render_success(self):
|
||
"""正常渲染流程。"""
|
||
clips = [_make_clip("c1", "main", order=0)]
|
||
asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
|
||
svc = _make_service(clips, asset_paths)
|
||
|
||
with (
|
||
_patch_path_exists(),
|
||
patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
|
||
patch.object(svc, "_execute_ffmpeg") as mock_exec,
|
||
patch.object(svc, "_probe_output", return_value=(5.0, 1024, 1280, 720)),
|
||
):
|
||
result = svc.render()
|
||
|
||
assert isinstance(result, RenderResult)
|
||
assert result.duration == 5.0
|
||
assert result.file_size == 1024
|
||
assert result.width == 1280
|
||
assert result.height == 720
|
||
mock_exec.assert_called_once()
|