test(unit): 第86波 - worker层VoiceExtractor/CoverGenerator/VideoProcessor纯逻辑单测 (+63)

- VoiceExtractor: 命令构建 + 滤镜顺序 + 边界用例 + 异常传递 (19个)
- CoverGenerator: 时间钳制 + 智能选帧均匀分布 + 文件大小选最佳 (23个)
- VideoProcessor: 数据类 + fps解析 + concat文件格式 + 缩略图参数 (21个)
- 全部通过mock外部依赖(ffmpeg/run_ffmpeg)实现，纯逻辑验证，无需真实FFmpeg

tests/unit/test_cover_generator_pure.py

@@ -0,0 +1,539 @@
+"""CoverGenerator 纯逻辑单测 — 时间钳制 + 智能选帧算法.
+
+通过 mock run_ffmpeg 和 probe_video_info 验证纯逻辑部分，
+不实际执行 FFmpeg，确保测试轻量快速。
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from video_processing.cover_generator import (
+    CoverGenerator,
+    DEFAULT_COVER_HEIGHT,
+    DEFAULT_COVER_QUALITY,
+    DEFAULT_COVER_TIME,
+    DEFAULT_COVER_WIDTH,
+    SMART_COVER_FRAME_COUNT,
+)
+
+
+class TestCoverGeneratorConstants:
+    """常量默认值测试."""
+
+    def test_default_cover_time(self):
+        """默认抽帧时间为 1.0 秒."""
+        assert DEFAULT_COVER_TIME == 1.0
+
+    def test_default_dimensions(self):
+        """默认封面尺寸 1080x1920 (竖屏)."""
+        assert DEFAULT_COVER_WIDTH == 1080
+        assert DEFAULT_COVER_HEIGHT == 1920
+
+    def test_default_quality(self):
+        """默认质量为 5 (JPEG q:v, 越小越好)."""
+        assert DEFAULT_COVER_QUALITY == 5
+
+    def test_smart_cover_frame_count(self):
+        """智能封面默认抽 3 帧."""
+        assert SMART_COVER_FRAME_COUNT == 3
+
+
+class TestExtractFrameCommand:
+    """extract_frame 命令构建测试."""
+
+    def _probe_video_info_mock(self, duration=10.0):
+        """创建 probe_video_info 的 mock."""
+        return {"duration": duration, "width": 1920, "height": 1080, "fps": 25.0}
+
+    def test_default_params_command(self, tmp_path):
+        """默认参数下 FFmpeg 命令正确."""
+        video_file = tmp_path / "test.mp4"
+        video_file.write_bytes(b"fake")
+        output_file = tmp_path / "cover.jpg"
+
+        with (
+            patch(
+                "video_processing.cover_generator.probe_video_info",
+                return_value=self._probe_video_info_mock(),
+            ),
+            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
+        ):
+            # 让 output_path 在 run_ffmpeg 后存在
+            def fake_run(cmd):
+                output_file.write_bytes(b"fake jpg")
+
+            mock_run.side_effect = fake_run
+            result = CoverGenerator.extract_frame(str(video_file), str(output_file))
+
+        assert result == Path(output_file)
+        mock_run.assert_called_once()
+        cmd = mock_run.call_args[0][0]
+
+        # 基本结构
+        assert cmd[0].endswith("ffmpeg") or "ffmpeg" in cmd[0]
+        assert "-y" in cmd
+        assert "-vframes" in cmd
+        assert cmd[cmd.index("-vframes") + 1] == "1"
+        assert "-f" in cmd
+        assert "mjpeg" in cmd[cmd.index("-f") + 1]
+
+        # 时间点
+        ss_idx = cmd.index("-ss")
+        assert float(cmd[ss_idx + 1]) == pytest.approx(DEFAULT_COVER_TIME, abs=0.001)
+
+        # 输入文件
+        i_idx = cmd.index("-i")
+        assert cmd[i_idx + 1] == str(video_file)
+
+        # 输出文件
+        assert cmd[-1] == str(output_file)
+
+        # scale + crop 滤镜
+        vf_idx = cmd.index("-vf")
+        vf_value = cmd[vf_idx + 1]
+        assert "scale=" in vf_value
+        assert "crop=" in vf_value
+        assert "force_original_aspect_ratio=increase" in vf_value
+
+    def test_custom_time(self, tmp_path):
+        """自定义抽帧时间点."""
+        video_file = tmp_path / "test.mp4"
+        video_file.write_bytes(b"fake")
+        output_file = tmp_path / "cover.jpg"
+
+        with (
+            patch(
+                "video_processing.cover_generator.probe_video_info",
+                return_value=self._probe_video_info_mock(duration=30.0),
+            ),
+            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
+        ):
+
+            def fake_run(cmd):
+                output_file.write_bytes(b"fake jpg")
+
+            mock_run.side_effect = fake_run
+            CoverGenerator.extract_frame(str(video_file), str(output_file), time_sec=5.5)
+
+        cmd = mock_run.call_args[0][0]
+        ss_idx = cmd.index("-ss")
+        assert float(cmd[ss_idx + 1]) == pytest.approx(5.5, abs=0.001)
+
+    def test_custom_dimensions(self, tmp_path):
+        """自定义输出尺寸."""
+        video_file = tmp_path / "test.mp4"
+        video_file.write_bytes(b"fake")
+        output_file = tmp_path / "cover.jpg"
+
+        with (
+            patch(
+                "video_processing.cover_generator.probe_video_info",
+                return_value=self._probe_video_info_mock(),
+            ),
+            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
+        ):
+
+            def fake_run(cmd):
+                output_file.write_bytes(b"fake jpg")
+
+            mock_run.side_effect = fake_run
+            CoverGenerator.extract_frame(str(video_file), str(output_file), width=1920, height=1080)
+
+        cmd = mock_run.call_args[0][0]
+        vf_idx = cmd.index("-vf")
+        vf_value = cmd[vf_idx + 1]
+        assert "scale=1920:1080:" in vf_value
+        assert "crop=1920:1080" in vf_value
+
+    def test_custom_quality(self, tmp_path):
+        """自定义 JPEG 质量."""
+        video_file = tmp_path / "test.mp4"
+        video_file.write_bytes(b"fake")
+        output_file = tmp_path / "cover.jpg"
+
+        with (
+            patch(
+                "video_processing.cover_generator.probe_video_info",
+                return_value=self._probe_video_info_mock(),
+            ),
+            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
+        ):
+
+            def fake_run(cmd):
+                output_file.write_bytes(b"fake jpg")
+
+            mock_run.side_effect = fake_run
+            CoverGenerator.extract_frame(str(video_file), str(output_file), quality=2)
+
+        cmd = mock_run.call_args[0][0]
+        q_idx = cmd.index("-q:v")
+        assert cmd[q_idx + 1] == "2"
+
+    def test_time_exceeds_duration_clamps_to_midpoint(self, tmp_path):
+        """抽帧时间超过视频时长时，钳制到中间帧."""
+        video_file = tmp_path / "test.mp4"
+        video_file.write_bytes(b"fake")
+        output_file = tmp_path / "cover.jpg"
+
+        with (
+            patch(
+                "video_processing.cover_generator.probe_video_info",
+                return_value=self._probe_video_info_mock(duration=5.0),
+            ),
+            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
+        ):
+
+            def fake_run(cmd):
+                output_file.write_bytes(b"fake jpg")
+
+            mock_run.side_effect = fake_run
+            CoverGenerator.extract_frame(str(video_file), str(output_file), time_sec=10.0)
+
+        cmd = mock_run.call_args[0][0]
+        ss_idx = cmd.index("-ss")
+        # 钳制到 duration/2 = 2.5
+        assert float(cmd[ss_idx + 1]) == pytest.approx(2.5, abs=0.001)
+
+    def test_negative_time_clamps_to_zero(self, tmp_path):
+        """负时间钳制到 0."""
+        video_file = tmp_path / "test.mp4"
+        video_file.write_bytes(b"fake")
+        output_file = tmp_path / "cover.jpg"
+
+        with (
+            patch(
+                "video_processing.cover_generator.probe_video_info",
+                return_value=self._probe_video_info_mock(duration=10.0),
+            ),
+            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
+        ):
+
+            def fake_run(cmd):
+                output_file.write_bytes(b"fake jpg")
+
+            mock_run.side_effect = fake_run
+            CoverGenerator.extract_frame(str(video_file), str(output_file), time_sec=-2.0)
+
+        cmd = mock_run.call_args[0][0]
+        ss_idx = cmd.index("-ss")
+        assert float(cmd[ss_idx + 1]) == pytest.approx(0.0, abs=0.001)
+
+    def test_time_equals_duration_clamps_to_midpoint(self, tmp_path):
+        """时间点等于时长时钳制到中间帧."""
+        video_file = tmp_path / "test.mp4"
+        video_file.write_bytes(b"fake")
+        output_file = tmp_path / "cover.jpg"
+
+        with (
+            patch(
+                "video_processing.cover_generator.probe_video_info",
+                return_value=self._probe_video_info_mock(duration=10.0),
+            ),
+            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
+        ):
+
+            def fake_run(cmd):
+                output_file.write_bytes(b"fake jpg")
+
+            mock_run.side_effect = fake_run
+            CoverGenerator.extract_frame(str(video_file), str(output_file), time_sec=10.0)
+
+        cmd = mock_run.call_args[0][0]
+        ss_idx = cmd.index("-ss")
+        assert float(cmd[ss_idx + 1]) == pytest.approx(5.0, abs=0.001)
+
+    def test_zero_duration_video(self, tmp_path):
+        """视频时长为 0 时的行为（不钳制，用原始时间）."""
+        video_file = tmp_path / "test.mp4"
+        video_file.write_bytes(b"fake")
+        output_file = tmp_path / "cover.jpg"
+
+        with (
+            patch(
+                "video_processing.cover_generator.probe_video_info",
+                return_value=self._probe_video_info_mock(duration=0.0),
+            ),
+            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
+        ):
+
+            def fake_run(cmd):
+                output_file.write_bytes(b"fake jpg")
+
+            mock_run.side_effect = fake_run
+            CoverGenerator.extract_frame(str(video_file), str(output_file), time_sec=0.5)
+
+        cmd = mock_run.call_args[0][0]
+        ss_idx = cmd.index("-ss")
+        assert float(cmd[ss_idx + 1]) == pytest.approx(0.5, abs=0.001)
+
+    def test_video_not_found_raises(self, tmp_path):
+        """视频文件不存在时抛出 FileNotFoundError."""
+        output_file = tmp_path / "cover.jpg"
+
+        with pytest.raises(FileNotFoundError):
+            CoverGenerator.extract_frame(str(tmp_path / "nonexistent.mp4"), str(output_file))
+
+    def test_output_creates_parent_dir(self, tmp_path):
+        """输出目录不存在时自动创建."""
+        video_file = tmp_path / "test.mp4"
+        video_file.write_bytes(b"fake")
+        out_dir = tmp_path / "deep" / "nested"
+        output_file = out_dir / "cover.jpg"
+
+        with (
+            patch(
+                "video_processing.cover_generator.probe_video_info",
+                return_value=self._probe_video_info_mock(),
+            ),
+            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
+        ):
+
+            def fake_run(cmd):
+                output_file.write_bytes(b"fake jpg")
+
+            mock_run.side_effect = fake_run
+            CoverGenerator.extract_frame(str(video_file), str(output_file))
+
+        assert out_dir.exists()
+        assert out_dir.is_dir()
+
+    def test_ffmpeg_failure_propagates(self, tmp_path):
+        """FFmpeg 失败时异常向上传递."""
+        video_file = tmp_path / "test.mp4"
+        video_file.write_bytes(b"fake")
+        output_file = tmp_path / "cover.jpg"
+
+        with (
+            patch(
+                "video_processing.cover_generator.probe_video_info",
+                return_value=self._probe_video_info_mock(),
+            ),
+            patch(
+                "video_processing.cover_generator.run_ffmpeg",
+                side_effect=RuntimeError("FFmpeg error"),
+            ),
+        ):
+            with pytest.raises(RuntimeError, match="FFmpeg error"):
+                CoverGenerator.extract_frame(str(video_file), str(output_file))
+
+
+class TestSmartCoverTimePoints:
+    """智能封面时间点计算测试."""
+
+    def test_single_frame_falls_back_to_default(self, tmp_path):
+        """只有 1 帧时退化为普通抽帧（取 DEFAULT_COVER_TIME 和 midpoint 中较小值）."""
+        video_file = tmp_path / "test.mp4"
+        video_file.write_bytes(b"fake")
+        output_file = tmp_path / "cover.jpg"
+
+        with (
+            patch(
+                "video_processing.cover_generator.probe_video_info",
+                return_value={"duration": 20.0},
+            ),
+            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
+        ):
+
+            def fake_run(cmd):
+                output_file.write_bytes(b"fake jpg")
+
+            mock_run.side_effect = fake_run
+            # frame_count=1 时退化为普通抽帧
+            CoverGenerator.extract_smart_cover(str(video_file), str(output_file), frame_count=1)
+
+        # 只调用一次（退化路径）
+        assert mock_run.call_count == 1
+        cmd = mock_run.call_args[0][0]
+        ss_idx = cmd.index("-ss")
+        # min(DEFAULT_COVER_TIME=1.0, duration/2=10.0) = 1.0
+        assert float(cmd[ss_idx + 1]) == pytest.approx(1.0, abs=0.001)
+
+    def test_zero_duration_falls_back(self, tmp_path):
+        """视频时长为 0 时退化为普通抽帧."""
+        video_file = tmp_path / "test.mp4"
+        video_file.write_bytes(b"fake")
+        output_file = tmp_path / "cover.jpg"
+
+        with (
+            patch(
+                "video_processing.cover_generator.probe_video_info",
+                return_value={"duration": 0.0},
+            ),
+            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
+        ):
+
+            def fake_run(cmd):
+                output_file.write_bytes(b"fake jpg")
+
+            mock_run.side_effect = fake_run
+            CoverGenerator.extract_smart_cover(str(video_file), str(output_file))
+
+        # 只调用一次（退化路径）
+        assert mock_run.call_count == 1
+
+    def test_three_frames_uniform_distribution(self, tmp_path):
+        """3 帧均匀分布在 5%~95% 区间."""
+        video_file = tmp_path / "test.mp4"
+        video_file.write_bytes(b"fake")
+        output_file = tmp_path / "cover.jpg"
+
+        call_times = []
+
+        with (
+            patch(
+                "video_processing.cover_generator.probe_video_info",
+                return_value={"duration": 100.0},
+            ),
+            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
+        ):
+
+            def fake_run(cmd):
+                # 记录抽帧时间
+                ss_idx = cmd.index("-ss")
+                call_times.append(float(cmd[ss_idx + 1]))
+                # 在输出路径写文件
+                output_arg = cmd[-1]
+                Path(output_arg).parent.mkdir(parents=True, exist_ok=True)
+                # 不同文件大小，让第三帧"最清晰"
+                idx = len(call_times) - 1
+                size = 1000 * (idx + 1)  # 递增的文件大小
+                Path(output_arg).write_bytes(b"x" * size)
+
+            mock_run.side_effect = fake_run
+            CoverGenerator.extract_smart_cover(str(video_file), str(output_file))
+
+        # 3 帧：5%、50%、95%
+        assert len(call_times) == 3
+        assert call_times[0] == pytest.approx(5.0, abs=0.1)  # 5%
+        assert call_times[1] == pytest.approx(50.0, abs=0.1)  # 50%
+        assert call_times[2] == pytest.approx(95.0, abs=0.1)  # 95%
+
+    def test_five_frames_distribution(self, tmp_path):
+        """5 帧均匀分布."""
+        video_file = tmp_path / "test.mp4"
+        video_file.write_bytes(b"fake")
+        output_file = tmp_path / "cover.jpg"
+
+        call_times = []
+
+        with (
+            patch(
+                "video_processing.cover_generator.probe_video_info",
+                return_value={"duration": 100.0},
+            ),
+            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
+        ):
+
+            def fake_run(cmd):
+                ss_idx = cmd.index("-ss")
+                call_times.append(float(cmd[ss_idx + 1]))
+                output_arg = cmd[-1]
+                Path(output_arg).parent.mkdir(parents=True, exist_ok=True)
+                idx = len(call_times) - 1
+                Path(output_arg).write_bytes(b"x" * (1000 * (idx + 1)))
+
+            mock_run.side_effect = fake_run
+            CoverGenerator.extract_smart_cover(str(video_file), str(output_file), frame_count=5)
+
+        assert len(call_times) == 5
+        # step = (95-5) / (5-1) = 22.5
+        # times: 5, 27.5, 50, 72.5, 95
+        assert call_times[0] == pytest.approx(5.0, abs=0.1)
+        assert call_times[1] == pytest.approx(27.5, abs=0.1)
+        assert call_times[2] == pytest.approx(50.0, abs=0.1)
+        assert call_times[3] == pytest.approx(72.5, abs=0.1)
+        assert call_times[4] == pytest.approx(95.0, abs=0.1)
+
+    def test_selects_largest_file_as_best(self, tmp_path):
+        """选择文件最大的帧作为最佳封面（清晰度近似）."""
+        video_file = tmp_path / "test.mp4"
+        video_file.write_bytes(b"fake")
+        output_file = tmp_path / "cover.jpg"
+
+        sizes = [5000, 15000, 8000]  # 第二帧最大
+
+        with (
+            patch(
+                "video_processing.cover_generator.probe_video_info",
+                return_value={"duration": 100.0},
+            ),
+            patch("video_processing.cover_generator.run_ffmpeg") as mock_run,
+        ):
+            call_idx = [0]
+
+            def fake_run(cmd):
+                output_arg = cmd[-1]
+                Path(output_arg).parent.mkdir(parents=True, exist_ok=True)
+                idx = call_idx[0]
+                Path(output_arg).write_bytes(b"x" * sizes[idx])
+                call_idx[0] += 1
+
+            mock_run.side_effect = fake_run
+            result = CoverGenerator.extract_smart_cover(str(video_file), str(output_file))
+
+        # 第二帧（索引1）应该是最佳
+        assert result == output_file
+        # 输出文件大小应等于第二帧大小
+        assert output_file.stat().st_size == 15000
+
+
+class TestProcessCustomCover:
+    """自定义封面处理测试."""
+
+    def test_custom_cover_resize_command(self, tmp_path):
+        """自定义封面调整尺寸命令正确."""
+        input_file = tmp_path / "upload.jpg"
+        input_file.write_bytes(b"fake")
+        output_file = tmp_path / "cover.jpg"
+
+        with patch("video_processing.cover_generator.run_ffmpeg") as mock_run:
+
+            def fake_run(cmd):
+                output_file.write_bytes(b"fake jpg")
+
+            mock_run.side_effect = fake_run
+            CoverGenerator.process_custom_cover(str(input_file), str(output_file))
+
+        mock_run.assert_called_once()
+        cmd = mock_run.call_args[0][0]
+
+        assert "-i" in cmd
+        assert cmd[cmd.index("-i") + 1] == str(input_file)
+        assert cmd[-1] == str(output_file)
+
+        # scale + crop
+        vf_idx = cmd.index("-vf")
+        vf_value = cmd[vf_idx + 1]
+        assert "scale=" in vf_value
+        assert "crop=" in vf_value
+
+    def test_custom_cover_not_found_raises(self, tmp_path):
+        """自定义封面文件不存在时抛出 FileNotFoundError."""
+        output_file = tmp_path / "cover.jpg"
+
+        with pytest.raises(FileNotFoundError):
+            CoverGenerator.process_custom_cover(str(tmp_path / "nonexistent.jpg"), str(output_file))
+
+    def test_custom_cover_custom_dimensions(self, tmp_path):
+        """自定义封面自定义输出尺寸."""
+        input_file = tmp_path / "upload.jpg"
+        input_file.write_bytes(b"fake")
+        output_file = tmp_path / "cover.jpg"
+
+        with patch("video_processing.cover_generator.run_ffmpeg") as mock_run:
+
+            def fake_run(cmd):
+                output_file.write_bytes(b"fake jpg")
+
+            mock_run.side_effect = fake_run
+            CoverGenerator.process_custom_cover(str(input_file), str(output_file), width=800, height=600)
+
+        cmd = mock_run.call_args[0][0]
+        vf_idx = cmd.index("-vf")
+        vf_value = cmd[vf_idx + 1]
+        assert "scale=800:600:" in vf_value
+        assert "crop=800:600" in vf_value

tests/unit/test_video_processor_pure.py

@@ -0,0 +1,365 @@
+"""VideoProcessor 纯逻辑单测 — 数据类 + 输入校验 + 解析逻辑.
+
+通过 mock ffmpeg-python 库验证纯逻辑部分，
+不实际执行 FFmpeg，确保测试轻量快速。
+"""
+
+from __future__ import annotations
+
+import os
+from dataclasses import fields
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from video_processing.processor import VideoProcessor, VideoResult
+
+
+class TestVideoResultDataclass:
+    """VideoResult 数据类测试."""
+
+    def test_all_fields_exist(self):
+        """所有字段都存在."""
+        field_names = {f.name for f in fields(VideoResult)}
+        expected = {
+            "output_path",
+            "thumbnail_path",
+            "duration",
+            "width",
+            "height",
+            "fps",
+            "file_size",
+        }
+        assert expected.issubset(field_names)
+
+    def test_default_construction(self):
+        """正常构造 VideoResult."""
+        result = VideoResult(
+            output_path="/tmp/out.mp4",
+            thumbnail_path="/tmp/out.jpg",
+            duration=10.5,
+            width=1920,
+            height=1080,
+            fps=25.0,
+            file_size=1024000,
+        )
+        assert result.output_path == "/tmp/out.mp4"
+        assert result.thumbnail_path == "/tmp/out.jpg"
+        assert result.duration == 10.5
+        assert result.width == 1920
+        assert result.height == 1080
+        assert result.fps == 25.0
+        assert result.file_size == 1024000
+
+    def test_zero_values(self):
+        """零值/边界值构造."""
+        result = VideoResult(
+            output_path="",
+            thumbnail_path="",
+            duration=0.0,
+            width=0,
+            height=0,
+            fps=0.0,
+            file_size=0,
+        )
+        assert result.duration == 0.0
+        assert result.file_size == 0
+
+
+class TestVideoProcessorInit:
+    """VideoProcessor 初始化测试."""
+
+    def test_default_temp_dir(self):
+        """默认使用系统临时目录."""
+        import tempfile
+
+        vp = VideoProcessor()
+        assert vp.temp_dir == tempfile.gettempdir()
+
+    def test_custom_temp_dir(self):
+        """自定义临时目录."""
+        vp = VideoProcessor(temp_dir="/my/temp")
+        assert vp.temp_dir == "/my/temp"
+
+
+class TestVideoProcessorConcatenateValidation:
+    """concatenate_videos 输入校验测试."""
+
+    def test_empty_input_raises(self):
+        """空输入列表抛出 ValueError."""
+        vp = VideoProcessor()
+        with pytest.raises(ValueError, match="cannot be empty"):
+            vp.concatenate_videos([], "/tmp/output.mp4")
+
+    def test_none_input_raises(self):
+        """None 输入抛出异常."""
+        vp = VideoProcessor()
+        with pytest.raises((ValueError, TypeError)):
+            vp.concatenate_videos(None, "/tmp/output.mp4")  # type: ignore[arg-type]
+
+
+class TestVideoProcessorGetVideoInfoParsing:
+    """get_video_info 解析逻辑测试（mock ffmpeg.probe）."""
+
+    def _mock_probe(self, streams=None, fmt=None):
+        """创建 ffmpeg.probe 的 mock 返回值."""
+        return {
+            "streams": streams
+            or [{"codec_type": "video", "width": 1920, "height": 1080, "r_frame_rate": "25/1", "codec_name": "h264"}],
+            "format": fmt or {"duration": "10.5", "bit_rate": "5000000"},
+        }
+
+    def test_basic_info_parsing(self):
+        """基本视频信息解析正确."""
+        vp = VideoProcessor()
+        probe_data = self._mock_probe()
+
+        with patch("video_processing.processor.ffmpeg.probe", return_value=probe_data):
+            info = vp.get_video_info("/tmp/test.mp4")
+
+        assert info["duration"] == 10.5
+        assert info["width"] == 1920
+        assert info["height"] == 1080
+        assert info["fps"] == 25.0
+        assert info["codec"] == "h264"
+        assert info["bitrate"] == 5000000
+
+    def test_fps_fraction_parsing(self):
+        """分数帧率解析（如 30000/1001 = 29.97）."""
+        vp = VideoProcessor()
+        probe_data = self._mock_probe(
+            streams=[
+                {
+                    "codec_type": "video",
+                    "width": 1920,
+                    "height": 1080,
+                    "r_frame_rate": "30000/1001",
+                    "codec_name": "h264",
+                }
+            ]
+        )
+
+        with patch("video_processing.processor.ffmpeg.probe", return_value=probe_data):
+            info = vp.get_video_info("/tmp/test.mp4")
+
+        assert info["fps"] == pytest.approx(29.97, abs=0.01)
+
+    def test_fps_integer_string(self):
+        """整数字符串帧率（如 "60"）."""
+        vp = VideoProcessor()
+        probe_data = self._mock_probe(
+            streams=[{"codec_type": "video", "width": 1920, "height": 1080, "r_frame_rate": "60", "codec_name": "h264"}]
+        )
+
+        with patch("video_processing.processor.ffmpeg.probe", return_value=probe_data):
+            info = vp.get_video_info("/tmp/test.mp4")
+
+        assert info["fps"] == 60.0
+
+    def test_missing_r_frame_rate(self):
+        """缺少 r_frame_rate 时使用默认值."""
+        vp = VideoProcessor()
+        probe_data = self._mock_probe(
+            streams=[{"codec_type": "video", "width": 1920, "height": 1080, "codec_name": "h264"}]
+        )
+
+        with patch("video_processing.processor.ffmpeg.probe", return_value=probe_data):
+            info = vp.get_video_info("/tmp/test.mp4")
+
+        assert info["fps"] == 25.0
+
+    def test_no_video_stream(self):
+        """没有视频流时的行为."""
+        vp = VideoProcessor()
+        probe_data = {
+            "streams": [{"codec_type": "audio", "codec_name": "aac"}],
+            "format": {"duration": "10.0", "bit_rate": "128000"},
+        }
+
+        with patch("video_processing.processor.ffmpeg.probe", return_value=probe_data):
+            with pytest.raises(StopIteration):
+                vp.get_video_info("/tmp/test.mp4")
+
+    def test_float_duration(self):
+        """浮点时长解析."""
+        vp = VideoProcessor()
+        probe_data = self._mock_probe(fmt={"duration": "123.456", "bit_rate": "0"})
+
+        with patch("video_processing.processor.ffmpeg.probe", return_value=probe_data):
+            info = vp.get_video_info("/tmp/test.mp4")
+
+        assert info["duration"] == pytest.approx(123.456, abs=0.001)
+
+    def test_bitrate_zero(self):
+        """码率为 0 时."""
+        vp = VideoProcessor()
+        probe_data = self._mock_probe(fmt={"duration": "10.0", "bit_rate": "0"})
+
+        with patch("video_processing.processor.ffmpeg.probe", return_value=probe_data):
+            info = vp.get_video_info("/tmp/test.mp4")
+
+        assert info["bitrate"] == 0
+
+    def test_ffmpeg_probe_error_raises(self):
+        """ffmpeg.probe 失败时抛出 RuntimeError."""
+        vp = VideoProcessor()
+
+        import ffmpeg
+
+        with patch(
+            "video_processing.processor.ffmpeg.probe",
+            side_effect=ffmpeg.Error([], b"", b"No such file"),
+        ):
+            with pytest.raises(RuntimeError, match="probe error"):
+                vp.get_video_info("/tmp/nonexistent.mp4")
+
+
+class TestVideoProcessorGenerateThumbnail:
+    """generate_thumbnail 测试."""
+
+    def _build_mock_chain(self):
+        """构建 ffmpeg.input → .output → .overwrite_output → .run 调用链."""
+        mock_input_node = MagicMock()
+        mock_output_node = MagicMock()
+        mock_overwrite_node = MagicMock()
+        mock_input_node.output.return_value = mock_output_node
+        mock_output_node.overwrite_output.return_value = mock_overwrite_node
+        return mock_input_node, mock_output_node, mock_overwrite_node
+
+    def test_default_output_path(self):
+        """默认输出路径为视频路径 + _thumb.jpg."""
+        vp = VideoProcessor()
+        mock_input_node, _mock_output, mock_overwrite = self._build_mock_chain()
+
+        with patch("video_processing.processor.ffmpeg.input", return_value=mock_input_node) as mock_ff_input:
+            result = vp.generate_thumbnail("/tmp/video.mp4")
+
+        assert result == "/tmp/video_thumb.jpg"
+        mock_ff_input.assert_called_once_with("/tmp/video.mp4", ss=1.0)
+        mock_input_node.output.assert_called_once()
+        # 验证输出路径和参数
+        output_args = mock_input_node.output.call_args
+        assert output_args[0][0] == "/tmp/video_thumb.jpg"
+        assert output_args[1].get("vframes") == 1
+        assert output_args[1].get("format") == "image2"
+        assert output_args[1].get("vcodec") == "mjpeg"
+
+    def test_custom_output_path(self):
+        """自定义输出路径."""
+        vp = VideoProcessor()
+        mock_input_node, _mock_output, mock_overwrite = self._build_mock_chain()
+
+        with patch("video_processing.processor.ffmpeg.input", return_value=mock_input_node) as mock_ff_input:
+            result = vp.generate_thumbnail("/tmp/video.mp4", output_path="/custom/thumb.jpg")
+
+        assert result == "/custom/thumb.jpg"
+
+    def test_custom_timestamp(self):
+        """自定义截图时间点."""
+        vp = VideoProcessor()
+        mock_input_node, _mock_output, mock_overwrite = self._build_mock_chain()
+
+        with patch("video_processing.processor.ffmpeg.input", return_value=mock_input_node) as mock_ff_input:
+            vp.generate_thumbnail("/tmp/video.mp4", timestamp=3.5)
+
+        # 验证 ss 参数
+        mock_ff_input.assert_called_once_with("/tmp/video.mp4", ss=3.5)
+
+    def test_ffmpeg_error_raises_runtime(self):
+        """FFmpeg 失败时抛出 RuntimeError."""
+        vp = VideoProcessor()
+
+        import ffmpeg
+
+        mock_input_node, mock_output, mock_overwrite = self._build_mock_chain()
+        mock_overwrite.run.side_effect = ffmpeg.Error([], b"", b"Output file #0 does not contain any stream")
+
+        with patch("video_processing.processor.ffmpeg.input", return_value=mock_input_node):
+            with pytest.raises(RuntimeError, match="thumbnail error"):
+                vp.generate_thumbnail("/tmp/video.mp4")
+
+
+class TestVideoProcessorConcatFileFormat:
+    """concat 临时文件格式验证."""
+
+    def test_concat_file_format(self, tmp_path):
+        """concat 临时文件格式符合 FFmpeg concat demuxer 规范."""
+        import os
+
+        vp = VideoProcessor(temp_dir=str(tmp_path))
+
+        written_content = {}
+
+        def fake_input(path, *args, **kwargs):
+            mock_node = MagicMock()
+            mock_output = MagicMock()
+            mock_overwrite = MagicMock()
+            mock_node.output.return_value = mock_output
+            mock_output.overwrite_output.return_value = mock_overwrite
+
+            if kwargs.get("format") == "concat":
+                # 读取 concat 文件内容
+                with open(path) as f:
+                    written_content["concat"] = f.read()
+            return mock_node
+
+        mock_probe = MagicMock(
+            return_value={
+                "streams": [{"codec_type": "video", "width": 1920, "height": 1080, "r_frame_rate": "25/1"}],
+                "format": {"duration": "5.0", "bit_rate": "1000000"},
+            }
+        )
+
+        with (
+            patch("video_processing.processor.ffmpeg.input", side_effect=fake_input),
+            patch("video_processing.processor.ffmpeg.probe", mock_probe),
+            patch("video_processing.processor.os.path.getsize", return_value=1024),
+        ):
+            with patch.object(VideoProcessor, "generate_thumbnail", return_value="/tmp/thumb.jpg"):
+                vp.concatenate_videos(
+                    ["/tmp/a.mp4", "/tmp/b.mp4", "/tmp/c.mp4"],
+                    str(tmp_path / "output.mp4"),
+                )
+
+        # 验证 concat 文件格式
+        assert "concat" in written_content
+        lines = written_content["concat"].strip().split("\n")
+        assert len(lines) == 3
+        assert lines[0].startswith("file '")
+        assert "a.mp4'" in lines[0]
+        assert "b.mp4'" in lines[1]
+        assert "c.mp4'" in lines[2]
+        # 使用绝对路径
+        first_path = lines[0].replace("file '", "").rstrip("'")
+        assert os.path.isabs(first_path)
+
+    def test_concat_creates_output_directory(self, tmp_path):
+        """输出目录不存在时自动创建."""
+        vp = VideoProcessor(temp_dir=str(tmp_path))
+
+        out_dir = tmp_path / "deep" / "output"
+        out_file = out_dir / "result.mp4"
+
+        mock_node = MagicMock()
+        mock_output = MagicMock()
+        mock_overwrite = MagicMock()
+        mock_node.output.return_value = mock_output
+        mock_output.overwrite_output.return_value = mock_overwrite
+
+        mock_probe = MagicMock(
+            return_value={
+                "streams": [{"codec_type": "video", "width": 1920, "height": 1080, "r_frame_rate": "25/1"}],
+                "format": {"duration": "5.0", "bit_rate": "1000000"},
+            }
+        )
+
+        with (
+            patch("video_processing.processor.ffmpeg.input", return_value=mock_node),
+            patch("video_processing.processor.ffmpeg.probe", mock_probe),
+            patch("video_processing.processor.os.path.getsize", return_value=1024),
+        ):
+            with patch.object(VideoProcessor, "generate_thumbnail", return_value=str(out_dir / "thumb.jpg")):
+                vp.concatenate_videos(["/tmp/a.mp4"], str(out_file))
+
+        assert out_dir.exists()
+        assert out_dir.is_dir()

tests/unit/test_voice_extractor_pure.py

@@ -0,0 +1,260 @@
+"""VoiceExtractor 纯逻辑单测 — 命令构建 + 边界用例.
+
+通过 mock run_ffmpeg 验证 FFmpeg 命令参数是否正确，
+不实际执行 FFmpeg，确保测试轻量快速。
+"""
+
+from __future__ import annotations
+
+import os
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from worker_app.tasks.voice_extraction import VoiceExtractor
+
+
+class TestVoiceExtractorExtractVoiceCommand:
+    """extract_voice 命令构建测试."""
+
+    def test_default_params_correct_command(self):
+        """默认参数下 FFmpeg 命令正确."""
+        extractor = VoiceExtractor()
+        with patch.object(VoiceExtractor, "_run_ffmpeg") as mock_run:
+            result = extractor.extract_voice("/tmp/input.mp4", "/tmp/output.mp3")
+
+        assert result == "/tmp/output.mp3"
+        mock_run.assert_called_once()
+        cmd = mock_run.call_args[0][0]
+
+        # 基本结构验证
+        assert cmd[0] == "ffmpeg"
+        assert "-y" in cmd
+        assert cmd[cmd.index("-i") + 1] == "/tmp/input.mp4"
+        assert "-vn" in cmd  # 无视频流
+        assert cmd[-1] == "/tmp/output.mp3"
+
+        # 音频滤镜验证
+        af_idx = cmd.index("-af")
+        af_value = cmd[af_idx + 1]
+        assert "highpass=f=200" in af_value
+        assert "afftdn=bn=20" in af_value
+        assert "bandpass=f=300:width_type=h:width=3000" in af_value
+        assert "loudnorm" in af_value
+
+        # 编码验证
+        assert "libmp3lame" in cmd
+        assert "-q:a" in cmd
+        assert cmd[cmd.index("-q:a") + 1] == "2"
+
+    def test_custom_highpass(self):
+        """自定义 highpass 频率."""
+        extractor = VoiceExtractor()
+        with patch.object(VoiceExtractor, "_run_ffmpeg") as mock_run:
+            extractor.extract_voice("/tmp/in.mp4", "/tmp/out.mp3", highpass=500)
+
+        cmd = mock_run.call_args[0][0]
+        af_value = cmd[cmd.index("-af") + 1]
+        assert "highpass=f=500" in af_value
+
+    def test_custom_bandpass_freq(self):
+        """自定义 bandpass 中心频率."""
+        extractor = VoiceExtractor()
+        with patch.object(VoiceExtractor, "_run_ffmpeg") as mock_run:
+            extractor.extract_voice("/tmp/in.mp4", "/tmp/out.mp3", bandpass_freq=500)
+
+        cmd = mock_run.call_args[0][0]
+        af_value = cmd[cmd.index("-af") + 1]
+        assert "bandpass=f=500:" in af_value
+
+    def test_custom_bandpass_width(self):
+        """自定义 bandpass 宽度."""
+        extractor = VoiceExtractor()
+        with patch.object(VoiceExtractor, "_run_ffmpeg") as mock_run:
+            extractor.extract_voice("/tmp/in.mp4", "/tmp/out.mp3", bandpass_width=5000)
+
+        cmd = mock_run.call_args[0][0]
+        af_value = cmd[cmd.index("-af") + 1]
+        assert "width=5000" in af_value
+
+    def test_custom_noise_reduction(self):
+        """自定义降噪强度."""
+        extractor = VoiceExtractor()
+        with patch.object(VoiceExtractor, "_run_ffmpeg") as mock_run:
+            extractor.extract_voice("/tmp/in.mp4", "/tmp/out.mp3", noise_reduction=30)
+
+        cmd = mock_run.call_args[0][0]
+        af_value = cmd[cmd.index("-af") + 1]
+        assert "afftdn=bn=30" in af_value
+
+    def test_filter_order_is_correct(self):
+        """滤镜顺序：highpass → 降噪 → bandpass → loudnorm."""
+        extractor = VoiceExtractor()
+        with patch.object(VoiceExtractor, "_run_ffmpeg") as mock_run:
+            extractor.extract_voice("/tmp/in.mp4", "/tmp/out.mp3")
+
+        cmd = mock_run.call_args[0][0]
+        af_value = cmd[cmd.index("-af") + 1]
+
+        hp_pos = af_value.index("highpass")
+        dn_pos = af_value.index("afftdn")
+        bp_pos = af_value.index("bandpass")
+        ln_pos = af_value.index("loudnorm")
+
+        assert hp_pos < dn_pos < bp_pos < ln_pos
+
+    def test_creates_output_directory(self, tmp_path):
+        """输出目录不存在时自动创建."""
+        out_dir = tmp_path / "nested" / "deep"
+        out_file = out_dir / "voice.mp3"
+
+        extractor = VoiceExtractor()
+        with patch.object(VoiceExtractor, "_run_ffmpeg"):
+            extractor.extract_voice("/tmp/in.mp4", str(out_file))
+
+        assert out_dir.exists()
+        assert out_dir.is_dir()
+
+    def test_returns_output_path(self):
+        """返回值为输出路径."""
+        extractor = VoiceExtractor()
+        with patch.object(VoiceExtractor, "_run_ffmpeg"):
+            result = extractor.extract_voice("/tmp/in.mp4", "/tmp/voice.mp3")
+
+        assert result == "/tmp/voice.mp3"
+
+
+class TestVoiceExtractorExtractBackgroundCommand:
+    """extract_background 命令构建测试."""
+
+    def test_default_params_correct_command(self):
+        """默认参数下 FFmpeg 命令正确."""
+        extractor = VoiceExtractor()
+        with patch.object(VoiceExtractor, "_run_ffmpeg") as mock_run:
+            result = extractor.extract_background("/tmp/input.mp4", "/tmp/output.mp3")
+
+        assert result == "/tmp/output.mp3"
+        mock_run.assert_called_once()
+        cmd = mock_run.call_args[0][0]
+
+        # 基本结构
+        assert cmd[0] == "ffmpeg"
+        assert "-y" in cmd
+        assert cmd[cmd.index("-i") + 1] == "/tmp/input.mp4"
+        assert "-vn" in cmd
+        assert cmd[-1] == "/tmp/output.mp3"
+
+        # 音频滤镜
+        af_idx = cmd.index("-af")
+        af_value = cmd[af_idx + 1]
+        assert "lowpass=f=200" in af_value
+        assert "loudnorm" in af_value
+
+        # 编码
+        assert "libmp3lame" in cmd
+
+    def test_custom_lowpass_freq(self):
+        """自定义 lowpass 频率."""
+        extractor = VoiceExtractor()
+        with patch.object(VoiceExtractor, "_run_ffmpeg") as mock_run:
+            extractor.extract_background("/tmp/in.mp4", "/tmp/out.mp3", lowpass=500)
+
+        cmd = mock_run.call_args[0][0]
+        af_value = cmd[cmd.index("-af") + 1]
+        assert "lowpass=f=500" in af_value
+
+    def test_filter_order_background(self):
+        """背景音滤镜顺序：lowpass → loudnorm."""
+        extractor = VoiceExtractor()
+        with patch.object(VoiceExtractor, "_run_ffmpeg") as mock_run:
+            extractor.extract_background("/tmp/in.mp4", "/tmp/out.mp3")
+
+        cmd = mock_run.call_args[0][0]
+        af_value = cmd[cmd.index("-af") + 1]
+
+        lp_pos = af_value.index("lowpass")
+        ln_pos = af_value.index("loudnorm")
+        assert lp_pos < ln_pos
+
+    def test_background_creates_output_directory(self, tmp_path):
+        """背景音输出目录不存在时自动创建."""
+        out_dir = tmp_path / "bgm" / "tracks"
+        out_file = out_dir / "bg.mp3"
+
+        extractor = VoiceExtractor()
+        with patch.object(VoiceExtractor, "_run_ffmpeg"):
+            extractor.extract_background("/tmp/in.mp4", str(out_file))
+
+        assert out_dir.exists()
+
+
+class TestVoiceExtractorEdgeCases:
+    """边界情况测试."""
+
+    def test_zero_highpass(self):
+        """highpass=0 时的行为（极端低值）."""
+        extractor = VoiceExtractor()
+        with patch.object(VoiceExtractor, "_run_ffmpeg") as mock_run:
+            extractor.extract_voice("/tmp/in.mp4", "/tmp/out.mp3", highpass=0)
+
+        cmd = mock_run.call_args[0][0]
+        af_value = cmd[cmd.index("-af") + 1]
+        assert "highpass=f=0" in af_value
+
+    def test_zero_bandpass_freq(self):
+        """bandpass_freq=0 时的极端情况."""
+        extractor = VoiceExtractor()
+        with patch.object(VoiceExtractor, "_run_ffmpeg") as mock_run:
+            extractor.extract_voice("/tmp/in.mp4", "/tmp/out.mp3", bandpass_freq=0)
+
+        cmd = mock_run.call_args[0][0]
+        af_value = cmd[cmd.index("-af") + 1]
+        assert "bandpass=f=0:" in af_value
+
+    def test_very_high_noise_reduction(self):
+        """极高降噪强度."""
+        extractor = VoiceExtractor()
+        with patch.object(VoiceExtractor, "_run_ffmpeg") as mock_run:
+            extractor.extract_voice("/tmp/in.mp4", "/tmp/out.mp3", noise_reduction=100)
+
+        cmd = mock_run.call_args[0][0]
+        af_value = cmd[cmd.index("-af") + 1]
+        assert "afftdn=bn=100" in af_value
+
+    def test_negative_lowpass_allowed(self):
+        """lowpass 负值（由调用方保证合法性，函数不做校验）."""
+        extractor = VoiceExtractor()
+        with patch.object(VoiceExtractor, "_run_ffmpeg") as mock_run:
+            extractor.extract_background("/tmp/in.mp4", "/tmp/out.mp3", lowpass=-10)
+
+        cmd = mock_run.call_args[0][0]
+        af_value = cmd[cmd.index("-af") + 1]
+        assert "lowpass=f=-10" in af_value
+
+    def test_run_ffmpeg_propagates_error(self):
+        """_run_ffmpeg 抛出异常时向上传递."""
+        extractor = VoiceExtractor()
+        with patch.object(VoiceExtractor, "_run_ffmpeg", side_effect=RuntimeError("FFmpeg failed")):
+            with pytest.raises(RuntimeError, match="FFmpeg failed"):
+                extractor.extract_voice("/tmp/in.mp4", "/tmp/out.mp3")
+
+    def test_voice_extractor_is_static_method(self):
+        """_run_ffmpeg 是静态方法，可在类上直接调用."""
+        # 验证 VoiceExtractor 可以直接实例化（无需参数）
+        extractor = VoiceExtractor()
+        assert extractor is not None
+
+    def test_multiple_extractions_same_instance(self):
+        """同一个实例可多次执行提取."""
+        extractor = VoiceExtractor()
+        call_count = 0
+
+        def fake_run(cmd):
+            nonlocal call_count
+            call_count += 1
+
+        with patch.object(VoiceExtractor, "_run_ffmpeg", side_effect=fake_run):
+            extractor.extract_voice("/tmp/a.mp4", "/tmp/a_voice.mp3")
+            extractor.extract_background("/tmp/a.mp4", "/tmp/a_bg.mp3")
+
+        assert call_count == 2