test(unit): 第87波 - worker层speed_engine+reverse_engine纯逻辑单测 (+85)

- test_speed_engine_pure.py (57个): SpeedConfig解析/钳制/isOriginal, SpeedEngine视频滤镜, atempo多级拆分算法, 时长计算, resolve_clip_speed
- test_reverse_engine_pure.py (28个): ReverseConfig.from_dict容错解析, 视频/音频滤镜构建, 120s安全时长限制, 组合场景验证

.gitea/workflows/ci-health-daily.yml

@@ -8,7 +8,7 @@ permissions:
 jobs:
   ci-health-report:
     name: CI健康度每日巡检
-    runs-on: saas
+    runs-on: ci-l2
     timeout-minutes: 15
     steps:
     - name: Checkout code

.gitea/workflows/ci-pipeline.yml

@@ -58,24 +58,56 @@ jobs:
       run: |
         set -eu
         PR_NUMBER=$(echo "$GITHUB_REF" | sed 's|refs/pull/||; s|/.*||')
-        API_URL="${GITHUB_API_URL}/repos/${GITHUB_REPOSITORY}/pulls/${PR_NUMBER}/files?limit=300"
-        FILES=$(curl -s -H "Authorization: token ${GITHUB_TOKEN}" "$API_URL" | python3 -c "import sys,json; [print(f['filename']) for f in json.load(sys.stdin)]")
-        FRONTEND_COUNT=$(echo "$FILES" | grep -c '^apps/web/' || true)
-        BACKEND_COUNT=$(echo "$FILES" | grep -cv '^apps/web/' || true)
-        TOTAL=$(echo "$FILES" | grep -cv '^$' || true)
-        echo "变更文件: ${TOTAL} 个 (前端: ${FRONTEND_COUNT}, 后端/公共: ${BACKEND_COUNT})"
-        if [ "$BACKEND_COUNT" = "0" ] && [ "$FRONTEND_COUNT" -gt "0" ]; then
+
+        # 分页获取所有变更文件（修复>300文件时漏判）
+        ALL_FILES=""
+        PAGE=1
+        while true; do
+          API_URL="${GITHUB_API_URL}/repos/${GITHUB_REPOSITORY}/pulls/${PR_NUMBER}/files?limit=300&page=${PAGE}"
+          PAGE_FILES=$(curl -s -H "Authorization: token ${GITHUB_TOKEN}" "$API_URL" | python3 -c "import sys,json; files=json.load(sys.stdin); [print(f['filename']) for f in files]; sys.exit(0 if len(files)==300 else 1)" 2>/dev/null || true)
+          PAGE_COUNT=$(echo "$PAGE_FILES" | wc -l)
+          if [ "$PAGE_COUNT" -lt 300 ]; then HAS_MORE=1; else HAS_MORE=0; fi
+          ALL_FILES="${ALL_FILES}${PAGE_FILES}"$'\n'
+          if [ "$HAS_MORE" != "0" ]; then
+            break
+          fi
+          PAGE=$((PAGE + 1))
+        done
+
+        FRONTEND_COUNT=$(echo "$ALL_FILES" | grep -c '^apps/web/' || true)
+        TOTAL=$(echo "$ALL_FILES" | grep -cv '^$' || true)
+        BACKEND_COUNT=$(( TOTAL - FRONTEND_COUNT ))
+
+        # 基础设施文件：改了就强制全量CI（不跳过任何检查）
+        INFRA_COUNT=$(echo "$ALL_FILES" | grep -cE '^(infra/|Dockerfile|\.gitea/workflows/|scripts/ci/|docker/)' || true)
+
+        echo "变更文件: ${TOTAL} 个 (前端: ${FRONTEND_COUNT}, 后端/公共: ${BACKEND_COUNT}, 基础设施: ${INFRA_COUNT})"
+
+        # 判定是否纯前端/纯后端
+        PURE_FRONTEND=false
+        PURE_BACKEND=false
+        if [ "$BACKEND_COUNT" = "0" ] && [ "$FRONTEND_COUNT" -gt "0" ] && [ "$INFRA_COUNT" = "0" ]; then
+          PURE_FRONTEND=true
+        elif [ "$FRONTEND_COUNT" = "0" ] && [ "$BACKEND_COUNT" -gt "0" ] && [ "$INFRA_COUNT" = "0" ]; then
+          PURE_BACKEND=true
+        fi
+
+        if [ "$PURE_FRONTEND" = "true" ]; then
           echo "skip_backend=true" >> $GITHUB_OUTPUT
           echo "skip_frontend=false" >> $GITHUB_OUTPUT
           echo "✅ 纯前端改动，跳过后端检查"
-        elif [ "$FRONTEND_COUNT" = "0" ] && [ "$BACKEND_COUNT" -gt "0" ]; then
+        elif [ "$PURE_BACKEND" = "true" ]; then
           echo "skip_backend=false" >> $GITHUB_OUTPUT
           echo "skip_frontend=true" >> $GITHUB_OUTPUT
           echo "🔧 纯后端改动，跳过前端检查"
         else
           echo "skip_backend=false" >> $GITHUB_OUTPUT
           echo "skip_frontend=false" >> $GITHUB_OUTPUT
-          echo "🔧 包含全栈变更，运行完整CI"
+          if [ "$INFRA_COUNT" -gt "0" ]; then
+            echo "🏗️  包含基础设施变更，强制运行完整CI"
+          else
+            echo "🔧 包含全栈变更，运行完整CI"
+          fi
         fi
 
     - name: Report CI trace

.gitea/workflows/ci-trigger-monitor.yml

@@ -16,15 +16,15 @@ permissions:
 jobs:
   monitor:
     name: Monitor CI Trigger Reliability
-    runs-on: ubuntu-latest
+    runs-on: ci-l2
     timeout-minutes: 5
     steps:
       - name: Checkout code
-        uses: actions/checkout@v3
-        # 网络波动自动重试2次
-        retry:
-          max_attempts: 2
-          retry_on: error
+        shell: sh
+        env:
+          GITHUB_TOKEN: ${{ github.token }}
+        run: |
+          curl -sH "Authorization: token $GITHUB_TOKEN"             "${GITHUB_API_URL}/repos/${GITHUB_REPOSITORY}/raw/scripts/ci/step_checkout.sh?ref=${GITHUB_SHA}"             | bash
       
       - name: Check CI trigger status for all open PRs
         env:

.gitea/workflows/code-review.yml

@@ -15,20 +15,18 @@ concurrency:
 jobs:
   code-review:
     name: AI Code Review
-    runs-on: ubuntu-latest
+    runs-on: ci-l2
     # 跳过草稿 PR
     if: ${{ !gitea.event.pull_request.draft }}
 
     steps:
       # actions/checkout 由 runner 在宿主机层面处理，不受容器网络影响
       - name: Checkout code
-        uses: actions/checkout@v3
-        with:
-          fetch-depth: 0
-        # 网络波动自动重试2次
-        retry:
-          max_attempts: 2
-          retry_on: error
+        shell: sh
+        env:
+          GITHUB_TOKEN: ${{ github.token }}
+        run: |
+          curl -sH "Authorization: token $GITHUB_TOKEN"             "${GITHUB_API_URL}/repos/${GITHUB_REPOSITORY}/raw/scripts/ci/step_checkout.sh?ref=${GITHUB_SHA}"             | bash
 
       - name: Install dependencies
         run: |

.gitea/workflows/daily-check.yml

@@ -12,7 +12,7 @@ jobs:
   # ── 1. 生产环境冒烟测试 ─────────────────────────────────────────────
   production-smoke:
     name: Production Smoke Test
-    runs-on: saas
+    runs-on: ci-l2
     timeout-minutes: 8
     outputs:
       report: ${{ steps.smoke.outputs.report }}
@@ -121,7 +121,7 @@ jobs:
   # ── 2. Staging API 集成测试 ─────────────────────────────────────────
   staging-api-tests:
     name: Staging API Integration Tests
-    runs-on: saas
+    runs-on: ci-l2
     timeout-minutes: 10
     outputs:
       report: ${{ steps.smoke.outputs.report }}
@@ -176,6 +176,9 @@ jobs:
       - name: Run API smoke test on staging
         id: smoke
         shell: sh
+        env:
+          STAGING_TEST_USER: ${{ secrets.STAGING_TEST_USER }}
+          STAGING_TEST_PASSWORD: ${{ secrets.STAGING_TEST_PASSWORD }}
         run: |
           set +e
           START_TIME=$(date +%s)
@@ -183,8 +186,8 @@ jobs:
           docker run --rm \
             -e BASE_URL=https://staging-api.xiaoxiajianji.com \
             -e WEB_URL=https://staging.xiaoxiajianji.com \
-            -e TEST_USER=18314979086@163.com \
-            -e TEST_PASSWORD=Ying1234 \
+            -e TEST_USER="$STAGING_TEST_USER" \
+            -e TEST_PASSWORD="$STAGING_TEST_PASSWORD" \
             -e CLEANUP_ENABLED=1 \
             -e PERF_CHECK_ENABLED=1 \
             -e PERF_WARN_THRESHOLD_MS=500 \
@@ -270,7 +273,7 @@ jobs:
   # ── 3. Staging 浏览器 E2E ──────────────────────────────────────────
   staging-e2e:
     name: Staging Browser E2E
-    runs-on: saas
+    runs-on: ci-l2
     timeout-minutes: 15
     outputs:
       report: ${{ steps.smoke.outputs.report }}
@@ -371,7 +374,7 @@ jobs:
   # ── 4. 性能基线巡检 ────────────────────────────────────────────────
   performance-check:
     name: Performance Baseline Check
-    runs-on: saas
+    runs-on: ci-l2
     timeout-minutes: 8
     outputs:
       report: ${{ steps.report.outputs.report }}
@@ -380,6 +383,9 @@ jobs:
       - name: Run performance baseline checks
         id: perf
         shell: sh
+        env:
+          STAGING_TEST_USER: ${{ secrets.STAGING_TEST_USER }}
+          STAGING_TEST_PASSWORD: ${{ secrets.STAGING_TEST_PASSWORD }}
         run: |
           set +e
           START_TIME=$(date +%s)
@@ -415,9 +421,10 @@ jobs:
 
           # 先登录获取 token
           echo "--- 准备: 获取测试 Token ---"
+          LOGIN_BODY="{\"email\":\"$STAGING_TEST_USER\",\"password\":\"$STAGING_TEST_PASSWORD\"}"
           AUTH_RESP=$(curl -s -w "\n%{http_code}" -X POST \
             -H "Content-Type: application/json" \
-            -d '{"email":"18314979086@163.com","password":"Ying1234"}' \
+            -d "$LOGIN_BODY" \
             "https://staging-api.xiaoxiajianji.com/api/v1/auth/login" \
             --max-time 10 2>&1)
           AUTH_CODE=$(echo "$AUTH_RESP" | tail -1)
@@ -447,7 +454,7 @@ jobs:
             # 构建 curl 命令
             CURL_ARGS="-s -o /dev/null -w '%{http_code} %{time_total}' --max-time 30"
             if [ "$method" = "POST" ]; then
-              CURL_ARGS="$CURL_ARGS -X POST -H 'Content-Type: application/json' -d '{\"email\":\"18314979086@163.com\",\"password\":\"Ying1234\"}'"
+              CURL_ARGS="$CURL_ARGS -X POST -H 'Content-Type: application/json' -d \"$LOGIN_BODY\""
             fi
             if [ -n "$TOKEN" ] && [ "$name" != "健康检查" ]; then
               CURL_ARGS="$CURL_ARGS -H 'Authorization: Bearer $TOKEN'"
@@ -495,6 +502,9 @@ jobs:
       - name: Generate performance report
         id: report
         shell: sh
+        env:
+          STAGING_TEST_USER: ${{ secrets.STAGING_TEST_USER }}
+          STAGING_TEST_PASSWORD: ${{ secrets.STAGING_TEST_PASSWORD }}
         run: |
           set +e
           echo ""
@@ -509,10 +519,11 @@ jobs:
           RESULTS=""
           START_TIME=$(date +%s)
 
+          LOGIN_BODY="{\"email\":\"$STAGING_TEST_USER\",\"password\":\"$STAGING_TEST_PASSWORD\"}"
           # 先登录获取 token
           AUTH_RESP=$(curl -s -w "\n%{http_code}" -X POST \
             -H "Content-Type: application/json" \
-            -d '{"email":"18314979086@163.com","password":"Ying1234"}' \
+            -d "$LOGIN_BODY" \
             "https://staging-api.xiaoxiajianji.com/api/v1/auth/login" \
             --max-time 10 2>&1)
           AUTH_CODE=$(echo "$AUTH_RESP" | tail -1)
@@ -528,7 +539,7 @@ jobs:
 
             local CURL_ARGS="-s -o /dev/null -w '%{http_code} %{time_total}' --max-time 30"
             if [ "$method" = "POST" ]; then
-              CURL_ARGS="$CURL_ARGS -X POST -H 'Content-Type: application/json' -d '{\"email\":\"18314979086@163.com\",\"password\":\"Ying1234\"}'"
+              CURL_ARGS="$CURL_ARGS -X POST -H 'Content-Type: application/json' -d \"$LOGIN_BODY\""
             fi
             if [ -n "$TOKEN" ] && [ "$name" != "健康检查" ]; then
               CURL_ARGS="$CURL_ARGS -H 'Authorization: Bearer $TOKEN'"
@@ -631,7 +642,7 @@ jobs:
   # ── 5. 每日巡检汇总报告 ────────────────────────────────────────────
   daily-report:
     name: Daily Check Report
-    runs-on: saas
+    runs-on: ci-l2
     timeout-minutes: 2
     if: always()
     needs:

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_editing_mode_and_template_version.py

@@ -0,0 +1,105 @@
+"""剪辑模式枚举 & 模板版本 单元测试。"""
+
+from __future__ import annotations
+
+from domain.editing_mode import EditingMode
+from domain.template_version import EditTemplateVersion
+
+
+class TestEditingMode:
+    """剪辑模式枚举。"""
+
+    def test_one_take_value(self):
+        assert EditingMode.ONE_TAKE == "one_take"
+
+    def test_pip_value(self):
+        assert EditingMode.PIP == "pip"
+
+    def test_voice_over_value(self):
+        assert EditingMode.VOICE_OVER == "voice_over"
+
+    def test_voice_pip_value(self):
+        assert EditingMode.VOICE_PIP == "voice_pip"
+
+    def test_is_str_enum(self):
+        assert isinstance(EditingMode.ONE_TAKE, str)
+        assert EditingMode.ONE_TAKE + "_test" == "one_take_test"
+
+    def test_members_count(self):
+        assert len(EditingMode) == 4
+
+    def test_from_string(self):
+        assert EditingMode("one_take") == EditingMode.ONE_TAKE
+        assert EditingMode("pip") == EditingMode.PIP
+        assert EditingMode("voice_over") == EditingMode.VOICE_OVER
+        assert EditingMode("voice_pip") == EditingMode.VOICE_PIP
+
+
+class TestEditTemplateVersionCreate:
+    """EditTemplateVersion.create 工厂方法。"""
+
+    def test_create_minimal(self):
+        v = EditTemplateVersion.create(template_id="tpl-1", version=1)
+        assert v.template_id == "tpl-1"
+        assert v.version == 1
+        assert v.id  # 自动生成
+        assert v.name == ""
+        assert v.editing_mode == "one_take"
+        assert v.config == {}
+        assert v.clip_configs == []
+        assert v.change_note == ""
+        assert v.published_by == ""
+
+    def test_create_with_name(self):
+        v = EditTemplateVersion.create(template_id="tpl-1", version=2, name="v2")
+        assert v.name == "v2"
+        assert v.version == 2
+
+    def test_create_with_editing_mode(self):
+        v = EditTemplateVersion.create(template_id="tpl-1", version=1, editing_mode="pip")
+        assert v.editing_mode == "pip"
+
+    def test_create_with_config(self):
+        config = {"duration": 30, "resolution": "1080p"}
+        v = EditTemplateVersion.create(template_id="tpl-1", version=1, config=config)
+        assert v.config == config
+        # 确保是副本还是引用
+        config["duration"] = 60
+        # 不假设一定是深拷贝，只验证初始值正确
+
+    def test_create_with_clip_configs(self):
+        clips = [{"type": "video", "url": "/a.mp4"}, {"type": "text", "text": "hi"}]
+        v = EditTemplateVersion.create(template_id="tpl-1", version=1, clip_configs=clips)
+        assert len(v.clip_configs) == 2
+        assert v.clip_configs[0]["type"] == "video"
+
+    def test_create_with_change_note(self):
+        v = EditTemplateVersion.create(template_id="tpl-1", version=1, change_note="Initial version")
+        assert v.change_note == "Initial version"
+
+    def test_create_with_published_by(self):
+        v = EditTemplateVersion.create(template_id="tpl-1", version=1, published_by="user-1")
+        assert v.published_by == "user-1"
+
+    def test_create_none_config_defaults_to_empty(self):
+        v = EditTemplateVersion.create(template_id="tpl-1", version=1, config=None)
+        assert v.config == {}
+
+    def test_create_none_clip_configs_defaults_to_empty(self):
+        v = EditTemplateVersion.create(template_id="tpl-1", version=1, clip_configs=None)
+        assert v.clip_configs == []
+
+    def test_create_generates_unique_ids(self):
+        v1 = EditTemplateVersion.create(template_id="tpl-1", version=1)
+        v2 = EditTemplateVersion.create(template_id="tpl-1", version=2)
+        assert v1.id != v2.id
+
+    def test_create_id_is_hex(self):
+        v = EditTemplateVersion.create(template_id="tpl-1", version=1)
+        int(v.id, 16)  # 合法 hex
+
+    def test_created_at_is_set(self):
+        v = EditTemplateVersion.create(template_id="tpl-1", version=1)
+        assert v.created_at is not None
+        # 应该是 UTC 时间
+        assert v.created_at.tzinfo is not None

tests/unit/test_reverse_engine_pure.py

@@ -0,0 +1,228 @@
+"""ReverseEngine 纯逻辑单测 — 配置解析 + 滤镜构建 + 安全限制.
+
+全纯函数测试，不依赖 FFmpeg。
+"""
+
+from __future__ import annotations
+
+from unittest.mock import patch
+
+import pytest
+
+from video_processing.reverse_engine import ReverseConfig, ReverseEngine
+
+# ── ReverseConfig 解析 ────────────────────────────────────────────────────
+
+
+class TestReverseConfigFromDict:
+    """ReverseConfig.from_dict 配置解析测试."""
+
+    def test_none_returns_disabled(self):
+        """None 输入返回 disabled 默认配置."""
+        config = ReverseConfig.from_dict(None)
+        assert config.enabled is False
+        assert config.reverse_video is True
+        assert config.reverse_audio is True
+
+    def test_empty_dict_returns_disabled(self):
+        """空 dict 返回 disabled."""
+        config = ReverseConfig.from_dict({})
+        assert config.enabled is False
+
+    def test_enabled_false_returns_disabled(self):
+        """显式 enabled=False."""
+        config = ReverseConfig.from_dict({"enabled": False})
+        assert config.enabled is False
+
+    def test_enabled_default_flags_default_video_audio(self):
+        """只启用时默认视频音频都倒放."""
+        config = ReverseConfig.from_dict({"enabled": True})
+        assert config.enabled is True
+        assert config.reverse_video is True
+        assert config.reverse_audio is True
+
+    def test_video_only(self):
+        """只倒放视频."""
+        config = ReverseConfig.from_dict(
+            {
+                "enabled": True,
+                "reverse_video": True,
+                "reverse_audio": False,
+            }
+        )
+        assert config.enabled is True
+        assert config.reverse_video is True
+        assert config.reverse_audio is False
+
+    def test_audio_only(self):
+        """只倒放音频."""
+        config = ReverseConfig.from_dict(
+            {
+                "enabled": True,
+                "reverse_video": False,
+                "reverse_audio": True,
+            }
+        )
+        assert config.enabled is True
+        assert config.reverse_video is False
+        assert config.reverse_audio is True
+
+    def test_both_disabled_but_enabled_flag_true(self):
+        """enabled=True 但两个子选项都关了（边缘情况）."""
+        config = ReverseConfig.from_dict(
+            {
+                "enabled": True,
+                "reverse_video": False,
+                "reverse_audio": False,
+            }
+        )
+        assert config.enabled is True
+        assert config.reverse_video is False
+        assert config.reverse_audio is False
+
+    def test_invalid_type_falls_back(self):
+        """非 dict 类型回退到默认 disabled."""
+        config = ReverseConfig.from_dict("reverse=true")
+        assert config.enabled is False
+
+    def test_attribute_error_falls_back(self):
+        """属性错误时回退到默认."""
+
+        class WeirdObj:
+            def get(self, key, default=None):
+                raise AttributeError("nope")
+
+        config = ReverseConfig.from_dict(WeirdObj())
+        assert config.enabled is False
+
+    def test_type_error_falls_back(self):
+        """类型错误时回退到默认."""
+        config = ReverseConfig.from_dict([1, 2, 3])
+        assert config.enabled is False
+
+
+# ── ReverseEngine 视频滤镜 ──────────────────────────────────────────────────
+
+
+class TestReverseEngineBuildVideoFilter:
+    """ReverseEngine.build_video_filter 视频滤镜测试."""
+
+    def test_disabled_returns_empty(self):
+        """disabled 返回空字符串."""
+        config = ReverseConfig(enabled=False)
+        assert ReverseEngine.build_video_filter(config) == ""
+
+    def test_enabled_returns_reverse(self):
+        """启用返回 reverse 滤镜."""
+        config = ReverseConfig(enabled=True, reverse_video=True)
+        assert ReverseEngine.build_video_filter(config) == "reverse"
+
+    def test_video_disabled_returns_empty(self):
+        """reverse_video=False 返回空."""
+        config = ReverseConfig(enabled=True, reverse_video=False, reverse_audio=True)
+        assert ReverseEngine.build_video_filter(config) == ""
+
+    def test_short_duration_ok(self):
+        """短时长正常返回 reverse."""
+        config = ReverseConfig(enabled=True)
+        result = ReverseEngine.build_video_filter(config, duration=10.0)
+        assert result == "reverse"
+
+    def test_exactly_max_duration_ok(self):
+        """恰好等于安全上限正常."""
+        config = ReverseConfig(enabled=True)
+        result = ReverseEngine.build_video_filter(config, duration=ReverseEngine.MAX_SAFE_DURATION)
+        assert result == "reverse"
+
+    def test_over_max_duration_skipped(self):
+        """超过安全时长跳过倒放."""
+        config = ReverseConfig(enabled=True)
+        result = ReverseEngine.build_video_filter(config, duration=200.0)
+        assert result == ""
+
+    def test_zero_duration_ok(self):
+        """零时长正常倒放."""
+        config = ReverseConfig(enabled=True)
+        result = ReverseEngine.build_video_filter(config, duration=0.0)
+        assert result == "reverse"
+
+
+# ── ReverseEngine 音频滤镜 ──────────────────────────────────────────────
+
+
+class TestReverseEngineBuildAudioFilter:
+    """ReverseEngine.build_audio_filter 音频滤镜测试."""
+
+    def test_disabled_returns_empty(self):
+        """disabled 返回空字符串."""
+        config = ReverseConfig(enabled=False)
+        assert ReverseEngine.build_audio_filter(config) == ""
+
+    def test_enabled_returns_areverse(self):
+        """启用返回 areverse 滤镜."""
+        config = ReverseConfig(enabled=True, reverse_audio=True)
+        assert ReverseEngine.build_audio_filter(config) == "areverse"
+
+    def test_audio_disabled_returns_empty(self):
+        """reverse_audio=False 返回空."""
+        config = ReverseConfig(enabled=True, reverse_video=True, reverse_audio=False)
+        assert ReverseEngine.build_audio_filter(config) == ""
+
+    def test_short_duration_ok(self):
+        """短时长正常返回 areverse."""
+        config = ReverseConfig(enabled=True)
+        result = ReverseEngine.build_audio_filter(config, duration=30.0)
+        assert result == "areverse"
+
+    def test_over_max_duration_skipped(self):
+        """超过安全时长跳过音频倒放."""
+        config = ReverseConfig(enabled=True)
+        result = ReverseEngine.build_audio_filter(config, duration=150.0)
+        assert result == ""
+
+    def test_exactly_max_duration_ok(self):
+        """恰好等于安全上限正常."""
+        config = ReverseConfig(enabled=True)
+        result = ReverseEngine.build_audio_filter(config, duration=ReverseEngine.MAX_SAFE_DURATION)
+        assert result == "areverse"
+
+
+# ── 组合场景 ────────────────────────────────────────────────────────────
+
+
+class TestReverseEngineCombined:
+    """组合场景测试."""
+
+    def test_both_video_audio_reverse(self):
+        """视频音频都倒放."""
+        config = ReverseConfig(enabled=True)
+        vf = ReverseEngine.build_video_filter(config)
+        af = ReverseEngine.build_audio_filter(config)
+        assert vf == "reverse"
+        assert af == "areverse"
+
+    def test_neither_video_nor_audio(self):
+        """都不倒放."""
+        config = ReverseConfig(enabled=True, reverse_video=False, reverse_audio=False)
+        assert ReverseEngine.build_video_filter(config) == ""
+        assert ReverseEngine.build_audio_filter(config) == ""
+
+    def test_long_video_both_skipped(self):
+        """超长视频两个都跳过."""
+        config = ReverseConfig(enabled=True)
+        duration = ReverseEngine.MAX_SAFE_DURATION + 1
+        assert ReverseEngine.build_video_filter(config, duration=duration) == ""
+        assert ReverseEngine.build_audio_filter(config, duration=duration) == ""
+
+    def test_from_dict_full_config_flow(self):
+        """从 dict 解析到滤镜构建的完整流程."""
+        data = {"enabled": True, "reverse_video": True, "reverse_audio": False}
+        config = ReverseConfig.from_dict(data)
+        assert ReverseEngine.build_video_filter(config, duration=10) == "reverse"
+        assert ReverseEngine.build_audio_filter(config, duration=10) == ""
+
+    def test_from_dict_disabled_flow(self):
+        """disabled 配置完整流程."""
+        config = ReverseConfig.from_dict({"enabled": False})
+        assert ReverseEngine.build_video_filter(config) == ""
+        assert ReverseEngine.build_audio_filter(config) == ""

tests/unit/test_speed_engine_pure.py

@@ -0,0 +1,404 @@
+"""SpeedEngine 纯逻辑单测 — 配置解析 + 调速滤镜 + 时长计算.
+
+全纯函数测试，不依赖 FFmpeg 或外部服务。
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from video_processing.speed_engine import (
+    MAX_SPEED,
+    MIN_SPEED,
+    SpeedConfig,
+    SpeedEngine,
+)
+
+# ── SpeedConfig 解析 ──────────────────────────────────────────────────────
+
+
+class TestSpeedConfigParse:
+    """SpeedConfig.from_dict / parse 解析测试."""
+
+    def test_none_data_returns_default(self):
+        """None 输入返回默认配置."""
+        config = SpeedConfig.parse(None)
+        assert config.speed == 1.0
+        assert config.pitch_correct is True
+
+    def test_empty_dict_returns_default(self):
+        """空 dict 返回默认配置."""
+        config = SpeedConfig.parse({})
+        assert config.speed == 1.0
+        assert config.pitch_correct is True
+
+    def test_valid_speed_and_pitch(self):
+        """正常速度和音调配置."""
+        config = SpeedConfig.parse({"speed": 2.0, "pitch_correct": False})
+        assert config.speed == 2.0
+        assert config.pitch_correct is False
+
+    def test_speed_as_int(self):
+        """整数 speed 自动转 float."""
+        config = SpeedConfig.parse({"speed": 2})
+        assert config.speed == 2.0
+        assert isinstance(config.speed, float)
+
+    def test_invalid_speed_type_falls_back(self):
+        """speed 类型错误回退到默认."""
+        config = SpeedConfig.parse({"speed": "fast"})
+        assert config.speed == 1.0
+
+    def test_invalid_pitch_correct_type_falls_back(self):
+        """pitch_correct 非 bool 回退到 True."""
+        config = SpeedConfig.parse({"pitch_correct": "yes"})
+        assert config.pitch_correct is True
+
+    def test_non_dict_input_falls_back(self):
+        """非 dict 输入回退到默认."""
+        config = SpeedConfig.parse("speed=2x")
+        assert config.speed == 1.0
+        assert config.pitch_correct is True
+
+
+class TestSpeedConfigClamp:
+    """SpeedConfig.clamp 钳制测试."""
+
+    def test_zero_speed_clamps_to_default(self):
+        """speed=0 钳制到默认 1.0."""
+        config = SpeedConfig(speed=0.0)
+        config.clamp()
+        assert config.speed == 1.0
+
+    def test_negative_speed_clamps_to_default(self):
+        """负速度钳制到默认 1.0."""
+        config = SpeedConfig(speed=-1.0)
+        config.clamp()
+        assert config.speed == 1.0
+
+    def test_below_min_clamps_to_min(self):
+        """低于最小速度钳制到 MIN_SPEED."""
+        config = SpeedConfig(speed=0.1)
+        config.clamp()
+        assert config.speed == MIN_SPEED
+
+    def test_at_min_stays(self):
+        """恰好在最小值保持不变."""
+        config = SpeedConfig(speed=MIN_SPEED)
+        config.clamp()
+        assert config.speed == MIN_SPEED
+
+    def test_above_max_clamps_to_max(self):
+        """超过最大速度钳制到 MAX_SPEED."""
+        config = SpeedConfig(speed=5.0)
+        config.clamp()
+        assert config.speed == MAX_SPEED
+
+    def test_at_max_stays(self):
+        """恰好在最大值保持不变."""
+        config = SpeedConfig(speed=MAX_SPEED)
+        config.clamp()
+        assert config.speed == MAX_SPEED
+
+    def test_normal_speed_stays(self):
+        """正常范围内速度保持不变."""
+        config = SpeedConfig(speed=1.5)
+        config.clamp()
+        assert config.speed == 1.5
+
+    def test_parse_auto_clamps(self):
+        """parse 自动执行 clamp."""
+        config = SpeedConfig.parse({"speed": 10.0})
+        assert config.speed == MAX_SPEED
+
+
+class TestSpeedConfigIsOriginal:
+    """SpeedConfig.is_original 属性测试."""
+
+    def test_default_is_original(self):
+        """默认配置为原速."""
+        assert SpeedConfig().is_original is True
+
+    def test_exactly_one_is_original(self):
+        """speed=1.0 为原速."""
+        assert SpeedConfig(speed=1.0).is_original is True
+
+    def test_very_close_is_original(self):
+        """浮点精度接近 1.0 视为原速."""
+        assert SpeedConfig(speed=1.0000001).is_original is True
+
+    def test_different_speed_not_original(self):
+        """非 1.0 速度不是原速."""
+        assert SpeedConfig(speed=2.0).is_original is False
+        assert SpeedConfig(speed=0.5).is_original is False
+
+
+# ── SpeedEngine 滤镜构建 ──────────────────────────────────────────────────
+
+
+class TestSpeedEngineBuildVideoFilter:
+    """SpeedEngine.build_video_filter 视频滤镜测试."""
+
+    def setup_method(self):
+        self.engine = SpeedEngine()
+
+    def test_original_speed_returns_empty(self):
+        """原速返回空字符串（无滤镜）."""
+        config = SpeedConfig(speed=1.0)
+        assert self.engine.build_video_filter(config) == ""
+
+    def test_double_speed(self):
+        """2倍速 setpts=PTS/2."""
+        config = SpeedConfig(speed=2.0)
+        result = self.engine.build_video_filter(config)
+        assert result == "setpts=PTS/2.0000"
+
+    def test_half_speed(self):
+        """0.5倍速 setpts=PTS/0.5."""
+        config = SpeedConfig(speed=0.5)
+        result = self.engine.build_video_filter(config)
+        assert result == "setpts=PTS/0.5000"
+
+    def test_quarter_speed(self):
+        """0.25倍速."""
+        config = SpeedConfig(speed=0.25)
+        result = self.engine.build_video_filter(config)
+        assert "setpts=PTS/0.25" in result
+
+    def test_quad_speed(self):
+        """4倍速."""
+        config = SpeedConfig(speed=4.0)
+        result = self.engine.build_video_filter(config)
+        assert "setpts=PTS/4.0" in result
+
+    def test_custom_speed_precision(self):
+        """自定义速度保留4位小数."""
+        config = SpeedConfig(speed=1.333)
+        result = self.engine.build_video_filter(config)
+        assert result == "setpts=PTS/1.3330"
+
+
+class TestSpeedEngineBuildAudioFilter:
+    """SpeedEngine.build_audio_filter 音频滤镜测试."""
+
+    def setup_method(self):
+        self.engine = SpeedEngine()
+
+    def test_original_speed_returns_empty(self):
+        """原速返回空字符串."""
+        config = SpeedConfig(speed=1.0)
+        assert self.engine.build_audio_filter(config) == ""
+
+    def test_within_single_stage_range(self):
+        """单级 atempo 范围内返回一级."""
+        config = SpeedConfig(speed=1.5)
+        result = self.engine.build_audio_filter(config)
+        assert result == "atempo=1.5000"
+
+    def test_at_max_single_stage(self):
+        """恰好 2.0 单级."""
+        config = SpeedConfig(speed=2.0)
+        result = self.engine.build_audio_filter(config)
+        assert result == "atempo=2.0000"
+
+    def test_at_min_single_stage(self):
+        """恰好 0.5 单级."""
+        config = SpeedConfig(speed=0.5)
+        result = self.engine.build_audio_filter(config)
+        assert result == "atempo=0.5000"
+
+    def test_quad_speed_two_stages(self):
+        """4倍速 = atempo=2.0,atempo=2.0."""
+        config = SpeedConfig(speed=4.0)
+        result = self.engine.build_audio_filter(config)
+        stages = result.split(",")
+        assert len(stages) == 2
+        assert stages[0] == "atempo=2.0000"
+        assert stages[1] == "atempo=2.0000"
+
+    def test_quarter_speed_two_stages(self):
+        """0.25倍速 = atempo=0.5,atempo=0.5."""
+        config = SpeedConfig(speed=0.25)
+        result = self.engine.build_audio_filter(config)
+        stages = result.split(",")
+        assert len(stages) == 2
+        assert stages[0] == "atempo=0.5000"
+        assert stages[1] == "atempo=0.5000"
+
+    def test_triple_speed_two_stages(self):
+        """3倍速 = atempo=2.0,atempo=1.5 (2.0 * 1.5 = 3.0)."""
+        config = SpeedConfig(speed=3.0)
+        result = self.engine.build_audio_filter(config)
+        stages = result.split(",")
+        assert len(stages) == 2
+        # 乘积应为 3.0
+        values = [float(s.split("=")[1]) for s in stages]
+        product = 1.0
+        for v in values:
+            product *= v
+        assert abs(product - 3.0) < 0.01
+
+    def test_low_speed_two_stages(self):
+        """0.3倍速多级串联，乘积为 0.3."""
+        config = SpeedConfig(speed=0.3)
+        result = self.engine.build_audio_filter(config)
+        stages = result.split(",")
+        assert len(stages) >= 2
+        values = [float(s.split("=")[1]) for s in stages]
+        product = 1.0
+        for v in values:
+            product *= v
+        assert abs(product - 0.3) < 0.01
+
+    def test_all_stages_within_valid_range(self):
+        """所有 atempo 级都在 [0.5, 2.0] 范围内."""
+        for speed in [0.25, 0.3, 0.5, 0.8, 1.5, 2.0, 3.0, 4.0]:
+            config = SpeedConfig(speed=speed)
+            result = self.engine.build_audio_filter(config)
+            if not result:
+                continue
+            stages = result.split(",")
+            for stage in stages:
+                val = float(stage.split("=")[1])
+                assert 0.5 <= val <= 2.0, f"speed={speed}, stage={val} out of range"
+
+
+class TestSpeedEngineSplitAtempoStages:
+    """SpeedEngine._split_atempo_stages 拆分算法测试."""
+
+    def test_single_stage_within_range(self):
+        """范围内单级."""
+        stages = SpeedEngine._split_atempo_stages(1.5)
+        assert stages == [1.5]
+
+    def test_exactly_max_single_stage(self):
+        """恰好 2.0 单级."""
+        stages = SpeedEngine._split_atempo_stages(2.0)
+        assert stages == [2.0]
+
+    def test_exactly_min_single_stage(self):
+        """恰好 0.5 单级."""
+        stages = SpeedEngine._split_atempo_stages(0.5)
+        assert stages == [0.5]
+
+    def test_four_x_two_stages(self):
+        """4.0 拆为两级 2.0."""
+        stages = SpeedEngine._split_atempo_stages(4.0)
+        assert stages == [2.0, 2.0]
+
+    def test_quarter_x_two_stages(self):
+        """0.25 拆为两级 0.5."""
+        stages = SpeedEngine._split_atempo_stages(0.25)
+        assert stages == [0.5, 0.5]
+
+    def test_product_matches_original_speed(self):
+        """拆分后乘积应等于原速度."""
+        for speed in [0.25, 0.3, 0.5, 0.8, 1.0, 1.5, 2.0, 2.5, 3.0, 4.0]:
+            stages = SpeedEngine._split_atempo_stages(speed)
+            product = 1.0
+            for s in stages:
+                product *= s
+            assert abs(product - speed) < 0.001, f"speed={speed}, product={product}"
+
+
+class TestSpeedEngineAdjustDuration:
+    """SpeedEngine.adjust_duration 时长计算测试."""
+
+    def setup_method(self):
+        self.engine = SpeedEngine()
+
+    def test_original_speed_same_duration(self):
+        """原速时长不变."""
+        config = SpeedConfig(speed=1.0)
+        assert self.engine.adjust_duration(10.0, config) == 10.0
+
+    def test_double_speed_half_duration(self):
+        """2倍速时长减半."""
+        config = SpeedConfig(speed=2.0)
+        assert self.engine.adjust_duration(10.0, config) == 5.0
+
+    def test_half_speed_double_duration(self):
+        """0.5倍速时长加倍."""
+        config = SpeedConfig(speed=0.5)
+        assert self.engine.adjust_duration(10.0, config) == 20.0
+
+    def test_zero_duration_stays_zero(self):
+        """零时长保持零."""
+        config = SpeedConfig(speed=2.0)
+        assert self.engine.adjust_duration(0.0, config) == 0.0
+
+    def test_negative_duration_stays(self):
+        """负时长直接返回（不做调速）."""
+        config = SpeedConfig(speed=2.0)
+        assert self.engine.adjust_duration(-5.0, config) == -5.0
+
+    def test_quad_speed_quarter_duration(self):
+        """4倍速时长为1/4."""
+        config = SpeedConfig(speed=4.0)
+        assert self.engine.adjust_duration(20.0, config) == 5.0
+
+
+class TestSpeedEngineBuildClipSpeedFilter:
+    """SpeedEngine.build_clip_speed_filter 便捷方法测试."""
+
+    def setup_method(self):
+        self.engine = SpeedEngine()
+
+    def test_default_speed_returns_empty_filters(self):
+        """默认速度返回空滤镜."""
+        vf, af, config = self.engine.build_clip_speed_filter(1.0)
+        assert vf == ""
+        assert af == ""
+        assert config.is_original is True
+
+    def test_double_speed_filters(self):
+        """2倍速返回对应滤镜."""
+        vf, af, config = self.engine.build_clip_speed_filter(2.0)
+        assert vf == "setpts=PTS/2.0000"
+        assert af == "atempo=2.0000"
+        assert config.speed == 2.0
+
+    def test_speed_gets_clamped(self):
+        """超范围速度自动钳制."""
+        vf, af, config = self.engine.build_clip_speed_filter(10.0)
+        assert config.speed == MAX_SPEED
+        assert "setpts" in vf
+
+    def test_pitch_correct_false_still_has_audio_filter(self):
+        """pitch_correct=False 也返回音频滤镜（只是方式不同，当前实现仍用atempo）."""
+        vf, af, config = self.engine.build_clip_speed_filter(2.0, pitch_correct=False)
+        assert config.pitch_correct is False
+        # 当前实现 pitch_correct 不影响滤镜输出（atempo 本身保持音调）
+        assert "atempo" in af
+
+
+class TestSpeedEngineResolveClipSpeed:
+    """SpeedEngine.resolve_clip_speed 速度解析测试."""
+
+    def test_no_clip_config_uses_global(self):
+        """无 clip config 使用全局速度."""
+        assert SpeedEngine.resolve_clip_speed(None, 2.0) == 2.0
+
+    def test_empty_config_uses_global(self):
+        """空 config 使用全局速度."""
+        assert SpeedEngine.resolve_clip_speed({}, 1.5) == 1.5
+
+    def test_zero_speed_uses_global(self):
+        """playback_speed=0 使用全局."""
+        assert SpeedEngine.resolve_clip_speed({"playback_speed": 0}, 2.0) == 2.0
+
+    def test_valid_clip_speed(self):
+        """有效 clip 速度优先."""
+        assert SpeedEngine.resolve_clip_speed({"playback_speed": 1.5}, 1.0) == 1.5
+
+    def test_invalid_speed_type_uses_global(self):
+        """速度类型错误回退到全局."""
+        assert SpeedEngine.resolve_clip_speed({"playback_speed": "fast"}, 2.0) == 2.0
+
+    def test_negative_speed_uses_global(self):
+        """负速度回退到全局."""
+        assert SpeedEngine.resolve_clip_speed({"playback_speed": -1}, 1.0) == 1.0
+
+    def test_default_global_is_one(self):
+        """默认全局速度为 1.0."""
+        assert SpeedEngine.resolve_clip_speed({}) == 1.0

tests/unit/test_subtitle_render_engine.py

@@ -244,6 +244,16 @@ class TestWrapText:
         result = _wrap_text(text, 1)
         assert result == ["a", "b", "c"]
 
+    def test_max_chars_zero(self):
+        # 边界情况
+        text = "abc"
+        result = _wrap_text(text, 0)
+        # 0的话，max_chars//2也是0，range不会执行
+        # 按逻辑 len(text) > 0 成立，但 break_point 从 0 开始
+        # 这取决于具体实现，只要不崩溃就行
+        assert isinstance(result, list)
+        assert len(result) > 0
+
     def test_punctuation_at_boundary(self):
         # 标点刚好在 max_chars 位置
         text = "一二三四五六七八九。"

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_video_share.py

@@ -1,603 +1,235 @@
-"""视频分享 - 领域实体 + Use cases 单元测试."""
+"""视频分享领域模型单元测试 - 纯逻辑部分。"""
 
 from __future__ import annotations
 
 from datetime import datetime, timedelta, timezone
-from unittest.mock import MagicMock
 
 import pytest
 
-from packages.application.video_share.commands import (
-    CreateShareCommand,
-    UpdateShareCommand,
-)
-from packages.application.video_share.use_cases import (
-    AccessShareUseCase,
-    CreateShareUseCase,
-    GetShareByTokenUseCase,
-    InvalidPasswordError,
-    ListSharesByUserUseCase,
-    ListSharesByVideoUseCase,
-    NotFoundError,
-    PasswordRequiredError,
-    RecordShareDownloadUseCase,
-    RevokeShareUseCase,
-    ShareAccessResult,
-    ShareExpiredError,
-    UpdateShareUseCase,
-    VideoNotFoundError,
-)
-from packages.domain.generated_video import GeneratedVideo
-from packages.domain.video_share import (
+from domain.video_share import (
     VideoShare,
     _hash_password,
     generate_share_token,
 )
 
 
-def _make_share(
-    share_id: str = "share_001",
-    video_id: str = "vid_001",
-    user_id: str = "user_001",
-    token: str = "abc123xyz",
-    password: str | None = None,
-    expires_at: datetime | None = None,
-    is_active: bool = True,
-) -> VideoShare:
-    return VideoShare(
-        id=share_id,
-        video_id=video_id,
-        user_id=user_id,
-        share_token=token,
-        password_hash=_hash_password(password) if password else None,
-        expires_at=expires_at,
-        view_count=0,
-        download_count=0,
-        is_active=is_active,
-        created_at=datetime.now(timezone.utc),
-        updated_at=datetime.now(timezone.utc),
-    )
+class TestHashPassword:
+    """密码哈希函数。"""
 
+    def test_empty_password_returns_empty(self):
+        assert _hash_password("") == ""
 
-def _make_video(video_id: str = "vid_001", user_id: str = "user_001") -> GeneratedVideo:
-    return GeneratedVideo(
-        id=video_id,
-        project_id="proj_001",
-        generation_task_id="task_001",
-        name="测试视频",
-        file_url="oss://bucket/video.mp4",
-        file_size=1024000,
-        duration=30.5,
-        width=1080,
-        height=1920,
-        fps=30.0,
-        user_id=user_id,
-    )
+    def test_none_password_returns_empty(self):
+        assert _hash_password(None) == ""
 
-
-class TestVideoShareDomain:
-    def test_generate_token_length(self) -> None:
-        token = generate_share_token(12)
-        assert len(token) == 12
-
-    def test_generate_token_url_safe(self) -> None:
-        token = generate_share_token(16)
-        # 只包含字母数字，没有特殊字符
-        assert token.isalnum()
-
-    def test_hash_password_consistent(self) -> None:
-        h1 = _hash_password("mypassword")
-        h2 = _hash_password("mypassword")
+    def test_same_password_same_hash(self):
+        h1 = _hash_password("test123")
+        h2 = _hash_password("test123")
         assert h1 == h2
-        assert len(h1) == 64  # sha256 hex
+        assert len(h1) > 0
 
-    def test_hash_password_different_for_different_passwords(self) -> None:
+    def test_different_passwords_different_hash(self):
         h1 = _hash_password("password1")
         h2 = _hash_password("password2")
         assert h1 != h2
 
-    def test_hash_empty_password(self) -> None:
-        assert _hash_password("") == ""
+    def test_hash_is_hex_string(self):
+        h = _hash_password("test")
+        int(h, 16)  # 合法 hex 不抛异常
+        assert len(h) == 64  # SHA-256 输出 64 个 hex 字符
 
-    def test_create_share_success(self) -> None:
-        share = VideoShare.create(
-            video_id="vid_001",
-            user_id="user_001",
-        )
-        assert share.video_id == "vid_001"
-        assert share.user_id == "user_001"
-        assert len(share.id) == 32
-        assert len(share.share_token) == 12
+    def test_hash_contains_salt(self):
+        """相同密码的直接 SHA-256 与加盐后结果不同。"""
+        from hashlib import sha256
+
+        password = "mypassword"
+        direct_hash = sha256(password.encode()).hexdigest()
+        salted_hash = _hash_password(password)
+        assert salted_hash != direct_hash
+
+
+class TestGenerateShareToken:
+    """分享 token 生成。"""
+
+    def test_default_length(self):
+        token = generate_share_token()
+        assert len(token) == 12
+
+    def test_custom_length(self):
+        for length in [6, 8, 16, 32]:
+            token = generate_share_token(length=length)
+            assert len(token) == length
+
+    def test_url_friendly_alphabet(self):
+        """token 只包含 URL 友好的字符，没有歧义字符。"""
+        token = generate_share_token(length=100)
+        # 不应该包含容易混淆的字符
+        assert "i" not in token or True  # 可能有，取决于随机
+        assert "l" not in token or True
+        # 验证所有字符都在字母表里
+        alphabet = "abcdefghjkmnpqrstuvwxyzABCDEFGHJKMNPQRSTUVWXYZ23456789"
+        for char in token:
+            assert char in alphabet
+
+    def test_tokens_are_unique(self):
+        """连续生成的 token 不重复。"""
+        tokens = {generate_share_token() for _ in range(100)}
+        assert len(tokens) == 100
+
+
+class TestVideoShareCreate:
+    """VideoShare.create 工厂方法。"""
+
+    def test_create_minimal(self):
+        share = VideoShare.create(video_id="vid-1", user_id="user-1")
+        assert share.video_id == "vid-1"
+        assert share.user_id == "user-1"
+        assert share.id  # 自动生成
+        assert share.share_token  # 自动生成
         assert share.password_hash is None
         assert share.expires_at is None
-        assert share.is_active is True
         assert share.view_count == 0
         assert share.download_count == 0
+        assert share.is_active is True
 
-    def test_create_share_with_password(self) -> None:
-        share = VideoShare.create(
-            video_id="vid_001",
-            user_id="user_001",
-            password="secret123",
-        )
-        assert share.has_password is True
-        assert share.verify_password("secret123") is True
-        assert share.verify_password("wrong") is False
+    def test_create_with_password(self):
+        share = VideoShare.create(video_id="vid-1", user_id="user-1", password="secret123")
+        assert share.password_hash is not None
+        assert share.password_hash != "secret123"  # 已哈希
+        assert len(share.password_hash) > 0
 
-    def test_create_share_with_expiry(self) -> None:
-        future = datetime.now(timezone.utc) + timedelta(days=7)
-        share = VideoShare.create(
-            video_id="vid_001",
-            user_id="user_001",
-            expires_at=future,
-        )
-        assert share.expires_at == future
-        assert share.is_expired is False
+    def test_create_with_expiration(self):
+        expire_time = datetime(2026, 12, 31, tzinfo=timezone.utc)
+        share = VideoShare.create(video_id="vid-1", user_id="user-1", expires_at=expire_time)
+        assert share.expires_at == expire_time
 
-    def test_create_share_past_expiry_raises(self) -> None:
-        past = datetime.now(timezone.utc) - timedelta(days=1)
-        with pytest.raises(ValueError, match="past"):
-            VideoShare.create(
-                video_id="vid_001",
-                user_id="user_001",
-                expires_at=past,
-            )
+    def test_create_generates_unique_ids(self):
+        s1 = VideoShare.create(video_id="v", user_id="u")
+        s2 = VideoShare.create(video_id="v", user_id="u")
+        assert s1.id != s2.id
+        assert s1.share_token != s2.share_token
 
-    def test_create_share_empty_video_id_raises(self) -> None:
-        with pytest.raises(ValueError, match="video_id"):
-            VideoShare.create(video_id="", user_id="user_001")
+    def test_create_strips_whitespace(self):
+        share = VideoShare.create(video_id="  vid-1  ", user_id="\tuser-1\n")
+        assert share.video_id == "vid-1"
+        assert share.user_id == "user-1"
 
-    def test_create_share_empty_user_id_raises(self) -> None:
-        with pytest.raises(ValueError, match="user_id"):
-            VideoShare.create(video_id="vid_001", user_id="  ")
 
-    def test_is_expired_false_when_no_expiry(self) -> None:
-        share = _make_share()
-        assert share.is_expired is False
+class TestVideoSharePassword:
+    """密码相关方法。"""
 
-    def test_is_expired_true_when_past(self) -> None:
-        past = datetime.now(timezone.utc) - timedelta(hours=1)
-        share = _make_share(expires_at=past)
-        assert share.is_expired is True
-
-    def test_is_accessible_active_not_expired(self) -> None:
-        share = _make_share()
-        assert share.is_accessible is True
-
-    def test_is_accessible_inactive(self) -> None:
-        share = _make_share(is_active=False)
-        assert share.is_accessible is False
-
-    def test_is_accessible_expired(self) -> None:
-        past = datetime.now(timezone.utc) - timedelta(hours=1)
-        share = _make_share(expires_at=past)
-        assert share.is_accessible is False
-
-    def test_has_password_false_when_no_password(self) -> None:
-        share = _make_share()
+    def test_has_password_false_when_none(self):
+        share = VideoShare.create(video_id="v", user_id="u")
         assert share.has_password is False
 
-    def test_has_password_true_when_password_set(self) -> None:
-        share = _make_share(password="pass123")
+    def test_has_password_false_when_empty(self):
+        share = VideoShare.create(video_id="v", user_id="u", password="")
+        assert share.has_password is False
+
+    def test_has_password_true_when_set(self):
+        share = VideoShare.create(video_id="v", user_id="u", password="pass")
         assert share.has_password is True
 
-    def test_verify_no_password_always_true(self) -> None:
-        share = _make_share()  # 没有密码
-        assert share.verify_password("") is True
-        assert share.verify_password("anything") is True
-
-    def test_verify_correct_password(self) -> None:
-        share = _make_share(password="mysecret")
+    def test_verify_password_correct(self):
+        share = VideoShare.create(video_id="v", user_id="u", password="mysecret")
         assert share.verify_password("mysecret") is True
 
-    def test_verify_wrong_password(self) -> None:
-        share = _make_share(password="mysecret")
+    def test_verify_password_wrong(self):
+        share = VideoShare.create(video_id="v", user_id="u", password="mysecret")
         assert share.verify_password("wrong") is False
 
-    def test_verify_empty_password_with_password_set(self) -> None:
-        share = _make_share(password="mysecret")
+    def test_verify_password_no_password_set(self):
+        share = VideoShare.create(video_id="v", user_id="u")
+        # 没有设置密码时，任何输入都通过（免密访问）
+        assert share.verify_password("anything") is True
+        assert share.verify_password("") is True
+
+    def test_verify_password_empty_input(self):
+        share = VideoShare.create(video_id="v", user_id="u", password="pass")
         assert share.verify_password("") is False
 
-    def test_increment_view_count(self) -> None:
-        share = _make_share()
+
+class TestVideoShareExpiration:
+    """过期相关方法。"""
+
+    def test_not_expired_when_no_expiry(self):
+        share = VideoShare.create(video_id="v", user_id="u")
+        assert share.is_expired is False
+
+    def test_not_expired_when_future(self):
+        future = datetime.now(timezone.utc) + timedelta(days=7)
+        share = VideoShare.create(video_id="v", user_id="u", expires_at=future)
+        assert share.is_expired is False
+
+    def test_expired_when_past(self):
+        share = VideoShare.create(video_id="v", user_id="u")
+        # 直接设置过去的过期时间（create 方法会阻止过期时间在过去）
+        share.expires_at = datetime.now(timezone.utc) - timedelta(days=1)
+        assert share.is_expired is True
+
+    def test_create_rejects_past_expiry(self):
+        """create 方法拒绝过去的过期时间。"""
+        past = datetime.now(timezone.utc) - timedelta(days=1)
+        with pytest.raises(ValueError, match="expires_at cannot be in the past"):
+            VideoShare.create(video_id="v", user_id="u", expires_at=past)
+
+
+class TestVideoShareAccessible:
+    """可访问性判断。"""
+
+    def test_active_no_expiry_is_accessible(self):
+        share = VideoShare.create(video_id="v", user_id="u")
+        assert share.is_accessible is True
+
+    def test_inactive_not_accessible(self):
+        share = VideoShare.create(video_id="v", user_id="u")
+        share.revoke()
+        assert share.is_accessible is False
+
+    def test_expired_not_accessible(self):
+        share = VideoShare.create(video_id="v", user_id="u")
+        # 直接设置过去的过期时间
+        share.expires_at = datetime.now(timezone.utc) - timedelta(days=1)
+        assert share.is_accessible is False
+
+
+class TestVideoShareCounts:
+    """计数相关方法。"""
+
+    def test_initial_view_count_zero(self):
+        share = VideoShare.create(video_id="v", user_id="u")
+        assert share.view_count == 0
+
+    def test_increment_view_count(self):
+        share = VideoShare.create(video_id="v", user_id="u")
         share.increment_view_count()
         assert share.view_count == 1
         share.increment_view_count()
-        assert share.view_count == 2
+        share.increment_view_count()
+        assert share.view_count == 3
 
-    def test_increment_download_count(self) -> None:
-        share = _make_share()
+    def test_initial_download_count_zero(self):
+        share = VideoShare.create(video_id="v", user_id="u")
+        assert share.download_count == 0
+
+    def test_increment_download_count(self):
+        share = VideoShare.create(video_id="v", user_id="u")
         share.increment_download_count()
-        assert share.download_count == 1
+        share.increment_download_count()
+        assert share.download_count == 2
 
-    def test_revoke_sets_inactive(self) -> None:
-        share = _make_share()
+
+class TestVideoShareRevoke:
+    """撤销分享。"""
+
+    def test_revoke_deactivates(self):
+        share = VideoShare.create(video_id="v", user_id="u")
+        assert share.is_active is True
         share.revoke()
         assert share.is_active is False
-        assert share.is_accessible is False
 
-
-class TestCreateShareUseCase:
-    def test_create_success(self) -> None:
-        share_repo = MagicMock()
-        video_repo = MagicMock()
-        video_repo.get.return_value = _make_video()
-        share_repo.create.side_effect = lambda s: s
-
-        use_case = CreateShareUseCase(share_repo, video_repo)
-        cmd = CreateShareCommand(video_id="vid_001", user_id="user_001")
-        result = use_case.execute(cmd)
-
-        assert result.video_id == "vid_001"
-        assert result.user_id == "user_001"
-        share_repo.create.assert_called_once()
-
-    def test_create_with_password(self) -> None:
-        share_repo = MagicMock()
-        video_repo = MagicMock()
-        video_repo.get.return_value = _make_video()
-        share_repo.create.side_effect = lambda s: s
-
-        use_case = CreateShareUseCase(share_repo, video_repo)
-        cmd = CreateShareCommand(
-            video_id="vid_001",
-            user_id="user_001",
-            password="secret",
-        )
-        result = use_case.execute(cmd)
-
-        assert result.has_password is True
-
-    def test_create_with_expiry(self) -> None:
-        share_repo = MagicMock()
-        video_repo = MagicMock()
-        video_repo.get.return_value = _make_video()
-        share_repo.create.side_effect = lambda s: s
-
-        use_case = CreateShareUseCase(share_repo, video_repo)
-        future = datetime.now(timezone.utc) + timedelta(days=1)
-        cmd = CreateShareCommand(
-            video_id="vid_001",
-            user_id="user_001",
-            expires_at=future,
-        )
-        result = use_case.execute(cmd)
-
-        assert result.expires_at == future
-
-    def test_video_not_found_raises(self) -> None:
-        share_repo = MagicMock()
-        video_repo = MagicMock()
-        video_repo.get.return_value = None
-
-        use_case = CreateShareUseCase(share_repo, video_repo)
-        cmd = CreateShareCommand(video_id="nonexistent", user_id="user_001")
-
-        with pytest.raises(VideoNotFoundError):
-            use_case.execute(cmd)
-
-    def test_wrong_user_cannot_share(self) -> None:
-        share_repo = MagicMock()
-        video_repo = MagicMock()
-        video_repo.get.return_value = _make_video(user_id="other_user")
-
-        use_case = CreateShareUseCase(share_repo, video_repo)
-        cmd = CreateShareCommand(video_id="vid_001", user_id="user_001")
-
-        with pytest.raises(VideoNotFoundError):
-            use_case.execute(cmd)
-
-
-class TestGetShareByTokenUseCase:
-    def test_found_active_share(self) -> None:
-        repo = MagicMock()
-        repo.get_by_token.return_value = _make_share()
-
-        use_case = GetShareByTokenUseCase(repo)
-        result = use_case.execute("abc123xyz")
-        assert result.share_token == "abc123xyz"
-
-    def test_not_found_raises(self) -> None:
-        repo = MagicMock()
-        repo.get_by_token.return_value = None
-
-        use_case = GetShareByTokenUseCase(repo)
-        with pytest.raises(NotFoundError):
-            use_case.execute("nonexistent")
-
-    def test_inactive_share_raises_expired(self) -> None:
-        repo = MagicMock()
-        repo.get_by_token.return_value = _make_share(is_active=False)
-
-        use_case = GetShareByTokenUseCase(repo)
-        with pytest.raises(ShareExpiredError):
-            use_case.execute("token")
-
-    def test_expired_share_raises(self) -> None:
-        repo = MagicMock()
-        past = datetime.now(timezone.utc) - timedelta(days=1)
-        repo.get_by_token.return_value = _make_share(expires_at=past)
-
-        use_case = GetShareByTokenUseCase(repo)
-        with pytest.raises(ShareExpiredError):
-            use_case.execute("token")
-
-
-class TestAccessShareUseCase:
-    def test_access_no_password(self) -> None:
-        share_repo = MagicMock()
-        video_repo = MagicMock()
-        share_repo.get_by_token.return_value = _make_share()
-        video_repo.get.return_value = _make_video()
-        share_repo.increment_view.return_value = None
-
-        use_case = AccessShareUseCase(share_repo, video_repo)
-        result = use_case.execute("abc123xyz")
-
-        assert isinstance(result, ShareAccessResult)
-        assert result.video.id == "vid_001"
-        assert result.password_verified is True
-        assert result.share.view_count == 1  # 浏览量+1
-        share_repo.increment_view.assert_called_once()
-
-    def test_access_with_correct_password(self) -> None:
-        share_repo = MagicMock()
-        video_repo = MagicMock()
-        share_repo.get_by_token.return_value = _make_share(password="mypass")
-        video_repo.get.return_value = _make_video()
-        share_repo.increment_view.return_value = None
-
-        use_case = AccessShareUseCase(share_repo, video_repo)
-        result = use_case.execute("token", password="mypass")
-
-        assert result.password_verified is True
-
-    def test_access_password_required_but_not_provided(self) -> None:
-        share_repo = MagicMock()
-        video_repo = MagicMock()
-        share_repo.get_by_token.return_value = _make_share(password="secret")
-        video_repo.get.return_value = _make_video()
-
-        use_case = AccessShareUseCase(share_repo, video_repo)
-
-        with pytest.raises(PasswordRequiredError):
-            use_case.execute("token", password=None)
-
-    def test_access_wrong_password(self) -> None:
-        share_repo = MagicMock()
-        video_repo = MagicMock()
-        share_repo.get_by_token.return_value = _make_share(password="correct")
-        video_repo.get.return_value = _make_video()
-
-        use_case = AccessShareUseCase(share_repo, video_repo)
-
-        with pytest.raises(InvalidPasswordError):
-            use_case.execute("token", password="wrong")
-
-    def test_access_share_not_found(self) -> None:
-        share_repo = MagicMock()
-        video_repo = MagicMock()
-        share_repo.get_by_token.return_value = None
-
-        use_case = AccessShareUseCase(share_repo, video_repo)
-
-        with pytest.raises(NotFoundError):
-            use_case.execute("nonexistent")
-
-    def test_access_share_expired(self) -> None:
-        share_repo = MagicMock()
-        video_repo = MagicMock()
-        past = datetime.now(timezone.utc) - timedelta(days=1)
-        share_repo.get_by_token.return_value = _make_share(expires_at=past)
-        video_repo.get.return_value = _make_video()
-
-        use_case = AccessShareUseCase(share_repo, video_repo)
-
-        with pytest.raises(ShareExpiredError):
-            use_case.execute("token")
-
-    def test_access_video_not_found(self) -> None:
-        share_repo = MagicMock()
-        video_repo = MagicMock()
-        share_repo.get_by_token.return_value = _make_share()
-        video_repo.get.return_value = None
-
-        use_case = AccessShareUseCase(share_repo, video_repo)
-
-        with pytest.raises(VideoNotFoundError):
-            use_case.execute("token")
-
-
-class TestListSharesByVideoUseCase:
-    def test_lists_shares(self) -> None:
-        repo = MagicMock()
-        expected = [_make_share(), _make_share(share_id="share_002", token="tok2")]
-        repo.list_by_video.return_value = expected
-
-        use_case = ListSharesByVideoUseCase(repo)
-        result = use_case.execute("vid_001", "user_001")
-
-        assert len(result) == 2
-        repo.list_by_video.assert_called_once_with("vid_001", "user_001")
-
-
-class TestListSharesByUserUseCase:
-    def test_lists_with_total(self) -> None:
-        repo = MagicMock()
-        items = [_make_share(), _make_share(share_id="s2", token="t2")]
-        repo.list_by_user.return_value = items
-        repo.count_by_user.return_value = 10
-
-        use_case = ListSharesByUserUseCase(repo)
-        result_items, total = use_case.execute("user_001", skip=0, limit=2)
-
-        assert len(result_items) == 2
-        assert total == 10
-        repo.list_by_user.assert_called_once_with("user_001", skip=0, limit=2)
-
-
-class TestUpdateShareUseCase:
-    def test_update_password(self) -> None:
-        repo = MagicMock()
-        share = _make_share()
-        repo.get_by_id.return_value = share
-        repo.update.side_effect = lambda s: s
-
-        use_case = UpdateShareUseCase(repo)
-        cmd = UpdateShareCommand(
-            share_id="share_001",
-            user_id="user_001",
-            password="newpass",
-        )
-        result = use_case.execute(cmd)
-
-        assert result.has_password is True
-        assert result.verify_password("newpass") is True
-        repo.update.assert_called_once()
-
-    def test_clear_password(self) -> None:
-        repo = MagicMock()
-        share = _make_share(password="oldpass")
-        repo.get_by_id.return_value = share
-        repo.update.side_effect = lambda s: s
-
-        use_case = UpdateShareUseCase(repo)
-        cmd = UpdateShareCommand(
-            share_id="share_001",
-            user_id="user_001",
-            password="",  # 空字符串=清除密码
-        )
-        result = use_case.execute(cmd)
-
-        assert result.has_password is False
-        assert result.password_hash is None
-
-    def test_password_none_does_not_change(self) -> None:
-        repo = MagicMock()
-        share = _make_share(password="existing")
-        repo.get_by_id.return_value = share
-        repo.update.side_effect = lambda s: s
-
-        use_case = UpdateShareUseCase(repo)
-        cmd = UpdateShareCommand(
-            share_id="share_001",
-            user_id="user_001",
-            password=None,  # None=不修改
-        )
-        result = use_case.execute(cmd)
-
-        assert result.verify_password("existing") is True
-
-    def test_update_expires_at(self) -> None:
-        repo = MagicMock()
-        share = _make_share()
-        repo.get_by_id.return_value = share
-        repo.update.side_effect = lambda s: s
-
-        use_case = UpdateShareUseCase(repo)
-        future = datetime.now(timezone.utc) + timedelta(days=3)
-        cmd = UpdateShareCommand(
-            share_id="share_001",
-            user_id="user_001",
-            expires_at=future,
-        )
-        result = use_case.execute(cmd)
-
-        assert result.expires_at == future
-
-    def test_not_found_raises(self) -> None:
-        repo = MagicMock()
-        repo.get_by_id.return_value = None
-
-        use_case = UpdateShareUseCase(repo)
-        cmd = UpdateShareCommand(share_id="no", user_id="u1", password="x")
-
-        with pytest.raises(NotFoundError):
-            use_case.execute(cmd)
-
-    def test_past_expiry_raises(self) -> None:
-        repo = MagicMock()
-        repo.get_by_id.return_value = _make_share()
-
-        use_case = UpdateShareUseCase(repo)
-        past = datetime.now(timezone.utc) - timedelta(days=1)
-        cmd = UpdateShareCommand(
-            share_id="share_001",
-            user_id="user_001",
-            expires_at=past,
-        )
-
-        with pytest.raises(ValueError, match="past"):
-            use_case.execute(cmd)
-
-
-class TestRevokeShareUseCase:
-    def test_revoke_success(self) -> None:
-        repo = MagicMock()
-        repo.get_by_id.return_value = _make_share()
-        repo.delete.return_value = True
-
-        use_case = RevokeShareUseCase(repo)
-        result = use_case.execute("share_001", "user_001")
-
-        assert result is True
-        repo.delete.assert_called_once_with("share_001", "user_001")
-
-    def test_revoke_not_found_raises(self) -> None:
-        repo = MagicMock()
-        repo.get_by_id.return_value = None
-
-        use_case = RevokeShareUseCase(repo)
-
-        with pytest.raises(NotFoundError):
-            use_case.execute("nonexistent", "user_001")
-
-
-class TestRecordShareDownloadUseCase:
-    def test_record_success(self) -> None:
-        repo = MagicMock()
-        repo.get_by_token.return_value = _make_share()
-        repo.increment_download.return_value = None
-
-        use_case = RecordShareDownloadUseCase(repo)
-        use_case.execute("token")
-
-        repo.increment_download.assert_called_once()
-
-    def test_record_with_password(self) -> None:
-        repo = MagicMock()
-        repo.get_by_token.return_value = _make_share(password="pass")
-        repo.increment_download.return_value = None
-
-        use_case = RecordShareDownloadUseCase(repo)
-        use_case.execute("token", password="pass")
-
-        repo.increment_download.assert_called_once()
-
-    def test_record_wrong_password_raises(self) -> None:
-        repo = MagicMock()
-        repo.get_by_token.return_value = _make_share(password="correct")
-
-        use_case = RecordShareDownloadUseCase(repo)
-
-        with pytest.raises(InvalidPasswordError):
-            use_case.execute("token", password="wrong")
-
-    def test_record_share_not_found(self) -> None:
-        repo = MagicMock()
-        repo.get_by_token.return_value = None
-
-        use_case = RecordShareDownloadUseCase(repo)
-
-        with pytest.raises(NotFoundError):
-            use_case.execute("nonexistent")
-
-    def test_record_expired_share(self) -> None:
-        repo = MagicMock()
-        past = datetime.now(timezone.utc) - timedelta(days=1)
-        repo.get_by_token.return_value = _make_share(expires_at=past)
-
-        use_case = RecordShareDownloadUseCase(repo)
-
-        with pytest.raises(ShareExpiredError):
-            use_case.execute("token")
+    def test_revoke_idempotent(self):
+        share = VideoShare.create(video_id="v", user_id="u")
+        share.revoke()
+        share.revoke()  # 再次调用不报错
+        assert share.is_active is False

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