test(wave138): video_filter_builder domain 层 +71 单测 #1053

Closed
xiaoxia wants to merge 2 commits from test/wave138-video-filter-builder into develop
+714
View File
@@ -0,0 +1,714 @@
"""video_filter_builder 单测.
domain 层纯逻辑模块,0 FFmpeg 依赖,快速轻量。
"""
from __future__ import annotations
from unittest.mock import MagicMock
import pytest
from packages.domain.video_filter_builder import (
DEFAULT_CLIP_DURATION,
DEFAULT_FPS,
DEFAULT_OUTPUT_HEIGHT,
DEFAULT_OUTPUT_WIDTH,
DEFAULT_TRANSITION_DURATION,
XFADE_TRANSITION_MAP,
ClipFilterChain,
build_clip_filter,
build_concat_filter,
build_filter_complex,
build_xfade_filter,
chain_filters,
has_audio,
)
# ── 工具函数 ──────────────────────────────────────────────────────────────────
def _make_chain(
clip_id: str = "c1",
input_index: int = 0,
duration: float = 3.0,
has_audio: bool = True,
filters: list[str] | None = None,
) -> ClipFilterChain:
"""快速创建 ClipFilterChain."""
if filters is None:
filters = ["scale=1280:720", "fps=25", "trim=0:3"]
return ClipFilterChain(
clip_id=clip_id,
input_index=input_index,
video_label=f"v{input_index}",
audio_label=f"a{input_index}" if has_audio else None,
filters=filters,
duration=duration,
)
def _mock_clip(
clip_id: str = "c1",
duration: float = 5.0,
start_time: float = 0.0,
clip_type: str = "video",
) -> MagicMock:
"""创建 mock 的 EditPlanClip."""
clip = MagicMock()
clip.id = clip_id
clip.duration = duration
clip.start_time = start_time
clip.clip_type = clip_type
return clip
# ═══════════════════════════════════════════════════════════════════════════════
# 常量测试
# ═══════════════════════════════════════════════════════════════════════════════
class TestConstants:
"""常量默认值测试."""
def test_default_resolution(self):
"""默认分辨率为 1280x720."""
assert DEFAULT_OUTPUT_WIDTH == 1280
assert DEFAULT_OUTPUT_HEIGHT == 720
def test_default_fps(self):
"""默认帧率 25."""
assert DEFAULT_FPS == 25
def test_default_transition_duration(self):
"""默认转场时长 0.5s."""
assert DEFAULT_TRANSITION_DURATION == 0.5
def test_default_clip_duration(self):
"""默认片段时长 5s."""
assert DEFAULT_CLIP_DURATION == 5.0
def test_xfade_map_contains_common_transitions(self):
"""xfade 转场映射包含常见类型."""
assert "fade" in XFADE_TRANSITION_MAP.values()
assert "slideleft" in XFADE_TRANSITION_MAP.values()
assert "slideright" in XFADE_TRANSITION_MAP.values()
assert "dissolve" in XFADE_TRANSITION_MAP.values()
assert "wipeleft" in XFADE_TRANSITION_MAP.values()
assert len(XFADE_TRANSITION_MAP) >= 5
# ═══════════════════════════════════════════════════════════════════════════════
# ClipFilterChain 数据类
# ═══════════════════════════════════════════════════════════════════════════════
class TestClipFilterChain:
"""数据类结构测试."""
def test_creation(self):
"""创建 ClipFilterChain."""
chain = ClipFilterChain(
clip_id="c1",
input_index=0,
video_label="v0",
audio_label="a0",
filters=["scale=1280:720"],
duration=5.0,
)
assert chain.clip_id == "c1"
assert chain.input_index == 0
assert chain.video_label == "v0"
assert chain.audio_label == "a0"
assert chain.filters == ["scale=1280:720"]
assert chain.duration == 5.0
def test_no_audio(self):
"""无音频流."""
chain = _make_chain(has_audio=False)
assert chain.audio_label is None
def test_frozen(self):
"""frozen dataclass 不可修改."""
chain = _make_chain()
with pytest.raises(Exception):
chain.duration = 10.0 # type: ignore
# ═══════════════════════════════════════════════════════════════════════════════
# chain_filters
# ═══════════════════════════════════════════════════════════════════════════════
class TestChainFilters:
"""滤镜串联测试."""
def test_single_filter(self):
"""单个滤镜."""
result = chain_filters(["scale=1280:720"], "v0")
assert result == "[0:v]scale=1280:720[v0]"
def test_multiple_filters(self):
"""多个滤镜用逗号连接."""
result = chain_filters(["scale=1280:720", "fps=25", "trim=0:5"], "v0")
assert "scale=1280:720,fps=25,trim=0:5" in result
assert result.startswith("[0:v]")
assert result.endswith("[v0]")
def test_empty_filters(self):
"""空滤镜列表."""
result = chain_filters([], "out")
assert result == "[0:v][out]"
def test_custom_input_label(self):
"""自定义输入标签."""
result = chain_filters(["fps=30"], "v1", input_label="1:v")
assert result.startswith("[1:v]")
assert result.endswith("[v1]")
def test_custom_output_label(self):
"""自定义输出标签."""
result = chain_filters(["scale=640:480"], "my_label")
assert result.endswith("[my_label]")
# ═══════════════════════════════════════════════════════════════════════════════
# has_audio
# ═══════════════════════════════════════════════════════════════════════════════
class TestHasAudio:
"""音频判断测试."""
def test_all_have_audio(self):
"""全部有音频."""
chains = [_make_chain(has_audio=True), _make_chain(has_audio=True)]
assert has_audio(chains) is True
def test_none_have_audio(self):
"""全部无音频."""
chains = [_make_chain(has_audio=False), _make_chain(has_audio=False)]
assert has_audio(chains) is False
def test_partial_audio(self):
"""部分有音频."""
chains = [_make_chain(has_audio=True), _make_chain(has_audio=False)]
assert has_audio(chains) is True
def test_empty_list(self):
"""空列表."""
assert has_audio([]) is False
def test_single_with_audio(self):
"""单个有音频."""
assert has_audio([_make_chain(has_audio=True)]) is True
def test_single_without_audio(self):
"""单个无音频."""
assert has_audio([_make_chain(has_audio=False)]) is False
# ═══════════════════════════════════════════════════════════════════════════════
# build_clip_filter
# ═══════════════════════════════════════════════════════════════════════════════
class TestBuildClipFilter:
"""单片段滤镜链构建测试."""
def test_basic_video_clip(self):
"""基础视频片段."""
clip = _mock_clip(duration=5.0, start_time=0.0)
chain = build_clip_filter(clip, 0, 1280, 720, 25)
assert chain.clip_id == "c1"
assert chain.input_index == 0
assert chain.video_label == "v0"
assert chain.audio_label == "a0"
assert chain.duration == 5.0
assert len(chain.filters) >= 5
def test_contains_scale_filter(self):
"""包含 scale 滤镜."""
clip = _mock_clip()
chain = build_clip_filter(clip, 0, 1280, 720, 25)
assert any("scale=1280:720" in f for f in chain.filters)
def test_contains_pad_filter(self):
"""包含 pad 滤镜(居中黑边)."""
clip = _mock_clip()
chain = build_clip_filter(clip, 0, 1280, 720, 25)
assert any("pad=1280:720" in f for f in chain.filters)
def test_contains_format_filter(self):
"""包含 format 滤镜(yuv420p."""
clip = _mock_clip()
chain = build_clip_filter(clip, 0, 1280, 720, 25)
assert any("format=yuv420p" in f for f in chain.filters)
def test_contains_fps_filter(self):
"""包含 fps 滤镜."""
clip = _mock_clip()
chain = build_clip_filter(clip, 0, 1280, 720, 30)
assert any("fps=30" in f for f in chain.filters)
def test_zero_fps_skipped(self):
"""fps=0 时跳过 fps 滤镜."""
clip = _mock_clip()
chain = build_clip_filter(clip, 0, 1280, 720, 0)
assert not any(f.startswith("fps=") for f in chain.filters)
def test_negative_fps_skipped(self):
"""负 fps 跳过."""
clip = _mock_clip()
chain = build_clip_filter(clip, 0, 1280, 720, -1)
assert not any(f.startswith("fps=") for f in chain.filters)
def test_start_time_offset(self):
"""有 start_time 时 setpts 带偏移."""
clip = _mock_clip(start_time=2.0)
chain = build_clip_filter(clip, 0, 1280, 720, 25)
assert any("PTS-STARTPTS+2.0/TB" in f for f in chain.filters)
def test_zero_start_time_no_offset(self):
"""start_time=0 时无偏移."""
clip = _mock_clip(start_time=0.0)
chain = build_clip_filter(clip, 0, 1280, 720, 25)
assert any("setpts=PTS-STARTPTS" in f for f in chain.filters)
# 不含 +N/TB 偏移
setpts_filters = [f for f in chain.filters if f.startswith("setpts=")]
# 第一个 setpts 是重置的(不含偏移),trim 后还有一个
assert len(setpts_filters) >= 1
def test_contains_trim_filter(self):
"""包含 trim 滤镜."""
clip = _mock_clip(duration=5.0)
chain = build_clip_filter(clip, 0, 1280, 720, 25)
assert any("trim=0:5.0" in f for f in chain.filters)
def test_negative_duration_uses_default(self):
"""duration<=0 时使用默认时长."""
clip = _mock_clip(duration=0.0)
chain = build_clip_filter(clip, 0, 1280, 720, 25)
assert chain.duration == DEFAULT_CLIP_DURATION
assert any(f"trim=0:{DEFAULT_CLIP_DURATION}" in f for f in chain.filters)
def test_title_clip_no_audio(self):
"""title 类型片段无音频."""
clip = _mock_clip(clip_type="title")
chain = build_clip_filter(clip, 0, 1280, 720, 25)
assert chain.audio_label is None
def test_subtitle_clip_no_audio(self):
"""subtitle 类型片段无音频."""
clip = _mock_clip(clip_type="subtitle")
chain = build_clip_filter(clip, 0, 1280, 720, 25)
assert chain.audio_label is None
def test_video_clip_has_audio(self):
"""video 类型片段有音频."""
clip = _mock_clip(clip_type="video")
chain = build_clip_filter(clip, 0, 1280, 720, 25)
assert chain.audio_label == "a0"
def test_image_clip_has_audio(self):
"""image 类型默认有音频标签(实际无音流由调用方判断)."""
clip = _mock_clip(clip_type="image")
chain = build_clip_filter(clip, 0, 1280, 720, 25)
# 只有 title/subtitle 被排除
assert chain.audio_label is not None
def test_input_index_matches_label(self):
"""input_index 对应标签编号."""
clip = _mock_clip()
chain = build_clip_filter(clip, 3, 1280, 720, 25)
assert chain.input_index == 3
assert chain.video_label == "v3"
assert chain.audio_label == "a3"
def test_filter_order(self):
"""滤镜顺序:scale → pad → format → fps → setpts → trim."""
clip = _mock_clip(duration=5.0, start_time=1.0)
chain = build_clip_filter(clip, 0, 1280, 720, 25)
filter_names = [f.split("=")[0] for f in chain.filters]
# scale 在 pad 前
assert filter_names.index("scale") < filter_names.index("pad")
# pad 在 format 前
assert filter_names.index("pad") < filter_names.index("format")
# format 在 fps 前
assert filter_names.index("format") < filter_names.index("fps")
# ═══════════════════════════════════════════════════════════════════════════════
# build_concat_filter
# ═══════════════════════════════════════════════════════════════════════════════
class TestBuildConcatFilter:
"""concat 滤镜构建测试."""
def test_empty_returns_empty(self):
"""空列表返回空."""
result, duration = build_concat_filter([])
assert result == ""
assert duration == 0.0
def test_single_clip(self):
"""单个片段."""
chain = _make_chain(clip_id="c1", input_index=0, duration=3.0)
result, total = build_concat_filter([chain])
assert "[0:v]" in result
assert "concat=n=1:v=1:a=0" in result
assert "[outv]" in result
assert total == 3.0
def test_two_clips(self):
"""两个片段 concat."""
chains = [
_make_chain(clip_id="c1", input_index=0, duration=3.0),
_make_chain(clip_id="c2", input_index=1, duration=2.0),
]
result, total = build_concat_filter(chains)
assert "[0:v]" in result
assert "[1:v]" in result
assert "concat=n=2:v=1:a=0[outv]" in result
assert total == 5.0
def test_three_clips(self):
"""三个片段."""
chains = [
_make_chain(clip_id="c1", input_index=0, duration=2.0),
_make_chain(clip_id="c2", input_index=1, duration=3.0),
_make_chain(clip_id="c3", input_index=2, duration=1.0),
]
result, total = build_concat_filter(chains)
assert "concat=n=3:v=1:a=0[outv]" in result
assert total == 6.0
def test_total_duration_sum(self):
"""总时长 = 各片段时长之和."""
chains = [
_make_chain(duration=1.5),
_make_chain(duration=2.5),
_make_chain(duration=3.0),
]
_, total = build_concat_filter(chains)
assert abs(total - 7.0) < 0.001
def test_audio_concat_with_audio(self):
"""有音频时包含音频 concat."""
chains = [
_make_chain(input_index=0, has_audio=True),
_make_chain(input_index=1, has_audio=True),
]
result, _ = build_concat_filter(chains)
assert "[outa]" in result
assert "concat=n=2:v=0:a=1[outa]" in result
def test_audio_normalization(self):
"""音频经过 aformat 归一化."""
chains = [_make_chain(input_index=0, has_audio=True)]
result, _ = build_concat_filter(chains)
assert "aformat=sample_rates=48000" in result
assert "channel_layouts=stereo" in result
assert "sample_fmts=fltp" in result
def test_no_audio_concat(self):
"""无音频时不生成音频 concat."""
chains = [
_make_chain(input_index=0, has_audio=False),
_make_chain(input_index=1, has_audio=False),
]
result, _ = build_concat_filter(chains)
assert "[outa]" not in result
assert "aformat" not in result
def test_partial_audio_only_includes_audio_chains(self):
"""部分有音频时,只对有音频的片段做 concat."""
chains = [
_make_chain(input_index=0, has_audio=True),
_make_chain(input_index=1, has_audio=False),
_make_chain(input_index=2, has_audio=True),
]
result, _ = build_concat_filter(chains)
# 音频 concat 只有 2 个输入
assert "concat=n=2:v=0:a=1[outa]" in result
def test_video_labels_correct(self):
"""视频标签正确."""
chains = [
_make_chain(clip_id="a", input_index=0, duration=1.0),
_make_chain(clip_id="b", input_index=1, duration=1.0),
]
result, _ = build_concat_filter(chains)
assert "[v0]" in result
assert "[v1]" in result
def test_filter_chain_applied_per_clip(self):
"""每个片段都有独立的滤镜链."""
chains = [
_make_chain(input_index=0, filters=["scale=1280:720", "fps=25"]),
_make_chain(input_index=1, filters=["scale=1280:720", "fps=25"]),
]
result, _ = build_concat_filter(chains)
# 两个片段都有滤镜处理
assert result.count("scale=1280:720") == 2
# ═══════════════════════════════════════════════════════════════════════════════
# build_xfade_filter
# ═══════════════════════════════════════════════════════════════════════════════
class TestBuildXfadeFilter:
"""xfade 转场滤镜构建测试."""
def test_empty_returns_empty(self):
"""空列表返回空."""
result, duration = build_xfade_filter([], 0.5, [])
assert result == ""
assert duration == 0.0
def test_single_clip_copy(self):
"""单个片段用 copy 直接输出."""
chain = _make_chain(clip_id="c1", input_index=0, duration=3.0)
result, total = build_xfade_filter([chain], 0.5, [])
assert "copy[outv]" in result
assert total == 3.0
def test_two_clips_fade_transition(self):
"""两个片段 + fade 转场."""
chains = [
_make_chain(input_index=0, duration=3.0),
_make_chain(input_index=1, duration=2.0),
]
result, total = build_xfade_filter(chains, 0.5, ["cut", "fade"])
assert "xfade=transition=fade" in result
assert "duration=0.5" in result
assert "[outv]" in result
# 总时长 = 3 + 2 - 0.5 = 4.5
assert abs(total - 4.5) < 0.001
def test_three_clips_with_transitions(self):
"""三个片段 + 多个转场."""
chains = [
_make_chain(input_index=0, duration=3.0),
_make_chain(input_index=1, duration=2.0),
_make_chain(input_index=2, duration=4.0),
]
result, total = build_xfade_filter(chains, 0.5, ["cut", "fade", "slideleft"])
# 两个 xfade 转场
assert result.count("xfade=") == 2
assert "xf1" in result # 中间标签
# 总时长 = 3+2+4 - 0.5*2 = 8.0
assert abs(total - 8.0) < 0.001
def test_offset_calculation(self):
"""转场 offset 计算正确."""
chains = [
_make_chain(input_index=0, duration=5.0),
_make_chain(input_index=1, duration=3.0),
]
result, _ = build_xfade_filter(chains, 1.0, ["cut", "fade"])
# offset = 5.0 - 1.0*1 = 4.0
assert "offset=4.000" in result
def test_offset_never_negative(self):
"""offset 不为负."""
chains = [
_make_chain(input_index=0, duration=0.3),
_make_chain(input_index=1, duration=0.3),
]
result, _ = build_xfade_filter(chains, 1.0, ["cut", "fade"])
# offset = max(0, 0.3 - 1.0) = 0
assert "offset=0.000" in result
def test_transition_slide_left(self):
"""slideleft 转场."""
chains = [
_make_chain(input_index=0, duration=2.0),
_make_chain(input_index=1, duration=2.0),
]
result, _ = build_xfade_filter(chains, 0.5, ["cut", "slide_left"])
assert "xfade=transition=slideleft" in result
def test_transition_slide_right(self):
"""slideright 转场."""
chains = [
_make_chain(input_index=0, duration=2.0),
_make_chain(input_index=1, duration=2.0),
]
result, _ = build_xfade_filter(chains, 0.5, ["cut", "slide_right"])
assert "xfade=transition=slideright" in result
def test_transition_dissolve(self):
"""dissolve 转场."""
chains = [
_make_chain(input_index=0, duration=2.0),
_make_chain(input_index=1, duration=2.0),
]
result, _ = build_xfade_filter(chains, 0.5, ["cut", "dissolve"])
assert "xfade=transition=dissolve" in result
def test_unknown_transition_defaults_to_fade(self):
"""未知转场默认 fade."""
chains = [
_make_chain(input_index=0, duration=2.0),
_make_chain(input_index=1, duration=2.0),
]
result, _ = build_xfade_filter(chains, 0.5, ["cut", "unknown_transition"])
assert "xfade=transition=fade" in result
def test_total_duration_minus_overlap(self):
"""总时长 = sum - transition_duration * (n-1)."""
chains = [
_make_chain(duration=10.0),
_make_chain(duration=10.0),
_make_chain(duration=10.0),
]
_, total = build_xfade_filter(chains, 1.0, ["cut", "fade", "wipe"])
# 30 - 2 = 28
assert abs(total - 28.0) < 0.001
def test_total_duration_never_negative(self):
"""总时长不为负."""
chains = [
_make_chain(duration=0.1),
_make_chain(duration=0.1),
]
_, total = build_xfade_filter(chains, 10.0, ["cut", "fade"])
assert total >= 0.0
def test_audio_with_xfade_path(self):
"""xfade 路径下音频也做 concat."""
chains = [
_make_chain(input_index=0, has_audio=True, duration=2.0),
_make_chain(input_index=1, has_audio=True, duration=3.0),
]
result, _ = build_xfade_filter(chains, 0.5, ["cut", "fade"])
assert "[outa]" in result
assert "aformat=" in result
def test_single_xfade_no_audio_processing(self):
"""单片段 xfade 路径不处理音频(与原实现一致)."""
chain = _make_chain(input_index=0, has_audio=True, duration=3.0)
result, _ = build_xfade_filter([chain], 0.5, [])
# 单片段 xfade 只有视频 copy,不处理音频
assert "copy[outv]" in result
assert "[outa]" not in result
assert "acopy" not in result
def test_no_audio_xfade(self):
"""无音频时不生成 [outa]."""
chains = [
_make_chain(input_index=0, has_audio=False, duration=2.0),
_make_chain(input_index=1, has_audio=False, duration=2.0),
]
result, _ = build_xfade_filter(chains, 0.5, ["cut", "fade"])
assert "[outa]" not in result
def test_intermediate_labels(self):
"""多片段时有中间 xf 标签."""
chains = [
_make_chain(input_index=0, duration=1.0),
_make_chain(input_index=1, duration=1.0),
_make_chain(input_index=2, duration=1.0),
_make_chain(input_index=3, duration=1.0),
]
result, _ = build_xfade_filter(chains, 0.3, ["cut", "fade", "wipe", "dissolve"])
assert "[xf1]" in result
assert "[xf2]" in result
assert "[outv]" in result
def test_transitions_shorter_than_clips(self):
"""transitions 列表比片段短时,后续用默认值."""
chains = [
_make_chain(input_index=0, duration=2.0),
_make_chain(input_index=1, duration=2.0),
_make_chain(input_index=2, duration=2.0),
]
# 只给一个转场(索引1有效,索引2越界)
result, _ = build_xfade_filter(chains, 0.5, ["cut", "fade"])
# 第2个转场(索引2)未知 → 默认 fade
assert result.count("xfade=transition=fade") == 2
# ═══════════════════════════════════════════════════════════════════════════════
# build_filter_complex
# ═══════════════════════════════════════════════════════════════════════════════
class TestBuildFilterComplex:
"""完整 filter_complex 构建(策略选择)测试."""
def test_empty_returns_empty(self):
"""空列表返回空."""
result, duration = build_filter_complex([], 1280, 720, 0.5, [])
assert result == ""
assert duration == 0.0
def test_single_clip_chain_mode(self):
"""单片段走单链模式."""
chain = _make_chain(input_index=0, duration=3.0, has_audio=True)
result, total = build_filter_complex([chain], 1280, 720, 0.5, [])
assert "[0:v]" in result
assert "[0:a]" in result
assert total == 3.0
def test_single_clip_no_audio(self):
"""单片段无音频."""
chain = _make_chain(input_index=0, duration=3.0, has_audio=False)
result, _ = build_filter_complex([chain], 1280, 720, 0.5, [])
assert "[0:a]" not in result
def test_multiple_clips_all_cut_uses_concat(self):
"""多片段 + 全 cut → 走 concat(高效模式)."""
chains = [
_make_chain(input_index=0, duration=2.0),
_make_chain(input_index=1, duration=3.0),
]
result, total = build_filter_complex(chains, 1280, 720, 0.5, ["cut", "cut"])
# concat 模式
assert "concat=n=2:v=1:a=0" in result
assert "xfade" not in result
assert total == 5.0
def test_multiple_clips_with_transition_uses_xfade(self):
"""多片段 + 有转场 → 走 xfade."""
chains = [
_make_chain(input_index=0, duration=2.0),
_make_chain(input_index=1, duration=3.0),
]
result, total = build_filter_complex(chains, 1280, 720, 0.5, ["cut", "fade"])
assert "xfade=" in result
assert abs(total - 4.5) < 0.001
def test_transition_effect_enum_value(self):
"""使用 TransitionEffect 枚举值也能正确判断."""
from packages.domain.template_clip_config import TransitionEffect
chains = [
_make_chain(input_index=0, duration=2.0),
_make_chain(input_index=1, duration=2.0),
]
# 传 TransitionEffect.CUT(不是字符串 "cut"
result, _ = build_filter_complex(
chains,
1280,
720,
0.5,
[TransitionEffect.CUT, TransitionEffect.CUT],
)
# 都是 cut → 走 concat
assert "concat=n=2:v=1:a=0" in result
def test_mixed_cut_and_transition(self):
"""混合 cut 和转场 → 走 xfade."""
chains = [
_make_chain(input_index=0, duration=1.0),
_make_chain(input_index=1, duration=1.0),
_make_chain(input_index=2, duration=1.0),
]
result, total = build_filter_complex(chains, 1280, 720, 0.5, ["cut", "fade", "cut"])
# 只要有一个非 cut 转场就走 xfade
assert "xfade=" in result
assert abs(total - 2.0) < 0.001