xiaoxia-saas/tests/integration/test_four_mode_rendering.py

"""剪辑模式渲染集成测试.

验证剪辑模式（ONE_TAKE / VOICE_OVER）通过
_build_plan_and_clips_from_task + UnifiedRenderService 的完整渲染流程。
注：PIP / VOICE_PIP 已下线，统一映射为 ONE_TAKE。

需要 ffmpeg 可用；CI 无 ffmpeg 时自动跳过。
"""

from __future__ import annotations

import shutil
import subprocess
import tempfile
from pathlib import Path

import pytest
from video_processing.unified_render_service import (
    RenderResult,
    UnifiedRenderService,
    _resolve_layer_role,
)
from worker_app.tasks.generation import _build_plan_and_clips_from_task

pytestmark = pytest.mark.skipif(
    not shutil.which("ffmpeg"),
    reason="ffmpeg not available",
)


# ── 辅助函数 ──────────────────────────────────────────────────────────────────


def _generate_test_video(path: Path, duration: float = 3.0, color: str = "red") -> None:
    """生成一个纯色测试视频。"""
    cmd = [
        "ffmpeg",
        "-y",
        "-f",
        "lavfi",
        "-i",
        f"color=c={color}:s=640x360:d={duration}:r=25",
        "-c:v",
        "libx264",
        "-pix_fmt",
        "yuv420p",
        "-movflags",
        "+faststart",
        str(path),
    ]
    subprocess.run(cmd, check=True, capture_output=True, timeout=30)


def _render_with_mode(
    mode: str,
    num_clips: int = 3,
    duration: float = 2.0,
) -> tuple[RenderResult, Path]:
    """用指定模式生成测试视频并渲染，返回 (result, work_dir)。

    调用方负责清理 work_dir。
    """
    work_dir = Path(tempfile.mkdtemp(prefix="test_4mode_"))

    # 生成测试视频素材
    colors = ["red", "green", "blue", "yellow", "purple"]
    downloaded_paths: list[Path] = []
    for i in range(num_clips):
        p = work_dir / f"test_{i:03d}.mp4"
        _generate_test_video(p, duration=duration, color=colors[i % len(colors)])
        downloaded_paths.append(p)

    # 构建虚拟 plan + clips
    task_id = f"test_task_{mode}"
    plan, clips, asset_path_map = _build_plan_and_clips_from_task(
        task_id=task_id,
        downloaded_paths=downloaded_paths,
        mode=mode,
    )

    # 渲染
    service = UnifiedRenderService(
        plan=plan,
        clips=clips,
        asset_path_map=asset_path_map,
        work_dir=work_dir,
        output_width=640,
        output_height=360,
        output_fps=25,
    )
    result = service.render()
    return result, work_dir


# ── 测试 _build_plan_and_clips_from_task ──────────────────────────────────────


class TestBuildPlanAndClips:
    """测试 4 种模式的虚拟 plan 构建。"""

    def _make_paths(self, n: int) -> list[Path]:
        return [Path(f"/tmp/test_{i}.mp4") for i in range(n)]

    def test_one_take_mode(self):
        paths = self._make_paths(3)
        plan, clips, asset_map = _build_plan_and_clips_from_task("t1", paths, "one_take")

        assert plan.id == "t1"
        assert len(clips) == 3
        assert all(c.clip_type == "main" for c in clips)
        assert len(asset_map) == 3

    def test_pip_mode_maps_to_one_take(self):
        """PIP 已下线，映射为 one_take → 全部 main clips。"""
        paths = self._make_paths(3)
        plan, clips, asset_map = _build_plan_and_clips_from_task("t2", paths, "pip")

        assert len(clips) == 3
        assert all(c.clip_type == "main" for c in clips)

    def test_voice_over_mode(self):
        paths = self._make_paths(3)
        plan, clips, asset_map = _build_plan_and_clips_from_task("t3", paths, "voice_over")

        assert len(clips) == 3
        assert all(c.clip_type == "main" for c in clips)
        assert all(c.config.get("role") == "b_roll" for c in clips)

    def test_voice_pip_mode_maps_to_one_take(self):
        """VOICE_PIP 已下线，映射为 one_take → 全部 main clips。"""
        paths = self._make_paths(4)
        plan, clips, asset_map = _build_plan_and_clips_from_task("t4", paths, "voice_pip")

        assert len(clips) == 4
        assert all(c.clip_type == "main" for c in clips)

    def test_unknown_mode_defaults_to_one_take(self):
        paths = self._make_paths(2)
        plan, clips, asset_map = _build_plan_and_clips_from_task("t5", paths, "unknown_mode")

        assert len(clips) == 2
        assert all(c.clip_type == "main" for c in clips)

    def test_asset_path_map_keys_match_clip_asset_ids(self):
        paths = self._make_paths(3)
        _, clips, asset_map = _build_plan_and_clips_from_task("t6", paths, "one_take")

        clip_asset_ids = {c.asset_id for c in clips}
        map_keys = set(asset_map.keys())
        assert clip_asset_ids == map_keys


# ── 测试图层分组（4 模式） ────────────────────────────────────────────────────


class TestFourModeLayerGrouping:
    """验证 4 种模式的 clip_type 分布经 _resolve_layer_role 后产生正确的图层。"""

    def test_one_take_layers(self):
        """ONE_TAKE: 3 main → 1 main layer。"""
        paths = [Path(f"/tmp/ot_{i}.mp4") for i in range(3)]
        _, clips, _ = _build_plan_and_clips_from_task("ot", paths, "one_take")

        roles = {_resolve_layer_role(c.clip_type, c.config) for c in clips}
        assert roles == {"main"}

    def test_pip_layers_now_one_take(self):
        """PIP 已下线 → one_take: 3 main → 1 main layer。"""
        paths = [Path(f"/tmp/pip_{i}.mp4") for i in range(3)]
        _, clips, _ = _build_plan_and_clips_from_task("pip", paths, "pip")

        roles = {_resolve_layer_role(c.clip_type, c.config) for c in clips}
        assert roles == {"main"}

    def test_voice_over_layers(self):
        """VOICE_OVER: 3 main(b_roll) → broll。"""
        paths = [Path(f"/tmp/vo_{i}.mp4") for i in range(3)]
        _, clips, _ = _build_plan_and_clips_from_task("vo", paths, "voice_over")

        roles = {_resolve_layer_role(c.clip_type, c.config) for c in clips}
        assert roles == {"broll"}

    def test_voice_pip_layers_now_one_take(self):
        """VOICE_PIP 已下线 → one_take: 4 main → main layer。"""
        paths = [Path(f"/tmp/vpip_{i}.mp4") for i in range(4)]
        _, clips, _ = _build_plan_and_clips_from_task("vpip", paths, "voice_pip")

        roles = {_resolve_layer_role(c.clip_type, c.config) for c in clips}
        assert roles == {"main"}


# ── 端到端渲染测试（需要 ffmpeg） ─────────────────────────────────────────────


class TestEndToEndRendering:
    """4 种模式的完整渲染测试，验证输出文件存在且时长合理。"""

    def test_one_take_render(self):
        result, work_dir = _render_with_mode("one_take", num_clips=2, duration=2.0)
        try:
            assert result.output_path.exists()
            assert result.file_size > 0
            assert result.duration > 0
            assert result.width == 640
            assert result.height == 360
        finally:
            shutil.rmtree(work_dir, ignore_errors=True)

    def test_pip_render(self):
        result, work_dir = _render_with_mode("pip", num_clips=2, duration=2.0)
        try:
            assert result.output_path.exists()
            assert result.file_size > 0
            assert result.duration > 0
        finally:
            shutil.rmtree(work_dir, ignore_errors=True)

    def test_voice_over_render(self):
        result, work_dir = _render_with_mode("voice_over", num_clips=2, duration=2.0)
        try:
            assert result.output_path.exists()
            assert result.file_size > 0
            assert result.duration > 0
        finally:
            shutil.rmtree(work_dir, ignore_errors=True)

    def test_voice_pip_render(self):
        result, work_dir = _render_with_mode("voice_pip", num_clips=3, duration=2.0)
        try:
            assert result.output_path.exists()
            assert result.file_size > 0
            assert result.duration > 0
        finally:
            shutil.rmtree(work_dir, ignore_errors=True)