"""xfade_builder 单测. domain 层 XFade 转场滤镜构建纯逻辑模块,0 FFmpeg 依赖。 覆盖:转场名称映射、滤镜链串联、xfade 滤镜链构建(含 duration 钳制)。 """ from __future__ import annotations from unittest.mock import MagicMock import pytest from packages.domain.xfade_builder import ( DEFAULT_TRANSITION_DURATION, SUPPORTED_TRANSITIONS, XFADE_TRANSITION_MAP, XFade_TRANSITION_NAMES, build_xfade_filter_chain, chain_filters, resolve_xfade_transition, ) # ═══════════════════════════════════════════════════════════════════════════════ # 常量测试 # ═══════════════════════════════════════════════════════════════════════════════ class TestConstants: """常量测试.""" def test_default_duration(self): """默认转场时长 0.5s.""" assert DEFAULT_TRANSITION_DURATION == 0.5 def test_transition_map_not_empty(self): """转场映射非空.""" assert len(XFADE_TRANSITION_MAP) > 10 def test_supported_transitions(self): """支持的转场数量与映射键一致.""" assert len(SUPPORTED_TRANSITIONS) == len(XFADE_TRANSITION_MAP) def test_output_names_subset(self): """输出名称是映射值的集合.""" assert XFade_TRANSITION_NAMES == set(XFADE_TRANSITION_MAP.values()) def test_fade_in_map(self): """fade 是基础转场.""" assert "fade" in XFADE_TRANSITION_MAP assert XFADE_TRANSITION_MAP["fade"] == "fade" def test_dissolve_aliases(self): """dissolve 有多个别名.""" assert XFADE_TRANSITION_MAP["dissolve"] == "dissolve" assert XFADE_TRANSITION_MAP["crossfade"] == "dissolve" assert XFADE_TRANSITION_MAP["crossdissolve"] == "dissolve" def test_slide_directions(self): """4 方向滑动都有映射.""" assert XFADE_TRANSITION_MAP["slideleft"] == "slideleft" assert XFADE_TRANSITION_MAP["slideright"] == "slideright" assert XFADE_TRANSITION_MAP["slideup"] == "slideup" assert XFADE_TRANSITION_MAP["slidedown"] == "slidedown" def test_slide_underscore_aliases(self): """下划线别名也支持.""" assert XFADE_TRANSITION_MAP["slide_left"] == "slideleft" assert XFADE_TRANSITION_MAP["slide_right"] == "slideright" assert XFADE_TRANSITION_MAP["slide_up"] == "slideup" assert XFADE_TRANSITION_MAP["slide_down"] == "slidedown" def test_slide_default_direction(self): """slide 默认向左滑.""" assert XFADE_TRANSITION_MAP["slide"] == "slideleft" def test_wipe_directions(self): """4 方向擦除.""" assert XFADE_TRANSITION_MAP["wipeleft"] == "wipeleft" assert XFADE_TRANSITION_MAP["wiperight"] == "wiperight" assert XFADE_TRANSITION_MAP["wipeup"] == "wipeup" assert XFADE_TRANSITION_MAP["wipedown"] == "wipedown" def test_wipe_default(self): """wipe 默认向左擦.""" assert XFADE_TRANSITION_MAP["wipe"] == "wipeleft" def test_zoom(self): """缩放转场.""" assert XFADE_TRANSITION_MAP["zoom"] == "zoomin" assert XFADE_TRANSITION_MAP["zoomin"] == "zoomin" assert XFADE_TRANSITION_MAP["zoomout"] == "zoomout" def test_circle_rect(self): """圆形/矩形裁剪.""" assert XFADE_TRANSITION_MAP["circle"] == "circlecrop" assert XFADE_TRANSITION_MAP["circlecrop"] == "circlecrop" assert XFADE_TRANSITION_MAP["rect"] == "rectcrop" assert XFADE_TRANSITION_MAP["rectcrop"] == "rectcrop" # ═══════════════════════════════════════════════════════════════════════════════ # 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", "format=yuv420p"], "out") assert "scale=1280:720,fps=25,format=yuv420p" in result 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="2:v") assert result.startswith("[2:v]") def test_custom_output_label(self): """自定义输出标签.""" result = chain_filters(["scale=640:480"], "my_output") assert result.endswith("[my_output]") def test_preserves_filter_order(self): """保持滤镜顺序.""" filters = ["a", "b", "c", "d"] result = chain_filters(filters, "out") idx_a = result.index("a") idx_b = result.index("b") idx_c = result.index("c") idx_d = result.index("d") assert idx_a < idx_b < idx_c < idx_d # ═══════════════════════════════════════════════════════════════════════════════ # resolve_xfade_transition # ═══════════════════════════════════════════════════════════════════════════════ class TestResolveXfadeTransition: """转场名称解析测试.""" def test_fade(self): """fade → fade.""" assert resolve_xfade_transition("fade") == "fade" def test_dissolve(self): """dissolve → dissolve.""" assert resolve_xfade_transition("dissolve") == "dissolve" def test_slide_left_alias(self): """slide_left 别名.""" assert resolve_xfade_transition("slide_left") == "slideleft" def test_slide_default(self): """slide 默认向左.""" assert resolve_xfade_transition("slide") == "slideleft" def test_zoom_default(self): """zoom 默认 zoomin.""" assert resolve_xfade_transition("zoom") == "zoomin" def test_wipe_default(self): """wipe 默认向左擦.""" assert resolve_xfade_transition("wipe") == "wipeleft" def test_unknown_falls_back_to_fade(self): """未知转场回退到 fade.""" assert resolve_xfade_transition("unknown_effect") == "fade" def test_empty_string_fades(self): """空字符串回退到 fade.""" assert resolve_xfade_transition("") == "fade" def test_enum_value(self): """支持枚举(有 .value 属性).""" mock_enum = MagicMock() mock_enum.value = "dissolve" assert resolve_xfade_transition(mock_enum) == "dissolve" def test_enum_unknown_value_fades(self): """枚举值未知时回退到 fade.""" mock_enum = MagicMock() mock_enum.value = "not_a_real_effect" assert resolve_xfade_transition(mock_enum) == "fade" def test_circle_alias(self): """circle 别名.""" assert resolve_xfade_transition("circle") == "circlecrop" def test_rect_alias(self): """rect 别名.""" assert resolve_xfade_transition("rect") == "rectcrop" # ═══════════════════════════════════════════════════════════════════════════════ # build_xfade_filter_chain — 基础结构 # ═══════════════════════════════════════════════════════════════════════════════ class TestBuildXfadeFilterChainBasic: """xfade 滤镜链基础结构测试.""" def test_empty_clips(self): """空片段返回空.""" result, duration = build_xfade_filter_chain([], [], []) assert result == "" assert duration == 0.0 def test_single_clip_copy(self): """单个片段用 copy.""" result, duration = build_xfade_filter_chain([5.0], ["v0"], []) assert "[v0]copy[outv]" in result assert duration == 5.0 def test_two_clips_fade(self): """两个片段 + fade 转场.""" result, total = build_xfade_filter_chain([5.0, 3.0], ["v0", "v1"], ["cut", "fade"]) assert "xfade=transition=fade" in result assert ":duration=0.500" in result assert "[outv]" in result # 总时长 = 5 + 3 - 0.5 = 7.5 assert abs(total - 7.5) < 0.01 def test_three_clips(self): """三个片段有 2 个 xfade.""" result, total = build_xfade_filter_chain( [4.0, 3.0, 5.0], ["v0", "v1", "v2"], ["cut", "fade", "dissolve"], ) assert result.count("xfade=") == 2 assert "xf1" in result # 中间标签 # 总时长 = 4 + 3 + 5 - 0.5*2 = 11.0 assert abs(total - 11.0) < 0.01 def test_output_label_custom(self): """自定义输出标签.""" result, _ = build_xfade_filter_chain( [5.0, 3.0], ["v0", "v1"], ["cut", "fade"], output_label="final", ) assert result.endswith("[final]") assert "[outv]" not in result def test_custom_transition_duration(self): """自定义转场时长.""" result, total = build_xfade_filter_chain( [5.0, 3.0], ["v0", "v1"], ["cut", "fade"], transition_duration=1.0, ) assert ":duration=1.000" in result assert abs(total - 7.0) < 0.01 # 5+3-1 = 7 def test_intermediate_labels(self): """多片段使用中间 xf 标签.""" result, _ = build_xfade_filter_chain( [1.0, 1.0, 1.0, 1.0, 1.0], ["v0", "v1", "v2", "v3", "v4"], ["cut"] * 5, ) # 5 个片段 = 4 个 xfade,中间标签 xf1, xf2, xf3 assert "[xf1]" in result assert "[xf2]" in result assert "[xf3]" in result def test_video_labels_used(self): """使用传入的视频标签.""" result, _ = build_xfade_filter_chain( [2.0, 2.0], ["clip_a", "clip_b"], ["cut", "fade"], ) assert "[clip_a]" in result assert "[clip_b]" in result # ═══════════════════════════════════════════════════════════════════════════════ # build_xfade_filter_chain — offset 计算 # ═══════════════════════════════════════════════════════════════════════════════ class TestBuildXfadeFilterChainOffset: """xfade offset 计算测试.""" def test_two_clips_offset(self): """两片段 offset = dur0 - td.""" result, _ = build_xfade_filter_chain( [10.0, 5.0], ["v0", "v1"], ["cut", "fade"], transition_duration=1.0, ) # offset = 10 - 1*1 = 9 assert ":offset=9.000" in result def test_three_clips_second_offset(self): """三片段第二个 offset.""" result, _ = build_xfade_filter_chain( [5.0, 4.0, 3.0], ["v0", "v1", "v2"], ["cut", "fade", "fade"], transition_duration=0.5, ) # 第一个 xfade offset = 5 - 0.5*1 = 4.5 # 第二个:cumulative = 5+4 = 9, offset = 9 - 0.5*2 = 8 assert ":offset=4.500" in result assert ":offset=8.000" in result def test_offset_never_negative(self): """offset 不为负.""" result, _ = build_xfade_filter_chain( [0.1, 0.1], ["v0", "v1"], ["cut", "fade"], transition_duration=1.0, ) # 找 offset 的值 import re offsets = re.findall(r"offset=([\d.]+)", result) for off in offsets: assert float(off) >= 0.0 def test_short_first_clip_clamps_td(self): """第一个片段很短时,转场时长被钳制.""" result, _ = build_xfade_filter_chain( [0.3, 2.0], ["v0", "v1"], ["cut", "fade"], transition_duration=1.0, ) # 第一片段只有 0.3s,offset ≈ 0, available ≈ 0.3, td 被钳制 import re durations = re.findall(r"duration=([\d.]+)", result) # 第一个 duration 是 xfade 的 duration xfade_dur = float(durations[0]) assert xfade_dur <= 0.3 # 不能超过第一个片段时长 # ═══════════════════════════════════════════════════════════════════════════════ # build_xfade_filter_chain — duration 钳制(防 exit 234) # ═══════════════════════════════════════════════════════════════════════════════ class TestBuildXfadeFilterChainClamping: """duration 钳制逻辑测试(防 FFmpeg exit 234).""" def test_td_not_exceed_first_input(self): """转场时长不超过第一个输入的可用时长.""" # 第一个片段 1s,转场 2s → 被钳制 result, total = build_xfade_filter_chain( [1.0, 3.0], ["v0", "v1"], ["cut", "fade"], transition_duration=2.0, ) import re durations = re.findall(r"xfade=transition=fade:duration=([\d.]+)", result) assert float(durations[0]) <= 1.0 # 总时长不会比 1+3 = 4 还大(钳制后 td < 2) assert total < 4.0 def test_td_not_exceed_second_input(self): """转场时长不超过第二个片段时长.""" result, _ = build_xfade_filter_chain( [3.0, 0.2], ["v0", "v1"], ["cut", "fade"], transition_duration=1.0, ) import re durations = re.findall(r"xfade=transition=fade:duration=([\d.]+)", result) assert float(durations[0]) <= 0.2 def test_td_minimum_1ms(self): """td 至少 1ms.""" result, _ = build_xfade_filter_chain( [0.0001, 0.0001], ["v0", "v1"], ["cut", "fade"], transition_duration=0.0, ) import re durations = re.findall(r"xfade=transition=fade:duration=([\d.]+)", result) if durations: assert float(durations[0]) >= 0.001 def test_many_short_clips(self): """多个极短片段.""" n = 5 durations = [0.2] * n labels = [f"v{i}" for i in range(n)] result, total = build_xfade_filter_chain( durations, labels, ["cut"] * n, transition_duration=0.5, ) # 4 个转场 assert result.count("xfade=") == 4 # 总时长合理:sum = 1.0,减去被钳制的转场 assert total > 0 assert total <= sum(durations) def test_second_xfade_first_input_is_accumulated(self): """第二个 xfade 的第一个输入时长是累积值(考虑之前的转场扣减).""" # 三个片段,转场比较长,验证第二步钳制 result, total = build_xfade_filter_chain( [2.0, 2.0, 2.0], ["v0", "v1", "v2"], ["cut", "fade", "fade"], transition_duration=1.0, ) # 第一个 xfade: first_input_dur = 2.0, td = min(1.0, 2.0-offset) # offset = 2 - 1*1 = 1.0, available = 2.0 - 1.0 = 1.0, td = 1.0 # 第二个 xfade: first_input_dur = (2+2) - 1.0 = 3.0(累积 - 已用转场) # offset = 4 - 1*2 = 2.0, available = 3.0 - 2.0 = 1.0, td = min(1.0, 1.0, 2.0) = 1.0 import re dur_match = re.findall(r":duration=([\d.]+)", result) # 两个 xfade,每个 duration 都是 1.0(正常情况) assert len(dur_match) == 2 assert float(dur_match[0]) == 1.0 assert float(dur_match[1]) == 1.0 def test_total_duration_positive(self): """总时长不为负.""" _, total = build_xfade_filter_chain( [0.1, 0.1], ["v0", "v1"], ["cut", "fade"], transition_duration=10.0, ) assert total >= 0.0 # ═══════════════════════════════════════════════════════════════════════════════ # build_xfade_filter_chain — 转场类型 # ═══════════════════════════════════════════════════════════════════════════════ class TestBuildXfadeFilterChainTransitions: """不同转场类型测试.""" def test_slideleft_transition(self): """slideleft 转场.""" result, _ = build_xfade_filter_chain([3.0, 2.0], ["v0", "v1"], ["cut", "slideleft"]) assert "xfade=transition=slideleft" in result def test_slide_left_alias_resolved(self): """slide_left 别名解析正确.""" result, _ = build_xfade_filter_chain([3.0, 2.0], ["v0", "v1"], ["cut", "slide_left"]) assert "xfade=transition=slideleft" in result def test_dissolve_transition(self): """dissolve 转场.""" result, _ = build_xfade_filter_chain([3.0, 2.0], ["v0", "v1"], ["cut", "dissolve"]) assert "xfade=transition=dissolve" in result def test_zoom_transition(self): """zoom 转场 → zoomin.""" result, _ = build_xfade_filter_chain([3.0, 2.0], ["v0", "v1"], ["cut", "zoom"]) assert "xfade=transition=zoomin" in result def test_wipe_transition(self): """wipe 转场.""" result, _ = build_xfade_filter_chain([3.0, 2.0], ["v0", "v1"], ["cut", "wipeup"]) assert "xfade=transition=wipeup" in result def test_unknown_transition_fade(self): """未知转场回退到 fade.""" result, _ = build_xfade_filter_chain([3.0, 2.0], ["v0", "v1"], ["cut", "nonexistent"]) assert "xfade=transition=fade" in result def test_cut_resolved_as_fade(self): """cut 也回退到 fade(调用方应对 cut 做特殊处理,但这里也能工作).""" result, _ = build_xfade_filter_chain([3.0, 2.0], ["v0", "v1"], ["cut", "cut"]) # cut 不在映射里,回退到 fade assert "xfade=transition=fade" in result def test_transitions_shorter_than_clips(self): """transitions 比片段少时,超出部分用 cut/fade.""" result, _ = build_xfade_filter_chain( [1.0, 1.0, 1.0, 1.0], ["v0", "v1", "v2", "v3"], ["cut", "fade"], # 只有 2 个转场 ) # 4 个片段 = 3 个 xfade,第三个用默认(cut→fade) assert result.count("xfade=") == 3 # ═══════════════════════════════════════════════════════════════════════════════ # build_xfade_filter_chain — 总时长验证 # ═══════════════════════════════════════════════════════════════════════════════ class TestBuildXfadeFilterChainTotalDuration: """总时长计算验证.""" def test_two_equal_clips_default_td(self): """两个等长片段 + 默认 0.5s 转场.""" _, total = build_xfade_filter_chain( [5.0, 5.0], ["v0", "v1"], ["cut", "fade"], ) assert abs(total - 9.5) < 0.01 # 5+5-0.5 def test_three_clips_two_transitions(self): """三个片段两个转场.""" _, total = build_xfade_filter_chain( [3.0, 4.0, 3.0], ["v0", "v1", "v2"], ["cut"] * 3, transition_duration=0.5, ) # 10 - 1.0 = 9.0 assert abs(total - 9.0) < 0.01 def test_single_clip_no_transition_loss(self): """单个片段无转场扣减.""" _, total = build_xfade_filter_chain([10.0], ["v0"], []) assert total == 10.0 def test_zero_duration_clips(self): """0 时长片段不崩溃.""" result, total = build_xfade_filter_chain( [0.0, 0.0], ["v0", "v1"], ["cut", "fade"], ) assert total >= 0.0 assert "xfade=" in result # 仍然生成转场(td 被钳制到最小) def test_very_long_transition_clamped(self): """极长转场被钳制,总时长仍为正.""" _, total = build_xfade_filter_chain( [2.0, 2.0], ["v0", "v1"], ["cut", "fade"], transition_duration=100.0, ) # 总时长 > 0 assert total > 0.0 # 且 < sum(durations) = 4(因为被钳制但还有重叠) assert total < 4.0 def test_many_clips_linear_total(self): """多片段总时长近似线性增长.""" n = 10 durations = [1.0] * n labels = [f"v{i}" for i in range(n)] _, total = build_xfade_filter_chain( durations, labels, ["cut"] * n, transition_duration=0.1, ) # 10 - 9*0.1 = 9.1 assert abs(total - 9.1) < 0.05