test: 修复 generation_cover 测试 mock 路径

- run_generate_cover 需要 patch packages.shared.ai_service 而非 route 模块
  (因为代码使用函数内 inline import)
- 补充 test_step25_skips 和 test_step25_exception 缺失的 run_generate_cover mock

apps/api/app/api/routes/generation_cover.py

@@ -105,10 +105,6 @@ def generate_cover(
                             generation_task_id,
                             rendered_storage_key[:80],
                         )
-                        logger.info(
-                            plan_id,
-                            generation_task_id,
-                        )
             except Exception:
                 logger.warning(
                     "封面生成: 通过 generation_task_id 查找视频失败: plan_id=%s",
@@ -116,6 +112,33 @@ def generate_cover(
                     exc_info=True,
                 )
 
+    # 第 2.5 步：通过 plan_id 作为 source_edit_plan_id 查找关联的已完成预览任务
+    if not rendered_storage_key:
+        try:
+            gen_task_repo = SQLAlchemyGenerationTaskRepository(db)
+            logger.info("[封面生成] 步骤2.5: 通过 source_edit_plan_id 查找: plan_id=%s", plan_id)
+            preview_tasks = gen_task_repo.list_by_source_edit_plan(plan_id)
+            for pt in preview_tasks:
+                if getattr(pt, "status", "") == "completed" and getattr(pt, "is_preview", False):
+                    video_repo = get_generated_video_repository(db)
+                    use_case = ListGeneratedVideosByTaskUseCase(video_repo)
+                    videos = use_case.execute(pt.id)
+                    if videos:
+                        rendered_storage_key = getattr(videos[0], "file_url", "") or ""
+                        logger.info(
+                            "[封面生成] ✅ 步骤2.5找到视频: plan_id=%s task_id=%s url=%s",
+                            plan_id,
+                            pt.id,
+                            rendered_storage_key[:80],
+                        )
+                        break
+        except Exception:
+            logger.warning(
+                "封面生成: 通过 source_edit_plan_id 查找预览任务失败: plan_id=%s",
+                plan_id,
+                exc_info=True,
+            )
+
     # 第三步：按 user + template 查找最近的已完成预览任务（兜底）
     if not rendered_storage_key:
         try:

tests/unit/test_generation_cover.py

@@ -209,3 +209,237 @@ class TestUnifiedCoverPipelineEndpoint:
             # 应该走 run_generate_cover 回退
             mock_run.assert_called_once()
             assert result.cover["image_url"] == "https://fallback.com/cover.jpg"
+
+
+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, call, 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
+        mock_preview_task = MagicMock()
+        mock_preview_task.id = "preview-task-789"
+        mock_preview_task.status = "completed"
+        mock_preview_task.is_preview = True
+
+        # 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,
+            patch("packages.shared.ai_service.run_generate_cover") as mock_run,
+            patch("packages.shared.storage.get_shared_storage_service") as mock_storage_getter,
+        ):
+            mock_repo = MagicMock()
+            # Step 2: no generation_task_id, so get() is not called
+            mock_repo.get.return_value = None
+            # Step 2.5: found by source_edit_plan_id
+            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": ""}}
+
+            mock_storage_svc = MagicMock()
+            mock_storage_svc.get_url.return_value = "https://oss.example.com/rendered/plan-x/video.mp4"
+            mock_storage_getter.return_value = mock_storage_svc
+
+            # cover_type=ai_frame but no cover_url on task -> fallback to run_generate_cover
+            mock_run.return_value = {"type": "ai_frame", "image_url": "https://cover.com/final.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
+            mock_repo.list_by_source_edit_plan.assert_called_once_with("plan-x")
+            # Step 3 should NOT be called since step 2.5 found the video
+            mock_repo.list_latest_completed_preview.assert_not_called()
+
+    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
+
+        # 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_db = MagicMock()
+
+        # Step 3 fallback finds a valid preview task
+        mock_step3_task = MagicMock()
+        mock_step3_task.id = "step3-task"
+
+        mock_video = MagicMock()
+        mock_video.file_url = "rendered/step3/video.mp4"
+
+        body = GenerateCoverRequest(cover_type="ai_frame")
+
+        with (
+            patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
+            patch("app.api.routes.generation_cover.get_generated_video_repository") as mock_video_repo,
+            patch("app.api.routes.generation_cover.ListGeneratedVideosByTaskUseCase") as mock_usecase_cls,
+            patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize,
+            patch("packages.shared.storage.get_shared_storage_service") as mock_storage_getter,
+            patch("packages.shared.ai_service.run_generate_cover") as mock_run,
+        ):
+            mock_run.return_value = {"type": "ai_frame", "image_url": "https://cover.com/step3.jpg"}
+            mock_repo = MagicMock()
+            mock_repo.get.return_value = None
+            mock_repo.list_by_source_edit_plan.return_value = [mock_task_failed, mock_task_full]
+            # Step 3 returns a valid task
+            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": ""}}
+
+            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_once()
+            mock_repo.list_latest_completed_preview.assert_called_once()
+
+    def test_step25_exception_does_not_block_step3(self):
+        """步骤2.5异常时不影响步骤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_video = MagicMock()
+        mock_video.file_url = "rendered/step3/video.mp4"
+
+        mock_db = MagicMock()
+
+        body = GenerateCoverRequest(cover_type="ai_frame")
+
+        with (
+            patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls,
+            patch("app.api.routes.generation_cover.get_generated_video_repository") as mock_video_repo,
+            patch("app.api.routes.generation_cover.ListGeneratedVideosByTaskUseCase") as mock_usecase_cls,
+            patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize,
+            patch("packages.shared.storage.get_shared_storage_service") as mock_storage_getter,
+            patch("packages.shared.ai_service.run_generate_cover") as mock_run,
+        ):
+            mock_run.return_value = {"type": "ai_frame", "image_url": "https://cover.com/step3.jpg"}
+            mock_repo = MagicMock()
+            mock_repo.get.return_value = None
+            # Step 2.5 raises
+            mock_repo.list_by_source_edit_plan.side_effect = RuntimeError("db error")
+            # Step 3 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": ""}}
+
+            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_once()
+
+
+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"