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("app.api.routes.generation_cover.get_shared_storage_service") as mock_storage_getter,
            patch("packages.shared.mediakit_client.get_mediakit_client") as mock_mk_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

            # MediaKit 抽帧也返回 None，模拟最终失败
            mock_mk = MagicMock()
            mock_mk.is_available = True
            mock_mk.extract_frames.return_value = None
            mock_mk_getter.return_value = mock_mk

            # body 不传 asset_ids，步骤 E2 不会进入
            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

    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",
            )

    def test_cover_url_found_via_cover_candidates_image_url(self):
        """步骤D：plan.config.cover_candidates 有 image_url 时，直接使用第一个候选封面。"""
        from unittest.mock import MagicMock, patch

        from app.api.routes.generation_cover import GenerateCoverRequest

        mock_plan = MagicMock()
        # 步骤A/B/C 都找不到，进入步骤D
        mock_plan.config = {
            "rendered_storage_key": "rendered/plan-z/video.mp4",  # 必须有预览视频才能通过前置检查
            "cover_candidates": [
                {"image_url": "https://oss.example.com/candidates/cover-1.jpg", "score": 0.95},
                {"image_url": "https://oss.example.com/candidates/cover-2.jpg", "score": 0.80},
            ],
        }

        mock_plan_svc = MagicMock()
        mock_plan_svc.get_plan_or_raise.return_value = mock_plan
        mock_template_svc = MagicMock()
        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 = None
            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_normalize.return_value = {
                "cover": {"type": "ai_frame", "image_url": "https://oss.example.com/candidates/cover-1.jpg"}
            }

            from app.api.routes.generation_cover import generate_cover

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

            # 步骤D从 cover_candidates 第一个元素的 image_url 提取封面
            assert result.cover["image_url"] == "https://oss.example.com/candidates/cover-1.jpg"
            # 验证 plan.config 被更新（至少调用一次：rendered_storage_key + cover）
            assert mock_plan_svc.update_plan_config.call_count >= 1

    def test_cover_url_found_via_cover_candidates_url_key(self):
        """步骤D：cover_candidates 用 url 键（非 image_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-w/video.mp4",  # 必须有预览视频才能通过前置检查
            "cover_candidates": [
                {"url": "https://oss.example.com/candidates/alt-cover.jpg"},
            ],
        }

        mock_plan_svc = MagicMock()
        mock_plan_svc.get_plan_or_raise.return_value = mock_plan
        mock_template_svc = MagicMock()
        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 = None
            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_normalize.return_value = {
                "cover": {"type": "ai_frame", "image_url": "https://oss.example.com/candidates/alt-cover.jpg"}
            }

            from app.api.routes.generation_cover import generate_cover

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

            # 步骤D fallback 到 url 键
            assert result.cover["image_url"] == "https://oss.example.com/candidates/alt-cover.jpg"

    def test_cover_candidates_skips_non_dict_first_element(self):
        """步骤D：cover_candidates 第一个元素不是 dict 时，安全跳过不崩溃。"""
        from unittest.mock import MagicMock, patch

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

        mock_plan = MagicMock()
        mock_plan.config = {
            "rendered_storage_key": "rendered/plan-skip/video.mp4",  # 必须有预览视频才能通过前置检查
            "cover_candidates": ["not-a-dict", 42, None],
        }

        mock_plan_svc = MagicMock()
        mock_plan_svc.get_plan_or_raise.return_value = mock_plan
        mock_template_svc = MagicMock()
        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_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_repo.list_latest_completed_preview.return_value = []
            mock_repo_cls.return_value = mock_repo

            # storage fallback 也找不到封面
            mock_storage_svc = MagicMock()
            mock_storage_svc.get_url.return_value = ""
            mock_storage_getter.return_value = mock_storage_svc

            from app.api.routes.generation_cover import generate_cover

            # 所有步骤都失败，应返回 400
            with pytest.raises(HTTPException) as exc_info:
                generate_cover(
                    body=body,
                    template_id="template-skip",
                    plan_id="plan-skip",
                    services=(mock_template_svc, mock_plan_svc),
                    db=mock_db,
                    current_user=MagicMock(),
                )
            assert exc_info.value.status_code == 400


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("app.api.routes.generation_cover.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("app.api.routes.generation_cover.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"


class TestUploadCoverType:
    """测试 cover_type=upload 封面上传功能。"""

    def test_upload_cover_saves_url_directly(self):
        """cover_type=upload 时直接保存 cover_url，不需要预览视频。"""
        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_current_user = MagicMock()
        mock_current_user.user.id = "user-upload"

        body = GenerateCoverRequest(
            cover_type="upload",
            cover_url="https://oss.example.com/uploaded/cover.jpg",
        )

        with patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize:
            mock_normalize.return_value = {
                "cover": {"type": "upload", "image_url": "https://oss.example.com/uploaded/cover.jpg"}
            }

            from app.api.routes.generation_cover import generate_cover

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

            assert result.plan_id == "plan-upload"
            assert result.cover["type"] == "upload"
            assert result.cover["image_url"] == "https://oss.example.com/uploaded/cover.jpg"

            # 验证 plan config 被更新
            mock_plan_svc.update_plan_config.assert_called_once()
            call_args = mock_plan_svc.update_plan_config.call_args
            assert call_args[0][0] == "plan-upload"
            assert call_args[0][1]["cover"]["type"] == "upload"

    def test_upload_cover_without_url_returns_400(self):
        """cover_type=upload 但未提供 cover_url 时返回 400。"""
        from unittest.mock import MagicMock

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

        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()

        body = GenerateCoverRequest(cover_type="upload")
        # cover_url is None by default

        import pytest
        from app.api.routes.generation_cover import generate_cover

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

        assert exc_info.value.status_code == 400
        assert "cover_url" in exc_info.value.detail

    def test_upload_cover_schema_has_cover_url_field(self):
        """GenerateCoverRequest schema 包含 cover_url 字段。"""
        from app.api.routes.generation_cover import GenerateCoverRequest

        req = GenerateCoverRequest(cover_type="upload", cover_url="https://example.com/img.jpg")
        assert req.cover_url == "https://example.com/img.jpg"
        assert req.cover_type == "upload"

        # 默认值为 None
        req2 = GenerateCoverRequest()
        assert req2.cover_url is None

    def test_upload_cover_does_not_require_preview_video(self):
        """cover_type=upload 时不查找预览视频，即使 plan.config 为空也不报错。"""
        from unittest.mock import MagicMock, patch

        from app.api.routes.generation_cover import GenerateCoverRequest

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

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

        body = GenerateCoverRequest(
            cover_type="upload",
            cover_url="https://oss.example.com/uploaded/my-cover.png",
        )

        with patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize:
            mock_normalize.return_value = {
                "cover": {"type": "upload", "image_url": "https://oss.example.com/uploaded/my-cover.png"}
            }

            from app.api.routes.generation_cover import generate_cover

            # 不应该抛出 "请先生成预览视频" 的异常
            result = generate_cover(
                body=body,
                template_id="template-1",
                plan_id="plan-no-preview",
                services=(mock_template_svc, mock_plan_svc),
                db=MagicMock(),
                current_user=MagicMock(),
            )

            assert result.cover["image_url"] == "https://oss.example.com/uploaded/my-cover.png"
            # 验证没有调用任何预览视频查找逻辑
            # (normalize_plan_config 是唯一被调用的外部函数)

    def test_cover_extracted_from_source_asset_when_no_preview(self):
        """步骤E2：无后端渲染产物时，直接从用户选择的视频素材抽帧。"""
        from unittest.mock import MagicMock, patch

        from app.api.routes.generation_cover import GenerateCoverRequest

        mock_plan = MagicMock()
        mock_plan.config = {}  # 无 rendered_storage_key，无 generation_task_id

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

        # 模拟视频素材
        mock_asset = MagicMock()
        mock_asset.file_type = "video"
        mock_asset.storage_key = "uploads/source-clip.mp4"

        mock_asset_repo = MagicMock()
        mock_asset_repo.get.return_value = mock_asset

        mock_mk = MagicMock()
        mock_mk.is_available = True
        mock_mk.extract_frames.return_value = [{"image_url": "https://mediakit.internal/frame-abc.jpg"}]

        mock_storage = MagicMock()
        mock_storage.get_url.return_value = "https://oss.example.com/uploads/source-clip.mp4"

        body = GenerateCoverRequest(
            cover_type="ai_frame",
            asset_ids=["asset-video-1"],
        )

        with (
            patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
            patch(
                "packages.adapters.sqlalchemy_impl.asset_repository.SQLAlchemyAssetRepository",
                return_value=mock_asset_repo,
            ),
            patch("packages.shared.mediakit_client.get_mediakit_client", return_value=mock_mk),
            patch("packages.shared.storage.get_shared_storage_service", return_value=mock_storage),
            patch(
                "app.api.routes.generation_cover._persist_cover_frame",
                return_value="https://oss.example.com/covers/final.jpg",
            ),
            patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize,
        ):
            mock_repo = MagicMock()
            mock_repo.get.return_value = None
            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_normalize.return_value = {
                "cover": {"type": "ai_frame", "image_url": "https://oss.example.com/covers/final.jpg"}
            }

            from app.api.routes.generation_cover import generate_cover

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

            assert result.cover["image_url"] == "https://oss.example.com/covers/final.jpg"
            mock_mk.extract_frames.assert_called_once()
            # 确保用的是源素材 URL
            call_kwargs = mock_mk.extract_frames.call_args.kwargs
            assert "source-clip.mp4" in call_kwargs["video_url"]

    def test_e2_passes_plan_title_to_persist_for_overlay(self):
        """步骤E2：plan.config.title.text 存在时，作为 title_text 传给 _persist_cover_frame 叠加标题。"""
        from unittest.mock import MagicMock, patch

        from app.api.routes.generation_cover import GenerateCoverRequest

        mock_plan = MagicMock()
        mock_plan.config = {"title": {"enabled": True, "text": "我的视频标题"}}

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

        mock_asset = MagicMock()
        mock_asset.file_type = "video"
        mock_asset.storage_key = "uploads/src.mp4"
        mock_asset_repo = MagicMock()
        mock_asset_repo.get.return_value = mock_asset

        mock_mk = MagicMock()
        mock_mk.is_available = True
        mock_mk.extract_frames.return_value = [{"image_url": "https://mk/frame.jpg"}]

        mock_storage = MagicMock()
        mock_storage.get_url.return_value = "https://oss.example.com/uploads/src.mp4"

        body = GenerateCoverRequest(cover_type="ai_frame", asset_ids=["a1"])

        with (
            patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
            patch(
                "packages.adapters.sqlalchemy_impl.asset_repository.SQLAlchemyAssetRepository",
                return_value=mock_asset_repo,
            ),
            patch("packages.shared.mediakit_client.get_mediakit_client", return_value=mock_mk),
            patch("packages.shared.storage.get_shared_storage_service", return_value=mock_storage),
            patch(
                "app.api.routes.generation_cover._persist_cover_frame",
                return_value="https://oss.example.com/covers/final.jpg",
            ) as mock_persist,
            patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize,
        ):
            mock_repo = MagicMock()
            mock_repo.get.return_value = None
            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_normalize.return_value = {
                "cover": {"type": "ai_frame", "image_url": "https://oss.example.com/covers/final.jpg"}
            }

            from app.api.routes.generation_cover import generate_cover

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

            assert result.cover["image_url"] == "https://oss.example.com/covers/final.jpg"
            # 标题文字必须透传给持久化函数（用于源素材帧叠加标题）
            assert mock_persist.call_args.kwargs.get("title_text") == "我的视频标题"

    def test_e2_passes_full_title_style_to_persist(self):
        """步骤E2：plan.config.title 包含完整样式时，color/position/font_size 都传给 _persist_cover_frame。"""
        from unittest.mock import MagicMock, patch

        from app.api.routes.generation_cover import GenerateCoverRequest

        mock_plan = MagicMock()
        mock_plan.config = {
            "title": {
                "enabled": True,
                "text": "样式标题",
                "color": "#00ff00",
                "position": "top",
                "font_size": 42,
            }
        }

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

        mock_asset = MagicMock()
        mock_asset.file_type = "video"
        mock_asset.storage_key = "uploads/src.mp4"
        mock_asset_repo = MagicMock()
        mock_asset_repo.get.return_value = mock_asset

        mock_mk = MagicMock()
        mock_mk.is_available = True
        mock_mk.extract_frames.return_value = [{"image_url": "https://mk/frame.jpg"}]

        mock_storage = MagicMock()
        mock_storage.get_url.return_value = "https://oss.example.com/uploads/src.mp4"

        body = GenerateCoverRequest(cover_type="ai_frame", asset_ids=["a1"])

        with (
            patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
            patch(
                "packages.adapters.sqlalchemy_impl.asset_repository.SQLAlchemyAssetRepository",
                return_value=mock_asset_repo,
            ),
            patch("packages.shared.mediakit_client.get_mediakit_client", return_value=mock_mk),
            patch("packages.shared.storage.get_shared_storage_service", return_value=mock_storage),
            patch(
                "app.api.routes.generation_cover._persist_cover_frame",
                return_value="https://oss.example.com/covers/styled.jpg",
            ) as mock_persist,
            patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize,
        ):
            mock_repo = MagicMock()
            mock_repo.get.return_value = None
            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_normalize.return_value = {
                "cover": {"type": "ai_frame", "image_url": "https://oss.example.com/covers/styled.jpg"}
            }

            from app.api.routes.generation_cover import generate_cover

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

            assert result.cover["image_url"] == "https://oss.example.com/covers/styled.jpg"
            kwargs = mock_persist.call_args.kwargs
            assert kwargs["title_text"] == "样式标题"
            assert kwargs["title_color"] == "#00ff00"
            assert kwargs["title_position"] == "top"
            assert kwargs["title_font_size"] == 42

    def test_e2_title_style_fallback_font_color(self):
        """步骤E2：前端传 font_color 时能正确兼容读取。"""
        from unittest.mock import MagicMock, patch

        from app.api.routes.generation_cover import GenerateCoverRequest

        mock_plan = MagicMock()
        mock_plan.config = {
            "title": {
                "enabled": True,
                "text": "兼容标题",
                "font_color": "#123456",
                "position": "center",
            }
        }

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

        mock_asset = MagicMock()
        mock_asset.file_type = "video"
        mock_asset.storage_key = "uploads/src.mp4"
        mock_asset_repo = MagicMock()
        mock_asset_repo.get.return_value = mock_asset

        mock_mk = MagicMock()
        mock_mk.is_available = True
        mock_mk.extract_frames.return_value = [{"image_url": "https://mk/frame.jpg"}]

        mock_storage = MagicMock()
        mock_storage.get_url.return_value = "https://oss.example.com/uploads/src.mp4"

        body = GenerateCoverRequest(cover_type="ai_frame", asset_ids=["a1"])

        with (
            patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
            patch(
                "packages.adapters.sqlalchemy_impl.asset_repository.SQLAlchemyAssetRepository",
                return_value=mock_asset_repo,
            ),
            patch("packages.shared.mediakit_client.get_mediakit_client", return_value=mock_mk),
            patch("packages.shared.storage.get_shared_storage_service", return_value=mock_storage),
            patch(
                "app.api.routes.generation_cover._persist_cover_frame",
                return_value="https://oss.example.com/covers/compat.jpg",
            ) as mock_persist,
            patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize,
        ):
            mock_repo = MagicMock()
            mock_repo.get.return_value = None
            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_normalize.return_value = {
                "cover": {"type": "ai_frame", "image_url": "https://oss.example.com/covers/compat.jpg"}
            }

            from app.api.routes.generation_cover import generate_cover

            generate_cover(
                body=body,
                template_id="tpl",
                plan_id="plan-compat",
                services=(mock_template_svc, mock_plan_svc),
                db=mock_db,
                current_user=MagicMock(),
            )

            kwargs = mock_persist.call_args.kwargs
            assert kwargs["title_color"] == "#123456"
            assert kwargs["title_position"] == "center"
            assert kwargs["title_font_size"] is None

    def test_step_e_skips_non_video_assets(self):
        """步骤E2：asset_ids 里只有图片素材时，不调用 MediaKit 并返回 400。"""
        from unittest.mock import MagicMock, patch

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

        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_db = MagicMock()

        mock_image_asset = MagicMock()
        mock_image_asset.file_type = "image"
        mock_image_asset.storage_key = "uploads/photo.png"

        mock_asset_repo = MagicMock()
        mock_asset_repo.get.return_value = mock_image_asset

        mock_mk = MagicMock()
        mock_mk.is_available = True

        body = GenerateCoverRequest(
            cover_type="ai_frame",
            asset_ids=["asset-img-1"],
        )

        with (
            patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
            patch(
                "packages.adapters.sqlalchemy_impl.asset_repository.SQLAlchemyAssetRepository",
                return_value=mock_asset_repo,
            ),
            patch("packages.shared.mediakit_client.get_mediakit_client", return_value=mock_mk),
            patch("app.api.routes.generation_cover.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_repo.list_latest_completed_preview.return_value = []
            mock_repo_cls.return_value = mock_repo
            mock_storage_getter.return_value = MagicMock()

            from app.api.routes.generation_cover import generate_cover

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

            assert exc_info.value.status_code == 400
            mock_mk.extract_frames.assert_not_called()


class TestCoverFromFinalVideo:
    """测试封面从最终成片任务(is_preview=False)获取视频源。"""

    def test_generated_video_fields_in_schema(self):
        """请求体支持 generated_video_id 和 video_url 字段。"""
        from app.api.routes.generation_cover import GenerateCoverRequest

        req = GenerateCoverRequest(
            generated_video_id="gv-001",
            video_url="https://example.com/final.mp4",
        )
        assert req.generated_video_id == "gv-001"
        assert req.video_url == "https://example.com/final.mp4"

        # 默认 None
        req_default = GenerateCoverRequest()
        assert req_default.generated_video_id is None
        assert req_default.video_url is None

    def test_cover_uses_final_video_when_generated_video_id_provided(self):
        """传 generated_video_id 时，从该最终成片视频抽帧。"""
        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()

        # Generated video
        mock_gv = MagicMock()
        mock_gv.file_url = "rendered/final/video.mp4"
        mock_gv.generation_task_id = "task-final-001"
        mock_gv.user_id = "user-1"

        # 最终成片任务
        mock_final_task = MagicMock()
        mock_final_task.id = "task-final-001"
        mock_final_task.created_by_user_id = "user-1"
        mock_final_task.cover_url = ""

        mock_gv_repo = MagicMock()
        mock_gv_repo.get.return_value = mock_gv

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

        body = GenerateCoverRequest(
            cover_type="ai_frame",
            generated_video_id="gv-final-001",
        )

        with (
            patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
            patch(
                "app.api.routes.generation_cover.get_generated_video_repository",
                return_value=mock_gv_repo,
            ),
            patch("app.api.routes.generation_cover.get_shared_storage_service") as mock_storage_getter,
            patch("packages.shared.mediakit_client.get_mediakit_client") as mock_mk_getter,
            patch(
                "app.api.routes.generation_cover._persist_cover_frame",
                return_value="https://oss.example.com/covers/final-cover.jpg",
            ) as mock_persist,
            patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize,
        ):
            mock_repo = MagicMock()
            mock_repo.get.return_value = mock_final_task
            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/final/video.mp4"
            mock_storage_svc.public_url = "https://oss.example.com"
            mock_storage_svc.endpoint = "oss.example.com"
            mock_storage_getter.return_value = mock_storage_svc

            mock_mk = MagicMock()
            mock_mk.is_available = True
            mock_mk.extract_frames.return_value = [{"image_url": "https://mk/frame.jpg"}]
            mock_mk_getter.return_value = mock_mk

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

            from app.api.routes.generation_cover import generate_cover

            result = generate_cover(
                body=body,
                template_id="tpl-1",
                plan_id="plan-final",
                services=(mock_template_svc, mock_plan_svc),
                db=mock_db,
                current_user=mock_current_user,
            )

            assert result.cover["image_url"] == "https://oss.example.com/covers/final-cover.jpg"
            call_kwargs = mock_mk.extract_frames.call_args.kwargs
            assert "rendered/final/video.mp4" in call_kwargs["video_url"]
            assert call_kwargs["strategy"] == "SpecifiedFrames"
            assert call_kwargs["max_frames"] == 1
            assert call_kwargs["max_retries"] == 0
            mock_persist.assert_called_once()

    def test_cover_uses_video_url_directly(self):
        """传 video_url 时，直接从该 URL 抽帧。"""
        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_db = MagicMock()
        mock_current_user = MagicMock()
        mock_current_user.user.id = "user-1"

        body = GenerateCoverRequest(
            cover_type="ai_frame",
            video_url="https://oss.example.com/rendered/final/video.mp4",
        )

        with (
            patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
            patch("app.api.routes.generation_cover.get_shared_storage_service") as mock_storage_getter,
            patch("packages.shared.mediakit_client.get_mediakit_client") as mock_mk_getter,
            patch(
                "app.api.routes.generation_cover._persist_cover_frame",
                return_value="https://oss.example.com/covers/c.jpg",
            ),
            patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize,
        ):
            mock_repo = MagicMock()
            mock_repo.get.return_value = None
            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.public_url = "https://oss.example.com"
            mock_storage_svc.endpoint = "oss.example.com"
            mock_storage_getter.return_value = mock_storage_svc

            mock_mk = MagicMock()
            mock_mk.is_available = True
            mock_mk.extract_frames.return_value = [{"image_url": "https://mk/f.jpg"}]
            mock_mk_getter.return_value = mock_mk

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

            from app.api.routes.generation_cover import generate_cover

            generate_cover(
                body=body,
                template_id="tpl-1",
                plan_id="plan-url",
                services=(mock_template_svc, mock_plan_svc),
                db=mock_db,
                current_user=mock_current_user,
            )

            call_kwargs = mock_mk.extract_frames.call_args.kwargs
            assert "rendered/final/video.mp4" in call_kwargs["video_url"]

    def test_cover_prefers_final_task_over_preview_in_source_plan(self):
        """步骤3：source_edit_plan 关联任务中，优先使用 is_preview=False 的最终成片。"""
        from unittest.mock import MagicMock, patch

        from app.api.routes.generation_cover import GenerateCoverRequest

        mock_plan = MagicMock()
        mock_plan.config = {}  # 无 rendered_storage_key / generation_task_id

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

        # 一个预览任务 + 一个最终成片任务
        mock_preview = MagicMock()
        mock_preview.id = "task-preview"
        mock_preview.status = "completed"
        mock_preview.is_preview = True
        mock_preview.cover_url = ""

        mock_final = MagicMock()
        mock_final.id = "task-final"
        mock_final.status = "completed"
        mock_final.is_preview = False
        mock_final.cover_url = ""

        mock_video_preview = MagicMock()
        mock_video_preview.file_url = "rendered/preview/video.mp4"
        mock_video_final = MagicMock()
        mock_video_final.file_url = "rendered/final/video.mp4"

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

        body = GenerateCoverRequest(cover_type="ai_frame")

        with (
            patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
            patch("app.api.routes.generation_cover.ListGeneratedVideosByTaskUseCase") as mock_list_videos,
            patch("app.api.routes.generation_cover.get_shared_storage_service") as mock_storage_getter,
            patch("packages.shared.mediakit_client.get_mediakit_client") as mock_mk_getter,
            patch(
                "app.api.routes.generation_cover._persist_cover_frame",
                return_value="https://oss.example.com/covers/c.jpg",
            ),
            patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize,
        ):
            mock_repo = MagicMock()
            mock_repo.get.return_value = None
            # list_by_source_edit_plan 返回 [preview, final]，最终成片排在后面
            mock_repo.list_by_source_edit_plan.return_value = [mock_preview, mock_final]
            mock_repo.list_latest_completed_preview.return_value = []
            mock_repo_cls.return_value = mock_repo

            # 根据 task id 返回不同 video
            def get_videos(task_id):
                if task_id == "task-final":
                    return [mock_video_final]
                return [mock_video_preview]

            mock_use_case = MagicMock()
            mock_use_case.execute.side_effect = get_videos
            mock_list_videos.return_value = mock_use_case

            mock_storage_svc = MagicMock()
            mock_storage_svc.get_url.side_effect = lambda key: f"https://oss.example.com/{key}"
            mock_storage_svc.public_url = "https://oss.example.com"
            mock_storage_svc.endpoint = "oss.example.com"
            mock_storage_getter.return_value = mock_storage_svc

            mock_mk = MagicMock()
            mock_mk.is_available = True
            mock_mk.extract_frames.return_value = [{"image_url": "https://mk/f.jpg"}]
            mock_mk_getter.return_value = mock_mk

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

            from app.api.routes.generation_cover import generate_cover

            generate_cover(
                body=body,
                template_id="tpl-1",
                plan_id="plan-priority",
                services=(mock_template_svc, mock_plan_svc),
                db=mock_db,
                current_user=mock_current_user,
            )

            # 应该使用 final video URL 抽帧，而非 preview
            call_kwargs = mock_mk.extract_frames.call_args.kwargs
            assert "rendered/final/video.mp4" in call_kwargs["video_url"]
            assert "rendered/preview" not in call_kwargs["video_url"]

    def test_cover_generated_video_permission_denied(self):
        """generated_video_id 关联任务属于其他用户时，返回 403。"""
        from unittest.mock import MagicMock, patch

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

        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_gv = MagicMock()
        mock_gv.file_url = "rendered/other/video.mp4"
        mock_gv.generation_task_id = "task-other"
        mock_gv.user_id = ""  # 老数据无 user_id，走关联任务归属校验

        mock_other_task = MagicMock()
        mock_other_task.created_by_user_id = "other-user"

        mock_gv_repo = MagicMock()
        mock_gv_repo.get.return_value = mock_gv

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

        body = GenerateCoverRequest(
            cover_type="ai_frame",
            generated_video_id="gv-other",
        )

        with (
            patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
            patch(
                "app.api.routes.generation_cover.get_generated_video_repository",
                return_value=mock_gv_repo,
            ),
        ):
            mock_repo = MagicMock()
            mock_repo.get.return_value = mock_other_task
            mock_repo_cls.return_value = mock_repo

            from app.api.routes.generation_cover import generate_cover

            with pytest.raises(HTTPException) as exc_info:
                generate_cover(
                    body=body,
                    template_id="tpl-1",
                    plan_id="plan-perm",
                    services=(mock_template_svc, mock_plan_svc),
                    db=mock_db,
                    current_user=mock_current_user,
                )
            assert exc_info.value.status_code == 403

    def test_cover_video_url_ssrf_blocked(self):
        """video_url 指向内网/非白名单域名时被忽略，不向其发起抽帧请求。"""
        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_db = MagicMock()
        mock_current_user = MagicMock()
        mock_current_user.user.id = "user-1"

        # SSRF 攻击载荷：内网元数据地址
        body = GenerateCoverRequest(
            cover_type="ai_frame",
            video_url="http://100.100.100.200/latest/meta-data/",
        )

        with (
            patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
            patch("app.api.routes.generation_cover.get_shared_storage_service") as mock_storage_getter,
            patch("packages.shared.mediakit_client.get_mediakit_client") as mock_mk_getter,
            patch("app.api.routes.generation_cover._persist_cover_frame") as mock_persist,
            patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize,
        ):
            mock_repo = MagicMock()
            mock_repo.get.return_value = None
            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.public_url = "https://xiaoxia-media.oss-cn-hangzhou.aliyuncs.com"
            mock_storage_svc.endpoint = "oss-cn-hangzhou.aliyuncs.com"
            mock_storage_svc.get_url.side_effect = lambda k: f"https://xiaoxia-media.oss-cn-hangzhou.aliyuncs.com/{k}"
            mock_storage_getter.return_value = mock_storage_svc

            mock_mk = MagicMock()
            mock_mk.is_available = True
            mock_mk.extract_frames.return_value = [{"image_url": "https://mk/f.jpg"}]
            mock_mk_getter.return_value = mock_mk

            mock_normalize.return_value = {"cover": {"type": "ai_frame", "image_url": "https://mk/f.jpg"}}

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

            # 内网 URL 被白名单拦截后，无任何可用视频源 → 400（而不是向内网发请求）
            with pytest.raises(HTTPException) as exc_info:
                generate_cover(
                    body=body,
                    template_id="tpl-1",
                    plan_id="plan-ssrf",
                    services=(mock_template_svc, mock_plan_svc),
                    db=mock_db,
                    current_user=mock_current_user,
                )
            assert exc_info.value.status_code == 400
            # MediaKit 从未被要求抽帧该内网地址
            if mock_mk.extract_frames.called:
                called_url = mock_mk.extract_frames.call_args.kwargs.get("video_url", "")
                assert "100.100.100.200" not in called_url
                assert "meta-data" not in called_url

    def test_cover_generated_video_ownership_unverifiable_denied(self):
        """video 无 user_id 且关联任务不存在时，归属无法确认 → 403（防权限绕过）。"""
        from unittest.mock import MagicMock, patch

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

        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_gv = MagicMock()
        mock_gv.file_url = "rendered/mystery/video.mp4"
        mock_gv.generation_task_id = "task-gone"  # 关联任务已删除
        mock_gv.user_id = ""  # 老数据无 owner

        mock_gv_repo = MagicMock()
        mock_gv_repo.get.return_value = mock_gv

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

        body = GenerateCoverRequest(
            cover_type="ai_frame",
            generated_video_id="gv-mystery",
        )

        with (
            patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
            patch(
                "app.api.routes.generation_cover.get_generated_video_repository",
                return_value=mock_gv_repo,
            ),
        ):
            mock_repo = MagicMock()
            mock_repo.get.return_value = None  # 关联任务查不到
            mock_repo_cls.return_value = mock_repo

            from app.api.routes.generation_cover import generate_cover

            with pytest.raises(HTTPException) as exc_info:
                generate_cover(
                    body=body,
                    template_id="tpl-1",
                    plan_id="plan-orphan",
                    services=(mock_template_svc, mock_plan_svc),
                    db=mock_db,
                    current_user=mock_current_user,
                )
            assert exc_info.value.status_code == 403

    def test_is_trusted_media_url_blocks_internal_and_ipv6(self):
        """白名单函数：内网 IPv4/IPv6/元数据地址一律拒绝，自家 OSS 域名放行。"""
        from unittest.mock import MagicMock, patch

        from app.api.routes.generation_cover import _is_trusted_media_url

        mock_storage = MagicMock()
        mock_storage.public_url = "https://xiaoxia-media.oss-cn-hangzhou.aliyuncs.com"
        mock_storage.endpoint = "oss-cn-hangzhou.aliyuncs.com"

        with patch(
            "app.api.routes.generation_cover.get_shared_storage_service",
            return_value=mock_storage,
        ):
            # 内网 / 元数据 / IPv6 本地地址全部拒绝
            for bad in [
                "http://127.0.0.1/admin",
                "http://10.0.0.5/video.mp4",
                "http://192.168.1.1/video.mp4",
                "http://172.16.0.1/video.mp4",
                "http://169.254.169.254/latest/meta-data/",
                "http://[::1]:8080/video.mp4",
                "http://[fe80::1]/video.mp4",
                "http://[fc00::1]/video.mp4",
                "http://localhost/x",
                "ftp://oss-cn-hangzhou.aliyuncs.com/a.mp4",
                "",
            ]:
                assert _is_trusted_media_url(bad) is False, f"应拒绝: {bad}"

            # 自家 OSS 域名（含签名 URL 子路径、bucket 域名）放行
            for good in [
                "https://xiaoxia-media.oss-cn-hangzhou.aliyuncs.com/rendered/final/v.mp4",
                "https://xiaoxia-media.oss-cn-hangzhou.aliyuncs.com/rendered/v.mp4?Expires=123&Signature=abc",
            ]:
                assert _is_trusted_media_url(good) is True, f"应放行: {good}"

    def test_is_trusted_media_url_endpoint_with_scheme_parsed(self):
        """endpoint 配置带 http:// 前缀时也能正确提取主机名，不出现 .http 后缀绕过。"""
        from unittest.mock import MagicMock, patch

        from app.api.routes.generation_cover import _is_trusted_media_url

        mock_storage = MagicMock()
        mock_storage.public_url = "http://oss.internal.example.com:9000"
        mock_storage.endpoint = "http://oss.internal.example.com:9000"

        with patch(
            "app.api.routes.generation_cover.get_shared_storage_service",
            return_value=mock_storage,
        ):
            # 正确域名放行
            assert _is_trusted_media_url("http://oss.internal.example.com:9000/a/b.mp4") is True
            # 伪造后缀域名必须拒绝（修复前 split(':')[0] 会取到 'http' 导致绕过）
            assert _is_trusted_media_url("http://evil-http.com/x.mp4") is False
            assert _is_trusted_media_url("http://evil.http/x.mp4") is False