xiaoxia-saas/tests/unit/test_1280_preview_speedup.py

"""统一渲染管线 — 预览与确认生成使用相同品质参数。

验证点：
1. UnifiedRenderService 不再有 is_preview 参数
2. 所有渲染统一使用 medium preset + CRF 23
3. RenderAdapter 统一执行校验和缩略图生成
4. 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 TestUnifiedRenderServiceNoPreviewParam:
    """UnifiedRenderService 构造函数不再接受 is_preview 参数。"""

    def test_constructor_has_no_is_preview(self):
        import inspect

        from video_processing.unified_render_service import UnifiedRenderService

        sig = inspect.signature(UnifiedRenderService.__init__)
        param_names = list(sig.parameters.keys())
        assert (
            "is_preview" not in param_names
        ), f"is_preview should be removed from UnifiedRenderService.__init__, found params: {param_names}"

    def test_no_is_preview_attribute(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 not hasattr(
            svc, "is_preview"
        ), "UnifiedRenderService should not have is_preview attribute after unification"


# ── 2. FFmpeg 参数统一为 medium + CRF 23 ──


class TestUnifiedFFmpegPreset:
    """所有渲染统一使用 medium preset + CRF 23。"""

    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_uses_medium_crf23(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=1920,
            output_height=1080,
        )

        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 medium (no conditional)
        preset_idx = cmd.index("-preset")
        assert cmd[preset_idx + 1] == "medium", f"Expected medium, got {cmd[preset_idx + 1]}"

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


# ── 3. RenderAdapter 统一执行校验和缩略图 ──


class TestRenderAdapterUnifiedPostProcess:
    """RenderAdapter 不再跳过校验和缩略图。"""

    def test_render_adapter_no_is_preview_param(self):
        import inspect

        from video_processing.render_adapter import RenderAdapter

        # Check render_from_memory signature
        sig = inspect.signature(RenderAdapter.render_from_memory)
        param_names = list(sig.parameters.keys())
        assert (
            "is_preview" not in param_names
        ), f"is_preview should be removed from render_from_memory, found params: {param_names}"

    def test_no_preview_skip_validation_in_source(self):
        """渲染适配器源码中不再包含预览跳过校验的逻辑。"""
        with open("apps/worker/video_processing/render_adapter.py") as f:
            source = f.read()

        assert "预览模式：跳过输出校验" not in source, "Should not skip validation in any mode"
        assert "if not is_preview:" not in source, "Thumbnail should always be generated"


# ── 4. generation.py 不再有 is_preview 覆盖逻辑 ──


class TestWorkerGenerationNoPreviewOverride:
    """Worker generation.py 不再覆盖预览分辨率为 480p。"""

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

        assert 'resolution = "854x480"' not in source, "Should not override resolution to 480p in preview mode"
        assert 'bitrate = "1M"' not in source, "Should not override bitrate to 1M in preview mode"

    def test_parallel_download_still_works(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"


# ── 5. generation_preview.py 不再有 PREVIEW_RESOLUTION ──


class TestPreviewNoLowQualityConstants:
    """预览 API 不再定义低质量常量。"""

    def test_no_preview_resolution_constant(self):
        with open("apps/api/app/api/routes/generation_preview.py") as f:
            source = f.read()

        assert "PREVIEW_RESOLUTION" not in source, "PREVIEW_RESOLUTION constant should be removed"
        assert "_calc_preview_resolution" not in source, "_calc_preview_resolution function should be removed"