xiaoxia-saas/tests/unit/domain/test_xfade_builder.py

"""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