xiaoxia-saas/tests/integration/test_full_pipeline.py

"""全链路集成测试.

验证 PlanGeneratorService → UnifiedRenderService → 查重 的端到端流程。
"""

from __future__ import annotations

import shutil
import subprocess
import tempfile
from pathlib import Path

import pytest
from video_processing.unified_render_service import (
    UnifiedRenderService,
)
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) -> None:
    """生成一个测试视频。"""
    cmd = [
        "ffmpeg",
        "-y",
        "-f",
        "lavfi",
        "-i",
        f"color=c=blue:s=640x360:d={duration}:r=25",
        "-c:v",
        "libx264",
        "-pix_fmt",
        "yuv420p",
        str(path),
    ]
    subprocess.run(cmd, check=True, capture_output=True, timeout=30)


def _generate_test_audio(path: Path, duration: float = 5.0) -> None:
    """生成一个测试音频文件。"""
    cmd = [
        "ffmpeg",
        "-y",
        "-f",
        "lavfi",
        "-i",
        f"sine=frequency=440:duration={duration}",
        "-c:a",
        "aac",
        "-b:a",
        "128k",
        str(path),
    ]
    subprocess.run(cmd, check=True, capture_output=True, timeout=30)


class TestFullPipeline:
    """验证从虚拟 plan 构建到渲染输出的完整流程。"""

    def test_one_take_pipeline(self):
        """ONE_TAKE 模式完整流程。"""
        with tempfile.TemporaryDirectory() as tmpdir:
            work_dir = Path(tmpdir)

            # 生成测试素材
            paths = []
            for i in range(3):
                p = work_dir / f"clip_{i}.mp4"
                _generate_test_video(p, duration=2.0)
                paths.append(p)

            # 构建虚拟 plan
            plan, clips, asset_map = _build_plan_and_clips_from_task("pipeline_test", paths, "one_take")

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

            assert result.output_path.exists()
            assert result.duration > 0
            assert result.file_size > 0
            assert result.width == 640
            assert result.height == 360

    def test_pipeline_with_audio_mux(self):
        """渲染 + 混音后处理。"""
        with tempfile.TemporaryDirectory() as tmpdir:
            work_dir = Path(tmpdir)

            # 生成测试素材
            video_path = work_dir / "clip_0.mp4"
            _generate_test_video(video_path, duration=3.0)

            # 构建虚拟 plan
            plan, clips, asset_map = _build_plan_and_clips_from_task("audio_test", [video_path], "one_take")

            # 渲染
            service = UnifiedRenderService(
                plan=plan,
                clips=clips,
                asset_path_map=asset_map,
                work_dir=work_dir,
                output_width=640,
                output_height=360,
            )
            render_result = service.render()

            # 混音 - 直接用 ffmpeg（_mux_audio_track 已被清理）
            audio_path = work_dir / "voice.aac"
            _generate_test_audio(audio_path, duration=5.0)

            final_path = work_dir / "final.mp4"
            mux_cmd = [
                "ffmpeg",
                "-y",
                "-i",
                str(render_result.output_path),
                "-i",
                str(audio_path),
                "-c:v",
                "copy",
                "-c:a",
                "aac",
                "-shortest",
                str(final_path),
            ]
            subprocess.run(mux_cmd, check=True, capture_output=True, timeout=30)

            assert final_path.exists()
            assert final_path.stat().st_size > 0

    def test_single_clip_pipeline(self):
        """单 clip 渲染（无转场）。"""
        with tempfile.TemporaryDirectory() as tmpdir:
            work_dir = Path(tmpdir)

            video_path = work_dir / "single.mp4"
            _generate_test_video(video_path, duration=5.0)

            plan, clips, asset_map = _build_plan_and_clips_from_task("single_test", [video_path], "one_take")

            service = UnifiedRenderService(
                plan=plan,
                clips=clips,
                asset_path_map=asset_map,
                work_dir=work_dir,
                output_width=640,
                output_height=360,
            )
            result = service.render()

            assert result.output_path.exists()
            assert result.duration > 0

    def test_dedup_helper_integration(self):
        """验证 dedup_helpers.create_video_record_and_dedup 的导入和签名。"""
        # 只验证函数存在且签名正确（不实际调用，需要数据库）
        import inspect

        from video_processing.dedup_helpers import create_video_record_and_dedup

        sig = inspect.signature(create_video_record_and_dedup)
        params = set(sig.parameters.keys())
        expected = {
            "generation_task_id",
            "project_id",
            "batch_id",
            "file_url",
            "file_size",
            "duration",
            "video_path",
            "mode",
            "session",
            "width",
            "height",
            "fps",
        }
        assert expected.issubset(params), f"Missing params: {expected - params}"