diff --git a/tests/unit/test_ffmpeg_pure_utils.py b/tests/unit/test_ffmpeg_pure_utils.py new file mode 100755 index 000000000..825793ccd --- /dev/null +++ b/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]") diff --git a/tests/unit/test_oss_helpers_pure.py b/tests/unit/test_oss_helpers_pure.py new file mode 100755 index 000000000..349612121 --- /dev/null +++ b/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 diff --git a/tests/unit/test_templates_editor_utils.py b/tests/unit/test_templates_editor_utils.py new file mode 100755 index 000000000..c28adc564 --- /dev/null +++ b/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)