tests/unit/test_ffmpeg_pure_utils.py

@@ -0,0 +1,209 @@
+"""FFmpeg工具函数纯逻辑测试 — chain_filters / resolve_xfade_transition / build_xfade_filter_chain."""
+
+from __future__ import annotations
+
+import pytest
+
+from video_processing.ffmpeg_utils import (
+    XFADE_TRANSITION_MAP,
+    chain_filters,
+    resolve_xfade_transition,
+    build_xfade_filter_chain,
+)
+
+
+class TestChainFilters:
+    """chain_filters 滤镜串联测试."""
+
+    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 result == "[0:v]scale=1280:720,fps=25,format=yuv420p[out]"
+
+    def test_empty_filters(self):
+        """空滤镜列表."""
+        result = chain_filters([], "v0")
+        assert result == "[0:v][v0]"
+
+    def test_custom_input_label(self):
+        """自定义输入标签."""
+        result = chain_filters(["scale=640:480"], "v1", input_label="1:v")
+        assert result == "[1:v]scale=640:480[v1]"
+
+
+class TestResolveXfadeTransition:
+    """resolve_xfade_transition 转场名称映射测试."""
+
+    def test_direct_match_fade(self):
+        """fade直接匹配."""
+        assert resolve_xfade_transition("fade") == "fade"
+
+    def test_direct_match_dissolve(self):
+        """dissolve直接匹配."""
+        assert resolve_xfade_transition("dissolve") == "dissolve"
+
+    def test_alias_crossfade(self):
+        """crossfade别名→dissolve."""
+        assert resolve_xfade_transition("crossfade") == "dissolve"
+
+    def test_alias_slide_left(self):
+        """slide_left别名→slideleft."""
+        assert resolve_xfade_transition("slide_left") == "slideleft"
+
+    def test_unknown_fallback_to_fade(self):
+        """未知值回退到fade."""
+        assert resolve_xfade_transition("nonexistent_effect") == "fade"
+
+    def test_empty_string_fallback(self):
+        """空字符串回退."""
+        assert resolve_xfade_transition("") == "fade"
+
+    def test_enum_value_support(self):
+        """支持带value属性的枚举对象."""
+
+        class FakeEnum:
+            value = "slideup"
+
+        assert resolve_xfade_transition(FakeEnum()) == "slideup"
+
+    def test_all_map_keys_resolve(self):
+        """映射表中所有key都能解析到有效值."""
+        for key in XFADE_TRANSITION_MAP:
+            result = resolve_xfade_transition(key)
+            assert result and isinstance(result, str)
+            assert result != ""
+
+    def test_cut_is_special_fallback(self):
+        """cut不在映射表中→回退到fade（硬切由调用方处理）."""
+        # cut是特殊值，不在映射表里
+        result = resolve_xfade_transition("cut")
+        # 不在映射表里就fallback到fade
+        assert result == "fade"
+
+
+class TestBuildXfadeFilterChain:
+    """build_xfade_filter_chain 转场滤镜链构建测试."""
+
+    def test_zero_clips(self):
+        """0个片段→空字符串+0时长."""
+        filter_str, total_dur = build_xfade_filter_chain([], [], [])
+        assert filter_str == ""
+        assert total_dur == 0.0
+
+    def test_single_clip(self):
+        """1个片段→直接copy，总时长等于片段时长."""
+        filter_str, total_dur = build_xfade_filter_chain(
+            [10.0], ["v0"], [], output_label="outv"
+        )
+        assert "[v0]copy[outv]" in filter_str
+        assert total_dur == pytest.approx(10.0)
+
+    def test_two_clips_basic(self):
+        """2个片段基本转场."""
+        filter_str, total_dur = build_xfade_filter_chain(
+            [5.0, 5.0],
+            ["v0", "v1"],
+            ["", "fade"],
+            transition_duration=0.5,
+            output_label="outv",
+        )
+        assert "xfade=transition=fade" in filter_str
+        assert "offset=" in filter_str
+        # 总时长 = 5 + 5 - 转场重叠
+        assert total_dur == pytest.approx(9.5)
+
+    def test_three_clips_chain(self):
+        """3个片段形成链式转场."""
+        filter_str, total_dur = build_xfade_filter_chain(
+            [3.0, 4.0, 5.0],
+            ["v0", "v1", "v2"],
+            ["", "fade", "dissolve"],
+            transition_duration=0.5,
+            output_label="out",
+        )
+        # 应该有2个xfade操作
+        assert filter_str.count("xfade=") == 2
+        assert "transition=fade" in filter_str
+        assert "transition=dissolve" in filter_str
+        # 总时长 = 3+4+5 - 2*0.5 = 11
+        assert total_dur == pytest.approx(11.0)
+
+    def test_transition_duration_clamped_to_clip(self):
+        """转场时长不能超过单个片段时长."""
+        filter_str, total_dur = build_xfade_filter_chain(
+            [2.0, 1.0],
+            ["v0", "v1"],
+            ["", "fade"],
+            transition_duration=3.0,  # 比第二个片段还长
+            output_label="outv",
+        )
+        # 转场时长被钳制到第二个片段时长(1.0)
+        assert "duration=1.000" in filter_str
+        assert total_dur == pytest.approx(2.0)  # 2 + 1 - 1 = 2
+
+    def test_very_short_clip_min_transition(self):
+        """极短片段至少保留1ms转场."""
+        filter_str, total_dur = build_xfade_filter_chain(
+            [1.0, 0.0001],
+            ["v0", "v1"],
+            ["", "fade"],
+            transition_duration=0.5,
+            output_label="outv",
+        )
+        # 至少有1ms
+        assert "duration=0.001" in filter_str
+
+    def test_transition_offset_calculation(self):
+        """offset计算验证."""
+        filter_str, _ = build_xfade_filter_chain(
+            [10.0, 10.0],
+            ["v0", "v1"],
+            ["", "fade"],
+            transition_duration=1.0,
+            output_label="outv",
+        )
+        # offset = max(0, 10 - 1*1) = 9
+        assert "offset=9.000" in filter_str
+
+    def test_fewer_transitions_than_clips(self):
+        """转场列表比片段少时使用cut(fallback to fade)."""
+        filter_str, total_dur = build_xfade_filter_chain(
+            [5.0, 5.0, 5.0],
+            ["v0", "v1", "v2"],
+            ["fade"],  # 只有1个转场，第2个转场缺省
+            transition_duration=0.5,
+            output_label="out",
+        )
+        # 应该有2个xfade
+        assert filter_str.count("xfade=") == 2
+        # 第二个xfade的转场是cut→fade fallback
+        assert filter_str.count("transition=fade") == 2
+
+    def test_output_label_final_clip(self):
+        """最后一个xfade的输出标签是output_label."""
+        filter_str, _ = build_xfade_filter_chain(
+            [3.0, 4.0, 5.0],
+            ["v0", "v1", "v2"],
+            ["", "fade", "slideleft"],
+            output_label="final_v",
+        )
+        assert filter_str.rstrip().endswith("[final_v]")
+
+    def test_intermediate_labels(self):
+        """中间步骤使用xf1, xf2等标签（从i=1开始计数）."""
+        filter_str, _ = build_xfade_filter_chain(
+            [2.0, 3.0, 4.0, 5.0],
+            ["v0", "v1", "v2", "v3"],
+            ["", "fade", "fade", "fade"],
+            output_label="out",
+        )
+        # 4个片段3次xfade，中间标签是xf1, xf2
+        assert "[xf1]" in filter_str
+        assert "[xf2]" in filter_str
+        # 最后一个是[out]
+        assert filter_str.rstrip().endswith("[out]")

tests/unit/test_oss_helpers_pure.py

@@ -0,0 +1,153 @@
+"""OSS助手纯逻辑测试 — normalize_storage_key / resolve_asset_path 输入校验."""
+
+from __future__ import annotations
+
+import hashlib
+import os
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+from video_processing.oss_helpers import normalize_storage_key, resolve_asset_path
+
+
+class TestNormalizeStorageKey:
+    """normalize_storage_key 存储键标准化测试."""
+
+    def test_plain_key_passthrough(self):
+        """普通路径原样返回."""
+        assert normalize_storage_key("path/to/file.mp4") == "path/to/file.mp4"
+
+    def test_https_url_extracts_path(self):
+        """HTTPS URL提取path部分."""
+        result = normalize_storage_key(
+            "https://bucket.oss-cn-hangzhou.aliyuncs.com/path/to/file.mp4"
+        )
+        assert result == "path/to/file.mp4"
+
+    def test_http_url_extracts_path(self):
+        """HTTP URL提取path部分."""
+        result = normalize_storage_key(
+            "http://example.com/assets/video.mp4"
+        )
+        assert result == "assets/video.mp4"
+
+    def test_url_with_query_params(self):
+        """带query参数的URL只取path."""
+        result = normalize_storage_key(
+            "https://bucket.oss-cn-hangzhou.aliyuncs.com/file.mp4?token=abc&expires=123"
+        )
+        assert result == "file.mp4"
+
+    def test_leading_slash_stripped(self):
+        """开头斜杠被去掉."""
+        assert normalize_storage_key("/path/to/file.mp4") == "path/to/file.mp4"
+
+    def test_url_without_path(self):
+        """URL没有path部分返回空字符串."""
+        result = normalize_storage_key("https://example.com")
+        assert result == ""
+
+    def test_nested_path(self):
+        """多层嵌套路径."""
+        assert normalize_storage_key("a/b/c/d/file.mp4") == "a/b/c/d/file.mp4"
+
+    def test_empty_string(self):
+        """空字符串."""
+        assert normalize_storage_key("") == ""
+
+    def test_url_with_port(self):
+        """带端口的URL."""
+        result = normalize_storage_key("http://localhost:9000/bucket/file.mp4")
+        assert result == "bucket/file.mp4"
+
+
+class TestResolveAssetPathInputValidation:
+    """resolve_asset_path 输入校验测试（不涉及真实下载）."""
+
+    def test_empty_string_returns_none(self, tmp_path):
+        """空字符串返回None."""
+        assert resolve_asset_path("", tmp_path) is None
+
+    def test_none_returns_none(self, tmp_path):
+        """None返回None（类型检查）."""
+        assert resolve_asset_path(None, tmp_path) is None  # type: ignore
+
+    def test_non_string_returns_none(self, tmp_path):
+        """非字符串返回None."""
+        assert resolve_asset_path(123, tmp_path) is None  # type: ignore
+
+    def test_null_byte_rejected(self, tmp_path):
+        """包含空字节的asset_id被拒绝."""
+        assert resolve_asset_path("file\x00.mp4", tmp_path) is None
+
+    def test_path_traversal_rejected(self, tmp_path):
+        """包含../的路径遍历攻击被拒绝（第3步下载前检查）."""
+        # mock download_asset不被调用，因为路径包含..会直接返回None
+        with patch("video_processing.oss_helpers.download_asset") as mock_dl:
+            result = resolve_asset_path("../etc/passwd", tmp_path)
+            assert result is None
+            mock_dl.assert_not_called()
+
+    def test_absolute_path_key_rejected(self, tmp_path):
+        """以/开头的存储键在下载前检查被拒."""
+        with patch("video_processing.oss_helpers.download_asset") as mock_dl:
+            result = resolve_asset_path("/etc/passwd", tmp_path)
+            assert result is None
+            mock_dl.assert_not_called()
+
+    def test_cache_hit_returns_cached_path(self, tmp_path):
+        """缓存命中返回缓存路径."""
+        asset_id = "test-asset-123"
+        cache_hash = hashlib.sha256(asset_id.encode()).hexdigest()[:16]
+        cached_file = tmp_path / f"{cache_hash}.mp4"
+        cached_file.write_bytes(b"fake video data")
+
+        result = resolve_asset_path(asset_id, tmp_path)
+        assert result == cached_file
+        assert result.exists()
+
+    def test_cache_empty_file_not_considered_hit(self, tmp_path):
+        """空文件不算缓存命中."""
+        asset_id = "empty-cache-file"
+        cache_hash = hashlib.sha256(asset_id.encode()).hexdigest()[:16]
+        cached_file = tmp_path / f"{cache_hash}.mp4"
+        cached_file.touch()  # 空文件
+
+        with patch("video_processing.oss_helpers.download_asset", return_value=False):
+            result = resolve_asset_path(asset_id, tmp_path)
+            # 空文件不命中缓存，走下载，下载失败返回None
+            assert result is None
+
+    def test_download_success_returns_path(self, tmp_path):
+        """下载成功返回本地路径."""
+        asset_id = "remote-asset"
+        cache_hash = hashlib.sha256(asset_id.encode()).hexdigest()[:16]
+        expected_path = tmp_path / f"{cache_hash}.mp4"
+
+        def fake_download(storage_key, local_path):
+            Path(local_path).write_bytes(b"downloaded data")
+            return True
+
+        with patch("video_processing.oss_helpers.download_asset", side_effect=fake_download):
+            result = resolve_asset_path(asset_id, tmp_path)
+            assert result == expected_path
+            assert result.exists()
+            assert result.stat().st_size > 0
+
+    def test_download_failure_returns_none(self, tmp_path):
+        """下载失败返回None."""
+        with patch("video_processing.oss_helpers.download_asset", return_value=False):
+            result = resolve_asset_path("nonexistent-asset", tmp_path)
+            assert result is None
+
+    def test_work_dir_not_exists_creates_on_demand(self, tmp_path):
+        """work_dir不存在时也能处理."""
+        asset_id = "new-dir-asset"
+        new_dir = tmp_path / "subdir" / "nested"
+
+        with patch("video_processing.oss_helpers.download_asset", return_value=False):
+            # 不存在的work_dir，缓存检查也不会命中
+            result = resolve_asset_path(asset_id, new_dir)
+            assert result is None

tests/unit/test_templates_editor_utils.py

@@ -0,0 +1,243 @@
+"""模板编辑器工具函数测试 — _utils.py 纯函数."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+import pytest
+
+from app.api.routes.templates_editor._utils import (
+    _clip_type_to_scene_label,
+    _clip_value,
+    _format_time,
+    _get_adjust_trim,
+    _get_adjust_volume,
+    _get_clip_config,
+    _validate_trim,
+)
+
+
+class TestFormatTime:
+    """_format_time 秒数格式化测试."""
+
+    def test_zero(self):
+        """0秒."""
+        assert _format_time(0) == "0:00"
+
+    def test_less_than_minute(self):
+        """小于1分钟."""
+        assert _format_time(30) == "0:30"
+        assert _format_time(5) == "0:05"
+        assert _format_time(59) == "0:59"
+
+    def test_exact_minute(self):
+        """整分钟."""
+        assert _format_time(60) == "1:00"
+        assert _format_time(120) == "2:00"
+
+    def test_minutes_and_seconds(self):
+        """几分几秒."""
+        assert _format_time(65) == "1:05"
+        assert _format_time(125) == "2:05"
+        assert _format_time(600) == "10:00"
+
+    def test_float_seconds_truncated(self):
+        """浮点秒数取整."""
+        assert _format_time(65.9) == "1:05"
+        assert _format_time(65.1) == "1:05"
+
+
+class TestClipTypeToSceneLabel:
+    """_clip_type_to_scene_label 片段类型转标签测试."""
+
+    def test_intro(self):
+        assert _clip_type_to_scene_label("intro", "") == "开场"
+
+    def test_title(self):
+        assert _clip_type_to_scene_label("title", "") == "标题"
+
+    def test_product(self):
+        assert _clip_type_to_scene_label("product", "") == "产品展示"
+
+    def test_showcase(self):
+        assert _clip_type_to_scene_label("showcase", "") == "场景展示"
+
+    def test_scene(self):
+        assert _clip_type_to_scene_label("scene", "") == "场景"
+
+    def test_subtitle(self):
+        assert _clip_type_to_scene_label("subtitle", "") == "字幕"
+
+    def test_text(self):
+        assert _clip_type_to_scene_label("text", "") == "文字"
+
+    def test_cta(self):
+        assert _clip_type_to_scene_label("cta", "") == "结尾 CTA"
+
+    def test_outro(self):
+        assert _clip_type_to_scene_label("outro", "") == "结尾"
+
+    def test_voiceover(self):
+        assert _clip_type_to_scene_label("voiceover", "") == "配音"
+
+    def test_transition(self):
+        assert _clip_type_to_scene_label("transition", "") == "转场"
+
+    def test_unknown_type_returns_itself(self):
+        """未知类型返回类型名本身."""
+        assert _clip_type_to_scene_label("unknown_type", "") == "unknown_type"
+
+    def test_empty_type_fallback(self):
+        """空类型fallback到片段."""
+        assert _clip_type_to_scene_label("", "") == "片段"
+
+    def test_with_text_content(self):
+        """带文本内容时追加文本预览."""
+        result = _clip_type_to_scene_label("subtitle", "大家好今天")
+        assert "字幕 - 大家好今天" == result
+
+    def test_text_truncated_at_20_chars(self):
+        """文本超过20字符截断."""
+        long_text = "一二三四五六七八九十一二三四五六七八九十"
+        result = _clip_type_to_scene_label("text", long_text + "extra")
+        # 前20个字符
+        assert long_text in result
+        assert "extra" not in result
+
+    def test_text_with_only_whitespace(self):
+        """文本只有空白时不追加."""
+        result = _clip_type_to_scene_label("intro", "   ")
+        assert result == "开场"
+
+
+class TestValidateTrim:
+    """_validate_trim 裁剪校验测试."""
+
+    def test_valid_trim(self):
+        """合法裁剪."""
+        _validate_trim(1.0, 1.0, 5.0)  # 不抛异常
+
+    def test_zero_trim(self):
+        """不裁剪也合法."""
+        _validate_trim(0.0, 0.0, 5.0)
+
+    def test_trim_equals_total_raises(self):
+        """裁剪总时长等于总时长→抛异常."""
+        with pytest.raises(ValueError, match="不能大于等于"):
+            _validate_trim(2.5, 2.5, 5.0)
+
+    def test_trim_exceeds_total_raises(self):
+        """裁剪超过总时长→抛异常."""
+        with pytest.raises(ValueError):
+            _validate_trim(3.0, 3.0, 5.0)
+
+    def test_only_start_exceeds(self):
+        """只有start就超过."""
+        with pytest.raises(ValueError):
+            _validate_trim(6.0, 0.0, 5.0)
+
+    def test_only_end_exceeds(self):
+        """只有end就超过."""
+        with pytest.raises(ValueError):
+            _validate_trim(0.0, 6.0, 5.0)
+
+
+class TestClipValue:
+    """_clip_value 枚举/字符串值提取测试."""
+
+    def test_plain_string(self):
+        """普通字符串返回自身."""
+        assert _clip_value("hello") == "hello"
+
+    def test_enum_value(self):
+        """带value属性的对象返回value."""
+
+        class FakeEnum:
+            value = "enum_value"
+
+        assert _clip_value(FakeEnum()) == "enum_value"
+
+    def test_int_value(self):
+        """整数转字符串."""
+        assert _clip_value(42) == "42"
+
+
+@dataclass
+class FakeClip:
+    """测试用假Clip对象."""
+
+    id: str = "clip_1"
+    playback_speed: float = 1.0
+    duration: float = 10.0
+    config: dict | None = None
+
+
+class TestGetClipConfig:
+    """_get_clip_config 安全获取配置测试."""
+
+    def test_normal_config(self):
+        """正常dict配置."""
+        clip = FakeClip(config={"volume": 0.5})
+        assert _get_clip_config(clip) == {"volume": 0.5}
+
+    def test_none_config(self):
+        """config为None→返回空dict."""
+        clip = FakeClip(config=None)
+        assert _get_clip_config(clip) == {}
+
+    def test_non_dict_config(self):
+        """config不是dict→返回空dict."""
+        clip = FakeClip(config="not_a_dict")
+        assert _get_clip_config(clip) == {}
+
+    def test_no_config_attribute(self):
+        """没有config属性→返回空dict."""
+
+        class NoConfig:
+            pass
+
+        assert _get_clip_config(NoConfig()) == {}
+
+
+class TestGetAdjustVolume:
+    """_get_adjust_volume 获取音量测试."""
+
+    def test_default_volume(self):
+        """无配置默认1.0."""
+        clip = FakeClip(config={})
+        assert _get_adjust_volume(clip) == pytest.approx(1.0)
+
+    def test_custom_volume(self):
+        """自定义音量."""
+        clip = FakeClip(config={"volume": 0.7})
+        assert _get_adjust_volume(clip) == pytest.approx(0.7)
+
+    def test_none_config(self):
+        """None config."""
+        clip = FakeClip(config=None)
+        assert _get_adjust_volume(clip) == pytest.approx(1.0)
+
+
+class TestGetAdjustTrim:
+    """_get_adjust_trim 获取裁剪测试."""
+
+    def test_default_trim(self):
+        """无配置默认都是0."""
+        clip = FakeClip(config={})
+        start, end = _get_adjust_trim(clip)
+        assert start == pytest.approx(0.0)
+        assert end == pytest.approx(0.0)
+
+    def test_custom_trim(self):
+        """自定义裁剪."""
+        clip = FakeClip(config={"trim_start": 1.5, "trim_end": 2.0})
+        start, end = _get_adjust_trim(clip)
+        assert start == pytest.approx(1.5)
+        assert end == pytest.approx(2.0)
+
+    def test_none_config(self):
+        """None config."""
+        clip = FakeClip(config=None)
+        start, end = _get_adjust_trim(clip)
+        assert start == pytest.approx(0.0)
+        assert end == pytest.approx(0.0)