xiaoxia-saas/tests/unit/test_generation_cover.py

"""Tests for generation cover route — schema validation and import checks."""

from __future__ import annotations

import pytest
from pydantic import ValidationError


def test_generation_cover_router_importable():
    """新路由模块可以正确导入"""
    from app.api.routes.generation_cover import router

    assert router is not None
    # tags 应该是 Generation
    assert "Generation" in router.tags


def test_generation_cover_route_path():
    """路由路径应为 /generate-cover"""
    from app.api.routes.generation_cover import router

    paths = [route.path for route in router.routes]
    assert "/generate-cover" in paths


def test_generation_cover_schemas_importable():
    """Schema 可以从新模块导入"""
    from app.api.routes.generation_cover import GenerateCoverRequest, GenerateCoverResponse

    # 验证请求 schema 默认值
    req = GenerateCoverRequest()
    assert req.asset_ids == []
    assert req.cover_type == "ai_frame"
    assert req.frame_time is None

    # 验证响应 schema
    resp = GenerateCoverResponse(plan_id="p1", cover={"image_url": "http://x"})
    assert resp.plan_id == "p1"
    assert resp.cover["image_url"] == "http://x"


def test_generation_cover_schemas_not_in_templates_editor():
    """旧的 templates_editor/schemas.py 不再包含封面 schema"""
    from app.api.routes.templates_editor import schemas as te_schemas

    assert not hasattr(te_schemas, "GenerateCoverRequest")
    assert not hasattr(te_schemas, "GenerateCoverResponse")


def test_templates_editor_no_cover_router():
    """templates_editor 不再包含 cover_router"""
    from app.api.routes.templates_editor import _sub_routers

    # cover_router 应该已被移除
    for sub in _sub_routers:
        for route in sub.routes:
            assert "generate-cover" not in getattr(route, "path", ""), "templates_editor 不应再有 generate-cover 路由"


def test_api_router_has_generation_cover():
    """api_router 应该包含 /api/v1/generation/generate-cover 路径"""
    from app.api.router import api_router

    all_paths = []
    for route in api_router.routes:
        if hasattr(route, "path"):
            all_paths.append(route.path)
        # 嵌套 router
        if hasattr(route, "routes"):
            for sub_route in route.routes:
                if hasattr(sub_route, "path"):
                    all_paths.append(sub_route.path)

    # 应该能找到 generate-cover 路径
    cover_paths = [p for p in all_paths if "generate-cover" in p]
    assert len(cover_paths) > 0, f"未找到 generate-cover 路由, 所有路径: {all_paths[:20]}"


def test_generation_cover_request_validation():
    """验证请求 schema 的字段约束"""
    from app.api.routes.generation_cover import GenerateCoverRequest

    # frame_time 不允许负数
    with pytest.raises(ValidationError):
        GenerateCoverRequest(frame_time=-1.0)

    # 合法的 frame_time
    req = GenerateCoverRequest(frame_time=5.5)
    assert req.frame_time == 5.5

    # 自定义 cover_type
    req2 = GenerateCoverRequest(cover_type="upload", asset_ids=["a1", "a2"])
    assert req2.cover_type == "upload"
    assert req2.asset_ids == ["a1", "a2"]


class TestUnifiedCoverPipelineEndpoint:
    """测试统一封面管道在 generate_cover endpoint 中的逻辑 (lines 189-215)."""

    def test_cover_url_from_generation_task(self):
        """当 GenerationTask 有 cover_url 时，直接返回该 URL 作为封面。"""
        from unittest.mock import MagicMock, patch

        from app.api.routes.generation_cover import GenerateCoverRequest, GenerateCoverResponse

        # Mock plan with rendered_storage_key (so we skip the 3-step lookup)
        mock_plan = MagicMock()
        mock_plan.config = {
            "generation_task_id": "task-123",
            "rendered_storage_key": "rendered/plan-1/video.mp4",
        }

        # Mock plan_svc
        mock_plan_svc = MagicMock()
        mock_plan_svc.get_plan_or_raise.return_value = mock_plan

        # Mock template_svc
        mock_template_svc = MagicMock()

        # Mock generation task with cover_url
        mock_task = MagicMock()
        mock_task.cover_url = "https://oss.example.com/rendered/plan-1/cover.jpg"

        mock_db = MagicMock()

        body = GenerateCoverRequest(cover_type="ai_frame")

        with (
            patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
            patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize,
        ):
            mock_repo = MagicMock()
            mock_repo.get.return_value = mock_task
            mock_repo_cls.return_value = mock_repo

            # normalize_plan_config should return the config with cover
            mock_normalize.return_value = {
                "cover": {"type": "ai_frame", "image_url": "https://oss.example.com/rendered/plan-1/cover.jpg"}
            }

            from app.api.routes.generation_cover import generate_cover

            result = generate_cover(
                body=body,
                template_id="template-1",
                plan_id="plan-1",
                services=(mock_template_svc, mock_plan_svc),
                db=mock_db,
                current_user=MagicMock(),
            )

            # 验证返回的封面数据来自 GenerationTask.cover_url
            assert result.plan_id == "plan-1"
            assert result.cover["image_url"] == "https://oss.example.com/rendered/plan-1/cover.jpg"
            assert result.cover["type"] == "ai_frame"

    def test_cover_url_all_fallbacks_fail_returns_400(self):
        """当所有步骤都找不到 cover_url 时，返回 400 而非 500。"""
        from unittest.mock import MagicMock, patch

        from app.api.routes.generation_cover import GenerateCoverRequest
        from fastapi import HTTPException

        mock_plan = MagicMock()
        mock_plan.config = {
            "generation_task_id": "task-456",
            "rendered_storage_key": "rendered/plan-2/video.mp4",
        }

        mock_plan_svc = MagicMock()
        mock_plan_svc.get_plan_or_raise.return_value = mock_plan
        mock_template_svc = MagicMock()

        # Task has no cover_url
        mock_task = MagicMock()
        mock_task.cover_url = ""

        mock_db = MagicMock()

        body = GenerateCoverRequest(cover_type="ai_frame")

        with (
            patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
            patch("packages.shared.storage.get_shared_storage_service") as mock_storage_getter,
        ):
            mock_repo = MagicMock()
            mock_repo.get.return_value = mock_task
            # No tasks found by source_edit_plan_id or user+template
            mock_repo.list_by_source_edit_plan.return_value = []
            mock_repo.list_latest_completed_preview.return_value = []
            mock_repo_cls.return_value = mock_repo

            mock_storage_svc = MagicMock()
            mock_storage_svc.get_url.return_value = "https://oss.example.com/rendered/plan-2/video.mp4"
            mock_storage_getter.return_value = mock_storage_svc

            from app.api.routes.generation_cover import generate_cover

            with pytest.raises(HTTPException) as exc_info:
                generate_cover(
                    body=body,
                    template_id="template-2",
                    plan_id="plan-2",
                    services=(mock_template_svc, mock_plan_svc),
                    db=mock_db,
                    current_user=MagicMock(),
                )

            assert exc_info.value.status_code == 400
            assert "封面尚未生成" in exc_info.value.detail

    def test_cover_url_found_via_source_edit_plan(self):
        """步骤B：通过 source_edit_plan_id 找到预览任务的 cover_url。"""
        from unittest.mock import MagicMock, patch

        from app.api.routes.generation_cover import GenerateCoverRequest

        mock_plan = MagicMock()
        # No generation_task_id, so step A is skipped
        mock_plan.config = {
            "rendered_storage_key": "rendered/plan-x/video.mp4",
        }

        mock_plan_svc = MagicMock()
        mock_plan_svc.get_plan_or_raise.return_value = mock_plan
        mock_template_svc = MagicMock()

        # Preview task found by source_edit_plan_id with cover_url
        mock_preview_task = MagicMock()
        mock_preview_task.id = "preview-task-abc"
        mock_preview_task.status = "completed"
        mock_preview_task.cover_url = "https://oss.example.com/rendered/preview/cover.jpg"

        mock_db = MagicMock()

        body = GenerateCoverRequest(cover_type="ai_frame")

        with (
            patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
            patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize,
        ):
            mock_repo = MagicMock()
            mock_repo.list_by_source_edit_plan.return_value = [mock_preview_task]
            mock_repo_cls.return_value = mock_repo

            mock_normalize.return_value = {
                "cover": {"type": "ai_frame", "image_url": "https://oss.example.com/rendered/preview/cover.jpg"}
            }

            from app.api.routes.generation_cover import generate_cover

            result = generate_cover(
                body=body,
                template_id="template-x",
                plan_id="plan-x",
                services=(mock_template_svc, mock_plan_svc),
                db=mock_db,
                current_user=MagicMock(),
            )

            assert result.cover["image_url"] == "https://oss.example.com/rendered/preview/cover.jpg"
            mock_repo.list_by_source_edit_plan.assert_called_once_with("plan-x")

    def test_cover_url_found_via_user_template(self):
        """步骤C：通过 user+template 找到预览任务的 cover_url。"""
        from unittest.mock import MagicMock, patch

        from app.api.routes.generation_cover import GenerateCoverRequest

        mock_plan = MagicMock()
        mock_plan.config = {
            "rendered_storage_key": "rendered/plan-y/video.mp4",
        }

        mock_plan_svc = MagicMock()
        mock_plan_svc.get_plan_or_raise.return_value = mock_plan
        mock_template_svc = MagicMock()

        # Step B finds nothing, step C finds a task
        mock_preview_task = MagicMock()
        mock_preview_task.id = "preview-task-def"
        mock_preview_task.status = "completed"
        mock_preview_task.cover_url = "https://oss.example.com/rendered/user-template-cover.jpg"

        mock_current_user = MagicMock()
        mock_current_user.user.id = "user-123"
        mock_db = MagicMock()

        body = GenerateCoverRequest(cover_type="ai_frame")

        with (
            patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
            patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize,
        ):
            mock_repo = MagicMock()
            mock_repo.list_by_source_edit_plan.return_value = []
            mock_repo.list_latest_completed_preview.return_value = [mock_preview_task]
            mock_repo_cls.return_value = mock_repo

            mock_normalize.return_value = {
                "cover": {"type": "ai_frame", "image_url": "https://oss.example.com/rendered/user-template-cover.jpg"}
            }

            from app.api.routes.generation_cover import generate_cover

            result = generate_cover(
                body=body,
                template_id="template-y",
                plan_id="plan-y",
                services=(mock_template_svc, mock_plan_svc),
                db=mock_db,
                current_user=mock_current_user,
            )

            assert result.cover["image_url"] == "https://oss.example.com/rendered/user-template-cover.jpg"
            mock_repo.list_latest_completed_preview.assert_called_once_with(
                user_id="user-123",
                template_id="template-y",
            )


class TestSourceEditPlanFallback:
    """测试步骤 2.5：通过 source_edit_plan_id 查找预览视频兜底逻辑。"""

    def test_step25_finds_video_by_source_edit_plan_id(self):
        """当步骤1和步骤2都找不到时，步骤2.5通过source_edit_plan_id找到预览视频和封面。"""
        from unittest.mock import MagicMock, patch

        from app.api.routes.generation_cover import GenerateCoverRequest

        # plan.config 没有 rendered_storage_key 和 generation_task_id
        mock_plan = MagicMock()
        mock_plan.config = {}

        mock_plan_svc = MagicMock()
        mock_plan_svc.get_plan_or_raise.return_value = mock_plan
        mock_template_svc = MagicMock()

        # Mock preview task found by source_edit_plan_id — with cover_url
        mock_preview_task = MagicMock()
        mock_preview_task.id = "preview-task-789"
        mock_preview_task.status = "completed"
        mock_preview_task.is_preview = True
        mock_preview_task.cover_url = "https://oss.example.com/rendered/cover.jpg"

        # Mock generated video
        mock_video = MagicMock()
        mock_video.file_url = "rendered/plan-x/video.mp4"

        mock_db = MagicMock()

        body = GenerateCoverRequest(cover_type="ai_frame")

        with (
            patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
            patch("app.api.routes.generation_cover.get_generated_video_repository") as mock_video_repo,
            patch("app.api.routes.generation_cover.ListGeneratedVideosByTaskUseCase") as mock_usecase_cls,
            patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize,
        ):
            mock_repo = MagicMock()
            mock_repo.get.return_value = None
            # Both video lookup (step 2.5) and cover_url lookup (step B) use this
            mock_repo.list_by_source_edit_plan.return_value = [mock_preview_task]
            mock_repo_cls.return_value = mock_repo

            mock_usecase = MagicMock()
            mock_usecase.execute.return_value = [mock_video]
            mock_usecase_cls.return_value = mock_usecase

            mock_normalize.return_value = {
                "cover": {"type": "ai_frame", "image_url": "https://oss.example.com/rendered/cover.jpg"}
            }

            from app.api.routes.generation_cover import generate_cover

            result = generate_cover(
                body=body,
                template_id="template-1",
                plan_id="plan-x",
                services=(mock_template_svc, mock_plan_svc),
                db=mock_db,
                current_user=MagicMock(),
            )

            # Verify step 2.5 was called for video
            mock_repo.list_by_source_edit_plan.assert_called_with("plan-x")
            # Cover was found via unified pipeline step B
            assert result.cover["image_url"] == "https://oss.example.com/rendered/cover.jpg"

    def test_step25_skips_non_completed_or_non_preview_tasks(self):
        """步骤2.5跳过非completed或非is_preview的任务，继续到步骤3。"""
        from unittest.mock import MagicMock, patch

        from app.api.routes.generation_cover import GenerateCoverRequest

        mock_plan = MagicMock()
        mock_plan.config = {}

        mock_plan_svc = MagicMock()
        mock_plan_svc.get_plan_or_raise.return_value = mock_plan
        mock_template_svc = MagicMock()

        # Task that is not completed
        mock_task_failed = MagicMock()
        mock_task_failed.id = "task-failed"
        mock_task_failed.status = "failed"
        mock_task_failed.is_preview = True
        mock_task_failed.cover_url = ""

        # Task that is not preview
        mock_task_full = MagicMock()
        mock_task_full.id = "task-full"
        mock_task_full.status = "completed"
        mock_task_full.is_preview = False
        mock_task_full.cover_url = ""

        mock_db = MagicMock()

        # Step 3 fallback finds a valid preview task WITH cover_url
        mock_step3_task = MagicMock()
        mock_step3_task.id = "step3-task"
        mock_step3_task.status = "completed"
        mock_step3_task.cover_url = "https://oss.example.com/rendered/step3-cover.jpg"

        mock_video = MagicMock()
        mock_video.file_url = "rendered/step3/video.mp4"

        body = GenerateCoverRequest(cover_type="ai_frame")

        with (
            patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
            patch("app.api.routes.generation_cover.get_generated_video_repository") as mock_video_repo,
            patch("app.api.routes.generation_cover.ListGeneratedVideosByTaskUseCase") as mock_usecase_cls,
            patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize,
            patch("packages.shared.storage.get_shared_storage_service") as mock_storage_getter,
        ):
            mock_repo = MagicMock()
            mock_repo.get.return_value = None
            mock_repo.list_by_source_edit_plan.return_value = [mock_task_failed, mock_task_full]
            # Video step 3 and cover step C both use list_latest_completed_preview
            mock_repo.list_latest_completed_preview.return_value = [mock_step3_task]
            mock_repo_cls.return_value = mock_repo

            mock_usecase = MagicMock()
            mock_usecase.execute.return_value = [mock_video]
            mock_usecase_cls.return_value = mock_usecase

            mock_normalize.return_value = {
                "cover": {"type": "ai_frame", "image_url": "https://oss.example.com/rendered/step3-cover.jpg"}
            }

            mock_storage_svc = MagicMock()
            mock_storage_svc.get_url.return_value = "https://oss.example.com/rendered/step3/video.mp4"
            mock_storage_getter.return_value = mock_storage_svc

            from app.api.routes.generation_cover import generate_cover

            result = generate_cover(
                body=body,
                template_id="template-1",
                plan_id="plan-y",
                services=(mock_template_svc, mock_plan_svc),
                db=mock_db,
                current_user=MagicMock(),
            )

            # Step 2.5 found tasks but none matched -> step 3 should be called
            mock_repo.list_by_source_edit_plan.assert_called()
            mock_repo.list_latest_completed_preview.assert_called()
            assert result.cover["image_url"] == "https://oss.example.com/rendered/step3-cover.jpg"

    def test_step25_exception_does_not_block_step3(self):
        """步骤2.5异常时不影响步骤3兜底（视频和封面都通过步骤3找到）。"""
        from unittest.mock import MagicMock, patch

        from app.api.routes.generation_cover import GenerateCoverRequest

        mock_plan = MagicMock()
        mock_plan.config = {}

        mock_plan_svc = MagicMock()
        mock_plan_svc.get_plan_or_raise.return_value = mock_plan
        mock_template_svc = MagicMock()

        mock_step3_task = MagicMock()
        mock_step3_task.id = "step3-task"
        mock_step3_task.status = "completed"
        mock_step3_task.cover_url = "https://oss.example.com/rendered/step3-cover.jpg"

        mock_video = MagicMock()
        mock_video.file_url = "rendered/step3/video.mp4"

        mock_db = MagicMock()

        body = GenerateCoverRequest(cover_type="ai_frame")

        with (
            patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
            patch("app.api.routes.generation_cover.get_generated_video_repository") as mock_video_repo,
            patch("app.api.routes.generation_cover.ListGeneratedVideosByTaskUseCase") as mock_usecase_cls,
            patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize,
            patch("packages.shared.storage.get_shared_storage_service") as mock_storage_getter,
        ):
            mock_repo = MagicMock()
            mock_repo.get.return_value = None
            # Both video step 2.5 and cover step B raise
            mock_repo.list_by_source_edit_plan.side_effect = RuntimeError("db error")
            # Step 3 / step C succeeds
            mock_repo.list_latest_completed_preview.return_value = [mock_step3_task]
            mock_repo_cls.return_value = mock_repo

            mock_usecase = MagicMock()
            mock_usecase.execute.return_value = [mock_video]
            mock_usecase_cls.return_value = mock_usecase

            mock_normalize.return_value = {
                "cover": {"type": "ai_frame", "image_url": "https://oss.example.com/rendered/step3-cover.jpg"}
            }

            mock_storage_svc = MagicMock()
            mock_storage_svc.get_url.return_value = "https://oss.example.com/rendered/step3/video.mp4"
            mock_storage_getter.return_value = mock_storage_svc

            from app.api.routes.generation_cover import generate_cover

            result = generate_cover(
                body=body,
                template_id="template-1",
                plan_id="plan-z",
                services=(mock_template_svc, mock_plan_svc),
                db=mock_db,
                current_user=MagicMock(),
            )

            # Step 3 was called after step 2.5 failed
            mock_repo.list_latest_completed_preview.assert_called()
            assert result.cover["image_url"] == "https://oss.example.com/rendered/step3-cover.jpg"


class TestStrayLoggerRemoved:
    """验证多余的 logger.info(plan_id, generation_task_id) 已被删除。"""

    def test_no_stray_logger_call_in_source(self):
        """源码中不应存在 logger.info(plan_id, generation_task_id) 这样的调用。"""
        import inspect

        from app.api.routes import generation_cover

        source = inspect.getsource(generation_cover)
        # The stray call was logger.info(\n plan_id,\n generation_task_id,\n)
        # with no format string — should not exist
        assert (
            "logger.info(\n                            plan_id," not in source
        ), "Stray logger.info(plan_id, generation_task_id) should be removed"