xiaoxia-saas/tests/unit/test_1280_preview_speedup.py

"""#1280 预览视频生成加速 — 单元测试。

验证点：
1. UnifiedRenderService.is_preview 参数正确传递
2. 预览模式使用 ultrafast preset + crf 28
3. RenderAdapter.render_from_memory 正确传递 is_preview
4. 预览模式跳过 ASR 初始化
5. 预览模式跳过输出校验和缩略图
6. generation.py 并行下载逻辑
"""

from __future__ import annotations

import os
import tempfile
from pathlib import Path
from unittest.mock import MagicMock, patch

import pytest

# ── 1. UnifiedRenderService is_preview 参数 ──


class TestUnifiedRenderServicePreviewFlag:
    """is_preview 参数正确传递和存储。"""

    def test_default_is_preview_false(self):
        from video_processing.unified_render_service import UnifiedRenderService

        svc = UnifiedRenderService(
            plan=MagicMock(id="test"),
            clips=[],
            asset_path_map={},
            work_dir=Path(tempfile.mkdtemp()),
        )
        assert svc.is_preview is False

    def test_is_preview_true(self):
        from video_processing.unified_render_service import UnifiedRenderService

        svc = UnifiedRenderService(
            plan=MagicMock(id="test"),
            clips=[],
            asset_path_map={},
            work_dir=Path(tempfile.mkdtemp()),
            is_preview=True,
        )
        assert svc.is_preview is True

    def test_is_preview_false_explicit(self):
        from video_processing.unified_render_service import UnifiedRenderService

        svc = UnifiedRenderService(
            plan=MagicMock(id="test"),
            clips=[],
            asset_path_map={},
            work_dir=Path(tempfile.mkdtemp()),
            is_preview=False,
        )
        assert svc.is_preview is False


# ── 2. 预览模式 FFmpeg 参数 ──


class TestPreviewFFmpegPreset:
    """预览模式使用 ultrafast preset + crf 28。"""

    def _make_clip(self):
        from video_processing.unified_render_service import ResolvedClip

        return ResolvedClip(
            clip_id="c1",
            asset_id="a1",
            local_path=Path("/tmp/fake.mp4"),
            clip_type="main",
            order=0,
            start_time=0,
            duration=10.0,
            playback_speed=1.0,
            transition_effect="cut",
            transition_duration=0.0,
            config={},
        )

    @patch("video_processing.unified_render_service.run_ffmpeg")
    def test_execute_ffmpeg_preview_uses_ultrafast(self, mock_run):
        from video_processing.unified_render_service import (
            RenderLayer,
            UnifiedRenderService,
        )

        plan = MagicMock()
        plan.id = "test_plan"
        plan.config = {"export": {"resolution": "854x480"}}

        clip = self._make_clip()

        svc = UnifiedRenderService(
            plan=plan,
            clips=[clip],
            asset_path_map={"a1": Path("/tmp/fake.mp4")},
            work_dir=Path(tempfile.mkdtemp()),
            output_width=854,
            output_height=480,
            is_preview=True,
        )

        layers = [RenderLayer(role="main", clips=[clip])]
        filter_complex, input_args = svc._build_filter_complex(layers)
        output_path = Path(tempfile.mkdtemp()) / "out.mp4"
        svc._execute_ffmpeg(filter_complex, input_args, output_path)

        mock_run.assert_called_once()
        cmd = mock_run.call_args[0][0]

        # Check preset is ultrafast
        preset_idx = cmd.index("-preset")
        assert cmd[preset_idx + 1] == "ultrafast", f"Expected ultrafast, got {cmd[preset_idx + 1]}"

        # Check crf is 28
        crf_idx = cmd.index("-crf")
        assert cmd[crf_idx + 1] == "28", f"Expected crf 28, got {cmd[crf_idx + 1]}"

    @patch("video_processing.unified_render_service.run_ffmpeg")
    def test_execute_ffmpeg_normal_uses_medium(self, mock_run):
        from video_processing.unified_render_service import (
            RenderLayer,
            UnifiedRenderService,
        )

        plan = MagicMock()
        plan.id = "test_plan"
        plan.config = {}

        clip = self._make_clip()

        svc = UnifiedRenderService(
            plan=plan,
            clips=[clip],
            asset_path_map={"a1": Path("/tmp/fake.mp4")},
            work_dir=Path(tempfile.mkdtemp()),
            output_width=1280,
            output_height=720,
            is_preview=False,
        )

        layers = [RenderLayer(role="main", clips=[clip])]
        filter_complex, input_args = svc._build_filter_complex(layers)
        output_path = Path(tempfile.mkdtemp()) / "out.mp4"
        svc._execute_ffmpeg(filter_complex, input_args, output_path)

        mock_run.assert_called_once()
        cmd = mock_run.call_args[0][0]

        preset_idx = cmd.index("-preset")
        assert cmd[preset_idx + 1] == "medium"

        crf_idx = cmd.index("-crf")
        assert cmd[crf_idx + 1] == "23"


# ── 3. RenderAdapter passes is_preview ──


class TestRenderAdapterPreviewPassthrough:
    """RenderAdapter 正确传递 is_preview 参数。"""

    def test_render_from_memory_passes_is_preview(self):
        from video_processing.render_adapter import RenderAdapter

        db = MagicMock()
        adapter = RenderAdapter(db)

        plan = MagicMock()
        plan.id = "test_plan"
        plan.config = {"export": {"resolution": "854x480"}}

        clip = MagicMock()
        clip.id = "c1"

        with patch.object(adapter, "_do_render") as mock_do_render:
            mock_do_render.return_value = MagicMock(
                success=True,
                output_path=Path("/tmp/out.mp4"),
                thumbnail_url="",
                duration=5.0,
                file_size=1000,
                width=854,
                height=480,
                output_url="https://oss/test.mp4",
                rendered_clip_ids=["c1"],
                failed_clip_ids=[],
            )

            adapter.render_from_memory(
                plan=plan,
                clips=[clip],
                asset_path_map={"a1": Path("/tmp/fake.mp4")},
                is_preview=True,
            )

            mock_do_render.assert_called_once()
            _, kwargs = mock_do_render.call_args
            assert kwargs.get("is_preview") is True


# ── 4. 预览模式跳过 ASR ──


class TestPreviewSkipsASR:
    """预览模式跳过 ASR 初始化。"""

    def test_render_method_source_has_asr_skip(self):
        """_do_render 在 is_preview=True 时不调用 _get_asr_service。"""
        with open("apps/worker/video_processing/render_adapter.py") as f:
            source = f.read()

        assert (
            "None if (is_preview and not has_voice_id) else self._get_asr_service()" in source
        ), "Should skip ASR initialization in preview mode unless voice_id is provided"


# ── 5. 并行下载逻辑 ──


class TestParallelDownload:
    """generation.py 并行下载素材。"""

    def test_parallel_download_uses_thread_pool(self):
        with open("apps/worker/worker_app/tasks/generation.py") as f:
            source = f.read()

        assert "ThreadPoolExecutor" in source, "Should use ThreadPoolExecutor for parallel downloads"
        assert "as_completed" in source, "Should use as_completed for result collection"

    def test_parallel_download_preserves_order(self):
        with open("apps/worker/worker_app/tasks/generation.py") as f:
            source = f.read()

        assert "sorted(results_map.keys())" in source, "Should sort results by original index"


# ── 6. generation.py _render_video passes is_preview ──


class TestRenderVideoPassesPreview:
    """_render_video 正确传递 is_preview 到 render_from_memory。"""

    def test_render_video_passes_is_preview(self):
        with open("apps/worker/worker_app/tasks/generation.py") as f:
            source = f.read()

        assert "is_preview=is_preview" in source, "Should pass is_preview to render_from_memory"


# ── 7. Preview mode skips thumbnail and validation ──


class TestPreviewSkipsThumbnailAndValidation:
    """预览模式跳过缩略图生成和输出校验。"""

    def test_render_adapter_skips_thumbnail_in_preview(self):
        with open("apps/worker/video_processing/render_adapter.py") as f:
            source = f.read()

        assert "if not is_preview:" in source, "Thumbnail should be conditional on is_preview"

    def test_render_adapter_skips_validation_in_preview(self):
        with open("apps/worker/video_processing/render_adapter.py") as f:
            source = f.read()

        assert "预览模式：跳过输出校验" in source, "Should skip validation in preview mode"


# ── 8. Pass-through rendering uses ultrafast in preview ──


class TestPassThroughPreviewPreset:
    """直通渲染在预览模式也使用 ultrafast。"""

    def test_pass_through_has_preview_preset(self):
        with open("apps/worker/video_processing/unified_render_service.py") as f:
            source = f.read()

        # The pass_through method should also use ultrafast for preview
        # Count occurrences of "ultrafast" - should be at least 2 (execute_ffmpeg + pass_through)
        count = source.count('"ultrafast" if self.is_preview')
        assert count >= 2, f"Expected at least 2 ultrafast preset usages, found {count}"