style: auto-format with black + isort + prettier [skip ci-format-check]

使用 FakeGenTask 替代 MagicMock，通过追踪 _assigned 字典
准确判断 cover_url 是否被显式设置。

apps/worker/worker_app/tasks/edit_plan_generation.py

@@ -98,6 +98,7 @@ def _finalize_render_success(
     output_path: Path,
     engine: str,
     thumbnail_url: str = "",
+    cover_candidates: list[dict] | None = None,
 ) -> dict:
     """渲染成功后的统一收尾：查重 + 更新状态 + 返回结果。"""
     # 创建 GeneratedVideo 记录 + 查重
@@ -163,6 +164,19 @@ def _finalize_render_success(
                 clip_count=len(rendered_clip_ids),
             )
             gen_task.completed_at = datetime.now(timezone.utc)
+
+            # 回写封面 URL 到 GenerationTask，供封面生成接口读取
+            if cover_candidates:
+                first_cover = cover_candidates[0].get("image_url") or cover_candidates[0].get("url") or ""
+                if first_cover:
+                    gen_task.cover_url = first_cover
+                    logger.info(
+                        "预览渲染完成，回写 cover_url: plan_id=%s task_id=%s url=%s",
+                        plan_id,
+                        generation_task_id,
+                        first_cover[:80],
+                    )
+
             gen_task_repo.update(gen_task)
 
     logger.info(
@@ -288,6 +302,7 @@ def _render_with_unified(
         output_path=output_path,
         engine="unified",
         thumbnail_url=thumbnail_url,
+        cover_candidates=result.cover_candidates,
     )
 
 

tests/unit/test_cover_url_finalize.py

@@ -0,0 +1,132 @@
+"""Tests for cover_url backfill to GenerationTask.
+
+Verifies _finalize_render_success correctly writes cover_url
+from cover_candidates to gen_task.cover_url.
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+# Add worker app to sys.path
+_WORKER_ROOT = Path(__file__).resolve().parents[2] / "apps" / "worker"
+if str(_WORKER_ROOT) not in sys.path:
+    sys.path.insert(0, str(_WORKER_ROOT))
+
+
+class FakeGenTask:
+    """Simple stand-in for GenerationTask that tracks attribute assignment."""
+
+    def __init__(self):
+        object.__setattr__(self, "_assigned", {})
+        self.id = "task-1"
+        self.status = MagicMock()
+        self.status.value = "running"
+
+    def __setattr__(self, name, value):
+        if not name.startswith("_"):
+            self._assigned[name] = value
+        object.__setattr__(self, name, value)
+
+    def append_log(self, **kwargs):
+        pass
+
+
+def _make_plan():
+    plan = MagicMock()
+    plan.project_id = "proj-1"
+    plan.created_by_user_id = "user-1"
+    plan.config = {"batch_id": "batch-1", "mode": "edit_plan", "title": {"text": "test"}}
+    plan.mark_completed = MagicMock()
+    return plan
+
+
+def _call_finalize(cover_candidates=None, gen_task=None, plan=None):
+    from worker_app.tasks.edit_plan_generation import _finalize_render_success
+
+    plan = plan or _make_plan()
+    gen_task = gen_task or FakeGenTask()
+
+    plan_repo = MagicMock()
+    clip_repo = MagicMock()
+    gen_task_repo = MagicMock()
+    gen_task_repo.get.return_value = gen_task
+    db = MagicMock()
+
+    with patch("worker_app.tasks.edit_plan_generation.create_video_record_and_dedup"):
+        result = _finalize_render_success(
+            plan=plan,
+            plan_repo=plan_repo,
+            clip_repo=clip_repo,
+            gen_task_repo=gen_task_repo,
+            db=db,
+            plan_id="plan-1",
+            output_url="https://oss.example.com/output.mp4",
+            storage_key="rendered/plan-1/task-1.mp4",
+            duration=10.0,
+            file_size=1024,
+            width=1280,
+            height=720,
+            rendered_clip_ids=["clip-1"],
+            failed_clip_ids=[],
+            generation_task_id="task-1",
+            output_path=Path("/tmp/output.mp4"),
+            engine="unified",
+            thumbnail_url="",
+            cover_candidates=cover_candidates,
+        )
+
+    return result, gen_task, gen_task_repo
+
+
+class TestFinalizeCoverUrl:
+
+    def test_cover_url_set_from_image_url(self):
+        """cover_candidates with image_url should set gen_task.cover_url"""
+        candidates = [
+            {"image_url": "https://oss.example.com/cover1.jpg", "frame_time": 1.5},
+            {"image_url": "https://oss.example.com/cover2.jpg", "frame_time": 3.0},
+        ]
+        _, gen_task, gen_task_repo = _call_finalize(cover_candidates=candidates)
+        assert gen_task.cover_url == "https://oss.example.com/cover1.jpg"
+        gen_task_repo.update.assert_called()
+
+    def test_cover_url_fallback_to_url_key(self):
+        """Should fallback to 'url' key when 'image_url' is absent"""
+        candidates = [{"url": "https://oss.example.com/cover_url_key.jpg"}]
+        _, gen_task, _ = _call_finalize(cover_candidates=candidates)
+        assert gen_task.cover_url == "https://oss.example.com/cover_url_key.jpg"
+
+    def test_cover_url_not_set_when_empty_list(self):
+        """Empty cover_candidates should not set cover_url"""
+        _, gen_task, _ = _call_finalize(cover_candidates=[])
+        assert "cover_url" not in gen_task._assigned
+
+    def test_cover_url_not_set_when_none(self):
+        """None cover_candidates should not set cover_url"""
+        _, gen_task, _ = _call_finalize(cover_candidates=None)
+        assert "cover_url" not in gen_task._assigned
+
+    def test_cover_url_not_set_when_url_empty(self):
+        """Empty URL strings in candidates should not set cover_url"""
+        candidates = [{"image_url": "", "url": ""}]
+        _, gen_task, _ = _call_finalize(cover_candidates=candidates)
+        assert "cover_url" not in gen_task._assigned
+
+    def test_no_generation_task_no_crash(self):
+        """Should not crash when gen_task is None"""
+        candidates = [{"image_url": "https://oss.example.com/cover.jpg"}]
+        gen_task_repo = MagicMock()
+        gen_task_repo.get.return_value = None
+        result, _, _ = _call_finalize(cover_candidates=candidates)
+        assert result["status"] == "completed"
+
+    def test_image_url_priority_over_url(self):
+        """image_url should take priority over url key"""
+        candidates = [{"image_url": "https://a.jpg", "url": "https://b.jpg"}]
+        _, gen_task, _ = _call_finalize(cover_candidates=candidates)
+        assert gen_task.cover_url == "https://a.jpg"