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_fallback_when_no_cover_url(self):
        """当 GenerationTask 没有 cover_url 时，跳过统一管道走 run_generate_cover。"""
        from unittest.mock import MagicMock, patch

        from app.api.routes.generation_cover import GenerateCoverRequest

        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.ai_service.run_generate_cover") as mock_run,
            patch("packages.shared.storage.get_shared_storage_service") as mock_storage_getter,
        ):
            mock_repo = MagicMock()
            mock_repo.get.return_value = mock_task
            mock_repo_cls.return_value = mock_repo

            # Mock storage service
            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

            mock_run.return_value = {"type": "ai_frame", "image_url": "https://fallback.com/cover.jpg"}

            from app.api.routes.generation_cover import generate_cover

            result = 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(),
            )

            # 应该走 run_generate_cover 回退
            mock_run.assert_called_once()
            assert result.cover["image_url"] == "https://fallback.com/cover.jpg"