test(api): add unit tests for viral_video celery send_task call sites

Cover the 4 dispatch lines that missed diff coverage on PR #2105:
- generate  -> worker.run_viral_video_pipeline
- retry     -> worker.run_viral_video_pipeline
- confirm-intent -> worker.resume_viral_video_pipeline
- analyze-style  -> worker.run_video_style_analysis

tests/unit/test_viral_video_routes.py

@@ -0,0 +1,174 @@
+"""viral_video.py HTTP 端点单元测试（celery send_task 分支覆盖）。
+
+直接调用路由函数（不启动 TestClient），通过 patch 注入 repo/session/user，
+覆盖 4 个 celery_app.send_task(...) 调用点：
+
+- create_viral_video (generate)    -> worker.run_viral_video_pipeline
+- retry_viral_video_job (retry)    -> worker.run_viral_video_pipeline
+- confirm_intent                   -> worker.resume_viral_video_pipeline
+- analyze_style                    -> worker.run_video_style_analysis
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+
+def _auth_user(uid: str = "u1"):
+    return SimpleNamespace(user=SimpleNamespace(id=uid))
+
+
+def _make_job(job_id: str = "job-1", user_id: str = "u1", status: str = "pending", **kwargs):
+    from packages.domain.viral_video import ViralVideoStatus
+
+    job = MagicMock()
+    job.id = job_id
+    job.user_id = user_id
+    job.status = ViralVideoStatus(status) if isinstance(status, str) else status
+    job.images = kwargs.pop("images", ["img-1"])
+    job.industry = kwargs.pop("industry", "电商")
+    job.target_customer = kwargs.pop("target_customer", "年轻人")
+    for k, v in {
+        "persona_id": "",
+        "viral_structure": "",
+        "marketing_purpose": "",
+        "bgm_preference": "",
+        "duration": 30,
+        "user_copy_text": "",
+        "fusion_level": "ai_polish",
+        "reference_audio_path": "",
+        "reference_video_url": "",
+        "style_strength": "medium",
+        "style_template_id": "",
+        "retry_count": 0,
+        "error_msg": "",
+        "result_video_url": "",
+        "style_guide": None,
+        "created_at": None,
+        "started_at": None,
+        "completed_at": None,
+        "stage": "",
+        "progress": 0.0,
+        "intent_result": None,
+        "updated_at": None,
+    }.items():
+        setattr(job, k, kwargs.pop(k, v))
+    return job
+
+
+# ── generate ────────────────────────────────────────────────────────────
+
+
+class TestCreateViralVideo:
+    def _req(self, **kw):
+        from app.schemas.viral_video import CreateViralVideoRequest
+
+        d = {"images": ["https://x.com/a.jpg"], "industry": "电商", "target_customer": "年轻人"}
+        d.update(kw)
+        return CreateViralVideoRequest(**d)
+
+    def test_generate_dispatches_celery_task(self):
+        from app.api.routes import viral_video as vv_mod
+
+        req = self._req()
+        user = _auth_user("u1")
+        session = MagicMock()
+        saved_job = _make_job(job_id="job-new", user_id="u1", status="pending")
+        repo = MagicMock()
+
+        def fake_save(job):
+            job.id = saved_job.id
+
+        repo.save.side_effect = fake_save
+
+        with (
+            patch.object(vv_mod, "_get_job_repo", return_value=repo),
+            patch.object(vv_mod.celery_app, "send_task") as mock_send,
+        ):
+            resp = vv_mod.create_viral_video(req, authenticated_user=user, session=session)
+
+        mock_send.assert_called_once_with("worker.run_viral_video_pipeline", args=[saved_job.id])
+        assert resp.id == saved_job.id
+
+
+# ── retry ───────────────────────────────────────────────────────────────
+
+
+class TestRetryViralVideo:
+    def test_retry_dispatches_celery_task(self):
+        from packages.domain.viral_video import ViralVideoStatus
+        from app.api.routes import viral_video as vv_mod
+
+        user = _auth_user("u1")
+        session = MagicMock()
+        job = _make_job(job_id="job-retry", user_id="u1", status=ViralVideoStatus.FAILED, retry_count=1)
+        repo = MagicMock()
+        repo.get.return_value = job
+
+        with (
+            patch.object(vv_mod, "_get_job_repo", return_value=repo),
+            patch.object(vv_mod.celery_app, "send_task") as mock_send,
+        ):
+            resp = vv_mod.retry_viral_video_job("job-retry", authenticated_user=user, session=session)
+
+        assert job.status == ViralVideoStatus.PENDING
+        assert job.retry_count == 2
+        mock_send.assert_called_once_with("worker.run_viral_video_pipeline", args=["job-retry"])
+        assert resp.id == "job-retry"
+
+
+# ── confirm-intent ──────────────────────────────────────────────────────
+
+
+class TestConfirmIntent:
+    def test_confirm_intent_dispatches_resume_task(self):
+        from packages.domain.viral_video import ViralVideoStatus
+        from app.api.routes import viral_video as vv_mod
+        from app.schemas.viral_video import ConfirmIntentRequest
+
+        user = _auth_user("u1")
+        session = MagicMock()
+        job = _make_job(job_id="job-cfm", user_id="u1", status=ViralVideoStatus.WAIT_USER_CONFIRM)
+        repo = MagicMock()
+        repo.get.return_value = job
+        req = ConfirmIntentRequest(confirmed_copy="确认后的文案")
+
+        with (
+            patch.object(vv_mod, "_get_job_repo", return_value=repo),
+            patch.object(vv_mod.celery_app, "send_task") as mock_send,
+        ):
+            resp = vv_mod.confirm_intent("job-cfm", req, authenticated_user=user, session=session)
+
+        assert job.user_copy_text == "确认后的文案"
+        job.resume_from_confirm.assert_called_once()
+        mock_send.assert_called_once_with("worker.resume_viral_video_pipeline", args=["job-cfm"])
+        assert resp.id == "job-cfm"
+
+
+# ── analyze-style ───────────────────────────────────────────────────────
+
+
+class TestAnalyzeStyle:
+    def test_analyze_style_dispatches_analysis_task(self):
+        from app.api.routes import viral_video as vv_mod
+        from app.schemas.viral_video import AnalyzeStyleRequest
+
+        user = _auth_user("u1")
+        session = MagicMock()
+        job = _make_job(job_id="job-sty", user_id="u1", status="pending")
+        repo = MagicMock()
+        repo.get.return_value = job
+        req = AnalyzeStyleRequest(reference_video_url="https://x.com/ref.mp4", style_template_id="tpl-1")
+
+        with (
+            patch.object(vv_mod, "_get_job_repo", return_value=repo),
+            patch.object(vv_mod.celery_app, "send_task") as mock_send,
+        ):
+            resp = vv_mod.analyze_style("job-sty", req, authenticated_user=user, session=session)
+
+        assert job.reference_video_url == "https://x.com/ref.mp4"
+        assert job.style_template_id == "tpl-1"
+        mock_send.assert_called_once_with("worker.run_video_style_analysis", args=["job-sty"])
+        assert resp.job_id == "job-sty"
+        assert resp.status == "analyzing"