style: auto-format with black + isort + prettier [skip ci-format-check]

补全视频拼接引擎纯逻辑模块单元测试，覆盖15个纯函数：
- parse_fps: 帧率解析（数字/分数/字符串）
- format_fps_filter: fps滤镜格式化
- resolve_output_params: 输出参数解析（配置→探测→默认三级优先级）
- calculate_scaled_size: 等比缩放+填充偏移计算
- can_use_stream_copy: stream copy可行性判断
- generate_concat_file_list: concat demuxer文件列表生成
- build_scale_pad_filter: scale+pad滤镜构建
- build_fps_filter: fps滤镜构建
- build_setpts_filter: setpts滤镜构建
- build_concat_filter: concat滤镜构建
- build_single_segment_filter_chain: 单段视频预处理滤镜链
- validate_concat_config: 拼接配置验证
- validate_video_path: 视频路径安全验证
- estimate_total_duration: 总时长估算
- count_valid_segments: 有效段统计

85个测试全部通过。

tests/unit/test_concat_engine_pure.py

@@ -1,12 +1,12 @@
-"""视频拼接引擎纯逻辑单元测试."""
+"""concat_engine_pure 单元测试."""
 
-from __future__ import annotations
+from pathlib import Path
 
-import pytest
-from video_processing.concat_engine_pure import (
+from apps.worker.video_processing.concat_engine_pure import (
     build_concat_filter,
     build_fps_filter,
     build_scale_pad_filter,
+    build_setpts_filter,
     build_single_segment_filter_chain,
     calculate_scaled_size,
     can_use_stream_copy,
@@ -20,200 +20,203 @@ from video_processing.concat_engine_pure import (
     validate_video_path,
 )
 
-# ─────────────────────────────────────────────────────────────────────────────
-# 帧率解析测试
-# ─────────────────────────────────────────────────────────────────────────────
+# ── parse_fps ────────────────────────────────────────────────────────────────
 
 
 class TestParseFps:
-    """parse_fps 测试."""
-
-    def test_integer_fps(self):
-        """整数帧率."""
-        assert parse_fps(30) == 30.0
-
-    def test_float_fps(self):
-        """浮点帧率."""
-        assert parse_fps(29.97) == pytest.approx(29.97)
-
-    def test_string_integer(self):
-        """字符串整数."""
-        assert parse_fps("30") == 30.0
-
-    def test_string_fraction(self):
-        """分数字符串（30/1）."""
-        assert parse_fps("30/1") == 30.0
-
-    def test_fraction_24000_1001(self):
-        """23.976 帧率."""
-        result = parse_fps("24000/1001")
-        assert result == pytest.approx(23.976, rel=0.01)
-
-    def test_none_input(self):
-        """None 输入返回默认值."""
+    def test_none_returns_default(self):
         assert parse_fps(None) == 30.0
 
-    def test_empty_string(self):
-        """空字符串返回默认值."""
-        assert parse_fps("") == 30.0
+    def test_integer_value(self):
+        assert parse_fps(30) == 30.0
+        assert parse_fps(24) == 24.0
 
-    def test_invalid_string(self):
-        """无效字符串."""
-        assert parse_fps("abc") == 30.0
+    def test_float_value(self):
+        assert parse_fps(29.97) == 29.97
+
+    def test_string_integer(self):
+        assert parse_fps("30") == 30.0
+        assert parse_fps("  60  ") == 60.0  # 带空格
+
+    def test_string_fraction(self):
+        assert parse_fps("30/1") == 30.0
+        assert abs(parse_fps("24000/1001") - 23.976) < 0.01
 
     def test_zero_denominator(self):
-        """分母为 0."""
         assert parse_fps("30/0") == 30.0
 
+    def test_empty_string(self):
+        assert parse_fps("") == 30.0
+        assert parse_fps("   ") == 30.0
+
+    def test_invalid_string(self):
+        assert parse_fps("abc") == 30.0
+        assert parse_fps("30fps") == 30.0
+
     def test_negative_fps(self):
-        """负帧率."""
         assert parse_fps(-30) == -30.0
 
+    def test_zero_fps(self):
+        assert parse_fps(0) == 0.0
+
+
+# ── format_fps_filter ───────────────────────────────────────────────────────
+
 
 class TestFormatFpsFilter:
-    """format_fps_filter 测试."""
-
     def test_integer_fps(self):
-        """整数帧率."""
         assert format_fps_filter(30.0) == "fps=30"
 
-    def test_float_fps(self):
-        """浮点帧率."""
+    def test_near_integer_fps(self):
+        # 接近整数时用整数形式（注意：int(fps)是截断不是四舍五入）
+        assert format_fps_filter(30.0001) == "fps=30"
+        assert format_fps_filter(30.0005) == "fps=30"  # int(30.0005)=30
+
+    def test_non_integer_fps(self):
+        result = format_fps_filter(23.976)
+        assert result.startswith("fps=")
+        assert "23.976" in result
+
+    def test_float_precision(self):
         result = format_fps_filter(29.97)
         assert result.startswith("fps=")
-        assert "29.97" in result
+        # 三位小数
+        parts = result.split("=")[1]
+        assert len(parts.split(".")[1]) == 3
 
-    def test_near_integer(self):
-        """接近整数."""
-        assert format_fps_filter(30.0001) == "fps=30"
+    def test_one_fps(self):
+        assert format_fps_filter(1.0) == "fps=1"
 
 
-# ─────────────────────────────────────────────────────────────────────────────
-# 输出参数计算测试
-# ─────────────────────────────────────────────────────────────────────────────
+# ── resolve_output_params ───────────────────────────────────────────────────
 
 
 class TestResolveOutputParams:
-    """resolve_output_params 测试."""
-
-    def test_all_specified(self):
-        """全部显式指定."""
+    def test_config_specified(self):
         w, h, fps = resolve_output_params(1920, 1080, 60.0)
         assert w == 1920
         assert h == 1080
         assert fps == 60.0
 
-    def test_no_specified_use_defaults(self):
-        """全部未指定，用默认值."""
-        w, h, fps = resolve_output_params(0, 0, 0)
-        assert w == 1080
-        assert h == 1920
-        assert fps == 30.0
-
-    def test_use_first_video_info(self):
-        """用第一段视频信息."""
+    def test_fallback_to_first_video_info(self):
         info = {"width": 1280, "height": 720, "r_frame_rate": "24/1"}
         w, h, fps = resolve_output_params(0, 0, 0, info)
         assert w == 1280
         assert h == 720
         assert fps == 24.0
 
-    def test_partial_specified(self):
-        """部分指定，未指定的用探测值."""
+    def test_fallback_to_defaults(self):
+        w, h, fps = resolve_output_params(0, 0, 0)
+        assert w == 1080  # default_width
+        assert h == 1920  # default_height
+        assert fps == 30.0
+
+    def test_partial_config(self):
+        # 宽度配置了，高度和帧率用探测的
         info = {"width": 1280, "height": 720, "r_frame_rate": "24/1"}
         w, h, fps = resolve_output_params(1920, 0, 0, info)
-        assert w == 1920  # 指定的
-        assert h == 720  # 探测的
+        assert w == 1920
+        assert h == 720
         assert fps == 24.0
 
-    def test_zero_size_clamped(self):
-        """零尺寸被钳制."""
-        w, h, fps = resolve_output_params(0, 0, 0, {})
-        assert w >= 1
-        assert h >= 1
-        assert fps >= 1.0
-
     def test_custom_defaults(self):
-        """自定义默认值."""
-        w, h, fps = resolve_output_params(0, 0, 0, None, 640, 480, 25.0)
+        w, h, fps = resolve_output_params(
+            0,
+            0,
+            0,
+            default_width=640,
+            default_height=480,
+            default_fps=25.0,
+        )
         assert w == 640
         assert h == 480
         assert fps == 25.0
 
+    def test_minimum_size(self):
+        w, h, fps = resolve_output_params(0, 0, 0, {"width": 0, "height": 0, "r_frame_rate": "0/1"})
+        assert w >= 1
+        assert h >= 1
+        assert fps >= 1.0
+
+    def test_fps_fraction_in_info(self):
+        info = {"width": 1920, "height": 1080, "r_frame_rate": "24000/1001"}
+        _, _, fps = resolve_output_params(0, 0, 0, info)
+        assert abs(fps - 23.976) < 0.01
+
+
+# ── calculate_scaled_size ───────────────────────────────────────────────────
+
 
 class TestCalculateScaledSize:
-    """calculate_scaled_size 测试."""
-
     def test_same_ratio(self):
-        """比例相同."""
         sw, sh, ox, oy = calculate_scaled_size(1920, 1080, 1920, 1080)
         assert sw == 1920
         assert sh == 1080
         assert ox == 0
         assert oy == 0
 
-    def test_wider_source(self):
-        """源更宽，上下填黑边."""
+    def test_wider_source_pad_top_bottom(self):
+        # 源是16:9，目标是9:16竖屏 → 上下填黑边
         sw, sh, ox, oy = calculate_scaled_size(1920, 1080, 1080, 1920)
         assert sw == 1080  # 以宽度为准
-        assert sh < 1920  # 高度按比例
+        assert sh == 607  # 1080 * 1080 / 1920 = 607.5 → 607
         assert ox == 0
         assert oy > 0  # 垂直居中
 
-    def test_taller_source(self):
-        """源更高，左右填黑边."""
+    def test_taller_source_pad_left_right(self):
+        # 源是9:16竖屏，目标是16:9横屏 → 左右填黑边
         sw, sh, ox, oy = calculate_scaled_size(1080, 1920, 1920, 1080)
         assert sh == 1080  # 以高度为准
-        assert sw < 1920  # 宽度按比例
+        assert sw == 607  # 1080 * 1080 / 1920 = 607.5 → 607
         assert ox > 0  # 水平居中
         assert oy == 0
 
-    def test_zero_source(self):
-        """零尺寸源."""
-        sw, sh, ox, oy = calculate_scaled_size(0, 0, 100, 100)
-        assert sw == 100
-        assert sh == 100
-
-    def test_scale_down(self):
-        """缩小."""
-        sw, sh, ox, oy = calculate_scaled_size(1920, 1080, 640, 360)
-        assert sw == 640
-        assert sh == 360
+    def test_zero_source_size(self):
+        sw, sh, ox, oy = calculate_scaled_size(0, 0, 1920, 1080)
+        assert sw == 1920
+        assert sh == 1080
         assert ox == 0
         assert oy == 0
 
-    def test_scale_up(self):
-        """放大."""
+    def test_negative_source_size(self):
+        sw, sh, ox, oy = calculate_scaled_size(-1, -1, 1920, 1080)
+        assert sw == 1920
+        assert sh == 1080
+        assert ox == 0
+        assert oy == 0
+
+    def test_target_same_ratio_different_size(self):
+        # 比例相同，尺寸不同 → 直接缩放到目标大小
         sw, sh, ox, oy = calculate_scaled_size(640, 360, 1920, 1080)
         assert sw == 1920
         assert sh == 1080
+        assert ox == 0
+        assert oy == 0
 
 
-# ─────────────────────────────────────────────────────────────────────────────
-# stream copy 判断测试
-# ─────────────────────────────────────────────────────────────────────────────
+# ── can_use_stream_copy ─────────────────────────────────────────────────────
 
 
 class TestCanUseStreamCopy:
-    """can_use_stream_copy 测试."""
+    def test_force_reencode_false(self):
+        assert can_use_stream_copy([], 1920, 1080, 30.0, force_reencode=True) is False
 
-    def test_identical_segments(self):
-        """所有段参数相同，可以 stream copy."""
+    def test_empty_segments(self):
+        assert can_use_stream_copy([], 1920, 1080, 30.0) is False
+
+    def test_single_segment_matching_params(self):
+        segs = [{"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "30/1"}]
+        assert can_use_stream_copy(segs, 1920, 1080, 30.0) is True
+
+    def test_multiple_segments_same_params(self):
         segs = [
             {"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "30/1"},
             {"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "30/1"},
+            {"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "30/1"},
         ]
         assert can_use_stream_copy(segs, 1920, 1080, 30.0) is True
 
-    def test_force_reencode(self):
-        """强制重编码."""
-        segs = [
-            {"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "30/1"},
-        ]
-        assert can_use_stream_copy(segs, 1920, 1080, 30.0, force_reencode=True) is False
-
     def test_different_codec(self):
-        """编码不同."""
         segs = [
             {"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "30/1"},
             {"codec_name": "hevc", "width": 1920, "height": 1080, "r_frame_rate": "30/1"},
@@ -221,7 +224,6 @@ class TestCanUseStreamCopy:
         assert can_use_stream_copy(segs, 1920, 1080, 30.0) is False
 
     def test_different_resolution(self):
-        """分辨率不同."""
         segs = [
             {"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "30/1"},
             {"codec_name": "h264", "width": 1280, "height": 720, "r_frame_rate": "30/1"},
@@ -229,306 +231,349 @@ class TestCanUseStreamCopy:
         assert can_use_stream_copy(segs, 1920, 1080, 30.0) is False
 
     def test_different_fps(self):
-        """帧率不同."""
         segs = [
             {"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "30/1"},
             {"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "60/1"},
         ]
         assert can_use_stream_copy(segs, 1920, 1080, 30.0) is False
 
-    def test_target_differs(self):
-        """目标参数与源不同."""
-        segs = [
-            {"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "30/1"},
-        ]
-        assert can_use_stream_copy(segs, 1280, 720, 30.0) is False
-
-    def test_empty_segments(self):
-        """空列表."""
-        assert can_use_stream_copy([], 1920, 1080, 30.0) is False
-
-    def test_single_segment(self):
-        """单段."""
+    def test_target_differs_from_source(self):
         segs = [{"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "30/1"}]
-        assert can_use_stream_copy(segs, 1920, 1080, 30.0) is True
+        # 目标分辨率不同
+        assert can_use_stream_copy(segs, 1280, 720, 30.0) is False
+        # 目标帧率不同
+        assert can_use_stream_copy(segs, 1920, 1080, 60.0) is False
+
+    def test_fps_fraction_match(self):
+        segs = [{"codec_name": "h264", "width": 1920, "height": 1080, "r_frame_rate": "24000/1001"}]
+        assert can_use_stream_copy(segs, 1920, 1080, 23.976) is True
 
 
-# ─────────────────────────────────────────────────────────────────────────────
-# 文件列表生成测试
-# ─────────────────────────────────────────────────────────────────────────────
+# ── generate_concat_file_list ───────────────────────────────────────────────
 
 
 class TestGenerateConcatFileList:
-    """generate_concat_file_list 测试."""
-
     def test_single_file(self):
-        """单个文件."""
-        result = generate_concat_file_list(["/a.mp4"])
-        assert "file '/a.mp4'" in result
-        assert result.endswith("\n")
+        result = generate_concat_file_list(["/tmp/video.mp4"])
+        assert result == "file '/tmp/video.mp4'\n"
 
     def test_multiple_files(self):
-        """多个文件."""
         result = generate_concat_file_list(["/a.mp4", "/b.mp4", "/c.mp4"])
         lines = result.strip().split("\n")
         assert len(lines) == 3
         assert lines[0] == "file '/a.mp4'"
         assert lines[1] == "file '/b.mp4'"
         assert lines[2] == "file '/c.mp4'"
+        assert result.endswith("\n")
+
+    def test_escapes_single_quotes(self):
+        result = generate_concat_file_list(["/path/with'quote.mp4"])
+        # 单引号转义: '\''
+        assert "'\\''" in result
 
     def test_empty_list(self):
-        """空列表."""
         result = generate_concat_file_list([])
         assert result == "\n"
 
-    def test_path_with_single_quote(self):
-        """路径包含单引号（转义）."""
-        result = generate_concat_file_list(["/path/to/file's.mp4"])
-        # 单引号应该被转义
-        assert "'\\''" in result or file
-        assert "file '" in result
-
     def test_path_with_spaces(self):
-        """路径包含空格."""
-        result = generate_concat_file_list(["/path/to/my video.mp4"])
-        assert "my video" in result
+        result = generate_concat_file_list(["/path/to/video file.mp4"])
+        assert "file '/path/to/video file.mp4'" in result
 
 
-# ─────────────────────────────────────────────────────────────────────────────
-# 滤镜构建测试
-# ─────────────────────────────────────────────────────────────────────────────
+# ── build_scale_pad_filter ──────────────────────────────────────────────────
 
 
 class TestBuildScalePadFilter:
-    """scale+pad 滤镜测试."""
-
-    def test_contains_scale(self):
-        """包含 scale."""
-        result = build_scale_pad_filter(1920, 1080)
-        assert "scale=" in result
-
-    def test_contains_pad(self):
-        """包含 pad."""
-        result = build_scale_pad_filter(1920, 1080)
-        assert "pad=" in result
-        assert "1920:1080" in result
-
-    def test_force_original_aspect_ratio(self):
-        """保持宽高比."""
+    def test_basic_filter(self):
         result = build_scale_pad_filter(1920, 1080)
+        assert "scale=1920:1080" in result
         assert "force_original_aspect_ratio=decrease" in result
+        assert "pad=1920:1080" in result
+        assert "black" in result
+        assert "(ow-iw)/2" in result
+        assert "(oh-ih)/2" in result
 
-    def test_black_padding(self):
-        """黑边填充."""
-        result = build_scale_pad_filter(1920, 1080)
-        assert ":black" in result
+    def test_different_resolution(self):
+        result = build_scale_pad_filter(1080, 1920)
+        assert "scale=1080:1920" in result
+        assert "pad=1080:1920" in result
+
+    def test_ignores_source_size(self):
+        # src_w/src_h 目前不影响输出，都是用表达式
+        result1 = build_scale_pad_filter(1920, 1080)
+        result2 = build_scale_pad_filter(1920, 1080, src_w=1280, src_h=720)
+        assert result1 == result2
+
+
+# ── build_fps_filter ────────────────────────────────────────────────────────
 
 
 class TestBuildFpsFilter:
-    """fps 滤镜测试."""
-
     def test_integer_fps(self):
-        """整数帧率."""
         assert build_fps_filter(30.0) == "fps=30"
 
     def test_float_fps(self):
-        """浮点帧率."""
         result = build_fps_filter(29.97)
         assert result.startswith("fps=")
 
 
-class TestBuildConcatFilter:
-    """concat 滤镜测试."""
+# ── build_setpts_filter ─────────────────────────────────────────────────────
 
-    def test_two_inputs_with_audio(self):
-        """两路输入，有音频."""
-        result = build_concat_filter(2, has_audio=True)
-        assert "[0:v][0:a][1:v][1:a]concat=n=2:v=1:a=1" in result
+
+class TestBuildSetptsFilter:
+    def test_returns_correct_string(self):
+        assert build_setpts_filter() == "setpts=PTS-STARTPTS"
+
+
+# ── build_concat_filter ─────────────────────────────────────────────────────
+
+
+class TestBuildConcatFilter:
+    def test_zero_inputs(self):
+        assert build_concat_filter(0) == ""
+
+    def test_single_input_with_audio(self):
+        result = build_concat_filter(1)
+        assert "[0:v][0:a]" in result
+        assert "concat=n=1:v=1:a=1" in result
         assert "[concat_v][concat_a]" in result
 
-    def test_three_inputs_video_only(self):
-        """三路输入，无音频."""
-        result = build_concat_filter(3, has_audio=False)
-        assert "[0:v][1:v][2:v]concat=n=3:v=1:a=0" in result
+    def test_single_input_no_audio(self):
+        result = build_concat_filter(1, has_audio=False)
+        assert "[0:v]" in result
+        assert "concat=n=1:v=1:a=0" in result
         assert "[concat_v]" in result
+        assert "[concat_a]" not in result
 
-    def test_single_input(self):
-        """单路输入."""
-        result = build_concat_filter(1, has_audio=True)
-        assert "[0:v][0:a]concat=n=1:v=1:a=1" in result
+    def test_multiple_inputs_with_audio(self):
+        result = build_concat_filter(3)
+        assert "[0:v][0:a][1:v][1:a][2:v][2:a]" in result
+        assert "concat=n=3:v=1:a=1" in result
 
-    def test_zero_inputs(self):
-        """零输入."""
-        assert build_concat_filter(0) == ""
+    def test_multiple_inputs_no_audio(self):
+        result = build_concat_filter(3, has_audio=False)
+        assert "[0:v][1:v][2:v]" in result
+        assert "concat=n=3:v=1:a=0" in result
+
+    def test_negative_inputs(self):
+        assert build_concat_filter(-1) == ""
+
+
+# ── build_single_segment_filter_chain ───────────────────────────────────────
 
 
 class TestBuildSingleSegmentFilterChain:
-    """单段滤镜链测试."""
-
     def test_with_audio(self):
-        """有音频."""
         result = build_single_segment_filter_chain(1920, 1080, 30.0, 0)
-        assert "scale=" in result
-        assert "fps=" in result
-        assert "setpts=PTS-STARTPTS" in result
-        assert "asetpts=PTS-STARTPTS" in result
+        # 视频链
+        assert "[0:v]" in result
         assert "[v0]" in result
+        assert "scale=1920:1080" in result
+        assert "fps=30" in result
+        assert "setpts=PTS-STARTPTS" in result
+        # 音频链
+        assert "[0:a]" in result
         assert "[a0]" in result
+        assert "asetpts=PTS-STARTPTS" in result
+        # 用分号分隔
+        assert ";" in result
 
-    def test_video_only(self):
-        """无音频."""
-        result = build_single_segment_filter_chain(1920, 1080, 30.0, 1, has_audio=False)
-        assert "scale=" in result
-        assert "setpts=" in result
-        assert "asetpts" not in result
-        assert "[v1]" in result
+    def test_without_audio(self):
+        result = build_single_segment_filter_chain(1920, 1080, 30.0, 2, has_audio=False)
+        assert "[2:v]" in result
+        assert "[v2]" in result
+        assert "[2:a]" not in result
+        assert ";" not in result  # 没有音频就没有分号
 
-    def test_segment_index_in_labels(self):
-        """段索引在标签中."""
-        result = build_single_segment_filter_chain(1920, 1080, 30.0, 5)
-        assert "[5:v]" in result
-        assert "[v5]" in result
+    def test_segment_index_propagated(self):
+        for idx in [0, 5, 10]:
+            result = build_single_segment_filter_chain(1920, 1080, 30.0, idx)
+            assert f"[{idx}:v]" in result
+            assert f"[v{idx}]" in result
 
 
-# ─────────────────────────────────────────────────────────────────────────────
-# 配置验证测试
-# ─────────────────────────────────────────────────────────────────────────────
+# ── validate_concat_config ──────────────────────────────────────────────────
 
 
 class TestValidateConcatConfig:
-    """配置验证测试."""
-
     def test_valid_config(self):
-        """合法配置."""
         config = {
-            "segments": [{"video_path": "/a.mp4"}, {"video_path": "/b.mp4"}],
+            "segments": [
+                {"video_path": "/a.mp4"},
+                {"video_path": "/b.mp4"},
+            ],
             "output_width": 1920,
             "output_height": 1080,
             "output_fps": 30,
         }
-        ok, errors = validate_concat_config(config)
-        assert ok is True
-        assert len(errors) == 0
+        valid, errors = validate_concat_config(config)
+        assert valid is True
+        assert errors == []
+
+    def test_no_segments(self):
+        valid, errors = validate_concat_config({})
+        assert valid is False
+        assert any("至少需要一个" in e for e in errors)
 
     def test_empty_segments(self):
-        """空段列表."""
-        ok, errors = validate_concat_config({"segments": []})
-        assert ok is False
-        assert any("至少需要" in e or "视频段" in e for e in errors)
+        valid, errors = validate_concat_config({"segments": []})
+        assert valid is False
+        assert len(errors) >= 1
 
     def test_missing_video_path(self):
-        """缺少 video_path."""
-        config = {"segments": [{"video_path": "/a.mp4"}, {}]}
-        ok, errors = validate_concat_config(config)
-        assert ok is False
+        config = {"segments": [{"video_path": ""}]}
+        valid, errors = validate_concat_config(config)
+        assert valid is False
         assert any("video_path" in e for e in errors)
 
-    def test_negative_width(self):
-        """负宽度."""
-        config = {"segments": [{"video_path": "/a.mp4"}], "output_width": -100}
-        ok, errors = validate_concat_config(config)
-        assert ok is False
+    def test_multiple_missing_paths(self):
+        config = {
+            "segments": [
+                {"video_path": "/a.mp4"},
+                {"video_path": ""},
+                {"video_path": ""},
+            ]
+        }
+        valid, errors = validate_concat_config(config)
+        assert valid is False
+        path_errors = [e for e in errors if "video_path" in e]
+        assert len(path_errors) == 2
+
+    def test_negative_output_width(self):
+        config = {"segments": [{"video_path": "/a.mp4"}], "output_width": -1}
+        valid, errors = validate_concat_config(config)
+        assert valid is False
         assert any("output_width" in e for e in errors)
 
-    def test_negative_height(self):
-        """负高度."""
-        config = {"segments": [{"video_path": "/a.mp4"}], "output_height": -100}
-        ok, errors = validate_concat_config(config)
-        assert ok is False
+    def test_negative_output_height(self):
+        config = {"segments": [{"video_path": "/a.mp4"}], "output_height": -1}
+        valid, errors = validate_concat_config(config)
+        assert valid is False
         assert any("output_height" in e for e in errors)
 
-    def test_negative_fps(self):
-        """负帧率."""
-        config = {"segments": [{"video_path": "/a.mp4"}], "output_fps": -30}
-        ok, errors = validate_concat_config(config)
-        assert ok is False
+    def test_negative_output_fps(self):
+        config = {"segments": [{"video_path": "/a.mp4"}], "output_fps": -1}
+        valid, errors = validate_concat_config(config)
+        assert valid is False
         assert any("output_fps" in e for e in errors)
 
-    def test_zero_output_params_ok(self):
-        """零输出参数合法（表示自动探测）."""
-        config = {"segments": [{"video_path": "/a.mp4"}]}
-        ok, errors = validate_concat_config(config)
-        assert ok is True
+    def test_zero_output_params_valid(self):
+        # 0值表示未指定，是合法的
+        config = {
+            "segments": [{"video_path": "/a.mp4"}],
+            "output_width": 0,
+            "output_height": 0,
+            "output_fps": 0,
+        }
+        valid, errors = validate_concat_config(config)
+        assert valid is True
 
 
-# ─────────────────────────────────────────────────────────────────────────────
-# 路径验证测试
-# ─────────────────────────────────────────────────────────────────────────────
+# ── validate_video_path ─────────────────────────────────────────────────────
 
 
 class TestValidateVideoPath:
-    """视频路径验证测试."""
-
     def test_empty_path(self):
-        """空路径."""
-        ok, msg = validate_video_path("", "/work")
-        assert ok is False
-        assert "不能为空" in msg
+        valid, err = validate_video_path("", "/work")
+        assert valid is False
+        assert "不能为空" in err
 
-    def test_path_traversal(self):
-        """路径遍历."""
-        ok, msg = validate_video_path("../etc/passwd", "/work")
-        assert ok is False
-        assert "回溯" in msg or ".." in msg
+    def test_relative_path_valid(self):
+        valid, err = validate_video_path("video.mp4", "/work")
+        assert valid is True
+        assert err == ""
 
-    def test_valid_relative_path(self):
-        """相对路径（不检查边界）."""
-        ok, msg = validate_video_path("video.mp4", "/work")
-        assert ok is True
+    def test_relative_path_with_subdir(self):
+        valid, err = validate_video_path("sub/video.mp4", "/work")
+        assert valid is True
 
-    def test_valid_absolute_path(self):
-        """绝对路径在工作目录内."""
-        ok, msg = validate_video_path("/work/sub/video.mp4", "/work")
-        assert ok is True
+    def test_path_traversal_rejected(self):
+        valid, err = validate_video_path("../secret.mp4", "/work")
+        assert valid is False
+        assert ".." in err
 
-    def test_path_outside_work_dir(self):
-        """路径在工作目录外."""
-        ok, msg = validate_video_path("/etc/passwd", "/work")
-        assert ok is False
-        assert "工作目录" in msg
+    def test_nested_path_traversal_rejected(self):
+        valid, err = validate_video_path("sub/../../secret.mp4", "/work")
+        assert valid is False
+
+    def test_absolute_path_inside_workdir(self):
+        valid, err = validate_video_path("/work/sub/video.mp4", "/work")
+        assert valid is True
+
+    def test_absolute_path_outside_workdir(self):
+        valid, err = validate_video_path("/etc/passwd", "/work")
+        assert valid is False
+        assert "工作目录内" in err
+
+    def test_path_object_input(self):
+        valid, err = validate_video_path(Path("video.mp4"), Path("/work"))
+        assert valid is True
 
 
-# ─────────────────────────────────────────────────────────────────────────────
-# 工具函数测试
-# ─────────────────────────────────────────────────────────────────────────────
+# ── estimate_total_duration ─────────────────────────────────────────────────
 
 
 class TestEstimateTotalDuration:
-    """总时长估算测试."""
+    def test_single_segment(self):
+        assert estimate_total_duration([{"duration": 10.5}]) == 10.5
 
     def test_multiple_segments(self):
-        """多段视频."""
-        segs = [{"duration": 10}, {"duration": 20.5}, {"duration": 5}]
-        assert estimate_total_duration(segs) == pytest.approx(35.5)
+        segs = [
+            {"duration": 10},
+            {"duration": 20.5},
+            {"duration": 5.5},
+        ]
+        assert estimate_total_duration(segs) == 36.0
 
     def test_empty_list(self):
-        """空列表."""
         assert estimate_total_duration([]) == 0.0
 
-    def test_invalid_duration_skipped(self):
-        """无效时长跳过."""
-        segs = [{"duration": 10}, {"duration": "abc"}, {"duration": 20}]
-        assert estimate_total_duration(segs) == pytest.approx(30.0)
+    def test_missing_duration_field(self):
+        segs = [{"path": "a.mp4"}, {"duration": 10}]
+        assert estimate_total_duration(segs) == 10.0
 
-    def test_missing_duration(self):
-        """缺 duration 字段."""
-        segs = [{}, {"duration": 10}]
-        assert estimate_total_duration(segs) == pytest.approx(10.0)
+    def test_invalid_duration_skipped(self):
+        segs = [
+            {"duration": 10},
+            {"duration": "abc"},
+            {"duration": 20},
+        ]
+        assert estimate_total_duration(segs) == 30.0
+
+    def test_string_duration(self):
+        segs = [{"duration": "15.5"}]
+        assert estimate_total_duration(segs) == 15.5
+
+    def test_negative_duration(self):
+        segs = [{"duration": -5}]
+        assert estimate_total_duration(segs) == -5.0
+
+
+# ── count_valid_segments ────────────────────────────────────────────────────
 
 
 class TestCountValidSegments:
-    """有效段统计测试."""
-
     def test_all_valid(self):
-        """全部有效."""
-        segs = [{"video_path": "/a.mp4"}, {"video_path": "/b.mp4"}]
+        segs = [
+            {"video_path": "/a.mp4"},
+            {"video_path": "/b.mp4"},
+        ]
         assert count_valid_segments(segs) == 2
 
     def test_some_invalid(self):
-        """部分无效."""
-        segs = [{"video_path": "/a.mp4"}, {}, {"video_path": ""}]
-        assert count_valid_segments(segs) == 1
+        segs = [
+            {"video_path": "/a.mp4"},
+            {"video_path": ""},
+            {"video_path": "/c.mp4"},
+        ]
+        assert count_valid_segments(segs) == 2
+
+    def test_none_valid(self):
+        segs = [
+            {"video_path": ""},
+            {"other_field": "x"},
+        ]
+        assert count_valid_segments(segs) == 0
 
     def test_empty_list(self):
-        """空列表."""
         assert count_valid_segments([]) == 0

tests/unit/test_shared_ai_service.py

@@ -0,0 +1,453 @@
+"""shared.ai_service 单元测试.
+
+主要测试纯逻辑部分：_parse_recommend_response / _fallback_recommend_clips / _call_ai_cover_service.
+"""
+
+from __future__ import annotations
+
+import json
+from unittest.mock import patch
+
+import pytest
+from shared.ai_service import (
+    _call_ai_cover_service,
+    _fallback_recommend_clips,
+    _parse_recommend_response,
+)
+
+# ── _parse_recommend_response 测试 ────────────────────────────────────────
+
+
+class TestParseRecommendResponseBasic:
+    """基础解析测试."""
+
+    def test_parse_valid_json(self):
+        content = json.dumps(
+            {
+                "clips": [
+                    {
+                        "clip_type": "intro",
+                        "order": 0,
+                        "text_content": "开场",
+                        "duration": 3.0,
+                        "transition_effect": "fade",
+                        "asset_id": "asset1",
+                        "start_time": 0.0,
+                        "config": {},
+                    },
+                    {
+                        "clip_type": "outro",
+                        "order": 1,
+                        "text_content": "结尾",
+                        "duration": 2.0,
+                        "transition_effect": "fade",
+                        "asset_id": "",
+                        "start_time": 0.0,
+                        "config": {},
+                    },
+                ],
+                "title": "测试视频",
+                "confidence": 0.85,
+            }
+        )
+        result = _parse_recommend_response(content, ["asset1"], 30.0)
+        assert result is not None
+        assert len(result["clips"]) == 2
+        assert result["confidence"] == 0.85
+        assert result["total_duration"] == 5.0
+        assert result["config"]["title"]["text"] == "测试视频"
+        assert result["config"]["title"]["ai_auto"] is True
+
+    def test_parse_none_returns_none(self):
+        result = _parse_recommend_response(None, ["a1"], 30.0)  # type: ignore[arg-type]
+        assert result is None
+
+    def test_parse_empty_string_returns_none(self):
+        result = _parse_recommend_response("", ["a1"], 30.0)
+        assert result is None
+
+    def test_parse_whitespace_only_returns_none(self):
+        result = _parse_recommend_response("   ", ["a1"], 30.0)
+        assert result is None
+
+    def test_parse_invalid_json_returns_none(self):
+        result = _parse_recommend_response("not json", ["a1"], 30.0)
+        assert result is None
+
+    def test_parse_non_dict_json_returns_none(self):
+        result = _parse_recommend_response("[1, 2, 3]", ["a1"], 30.0)
+        assert result is None
+
+
+class TestParseRecommendResponseClips:
+    """clips 解析测试."""
+
+    def test_parse_no_clips_returns_none(self):
+        content = json.dumps({"title": "test", "clips": []})
+        result = _parse_recommend_response(content, ["a1"], 30.0)
+        assert result is None
+
+    def test_parse_clips_not_list_returns_none(self):
+        content = json.dumps({"clips": "not a list"})
+        result = _parse_recommend_response(content, ["a1"], 30.0)
+        assert result is None
+
+    def test_parse_clips_sorted_by_order(self):
+        content = json.dumps(
+            {
+                "clips": [
+                    {"clip_type": "outro", "order": 2, "duration": 2, "asset_id": "a1"},
+                    {"clip_type": "intro", "order": 0, "duration": 3, "asset_id": "a1"},
+                    {"clip_type": "showcase", "order": 1, "duration": 5, "asset_id": "a1"},
+                ],
+            }
+        )
+        result = _parse_recommend_response(content, ["a1"], 30.0)
+        assert result is not None
+        assert len(result["clips"]) == 3
+        assert result["clips"][0]["clip_type"] == "intro"
+        assert result["clips"][1]["clip_type"] == "showcase"
+        assert result["clips"][2]["clip_type"] == "outro"
+
+    def test_parse_clips_renumbered_continuously(self):
+        content = json.dumps(
+            {
+                "clips": [
+                    {"clip_type": "intro", "order": 10, "duration": 2, "asset_id": "a1"},
+                    {"clip_type": "outro", "order": 20, "duration": 2, "asset_id": "a1"},
+                ],
+            }
+        )
+        result = _parse_recommend_response(content, ["a1"], 30.0)
+        assert result is not None
+        assert result["clips"][0]["order"] == 0
+        assert result["clips"][1]["order"] == 1
+
+    def test_parse_skips_invalid_clip_dicts(self):
+        content = json.dumps(
+            {
+                "clips": [
+                    {"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a1"},
+                    "not a dict",
+                    {"clip_type": "outro", "order": 2, "duration": 2, "asset_id": "a1"},
+                ],
+            }
+        )
+        result = _parse_recommend_response(content, ["a1"], 30.0)
+        assert result is not None
+        assert len(result["clips"]) == 2
+
+
+class TestParseRecommendResponseFields:
+    """各字段解析与边界测试."""
+
+    def test_parse_duration_clamped_min(self):
+        content = json.dumps(
+            {
+                "clips": [
+                    {"clip_type": "intro", "order": 0, "duration": 0.5, "asset_id": "a1"},
+                ],
+            }
+        )
+        result = _parse_recommend_response(content, ["a1"], 30.0)
+        assert result is not None
+        assert result["clips"][0]["duration"] == 1.0
+
+    def test_parse_duration_clamped_max(self):
+        content = json.dumps(
+            {
+                "clips": [
+                    {"clip_type": "intro", "order": 0, "duration": 100, "asset_id": "a1"},
+                ],
+            }
+        )
+        result = _parse_recommend_response(content, ["a1"], 30.0)
+        assert result is not None
+        assert result["clips"][0]["duration"] == 30.0
+
+    def test_parse_start_time_clamped_min(self):
+        content = json.dumps(
+            {
+                "clips": [
+                    {"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a1", "start_time": -5.0},
+                ],
+            }
+        )
+        result = _parse_recommend_response(content, ["a1"], 30.0)
+        assert result is not None
+        assert result["clips"][0]["start_time"] == 0.0
+
+    def test_parse_asset_id_not_in_list_empty(self):
+        content = json.dumps(
+            {
+                "clips": [
+                    {"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "unknown_asset"},
+                ],
+            }
+        )
+        result = _parse_recommend_response(content, ["a1", "a2"], 30.0)
+        assert result is not None
+        assert result["clips"][0]["asset_id"] == ""
+
+    def test_parse_asset_id_in_list_kept(self):
+        content = json.dumps(
+            {
+                "clips": [
+                    {"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a2"},
+                ],
+            }
+        )
+        result = _parse_recommend_response(content, ["a1", "a2"], 30.0)
+        assert result is not None
+        assert result["clips"][0]["asset_id"] == "a2"
+
+    def test_parse_default_values(self):
+        content = json.dumps(
+            {
+                "clips": [
+                    {"order": 0},
+                ],
+            }
+        )
+        result = _parse_recommend_response(content, ["a1"], 30.0)
+        assert result is not None
+        clip = result["clips"][0]
+        assert clip["clip_type"] == "showcase"
+        assert clip["text_content"] == ""
+        assert clip["duration"] == 3.0
+        assert clip["transition_effect"] == "cut"
+        assert clip["asset_id"] == ""
+        assert clip["start_time"] == 0.0
+        assert clip["config"] == {}
+
+
+class TestParseRecommendResponseMarkdown:
+    """Markdown 代码块包裹的 JSON 测试."""
+
+    def test_parse_markdown_json(self):
+        content = (
+            "```json\n"
+            + json.dumps(
+                {
+                    "clips": [{"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a1"}],
+                    "title": "md test",
+                }
+            )
+            + "\n```"
+        )
+        result = _parse_recommend_response(content, ["a1"], 30.0)
+        assert result is not None
+        assert len(result["clips"]) == 1
+        assert result["config"]["title"]["text"] == "md test"
+
+    def test_parse_backticks_no_language(self):
+        content = (
+            "```\n"
+            + json.dumps(
+                {
+                    "clips": [{"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a1"}],
+                }
+            )
+            + "\n```"
+        )
+        result = _parse_recommend_response(content, ["a1"], 30.0)
+        assert result is not None
+        assert len(result["clips"]) == 1
+
+
+class TestParseRecommendResponseConfidence:
+    """confidence 解析测试."""
+
+    def test_parse_confidence_normal(self):
+        content = json.dumps(
+            {
+                "clips": [{"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a1"}],
+                "confidence": 0.85,
+            }
+        )
+        result = _parse_recommend_response(content, ["a1"], 30.0)
+        assert result is not None
+        assert result["confidence"] == 0.85
+
+    def test_parse_confidence_clamped_min(self):
+        content = json.dumps(
+            {
+                "clips": [{"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a1"}],
+                "confidence": -0.5,
+            }
+        )
+        result = _parse_recommend_response(content, ["a1"], 30.0)
+        assert result is not None
+        assert result["confidence"] == 0.0
+
+    def test_parse_confidence_clamped_max(self):
+        content = json.dumps(
+            {
+                "clips": [{"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a1"}],
+                "confidence": 1.5,
+            }
+        )
+        result = _parse_recommend_response(content, ["a1"], 30.0)
+        assert result is not None
+        assert result["confidence"] == 1.0
+
+    def test_parse_confidence_default(self):
+        content = json.dumps(
+            {
+                "clips": [{"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a1"}],
+            }
+        )
+        result = _parse_recommend_response(content, ["a1"], 30.0)
+        assert result is not None
+        assert result["confidence"] == 0.7
+
+
+class TestParseRecommendResponseConfig:
+    """config 生成测试."""
+
+    def test_parse_no_title_no_ai_auto(self):
+        content = json.dumps(
+            {
+                "clips": [{"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a1"}],
+            }
+        )
+        result = _parse_recommend_response(content, ["a1"], 30.0)
+        assert result is not None
+        # 没有 title 时，config 的 title.text 保持默认（DEFAULT_EDIT_PLAN_CONFIG 中的值）
+        assert "title" in result["config"]
+
+    def test_parse_config_is_deep_copy(self):
+        content = json.dumps(
+            {
+                "clips": [{"clip_type": "intro", "order": 0, "duration": 2, "asset_id": "a1"}],
+                "title": "test",
+            }
+        )
+        result1 = _parse_recommend_response(content, ["a1"], 30.0)
+        result2 = _parse_recommend_response(content, ["a1"], 30.0)
+        # 修改其中一个不影响另一个
+        result1["config"]["title"]["text"] = "modified"
+        assert result2["config"]["title"]["text"] != "modified"
+
+
+class TestParseRecommendResponseTotalDuration:
+    """total_duration 计算测试."""
+
+    def test_parse_total_duration_sum(self):
+        content = json.dumps(
+            {
+                "clips": [
+                    {"clip_type": "intro", "order": 0, "duration": 3.5, "asset_id": "a1"},
+                    {"clip_type": "showcase", "order": 1, "duration": 5.2, "asset_id": "a1"},
+                    {"clip_type": "outro", "order": 2, "duration": 2.0, "asset_id": "a1"},
+                ],
+            }
+        )
+        result = _parse_recommend_response(content, ["a1"], 30.0)
+        assert result is not None
+        assert result["total_duration"] == pytest.approx(10.7, abs=0.01)
+
+
+# ── _fallback_recommend_clips 测试 ────────────────────────────────────────
+
+
+class TestFallbackRecommendClips:
+    """本地降级推荐方案测试."""
+
+    def test_fallback_returns_dict_with_clips(self):
+        with patch("shared.ai_service.time.sleep"):
+            result = _fallback_recommend_clips("plan1", "tmpl1", ["a1", "a2"], "one_take", 30.0)
+        assert "clips" in result
+        assert "config" in result
+        assert "total_duration" in result
+        assert "confidence" in result
+
+    def test_fallback_has_intro_and_outro(self):
+        with patch("shared.ai_service.time.sleep"):
+            result = _fallback_recommend_clips("plan1", "tmpl1", ["a1", "a2"], "one_take", 30.0)
+        clips = result["clips"]
+        assert clips[0]["clip_type"] == "intro"
+        assert clips[-1]["clip_type"] == "outro"
+
+    def test_fallback_showcase_count_matches_assets(self):
+        with patch("shared.ai_service.time.sleep"):
+            result = _fallback_recommend_clips("plan1", "tmpl1", ["a1", "a2", "a3"], "one_take", 30.0)
+        showcase_clips = [c for c in result["clips"] if c["clip_type"] == "showcase"]
+        assert len(showcase_clips) == 3
+
+    def test_fallback_no_assets_still_works(self):
+        with patch("shared.ai_service.time.sleep"):
+            result = _fallback_recommend_clips("plan1", "tmpl1", [], "one_take", 30.0)
+        assert len(result["clips"]) >= 2  # 至少有intro和outro
+
+    def test_fallback_intro_uses_first_asset(self):
+        with patch("shared.ai_service.time.sleep"):
+            result = _fallback_recommend_clips("plan1", "tmpl1", ["a1", "a2"], "one_take", 30.0)
+        assert result["clips"][0]["asset_id"] == "a1"
+
+    def test_fallback_outro_has_empty_asset(self):
+        with patch("shared.ai_service.time.sleep"):
+            result = _fallback_recommend_clips("plan1", "tmpl1", ["a1"], "one_take", 30.0)
+        assert result["clips"][-1]["asset_id"] == ""
+
+    def test_fallback_confidence_in_range(self):
+        with patch("shared.ai_service.time.sleep"):
+            result = _fallback_recommend_clips("plan1", "tmpl1", ["a1"], "one_take", 30.0)
+        assert 0.75 <= result["confidence"] <= 0.95
+
+    def test_fallback_title_contains_asset_count(self):
+        with patch("shared.ai_service.time.sleep"):
+            result = _fallback_recommend_clips("plan1", "tmpl1", ["a1", "a2", "a3"], "one_take", 30.0)
+        assert "3" in result["config"]["title"]["text"]
+        assert result["config"]["title"]["ai_auto"] is True
+
+    def test_fallback_total_duration_matches(self):
+        with patch("shared.ai_service.time.sleep"):
+            result = _fallback_recommend_clips("plan1", "tmpl1", ["a1", "a2"], "one_take", 30.0)
+        total = sum(c["duration"] for c in result["clips"])
+        assert result["total_duration"] == round(total, 1)
+
+    def test_fallback_orders_are_sequential(self):
+        with patch("shared.ai_service.time.sleep"):
+            result = _fallback_recommend_clips("plan1", "tmpl1", ["a1", "a2", "a3"], "one_take", 30.0)
+        orders = [c["order"] for c in result["clips"]]
+        assert orders == list(range(len(result["clips"])))
+
+
+# ── _call_ai_cover_service 测试 ───────────────────────────────────────────
+
+
+class TestAiCoverService:
+    """AI封面生成服务测试."""
+
+    def test_cover_type_upload(self):
+        with patch("shared.ai_service.time.sleep"):
+            result = _call_ai_cover_service("plan1", ["a1"], "upload")
+        assert result["type"] == "upload"
+        assert result["image_url"] == ""
+
+    def test_cover_type_manual_with_frame_time(self):
+        with patch("shared.ai_service.time.sleep"):
+            result = _call_ai_cover_service("plan1", ["a1"], "manual", frame_time=5.5)
+        assert result["type"] == "manual"
+        assert result["frame_time"] == 5.5
+        assert "5.5" in result["image_url"]
+
+    def test_cover_type_ai_frame(self):
+        with patch("shared.ai_service.time.sleep"):
+            with patch("shared.ai_service.random.uniform", side_effect=[5.0, 0.9]):
+                result = _call_ai_cover_service("plan1", ["a1"], "ai_frame")
+        assert result["type"] == "ai_frame"
+        assert result["frame_time"] == 5.0
+        assert result["confidence"] == 0.9
+        assert "plan1" in result["image_url"]
+
+    def test_cover_type_ai_regenerate(self):
+        with patch("shared.ai_service.time.sleep"):
+            result = _call_ai_cover_service("plan1", ["a1"], "ai_regenerate")
+        assert result["type"] == "ai_frame"
+
+    def test_cover_frame_time_in_range(self):
+        with patch("shared.ai_service.time.sleep"):
+            result = _call_ai_cover_service("plan1", ["a1"], "ai_frame")
+        assert 1.0 <= result["frame_time"] <= 10.0