xiaoxia-saas/tests/unit/test_compose_video_cover_integration.py

"""测试 compose_video 任务中封面生成集成."""

from __future__ import annotations

from pathlib import Path
from unittest.mock import MagicMock, Mock, patch

import pytest


class TestComposeVideoCoverIntegration:
    """测试视频合成任务中的封面生成集成."""

    @pytest.fixture
    def mock_job_service(self):
        """模拟 JobService."""
        service = MagicMock()
        service.get_job.return_value = MagicMock(
            id="job_123",
            payload={"plan_id": "plan_456"},
        )
        return service

    @pytest.fixture
    def mock_db(self):
        """模拟数据库会话."""
        return MagicMock()

    @pytest.fixture
    def mock_render_result(self):
        """模拟渲染结果."""
        result = MagicMock()
        result.success = True
        result.output_path = Path("/tmp/output/video_123.mp4")
        result.output_url = "https://example.com/video_123.mp4"
        result.duration = 30.0
        result.clip_count = 5
        result.width = 1080
        result.height = 1920
        result.file_size = 1024000
        return result

    @pytest.fixture
    def mock_plan_with_cover_enabled(self):
        """模拟启用封面的 plan."""
        plan = MagicMock()
        plan.id = "plan_456"
        plan.config = {
            "cover_config": {
                "enabled": True,
                "mode": "smart",
            }
        }
        return plan

    @pytest.fixture
    def mock_plan_with_cover_disabled(self):
        """模拟禁用封面的 plan."""
        plan = MagicMock()
        plan.id = "plan_456"
        plan.config = {
            "cover_config": {
                "enabled": False,
            }
        }
        return plan

    @pytest.fixture
    def mock_plan_without_cover_config(self):
        """模拟没有 cover_config 的 plan."""
        plan = MagicMock()
        plan.id = "plan_456"
        plan.config = {}
        return plan

    def test_cover_generation_called_when_enabled(
        self,
        mock_job_service,
        mock_db,
        mock_render_result,
        mock_plan_with_cover_enabled,
    ):
        """测试封面生成在启用时被调用."""
        from worker_app.tasks.compose_video import _compose_with_unified_engine

        # 模拟 RenderAdapter
        with patch("video_processing.render_adapter.RenderAdapter") as MockAdapter:
            adapter_instance = MagicMock()
            adapter_instance.render_plan.return_value = mock_render_result
            adapter_instance.validate_plan.return_value = (True, [], [], 5, 5)
            MockAdapter.return_value = adapter_instance

            # 模拟 EditPlanRepository
            with patch(
                "packages.adapters.sqlalchemy_impl.edit_plan_repository.SQLAlchemyEditPlanRepository"
            ) as MockPlanRepo:
                plan_repo_instance = MagicMock()
                plan_repo_instance.get.return_value = mock_plan_with_cover_enabled
                MockPlanRepo.return_value = plan_repo_instance

                # 模拟 generate_cover_from_plan
                with patch("video_processing.cover_generator.generate_cover_from_plan") as mock_gen_cover:
                    mock_gen_cover.return_value = Path("/tmp/output/cover_plan_456.jpg")

                    # 执行
                    task = MagicMock()
                    result = _compose_with_unified_engine(
                        task, mock_job_service, mock_job_service.get_job(), "plan_456", mock_db
                    )

                    # 验证封面生成被调用
                    mock_gen_cover.assert_called_once()
                    call_args = mock_gen_cover.call_args
                    assert call_args[0][0] == mock_plan_with_cover_enabled  # plan
                    assert call_args[0][1] == mock_render_result.output_path  # video_path
                    assert call_args[0][2] == mock_render_result.output_path.parent  # output_dir

                    # 验证结果包含 cover_url
                    assert "cover_url" in result["result"]
                    assert result["result"]["cover_url"] == "/covers/plan_456.jpg"

    def test_cover_generation_skipped_when_disabled(
        self,
        mock_job_service,
        mock_db,
        mock_render_result,
        mock_plan_with_cover_disabled,
    ):
        """测试封面生成在禁用时被跳过."""
        from worker_app.tasks.compose_video import _compose_with_unified_engine

        with patch("video_processing.render_adapter.RenderAdapter") as MockAdapter:
            adapter_instance = MagicMock()
            adapter_instance.render_plan.return_value = mock_render_result
            adapter_instance.validate_plan.return_value = (True, [], [], 5, 5)
            MockAdapter.return_value = adapter_instance

            with patch(
                "packages.adapters.sqlalchemy_impl.edit_plan_repository.SQLAlchemyEditPlanRepository"
            ) as MockPlanRepo:
                plan_repo_instance = MagicMock()
                plan_repo_instance.get.return_value = mock_plan_with_cover_disabled
                MockPlanRepo.return_value = plan_repo_instance

                with patch("video_processing.cover_generator.generate_cover_from_plan") as mock_gen_cover:
                    task = MagicMock()
                    result = _compose_with_unified_engine(
                        task, mock_job_service, mock_job_service.get_job(), "plan_456", mock_db
                    )

                    # 验证封面生成未被调用
                    mock_gen_cover.assert_not_called()

                    # 验证结果中 cover_url 为 None
                    assert "cover_url" in result["result"]
                    assert result["result"]["cover_url"] is None

    def test_cover_generation_skipped_when_no_config(
        self,
        mock_job_service,
        mock_db,
        mock_render_result,
        mock_plan_without_cover_config,
    ):
        """测试没有 cover_config 时封面生成被跳过."""
        from worker_app.tasks.compose_video import _compose_with_unified_engine

        with patch("video_processing.render_adapter.RenderAdapter") as MockAdapter:
            adapter_instance = MagicMock()
            adapter_instance.render_plan.return_value = mock_render_result
            adapter_instance.validate_plan.return_value = (True, [], [], 5, 5)
            MockAdapter.return_value = adapter_instance

            with patch(
                "packages.adapters.sqlalchemy_impl.edit_plan_repository.SQLAlchemyEditPlanRepository"
            ) as MockPlanRepo:
                plan_repo_instance = MagicMock()
                plan_repo_instance.get.return_value = mock_plan_without_cover_config
                MockPlanRepo.return_value = plan_repo_instance

                with patch("video_processing.cover_generator.generate_cover_from_plan") as mock_gen_cover:
                    task = MagicMock()
                    result = _compose_with_unified_engine(
                        task, mock_job_service, mock_job_service.get_job(), "plan_456", mock_db
                    )

                    # 验证封面生成未被调用
                    mock_gen_cover.assert_not_called()

                    # 验证结果中 cover_url 为 None
                    assert "cover_url" in result["result"]
                    assert result["result"]["cover_url"] is None

    def test_cover_generation_failure_does_not_break_video(
        self,
        mock_job_service,
        mock_db,
        mock_render_result,
        mock_plan_with_cover_enabled,
    ):
        """测试封面生成失败不影响视频合成."""
        from worker_app.tasks.compose_video import _compose_with_unified_engine

        with patch("video_processing.render_adapter.RenderAdapter") as MockAdapter:
            adapter_instance = MagicMock()
            adapter_instance.render_plan.return_value = mock_render_result
            adapter_instance.validate_plan.return_value = (True, [], [], 5, 5)
            MockAdapter.return_value = adapter_instance

            with patch(
                "packages.adapters.sqlalchemy_impl.edit_plan_repository.SQLAlchemyEditPlanRepository"
            ) as MockPlanRepo:
                plan_repo_instance = MagicMock()
                plan_repo_instance.get.return_value = mock_plan_with_cover_enabled
                MockPlanRepo.return_value = plan_repo_instance

                with patch("video_processing.cover_generator.generate_cover_from_plan") as mock_gen_cover:
                    # 模拟封面生成抛出异常
                    mock_gen_cover.side_effect = Exception("FFmpeg failed")

                    task = MagicMock()
                    result = _compose_with_unified_engine(
                        task, mock_job_service, mock_job_service.get_job(), "plan_456", mock_db
                    )

                    # 验证视频合成仍然成功
                    assert result["status"] == "completed"
                    assert "result" in result
                    assert result["result"]["output_url"] == mock_render_result.output_url

                    # 验证结果中 cover_url 为 None
                    assert result["result"]["cover_url"] is None