diff --git a/apps/api/app/api/routes/generation_cover.py b/apps/api/app/api/routes/generation_cover.py index 125346c7c..56ba38fb7 100644 --- a/apps/api/app/api/routes/generation_cover.py +++ b/apps/api/app/api/routes/generation_cover.py @@ -49,6 +49,10 @@ class GenerateCoverRequest(BaseModel): ge=0.0, description="手动选帧时间点（秒），仅 cover_type=manual 时有效", ) + cover_url: Optional[str] = Field( + default=None, + description="上传的封面图片 URL，仅 cover_type=upload 时有效", + ) class GenerateCoverResponse(BaseModel): @@ -80,6 +84,29 @@ def generate_cover( _, plan_svc = services plan = plan_svc.get_plan_or_raise(plan_id) + # ── upload 类型：直接保存前端上传的封面图片，不需要预览视频 ────── + if body.cover_type == "upload": + if not body.cover_url: + raise HTTPException( + status_code=400, + detail="cover_type=upload 时必须提供 cover_url", + ) + cover_data = { + "type": "upload", + "image_url": body.cover_url, + } + current_config = dict(plan.config) if plan.config else {} + current_config["cover"] = cover_data + normalized = normalize_plan_config(current_config) + plan_svc.update_plan_config(plan_id, {"cover": normalized["cover"]}) + logger.info( + "封面上传完成: plan_id=%s cover_url=%s by user=%s", + plan_id, + body.cover_url[:80] if body.cover_url else "", + current_user.user.id, + ) + return GenerateCoverResponse(plan_id=plan_id, cover=cover_data) + # ── 3 步查找预览视频 URL ────────────────────────────────────────── # 第一步：从 plan.config 读取 logger.info("[封面生成] 步骤1: 从 plan.config 查找 rendered_storage_key: plan_id=%s", plan_id) diff --git a/apps/api/app/api/routes/templates_editor/generation.py b/apps/api/app/api/routes/templates_editor/generation.py index 804b66a14..8c49d946d 100755 --- a/apps/api/app/api/routes/templates_editor/generation.py +++ b/apps/api/app/api/routes/templates_editor/generation.py @@ -8,8 +8,9 @@ from __future__ import annotations +import json import logging -from typing import Any +from typing import Any, Optional from app.auth import AuthenticatedUser, get_current_user from app.core.celery_app import celery_app @@ -45,6 +46,7 @@ from ._fallback import ( from .dependencies import _check_queue_limits, get_draft_plan_id, get_editor_services from .schemas import ( ClipStatusItem, + EditPlanGenerateRequest, EditPlanGenerateResponse, EditPlanGenerationsResponse, EditPlanGenerationStatusResponse, @@ -57,6 +59,7 @@ router = APIRouter(tags=["Template Editor"]) @router.post("/generate", response_model=EditPlanGenerateResponse) def generate_editor_draft( template_id: str, + request: Optional[EditPlanGenerateRequest] = None, plan_id: str = Depends(get_draft_plan_id), services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services), db: Session = Depends(get_db_session), @@ -65,6 +68,7 @@ def generate_editor_draft( asset_repo: Any = Depends(get_asset_repository), ) -> EditPlanGenerateResponse: """触发模板草稿渲染生成""" + req = request or EditPlanGenerateRequest() _, plan_svc = services plan_check = plan_svc.get_plan_or_raise(plan_id) @@ -85,6 +89,36 @@ def generate_editor_draft( # 检查是否可复用已完成的预览产物（预览品质已与正式一致） gen_task_repo = SQLAlchemyGenerationTaskRepository(db) reusable_task = _find_reusable_preview_task(gen_task_repo, plan_id, plan_check) + if reusable_task: + # 复用预览产物：标记为正式产出，跳过渲染 + # 如果前端传了 title_config，需要创建新任务（因为预览任务的 custom_title 可能不同） + title_config_reuse = req.title_config or {} + title_text_reuse = (title_config_reuse.get("text") or "").strip() + existing_custom_title = getattr(reusable_task, "custom_title", "") or "" + if title_text_reuse and existing_custom_title: + # 如果新标题和已有标题不同，不能复用，走新建任务流程 + new_title_json = json.dumps(title_config_reuse, ensure_ascii=False) + if new_title_json != existing_custom_title: + logger.info( + "[模板生成] 标题已变更，跳过复用: task_id=%s", + reusable_task.id, + ) + reusable_task = None + elif title_text_reuse and not existing_custom_title: + # 原来没标题，现在有标题，不能复用 + logger.info( + "[模板生成] 新增标题，跳过复用: task_id=%s", + reusable_task.id, + ) + reusable_task = None + elif not title_text_reuse and existing_custom_title: + # 原来有标题，现在移除了，不能复用 + logger.info( + "[模板生成] 移除标题，跳过复用: task_id=%s", + reusable_task.id, + ) + reusable_task = None + if reusable_task: # 复用预览产物：标记为正式产出，跳过渲染 reusable_task.mark_confirmed() @@ -133,6 +167,21 @@ def generate_editor_draft( gen_task_use_case = CreateGenerationTaskUseCase(gen_task_repo) plan = plan_svc.get_plan_or_raise(plan_id) config_asset_ids = (plan.config or {}).get("asset_ids", []) + # 从 plan config 读取封面 URL（由 generate-cover 保存） + cover_url_from_config = (plan.config or {}).get("cover", {}).get("image_url", "") + + # 处理标题配置：序列化 title_config 为 JSON 存入 custom_title + title_config = req.title_config or {} + title_text = (title_config.get("text") or "").strip() + custom_title_value = "" + if title_text: + custom_title_value = json.dumps(title_config, ensure_ascii=False) + logger.info( + "[模板生成] 标题配置: text=%s, config_keys=%s", + title_text[:30], + list(title_config.keys()), + ) + gen_task = gen_task_use_case.execute( CreateGenerationTaskCommand( project_id=plan.project_id or "", @@ -140,6 +189,8 @@ def generate_editor_draft( created_by_user_id=current_user.user.id, source_edit_plan_id=plan_id, asset_ids=list(config_asset_ids) if config_asset_ids else [], + cover_url=cover_url_from_config, + custom_title=custom_title_value, ), ) diff --git a/apps/api/app/api/routes/templates_editor/schemas.py b/apps/api/app/api/routes/templates_editor/schemas.py index 2d653ed00..bc441d333 100755 --- a/apps/api/app/api/routes/templates_editor/schemas.py +++ b/apps/api/app/api/routes/templates_editor/schemas.py @@ -6,7 +6,7 @@ from __future__ import annotations import re as _re -from typing import Any, List, Optional +from typing import Any, Dict, List, Optional from app.schemas.generation_task import GenerationTaskResponse from pydantic import BaseModel, Field, validator @@ -44,6 +44,14 @@ class EditPlanGenerationStatusResponse(BaseModel): clips: List[ClipStatusItem] +class EditPlanGenerateRequest(BaseModel): + """模板编辑器触发生成请求体""" + title_config: Optional[Dict[str, Any]] = Field( + default_factory=dict, + description="标题配置（可选），渲染时烧录到视频中。支持字段: text/font/font_size/font_color/position/bold/stroke/shadow", + ) + + class EditPlanGenerateResponse(BaseModel): """剪辑计划触发生成响应体""" diff --git a/tests/unit/test_generate_title_and_cover.py b/tests/unit/test_generate_title_and_cover.py new file mode 100644 index 000000000..eea8a1644 --- /dev/null +++ b/tests/unit/test_generate_title_and_cover.py @@ -0,0 +1,369 @@ +"""Tests for /generate endpoint — custom_title and cover_url passing.""" + +from __future__ import annotations + +import json +from unittest.mock import MagicMock, patch + +import pytest + + +class TestGenerateEndpointTitleAndCover: + """测试 /generate 端点传递 custom_title 和 cover_url。""" + + def test_generate_passes_cover_url_from_plan_config(self): + """从 plan.config.cover.image_url 读取封面 URL 传递给生成任务。""" + from app.api.routes.templates_editor.generation import generate_editor_draft + from app.api.routes.templates_editor.schemas import EditPlanGenerateRequest + + mock_plan = MagicMock() + mock_plan.id = "plan-123" + mock_plan.project_id = "project-1" + mock_plan.template_id = "template-1" + mock_plan.status = MagicMock(value="editing") + mock_plan.config = { + "clips": [{"id": "c1"}], + "asset_ids": ["a1"], + "cover": {"type": "upload", "image_url": "https://oss.example.com/uploaded/cover.jpg"}, + } + mock_plan.updated_at = None + + mock_plan_svc = MagicMock() + mock_plan_svc.get_plan_or_raise.return_value = mock_plan + mock_plan_svc.can_generate.return_value = (True, "") + mock_plan_svc.mark_clips_ready.return_value = 1 + mock_template_svc = MagicMock() + + mock_gen_task = MagicMock() + mock_gen_task.id = "task-new" + mock_gen_task.project_id = "project-1" + + mock_current_user = MagicMock() + mock_current_user.user.id = "user-1" + + body = EditPlanGenerateRequest() # No title_config + + with ( + patch("app.api.routes.templates_editor.generation.SQLAlchemyGenerationTaskRepository") as mock_repo_cls, + patch("app.api.routes.templates_editor.generation.CreateGenerationTaskUseCase") as mock_usecase_cls, + patch("app.api.routes.templates_editor.generation._find_reusable_preview_task", return_value=None), + patch("app.api.routes.templates_editor.generation._auto_fallback_draft_to_editing"), + patch("app.api.routes.templates_editor.generation._auto_fallback_copy_template_clips"), + patch("app.api.routes.templates_editor.generation._auto_fallback_assign_assets", return_value=[]), + patch("app.api.routes.templates_editor.generation._auto_fallback_auto_material_mode"), + patch("app.api.routes.templates_editor.generation._check_queue_limits"), + patch("app.api.routes.templates_editor.generation.celery_app"), + patch("app.api.routes.templates_editor.generation.get_draft_plan_id", return_value="plan-123"), + ): + mock_repo = MagicMock() + mock_repo_cls.return_value = mock_repo + + mock_usecase = MagicMock() + mock_usecase.execute.return_value = mock_gen_task + mock_usecase_cls.return_value = mock_usecase + + mock_plan_svc.transition_status = MagicMock() + mock_plan_svc.get_plan_or_raise.return_value = mock_plan + + result = generate_editor_draft( + template_id="template-1", + request=body, + plan_id="plan-123", + services=(mock_template_svc, mock_plan_svc), + db=MagicMock(), + current_user=mock_current_user, + asset_library_repo=MagicMock(), + asset_repo=MagicMock(), + ) + + # Verify cover_url was passed to CreateGenerationTaskCommand + call_args = mock_usecase.execute.call_args + command = call_args[0][0] + assert command.cover_url == "https://oss.example.com/uploaded/cover.jpg" + assert command.custom_title == "" + + def test_generate_passes_custom_title_from_title_config(self): + """前端传 title_config 时，序列化为 JSON 存入 custom_title。""" + from app.api.routes.templates_editor.generation import generate_editor_draft + from app.api.routes.templates_editor.schemas import EditPlanGenerateRequest + + mock_plan = MagicMock() + mock_plan.id = "plan-456" + mock_plan.project_id = "project-1" + mock_plan.template_id = "template-1" + mock_plan.status = MagicMock(value="editing") + mock_plan.config = { + "clips": [{"id": "c1"}], + "asset_ids": ["a1"], + "cover": {"type": "ai_frame", "image_url": "https://oss.example.com/cover.jpg"}, + } + mock_plan.updated_at = None + + mock_plan_svc = MagicMock() + mock_plan_svc.get_plan_or_raise.return_value = mock_plan + mock_plan_svc.can_generate.return_value = (True, "") + mock_plan_svc.mark_clips_ready.return_value = 1 + mock_template_svc = MagicMock() + + mock_gen_task = MagicMock() + mock_gen_task.id = "task-title" + + mock_current_user = MagicMock() + mock_current_user.user.id = "user-1" + + title_config = { + "text": "测试标题", + "font_size": 36, + "font_color": "#ffffff", + "position": "center", + } + body = EditPlanGenerateRequest(title_config=title_config) + + with ( + patch("app.api.routes.templates_editor.generation.SQLAlchemyGenerationTaskRepository") as mock_repo_cls, + patch("app.api.routes.templates_editor.generation.CreateGenerationTaskUseCase") as mock_usecase_cls, + patch("app.api.routes.templates_editor.generation._find_reusable_preview_task", return_value=None), + patch("app.api.routes.templates_editor.generation._auto_fallback_draft_to_editing"), + patch("app.api.routes.templates_editor.generation._auto_fallback_copy_template_clips"), + patch("app.api.routes.templates_editor.generation._auto_fallback_assign_assets", return_value=[]), + patch("app.api.routes.templates_editor.generation._auto_fallback_auto_material_mode"), + patch("app.api.routes.templates_editor.generation._check_queue_limits"), + patch("app.api.routes.templates_editor.generation.celery_app"), + ): + mock_repo = MagicMock() + mock_repo_cls.return_value = mock_repo + + mock_usecase = MagicMock() + mock_usecase.execute.return_value = mock_gen_task + mock_usecase_cls.return_value = mock_usecase + + mock_plan_svc.transition_status = MagicMock() + mock_plan_svc.get_plan_or_raise.return_value = mock_plan + + result = generate_editor_draft( + template_id="template-1", + request=body, + plan_id="plan-456", + services=(mock_template_svc, mock_plan_svc), + db=MagicMock(), + current_user=mock_current_user, + asset_library_repo=MagicMock(), + asset_repo=MagicMock(), + ) + + # Verify custom_title was serialized to JSON + call_args = mock_usecase.execute.call_args + command = call_args[0][0] + parsed_title = json.loads(command.custom_title) + assert parsed_title["text"] == "测试标题" + assert parsed_title["font_size"] == 36 + assert command.cover_url == "https://oss.example.com/cover.jpg" + + def test_generate_empty_title_config_passes_empty_custom_title(self): + """title_config 为空时 custom_title 为空字符串。""" + from app.api.routes.templates_editor.generation import generate_editor_draft + from app.api.routes.templates_editor.schemas import EditPlanGenerateRequest + + mock_plan = MagicMock() + mock_plan.id = "plan-789" + mock_plan.project_id = "project-1" + mock_plan.template_id = "template-1" + mock_plan.status = MagicMock(value="editing") + mock_plan.config = {"clips": [{"id": "c1"}], "asset_ids": ["a1"]} + mock_plan.updated_at = None + + mock_plan_svc = MagicMock() + mock_plan_svc.get_plan_or_raise.return_value = mock_plan + mock_plan_svc.can_generate.return_value = (True, "") + mock_plan_svc.mark_clips_ready.return_value = 1 + mock_template_svc = MagicMock() + + mock_gen_task = MagicMock() + mock_gen_task.id = "task-no-title" + + body = EditPlanGenerateRequest() # No title_config + + with ( + patch("app.api.routes.templates_editor.generation.SQLAlchemyGenerationTaskRepository") as mock_repo_cls, + patch("app.api.routes.templates_editor.generation.CreateGenerationTaskUseCase") as mock_usecase_cls, + patch("app.api.routes.templates_editor.generation._find_reusable_preview_task", return_value=None), + patch("app.api.routes.templates_editor.generation._auto_fallback_draft_to_editing"), + patch("app.api.routes.templates_editor.generation._auto_fallback_copy_template_clips"), + patch("app.api.routes.templates_editor.generation._auto_fallback_assign_assets", return_value=[]), + patch("app.api.routes.templates_editor.generation._auto_fallback_auto_material_mode"), + patch("app.api.routes.templates_editor.generation._check_queue_limits"), + patch("app.api.routes.templates_editor.generation.celery_app"), + ): + mock_repo = MagicMock() + mock_repo_cls.return_value = mock_repo + + mock_usecase = MagicMock() + mock_usecase.execute.return_value = mock_gen_task + mock_usecase_cls.return_value = mock_usecase + + mock_plan_svc.transition_status = MagicMock() + mock_plan_svc.get_plan_or_raise.return_value = mock_plan + + result = generate_editor_draft( + template_id="template-1", + request=body, + plan_id="plan-789", + services=(mock_template_svc, mock_plan_svc), + db=MagicMock(), + current_user=MagicMock(), + asset_library_repo=MagicMock(), + asset_repo=MagicMock(), + ) + + call_args = mock_usecase.execute.call_args + command = call_args[0][0] + assert command.custom_title == "" + + +class TestGenerateEndpointRequestSchema: + """测试 EditPlanGenerateRequest schema。""" + + def test_schema_default_empty_title_config(self): + """默认 title_config 为空 dict。""" + from app.api.routes.templates_editor.schemas import EditPlanGenerateRequest + + req = EditPlanGenerateRequest() + assert req.title_config == {} + + def test_schema_accepts_title_config(self): + """可以传入标题配置。""" + from app.api.routes.templates_editor.schemas import EditPlanGenerateRequest + + req = EditPlanGenerateRequest(title_config={"text": "我的标题", "font_size": 48}) + assert req.title_config["text"] == "我的标题" + assert req.title_config["font_size"] == 48 + + +class TestGenerateTitleChangeSkipsReuse: + """测试标题变更时跳过预览产物复用。""" + + def _make_mocks(self, custom_title=""): + mock_plan = MagicMock() + mock_plan.id = "plan-reuse" + mock_plan.project_id = "project-1" + mock_plan.template_id = "template-1" + mock_plan.status = MagicMock(value="editing") + mock_plan.config = {"clips": [{"id": "c1"}], "asset_ids": ["a1"]} + mock_plan.updated_at = None + + mock_plan_svc = MagicMock() + mock_plan_svc.get_plan_or_raise.return_value = mock_plan + mock_plan_svc.can_generate.return_value = (True, "") + mock_plan_svc.mark_clips_ready.return_value = 1 + mock_template_svc = MagicMock() + + reusable_task = MagicMock() + reusable_task.id = "task-reusable" + reusable_task.is_completed = True + reusable_task.is_preview = True + reusable_task.custom_title = custom_title + reusable_task.project_id = "project-1" + reusable_task.source_edit_plan_id = "plan-reuse" + + mock_new_task = MagicMock() + mock_new_task.id = "task-new" + + return mock_plan, mock_plan_svc, mock_template_svc, reusable_task, mock_new_task + + def test_title_removed_skips_reuse(self): + """原来有标题，现在移除了 → 跳过复用，创建新任务。""" + from app.api.routes.templates_editor.generation import generate_editor_draft + from app.api.routes.templates_editor.schemas import EditPlanGenerateRequest + + mock_plan, mock_plan_svc, mock_template_svc, reusable_task, mock_new_task = self._make_mocks( + custom_title='{"text": "旧标题"}' + ) + + body = EditPlanGenerateRequest() # No title_config → title removed + + with ( + patch("app.api.routes.templates_editor.generation.SQLAlchemyGenerationTaskRepository") as mock_repo_cls, + patch("app.api.routes.templates_editor.generation.CreateGenerationTaskUseCase") as mock_usecase_cls, + patch("app.api.routes.templates_editor.generation._find_reusable_preview_task", return_value=reusable_task), + patch("app.api.routes.templates_editor.generation._auto_fallback_draft_to_editing"), + patch("app.api.routes.templates_editor.generation._auto_fallback_copy_template_clips"), + patch("app.api.routes.templates_editor.generation._auto_fallback_assign_assets", return_value=[]), + patch("app.api.routes.templates_editor.generation._auto_fallback_auto_material_mode"), + patch("app.api.routes.templates_editor.generation._check_queue_limits"), + patch("app.api.routes.templates_editor.generation.celery_app"), + patch("app.api.routes.templates_editor.generation.get_draft_plan_id", return_value="plan-reuse"), + ): + mock_repo = MagicMock() + mock_repo_cls.return_value = mock_repo + + mock_usecase = MagicMock() + mock_usecase.execute.return_value = mock_new_task + mock_usecase_cls.return_value = mock_usecase + + mock_plan_svc.transition_status = MagicMock() + mock_plan_svc.get_plan_or_raise.return_value = mock_plan + + result = generate_editor_draft( + template_id="template-1", + request=body, + plan_id="plan-reuse", + services=(mock_template_svc, mock_plan_svc), + db=MagicMock(), + current_user=MagicMock(), + asset_library_repo=MagicMock(), + asset_repo=MagicMock(), + ) + + # 应该创建新任务而不是复用 + mock_usecase.execute.assert_called_once() + # 不应该 mark_confirmed 在 reusable_task 上 + reusable_task.mark_confirmed.assert_not_called() + + def test_title_changed_skips_reuse(self): + """标题变更 → 跳过复用。""" + import json + + from app.api.routes.templates_editor.generation import generate_editor_draft + from app.api.routes.templates_editor.schemas import EditPlanGenerateRequest + + mock_plan, mock_plan_svc, mock_template_svc, reusable_task, mock_new_task = self._make_mocks( + custom_title=json.dumps({"text": "旧标题", "font_size": 36}, ensure_ascii=False) + ) + + body = EditPlanGenerateRequest(title_config={"text": "新标题", "font_size": 48}) + + with ( + patch("app.api.routes.templates_editor.generation.SQLAlchemyGenerationTaskRepository") as mock_repo_cls, + patch("app.api.routes.templates_editor.generation.CreateGenerationTaskUseCase") as mock_usecase_cls, + patch("app.api.routes.templates_editor.generation._find_reusable_preview_task", return_value=reusable_task), + patch("app.api.routes.templates_editor.generation._auto_fallback_draft_to_editing"), + patch("app.api.routes.templates_editor.generation._auto_fallback_copy_template_clips"), + patch("app.api.routes.templates_editor.generation._auto_fallback_assign_assets", return_value=[]), + patch("app.api.routes.templates_editor.generation._auto_fallback_auto_material_mode"), + patch("app.api.routes.templates_editor.generation._check_queue_limits"), + patch("app.api.routes.templates_editor.generation.celery_app"), + patch("app.api.routes.templates_editor.generation.get_draft_plan_id", return_value="plan-reuse"), + ): + mock_repo = MagicMock() + mock_repo_cls.return_value = mock_repo + + mock_usecase = MagicMock() + mock_usecase.execute.return_value = mock_new_task + mock_usecase_cls.return_value = mock_usecase + + mock_plan_svc.transition_status = MagicMock() + mock_plan_svc.get_plan_or_raise.return_value = mock_plan + + result = generate_editor_draft( + template_id="template-1", + request=body, + plan_id="plan-reuse", + services=(mock_template_svc, mock_plan_svc), + db=MagicMock(), + current_user=MagicMock(), + asset_library_repo=MagicMock(), + asset_repo=MagicMock(), + ) + + mock_usecase.execute.assert_called_once() + reusable_task.mark_confirmed.assert_not_called() diff --git a/tests/unit/test_generation_cover.py b/tests/unit/test_generation_cover.py index 3f0e448a5..a7163087f 100644 --- a/tests/unit/test_generation_cover.py +++ b/tests/unit/test_generation_cover.py @@ -552,3 +552,138 @@ class TestStrayLoggerRemoved: 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 是唯一被调用的外部函数)