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

@@ -310,6 +310,22 @@ def generate_cover(
                     exc_info=True,
                 )
 
+        # 步骤 D：从 plan.config.cover_candidates 读取（Worker 渲染时写入）
+        if not cover_url_from_task:
+            _candidates = (plan.config or {}).get("cover_candidates") or []
+            if isinstance(_candidates, list) and _candidates:
+                _first = _candidates[0]
+                if isinstance(_first, dict):
+                    cover_url_from_task = (
+                        _first.get("image_url") or _first.get("url") or ""
+                    )
+                    if cover_url_from_task:
+                        logger.info(
+                            "[封面生成] 统一管道封面(步骤D-cover_candidates): plan_id=%s url=%s",
+                            plan_id,
+                            cover_url_from_task[:80],
+                        )
+
         if cover_url_from_task:
             # 标题已在预览视频渲染时烧录（ASS字幕），封面帧自然包含标题
             cover_data = {
@@ -325,7 +341,7 @@ def generate_cover(
             return GenerateCoverResponse(plan_id=plan_id, cover=cover_data)
 
         logger.warning(
-            "[封面生成] 统一管道未找到 cover_url: plan_id=%s",
+            "[封面生成] 统一管道未找到 cover_url (A/B/C/D均未命中): plan_id=%s",
             plan_id,
         )
         # ai_frame/ai_regenerate 类型必须从渲染管道获取，不再回退到 AI 服务

apps/web/e2e/core-generation.spec.ts

@@ -231,10 +231,7 @@ test.describe("Core generation flow", () => {
       (response) => {
         const url = response.url()
         const path = new URL(url).pathname
-        return (
-          response.request().method() === "POST" &&
-          path.endsWith("/generation/tasks")
-        )
+        return response.request().method() === "POST" && path.endsWith("/generation/tasks")
       },
       { timeout: 30_000 },
     )

tests/unit/test_generation_cover.py

@@ -318,6 +318,157 @@ class TestUnifiedCoverPipelineEndpoint:
                 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("packages.shared.storage.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 查找预览视频兜底逻辑。"""

tests/unit/test_generation_worker_fixes.py

@@ -13,6 +13,18 @@ os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret")
 os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
 
 
+def _patch_session_local(mock_session):
+    """Patch worker_app.db.SessionLocal robustly even when other tests
+    have pre-registered a MagicMock for worker_app.db in sys.modules.
+    Uses patch.dict to inject a clean module so that
+    'from worker_app.db import SessionLocal' resolves correctly."""
+    from types import ModuleType
+
+    _fresh_db = ModuleType("worker_app.db")
+    _fresh_db.SessionLocal = lambda *a, **kw: mock_session
+    return patch.dict(sys.modules, {"worker_app.db": _fresh_db})
+
+
 class TestLoadTemplateSegmentDurations:
     """_load_template_segment_durations 单元测试 (covers lines 198-226)."""
 
@@ -40,8 +52,7 @@ class TestLoadTemplateSegmentDurations:
         mock_session = MagicMock()
         mock_session.query.return_value = mock_query
 
-        # Patch at the source module since it's imported inside the function
-        with patch("worker_app.db.SessionLocal", return_value=mock_session):
+        with _patch_session_local(mock_session):
             result = _load_template_segment_durations("tpl_123")
 
         assert result == [5.0, 8.0, 3.0]
@@ -63,16 +74,24 @@ class TestLoadTemplateSegmentDurations:
         mock_session = MagicMock()
         mock_session.query.return_value = mock_query
 
-        with patch("worker_app.db.SessionLocal", return_value=mock_session):
+        with _patch_session_local(mock_session):
             result = _load_template_segment_durations("tpl_456")
 
         assert result == [5.0]
 
     def test_db_error_returns_empty(self):
         """数据库异常返回空列表，不抛出。"""
+        from types import ModuleType
+
         from worker_app.tasks.generation import _load_template_segment_durations
 
-        with patch("worker_app.db.SessionLocal", side_effect=Exception("DB down")):
+        _err_db = ModuleType("worker_app.db")
+
+        def _raise(*a, **kw):
+            raise Exception("DB down")
+
+        _err_db.SessionLocal = _raise
+        with patch.dict(sys.modules, {"worker_app.db": _err_db}):
             result = _load_template_segment_durations("tpl_789")
 
         assert result == []
@@ -86,7 +105,7 @@ class TestLoadTemplateSegmentDurations:
         mock_session = MagicMock()
         mock_session.query.return_value = mock_query
 
-        with patch("worker_app.db.SessionLocal", return_value=mock_session):
+        with _patch_session_local(mock_session):
             result = _load_template_segment_durations("tpl_empty")
 
         assert result == []

tests/unit/test_health_routes.py

@@ -1,7 +1,6 @@
 """Unit tests for apps/api/app/api/routes/health.py
 
 覆盖 _check_database() 和 _check_migrations() 中 psycopg3 连接逻辑。
-确保增量覆盖率 ≥ 60%（目标覆盖 lines 52, 127）。
 """
 
 from unittest.mock import AsyncMock, MagicMock, patch
@@ -9,63 +8,70 @@ from unittest.mock import AsyncMock, MagicMock, patch
 import pytest
 
 
+def _make_cursor(fetchone_result=None):
+    """Create a mock cursor with context manager support."""
+    cur = MagicMock()
+    cur.__enter__ = MagicMock(return_value=cur)
+    cur.__exit__ = MagicMock(return_value=False)
+    if fetchone_result is not None:
+        cur.fetchone.return_value = fetchone_result
+    return cur
+
+
+def _make_conn(cursor_result=None):
+    conn = MagicMock()
+    conn.cursor.return_value = cursor_result or _make_cursor()
+    return conn
+
+
 @pytest.mark.asyncio
 class TestCheckDatabase:
-    """Tests for _check_database() health check function."""
 
-    @patch("apps.api.app.api.routes.health.settings")
-    @patch("apps.api.app.api.routes.health.psycopg.connect")
-    async def test_check_database_success(self, mock_connect, mock_settings):
-        """PostgreSQL 连接成功时返回 healthy。"""
+    async def test_check_database_success(self):
+        mock_cur = _make_cursor(fetchone_result=(1,))
+        mock_conn = _make_conn(mock_cur)
+        mock_settings = MagicMock()
         mock_settings.USE_IN_MEMORY_DB = False
         mock_settings.DATABASE_URL = "postgresql+psycopg://test:test@localhost/test"
 
-        # Mock connection and cursor
-        mock_conn = MagicMock()
-        mock_cursor = MagicMock()
-        mock_cursor.__enter__ = MagicMock(return_value=mock_cursor)
-        mock_cursor.__exit__ = MagicMock(return_value=False)
-        mock_cursor.fetchone.return_value = (1,)
-        mock_conn.cursor.return_value = mock_cursor
-        mock_connect.return_value = mock_conn
+        from apps.api.app.api.routes import health
 
-        from apps.api.app.api.routes.health import _check_database
-
-        result = await _check_database()
+        with patch.object(health, "psycopg") as mock_psycopg,              patch.object(health, "settings", mock_settings):
+            mock_psycopg.connect.return_value = mock_conn
+            result = await health._check_database()
 
         assert result["status"] == "healthy"
         assert result["type"] == "postgresql"
         assert result["message"] == "Database connection successful"
-        mock_connect.assert_called_once_with(
+        mock_psycopg.connect.assert_called_once_with(
             "postgresql+psycopg://test:test@localhost/test", connect_timeout=3
         )
-        mock_cursor.execute.assert_called_once_with("SELECT 1")
+        mock_cur.execute.assert_called_once_with("SELECT 1")
         mock_conn.close.assert_called_once()
 
-    @patch("apps.api.app.api.routes.health.settings")
-    @patch("apps.api.app.api.routes.health.psycopg.connect")
-    async def test_check_database_connection_failure(self, mock_connect, mock_settings):
-        """PostgreSQL 连接失败时返回 unhealthy。"""
+    async def test_check_database_connection_failure(self):
+        mock_settings = MagicMock()
         mock_settings.USE_IN_MEMORY_DB = False
         mock_settings.DATABASE_URL = "postgresql+psycopg://test:test@localhost/test"
-        mock_connect.side_effect = Exception("connection refused")
 
-        from apps.api.app.api.routes.health import _check_database
+        from apps.api.app.api.routes import health
 
-        result = await _check_database()
+        with patch.object(health, "psycopg") as mock_psycopg,              patch.object(health, "settings", mock_settings):
+            mock_psycopg.connect.side_effect = Exception("connection refused")
+            result = await health._check_database()
 
         assert result["status"] == "unhealthy"
         assert result["type"] == "postgresql"
         assert "connection refused" in result["message"]
 
-    @patch("apps.api.app.api.routes.health.settings")
-    async def test_check_database_in_memory(self, mock_settings):
-        """使用内存数据库时跳过 PostgreSQL 检查。"""
+    async def test_check_database_in_memory(self):
+        mock_settings = MagicMock()
         mock_settings.USE_IN_MEMORY_DB = True
 
-        from apps.api.app.api.routes.health import _check_database
+        from apps.api.app.api.routes import health
 
-        result = await _check_database()
+        with patch.object(health, "settings", mock_settings):
+            result = await health._check_database()
 
         assert result["status"] == "healthy"
         assert result["type"] == "in_memory"
@@ -73,80 +79,66 @@ class TestCheckDatabase:
 
 @pytest.mark.asyncio
 class TestCheckMigrations:
-    """Tests for _check_migrations() health check function."""
 
-    @patch("apps.api.app.api.routes.health.settings")
-    @patch("apps.api.app.api.routes.health.psycopg.connect")
-    async def test_check_migrations_success(self, mock_connect, mock_settings):
-        """所有迁移表存在时返回 healthy。"""
+    async def test_check_migrations_success(self):
+        mock_cur = _make_cursor(fetchone_result=(5,))
+        mock_conn = _make_conn(mock_cur)
+        mock_settings = MagicMock()
         mock_settings.USE_IN_MEMORY_DB = False
         mock_settings.DATABASE_URL = "postgresql+psycopg://test:test@localhost/test"
 
-        mock_conn = MagicMock()
-        mock_cursor = MagicMock()
-        mock_cursor.__enter__ = MagicMock(return_value=mock_cursor)
-        mock_cursor.__exit__ = MagicMock(return_value=False)
-        mock_cursor.fetchone.return_value = (5,)  # 5 tables found
-        mock_conn.cursor.return_value = mock_cursor
-        mock_connect.return_value = mock_conn
+        from apps.api.app.api.routes import health
 
-        from apps.api.app.api.routes.health import _check_migrations
-
-        result = await _check_migrations()
+        with patch.object(health, "psycopg") as mock_psycopg,              patch.object(health, "settings", mock_settings):
+            mock_psycopg.connect.return_value = mock_conn
+            result = await health._check_migrations()
 
         assert result["status"] == "healthy"
         assert result["message"] == "Database migrations applied"
-        mock_connect.assert_called_once_with(
+        mock_psycopg.connect.assert_called_once_with(
             "postgresql+psycopg://test:test@localhost/test", connect_timeout=3
         )
         mock_conn.close.assert_called_once()
 
-    @patch("apps.api.app.api.routes.health.settings")
-    @patch("apps.api.app.api.routes.health.psycopg.connect")
-    async def test_check_migrations_missing_tables(self, mock_connect, mock_settings):
-        """迁移表不完整时返回 unhealthy。"""
+    async def test_check_migrations_missing_tables(self):
+        mock_cur = _make_cursor(fetchone_result=(2,))
+        mock_conn = _make_conn(mock_cur)
+        mock_settings = MagicMock()
         mock_settings.USE_IN_MEMORY_DB = False
         mock_settings.DATABASE_URL = "postgresql+psycopg://test:test@localhost/test"
 
-        mock_conn = MagicMock()
-        mock_cursor = MagicMock()
-        mock_cursor.__enter__ = MagicMock(return_value=mock_cursor)
-        mock_cursor.__exit__ = MagicMock(return_value=False)
-        mock_cursor.fetchone.return_value = (2,)  # Only 2 of 5 tables
-        mock_conn.cursor.return_value = mock_cursor
-        mock_connect.return_value = mock_conn
+        from apps.api.app.api.routes import health
 
-        from apps.api.app.api.routes.health import _check_migrations
-
-        result = await _check_migrations()
+        with patch.object(health, "psycopg") as mock_psycopg,              patch.object(health, "settings", mock_settings):
+            mock_psycopg.connect.return_value = mock_conn
+            result = await health._check_migrations()
 
         assert result["status"] == "unhealthy"
         assert "Missing tables" in result["message"]
         assert "2/5" in result["message"]
 
-    @patch("apps.api.app.api.routes.health.settings")
-    @patch("apps.api.app.api.routes.health.psycopg.connect")
-    async def test_check_migrations_connection_failure(self, mock_connect, mock_settings):
-        """数据库连接失败时返回 unhealthy。"""
+    async def test_check_migrations_connection_failure(self):
+        mock_settings = MagicMock()
         mock_settings.USE_IN_MEMORY_DB = False
         mock_settings.DATABASE_URL = "postgresql+psycopg://test:test@localhost/test"
-        mock_connect.side_effect = Exception("connection refused")
 
-        from apps.api.app.api.routes.health import _check_migrations
+        from apps.api.app.api.routes import health
 
-        result = await _check_migrations()
+        with patch.object(health, "psycopg") as mock_psycopg,              patch.object(health, "settings", mock_settings):
+            mock_psycopg.connect.side_effect = Exception("connection refused")
+            result = await health._check_migrations()
 
         assert result["status"] == "unhealthy"
         assert "Migration check failed" in result["message"]
 
-    @patch("apps.api.app.api.routes.health.settings")
-    async def test_check_migrations_in_memory(self, mock_settings):
-        """使用内存数据库时跳过迁移检查。"""
+    async def test_check_migrations_in_memory(self):
+        mock_settings = MagicMock()
         mock_settings.USE_IN_MEMORY_DB = True
 
-        from apps.api.app.api.routes.health import _check_migrations
+        from apps.api.app.api.routes import health
 
-        result = await _check_migrations()
+        with patch.object(health, "settings", mock_settings):
+            result = await health._check_migrations()
 
         assert result["status"] == "healthy"
         assert "no migrations needed" in result["message"]
@@ -154,33 +146,27 @@ class TestCheckMigrations:
 
 @pytest.mark.asyncio
 class TestStartupCheck:
-    """Tests for startup_check() endpoint."""
 
-    @patch("apps.api.app.api.routes.health._check_migrations")
-    @patch("apps.api.app.api.routes.health._check_database")
-    async def test_startup_all_healthy(self, mock_db, mock_mig):
-        """所有检查通过时返回 started。"""
-        mock_db.return_value = {"status": "healthy"}
-        mock_mig.return_value = {"status": "healthy"}
+    async def test_startup_all_healthy(self):
+        from apps.api.app.api.routes import health
 
-        from apps.api.app.api.routes.health import startup_check
-
-        result = await startup_check()
+        with patch.object(health, "_check_migrations", new_callable=AsyncMock) as mock_mig,              patch.object(health, "_check_database", new_callable=AsyncMock) as mock_db:
+            mock_db.return_value = {"status": "healthy"}
+            mock_mig.return_value = {"status": "healthy"}
+            result = await health.startup_check()
 
         assert result["status"] == "started"
 
-    @patch("apps.api.app.api.routes.health._check_migrations")
-    @patch("apps.api.app.api.routes.health._check_database")
-    async def test_startup_db_unhealthy(self, mock_db, mock_mig):
-        """数据库不健康时返回 starting + 503。"""
-        mock_db.return_value = {"status": "unhealthy", "message": "fail"}
-        mock_mig.return_value = {"status": "healthy"}
+    async def test_startup_db_unhealthy(self):
+        import json
 
-        from apps.api.app.api.routes.health import startup_check
+        from apps.api.app.api.routes import health
 
-        result = await startup_check()
+        with patch.object(health, "_check_migrations", new_callable=AsyncMock) as mock_mig,              patch.object(health, "_check_database", new_callable=AsyncMock) as mock_db:
+            mock_db.return_value = {"status": "unhealthy", "message": "fail"}
+            mock_mig.return_value = {"status": "healthy"}
+            result = await health.startup_check()
 
         assert result.status_code == 503
-        import json
         body = json.loads(result.body)
         assert body["status"] == "starting"