diff --git a/tests/unit/domain/test_video_filter_builder.py b/tests/unit/domain/test_video_filter_builder.py new file mode 100755 index 000000000..f47d83e7b --- /dev/null +++ b/tests/unit/domain/test_video_filter_builder.py @@ -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