xiaoxia-saas/tests/unit/test_ffmpeg_encoding_optimization.py

"""Tests for FFmpeg encoding optimization (#1758).

验证：
1. 集中编码常量正确定义，支持环境变量覆盖
2. 所有渲染路径（_execute_ffmpeg / _render_pass_through / normalize_video / processor）
   使用统一的编码参数
3. preset 从 medium → fast，确保渲染速度提升
4. threads=0 自动检测 CPU 核心数
"""

from __future__ import annotations

import importlib
import os
import sys
from unittest.mock import MagicMock, patch

import pytest

# ── 常量定义与默认值 ─────────────────────────────────────────────────────────


class TestEncodingConstants:
    """测试集中编码常量的定义与默认值."""

    def test_default_preset_is_fast(self):
        """默认 preset 应为 fast（非 medium），确保速度提升."""
        from shared.ffmpeg_utils import FFMPEG_ENCODE_PRESET

        assert FFMPEG_ENCODE_PRESET == "fast"

    def test_default_crf_is_23(self):
        """默认 CRF 保持 23，画质不变."""
        from shared.ffmpeg_utils import FFMPEG_ENCODE_CRF

        assert FFMPEG_ENCODE_CRF == "23"

    def test_default_threads_is_auto(self):
        """默认线程数为 0（自动检测 CPU 核心数）."""
        from shared.ffmpeg_utils import FFMPEG_ENCODE_THREADS

        assert FFMPEG_ENCODE_THREADS == "0"

    def test_preset_is_not_medium(self):
        """确保 preset 不再是 medium（验证改动生效）."""
        from shared.ffmpeg_utils import FFMPEG_ENCODE_PRESET

        assert FFMPEG_ENCODE_PRESET != "medium"


# ── 环境变量覆盖 ─────────────────────────────────────────────────────────────


class TestEnvironmentVariableOverride:
    """测试环境变量覆盖编码参数."""

    def _reload_module(self):
        """重新加载模块以读取环境变量."""
        import shared.ffmpeg_utils

        return importlib.reload(shared.ffmpeg_utils)

    def test_override_preset_via_env(self, monkeypatch):
        """通过环境变量覆盖 preset."""
        monkeypatch.setenv("FFMPEG_ENCODE_PRESET", "ultrafast")
        mod = self._reload_module()
        assert mod.FFMPEG_ENCODE_PRESET == "ultrafast"
        # 恢复
        monkeypatch.delenv("FFMPEG_ENCODE_PRESET", raising=False)
        self._reload_module()

    def test_override_crf_via_env(self, monkeypatch):
        """通过环境变量覆盖 CRF."""
        monkeypatch.setenv("FFMPEG_ENCODE_CRF", "18")
        mod = self._reload_module()
        assert mod.FFMPEG_ENCODE_CRF == "18"
        monkeypatch.delenv("FFMPEG_ENCODE_CRF", raising=False)
        self._reload_module()

    def test_override_threads_via_env(self, monkeypatch):
        """通过环境变量覆盖线程数."""
        monkeypatch.setenv("FFMPEG_ENCODE_THREADS", "4")
        mod = self._reload_module()
        assert mod.FFMPEG_ENCODE_THREADS == "4"
        monkeypatch.delenv("FFMPEG_ENCODE_THREADS", raising=False)
        self._reload_module()


# ── Worker 层 re-export ──────────────────────────────────────────────────────


class TestWorkerReExport:
    """测试 worker 层 ffmpeg_utils 正确 re-export 编码常量."""

    def test_worker_reexports_preset(self):
        """worker ffmpeg_utils 应 re-export FFMPEG_ENCODE_PRESET."""
        from video_processing.ffmpeg_utils import FFMPEG_ENCODE_PRESET

        assert FFMPEG_ENCODE_PRESET == "fast"

    def test_worker_reexports_crf(self):
        """worker ffmpeg_utils 应 re-export FFMPEG_ENCODE_CRF."""
        from video_processing.ffmpeg_utils import FFMPEG_ENCODE_CRF

        assert FFMPEG_ENCODE_CRF == "23"

    def test_worker_reexports_threads(self):
        """worker ffmpeg_utils 应 re-export FFMPEG_ENCODE_THREADS."""
        from video_processing.ffmpeg_utils import FFMPEG_ENCODE_THREADS

        assert FFMPEG_ENCODE_THREADS == "0"


# ── _execute_ffmpeg 编码参数 ─────────────────────────────────────────────────


class TestExecuteFfmpegEncoding:
    """测试 _execute_ffmpeg 方法使用正确的编码参数."""

    def _get_execute_command(self):
        """提取 _execute_ffmpeg 构建的 FFmpeg 命令."""
        from video_processing.unified_render_service import UnifiedRenderService

        service = UnifiedRenderService.__new__(UnifiedRenderService)
        service.plan = MagicMock()
        service.plan.id = "test-plan"

        captured_cmd = []

        def mock_run_ffmpeg(cmd, **kwargs):
            captured_cmd.extend(cmd)

        with patch(
            "video_processing.unified_render_service.run_ffmpeg",
            side_effect=mock_run_ffmpeg,
        ):
            service._execute_ffmpeg(
                "test_filter_complex",
                ["-i", "input.mp4"],
                __import__("pathlib").Path("/tmp/output.mp4"),
            )

        return captured_cmd

    def test_execute_uses_fast_preset(self):
        """_execute_ffmpeg 应使用 fast preset."""
        cmd = self._get_execute_command()
        idx = cmd.index("-preset")
        assert cmd[idx + 1] == "fast"

    def test_execute_uses_crf_23(self):
        """_execute_ffmpeg 应使用 CRF 23."""
        cmd = self._get_execute_command()
        idx = cmd.index("-crf")
        assert cmd[idx + 1] == "23"

    def test_execute_uses_auto_threads(self):
        """_execute_ffmpeg 应启用多线程."""
        cmd = self._get_execute_command()
        idx = cmd.index("-threads")
        assert cmd[idx + 1] == "0"

    def test_execute_no_medium_preset(self):
        """_execute_ffmpeg 不应再使用 medium preset."""
        cmd = self._get_execute_command()
        assert "medium" not in cmd


# ── _render_pass_through 编码参数 ────────────────────────────────────────────


class TestRenderPassThroughEncoding:
    """测试 _render_pass_through 方法使用正确的编码参数."""

    def test_passthrough_command_contains_fast_preset(self):
        """_render_pass_through 命令应包含 fast preset."""
        # 通过源码检查确认参数已替换
        import inspect

        from video_processing.unified_render_service import UnifiedRenderService

        source = inspect.getsource(UnifiedRenderService._render_pass_through)
        assert "FFMPEG_ENCODE_PRESET" in source
        assert "FFMPEG_ENCODE_THREADS" in source
        assert '"medium"' not in source

    def test_passthrough_command_contains_threads(self):
        """_render_pass_through 命令应包含 threads 参数."""
        import inspect

        from video_processing.unified_render_service import UnifiedRenderService

        source = inspect.getsource(UnifiedRenderService._render_pass_through)
        assert "FFMPEG_ENCODE_THREADS" in source


# ── normalize_video 编码参数 ─────────────────────────────────────────────────


class TestNormalizeVideoEncoding:
    """测试 normalize_video 使用正确的编码参数."""

    def test_normalize_uses_fast_preset(self):
        """normalize_video 应使用 fast preset."""
        import inspect

        from video_processing.ffmpeg_utils import normalize_video

        source = inspect.getsource(normalize_video)
        assert "FFMPEG_ENCODE_PRESET" in source
        assert '"medium"' not in source

    def test_normalize_uses_configurable_crf(self):
        """normalize_video CRF 应使用可配置常量."""
        import inspect

        from video_processing.ffmpeg_utils import normalize_video

        source = inspect.getsource(normalize_video)
        assert "FFMPEG_ENCODE_CRF" in source

    def test_normalize_uses_threads(self):
        """normalize_video 应启用多线程."""
        import inspect

        from video_processing.ffmpeg_utils import normalize_video

        source = inspect.getsource(normalize_video)
        assert "FFMPEG_ENCODE_THREADS" in source


# ── random_edge_crop 编码参数 ────────────────────────────────────────────────


class TestRandomEdgeCropEncoding:
    """测试 random_edge_crop 函数使用统一编码参数."""

    def test_crop_uses_unified_preset(self):
        """random_edge_crop 应使用统一 preset 常量（非硬编码 fast）."""
        import inspect

        from video_processing.ffmpeg_utils import random_edge_crop

        source = inspect.getsource(random_edge_crop)
        assert "FFMPEG_ENCODE_PRESET" in source

    def test_crop_uses_unified_crf(self):
        """random_edge_crop 应使用统一 CRF 常量（非硬编码 18）."""
        import inspect

        from video_processing.ffmpeg_utils import random_edge_crop

        source = inspect.getsource(random_edge_crop)
        assert "FFMPEG_ENCODE_CRF" in source
        assert '"18"' not in source


# ── processor.py 编码参数 ────────────────────────────────────────────────────


class TestProcessorEncoding:
    """测试 VideoProcessor 使用统一编码参数."""

    def test_processor_uses_unified_preset(self):
        """VideoProcessor.concatenate_videos 应使用统一 preset."""
        import inspect

        from video_processing.processor import VideoProcessor

        source = inspect.getsource(VideoProcessor.concatenate_videos)
        assert "FFMPEG_ENCODE_PRESET" in source
        assert 'preset="medium"' not in source

    def test_processor_uses_unified_crf(self):
        """VideoProcessor.concatenate_videos 应使用统一 CRF."""
        import inspect

        from video_processing.processor import VideoProcessor

        source = inspect.getsource(VideoProcessor.concatenate_videos)
        assert "FFMPEG_ENCODE_CRF" in source
        assert "crf=23" not in source


# ── 编码参数一致性验证 ────────────────────────────────────────────────────────


class TestEncodingConsistency:
    """验证所有渲染路径使用同一套编码参数."""

    def test_all_paths_share_same_preset_source(self):
        """所有路径的 preset 均来自 FFMPEG_ENCODE_PRESET."""
        import inspect

        from video_processing.ffmpeg_utils import normalize_video, random_edge_crop
        from video_processing.processor import VideoProcessor
        from video_processing.unified_render_service import UnifiedRenderService

        sources = [
            inspect.getsource(UnifiedRenderService._execute_ffmpeg),
            inspect.getsource(UnifiedRenderService._render_pass_through),
            inspect.getsource(normalize_video),
            inspect.getsource(random_edge_crop),
            inspect.getsource(VideoProcessor.concatenate_videos),
        ]

        for i, src in enumerate(sources):
            assert "FFMPEG_ENCODE_PRESET" in src, f"渲染路径 #{i} 未使用 FFMPEG_ENCODE_PRESET"

    def test_no_hardcoded_medium_preset_anywhere(self):
        """所有渲染代码中不应再有硬编码的 medium preset."""
        import inspect

        from video_processing.ffmpeg_utils import normalize_video
        from video_processing.unified_render_service import UnifiedRenderService

        for fn in [
            UnifiedRenderService._execute_ffmpeg,
            UnifiedRenderService._render_pass_through,
            normalize_video,
        ]:
            src = inspect.getsource(fn)
            assert '"medium"' not in src, f"{fn.__name__} 仍有硬编码 medium preset"

    def test_no_hardcoded_crf_23_anywhere(self):
        """所有渲染代码中不应再有硬编码的 CRF 23."""
        import inspect

        from video_processing.ffmpeg_utils import normalize_video
        from video_processing.unified_render_service import UnifiedRenderService

        for fn in [
            UnifiedRenderService._execute_ffmpeg,
            UnifiedRenderService._render_pass_through,
            normalize_video,
        ]:
            src = inspect.getsource(fn)
            # 检查没有硬编码的 "23" 作为 CRF 值
            # 注意：行中可能有其他 "23"，只检查紧跟 -crf 之后的值
            lines = src.split("\n")
            for j, line in enumerate(lines):
                if '"-crf"' in line or "'-crf'" in line:
                    # 下一个非空行应该是常量引用而非硬编码数字
                    for k in range(j + 1, min(j + 3, len(lines))):
                        stripped = lines[k].strip().strip(",").strip('"').strip("'")
                        if stripped:
                            assert stripped != "23", f"{fn.__name__} 仍有硬编码 CRF 23"
                            break