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