From 8092d60111f72a576bae49cc9272b02f8f8cc8c7 Mon Sep 17 00:00:00 2001 From: xiaoxia-agent Date: Tue, 15 Sep 2026 02:10:31 +0800 Subject: [PATCH 1/5] =?UTF-8?q?fix(backend):=20#1867=20=E5=AF=B9=E5=8F=A3?= =?UTF-8?q?=E5=9E=8B=E5=8D=A1running=E4=B8=89=E5=B1=82=E9=98=B2=E5=BE=A1?= =?UTF-8?q?=E4=BF=AE=E5=A4=8D=EF=BC=88rebase=E5=88=B0=E6=9C=80=E6=96=B0dev?= =?UTF-8?q?elop=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 三层防御: 1. 前端轮询 GET /lipsync/jobs/{id}：距上次更新超过30秒时同步刷新一次（避免 background task 静默失败永久卡 running），否则后台异步刷新 2. refresh_job_status：DB commit 包 try/except 并 rollback，防止 DB 提交失败导致后台任务静默失败 3. TTS Celery 任务：MediaKit 提交成功后链式触发 poll_mediakit_status 兜底轮询任务（worker主动拉状态，最多60次×10s=10分钟，不依赖前端轮询） 解决冲突: rebase 到最新 develop（含 #1845/#1849/#1864 等已合并的lipsync优化），保留同步转存/异步转存/预合成等已有功能 Refs #1867 --- apps/api/app/api/routes/lipsync.py | 17 +++- apps/api/app/services/lipsync_service.py | 73 +++++++++------- apps/api/app/tasks/lipsync_tts.py | 104 +++++++++++++++++++++++ 3 files changed, 163 insertions(+), 31 deletions(-) diff --git a/apps/api/app/api/routes/lipsync.py b/apps/api/app/api/routes/lipsync.py index f4944ec62..eb045be7a 100644 --- a/apps/api/app/api/routes/lipsync.py +++ b/apps/api/app/api/routes/lipsync.py @@ -193,7 +193,22 @@ def get_lipsync_job( raise HTTPException(status_code=404, detail="任务不存在") if job.status not in ("completed", "failed"): - background.add_task(svc.refresh_job_status, job_id, current_user.user.id) + # 三层防御 ①：如果距上次更新超过 30 秒，同步刷新一次（避免 background task + # 静默失败导致前端永远看到 running）；否则挂后台异步刷新（避免阻塞轮询）。 + from datetime import datetime as _dt, timezone as _tz + + _now = _dt.now(_tz.utc) + _stale = job.updated_at is None or (_now - job.updated_at).total_seconds() > 30 + if _stale: + try: + refreshed = svc.refresh_job_status(job_id, current_user.user.id) + if refreshed is not None: + job = refreshed + except Exception as exc: # noqa: BLE001 + logger.error("同步刷新对口型状态失败 job_id=%s err=%s", job_id, exc, exc_info=True) + background.add_task(svc.refresh_job_status, job_id, current_user.user.id) + else: + background.add_task(svc.refresh_job_status, job_id, current_user.user.id) return job diff --git a/apps/api/app/services/lipsync_service.py b/apps/api/app/services/lipsync_service.py index a5f3fb12c..e39c5462a 100644 --- a/apps/api/app/services/lipsync_service.py +++ b/apps/api/app/services/lipsync_service.py @@ -489,39 +489,52 @@ class LipsyncService: mk_status = status_data.get("status", STATUS_RUNNING) logger.info("MediaKit 对口型状态 [%s]: %s", job_id, mk_status) - if mk_status == STATUS_COMPLETED: - result = status_data.get("result", {}) - job.status = STATUS_COMPLETED - temp_url = result.get("video_url", "") - job.output_video_url = temp_url - job.output_duration = result.get("duration", 0.0) - job.completed_at = datetime.now(timezone.utc) + try: + if mk_status == STATUS_COMPLETED: + result = status_data.get("result", {}) + job.status = STATUS_COMPLETED + temp_url = result.get("video_url", "") + job.output_video_url = temp_url + job.output_duration = result.get("duration", 0.0) + job.completed_at = datetime.now(timezone.utc) + job.updated_at = datetime.now(timezone.utc) + self.db.commit() + # 异步转存自家 OSS + try: + from app.tasks.lipsync_tts import persist_output_video_task + + persist_output_video_task.apply_async(args=(job_id, user_id, temp_url)) + except Exception as exc: + logger.warning( + "提交输出视频异步转存任务失败，保留临时 URL: job_id=%s err=%s", + job_id, + exc, + ) + elif mk_status == STATUS_FAILED: + error = status_data.get("error", {}) + job.status = "failed" + job.error_message = error.get("message", "任务执行失败") + job.error_code = error.get("code", "TaskFailed") + job.completed_at = datetime.now(timezone.utc) + else: + # 中间状态（running/processing/queued 等）同步到 DB，避免前端永远卡在 submitted + if isinstance(mk_status, str) and mk_status: + job.status = mk_status job.updated_at = datetime.now(timezone.utc) self.db.commit() - # 异步转存自家 OSS + except Exception as exc: # noqa: BLE001 - DB 提交失败必须记录日志并重试，否则后台任务静默失败 + logger.error( + "refresh_job_status 提交 DB 失败 job_id=%s mk_status=%s err=%s", + job_id, + mk_status, + exc, + exc_info=True, + ) try: - from app.tasks.lipsync_tts import persist_output_video_task - - persist_output_video_task.apply_async(args=(job_id, user_id, temp_url)) - except Exception as exc: - logger.warning( - "提交输出视频异步转存任务失败，保留临时 URL: job_id=%s err=%s", - job_id, - exc, - ) - self.db.refresh(job) - return job - elif mk_status == STATUS_FAILED: - error = status_data.get("error", {}) - job.status = "failed" - job.error_message = error.get("message", "任务执行失败") - job.error_code = error.get("code", "TaskFailed") - job.completed_at = datetime.now(timezone.utc) - else: - if isinstance(mk_status, str) and mk_status: - job.status = mk_status - job.updated_at = datetime.now(timezone.utc) - self.db.commit() + self.db.rollback() + except Exception: + pass + # DB commit 失败不 raise，返回当前 job 对象让下次轮询再试 self.db.refresh(job) return job diff --git a/apps/api/app/tasks/lipsync_tts.py b/apps/api/app/tasks/lipsync_tts.py index 2f75e9b41..109899582 100644 --- a/apps/api/app/tasks/lipsync_tts.py +++ b/apps/api/app/tasks/lipsync_tts.py @@ -282,6 +282,18 @@ def tts_synthesize_and_submit( job.error_code = exc.code logger.error("[lipsync_tts] 提交 MediaKit 失败: job_id=%s err=%s", job_id, exc) + # 三层防御 ③：链式触发 Celery 兜底轮询——MediaKit 提交成功后由 worker + # 主动拉取状态到终态，不依赖前端轮询触发的 FastAPI background task + # （background task 可能静默失败导致永久卡 running）。 + if job.status == "submitted" and job.mediakit_task_id: + try: + poll_mediakit_status.apply_async( + kwargs={"job_id": job_id, "user_id": user_id}, + countdown=10, # 10 秒后开始轮询，给 MediaKit 一点处理时间 + ) + except Exception as exc: # noqa: BLE001 + logger.warning("[lipsync_tts] 提交兜底轮询任务失败（不影响主流程）: job_id=%s err=%s", job_id, exc) + db.commit() except Exception: @@ -300,6 +312,98 @@ def tts_synthesize_and_submit( db.close() +@shared_task( + bind=True, + name="lipsync_tts.poll_mediakit_status", + max_retries=60, # 最多轮询 60 次 + default_retry_delay=10, # 每次间隔 10 秒（总兜底时长 10 分钟） +) +def poll_mediakit_status(self, job_id: str, user_id: str): + """Celery 兜底轮询：TTS 提交 MediaKit 后，由 worker 主动拉取状态直到终态。 + + 不依赖前端轮询，避免 background task 静默失败导致任务永久卡 running/submitted。 + """ + from sqlalchemy.orm import Session as DBSession + + from packages.adapters.sqlalchemy_impl.models import LipsyncJobModel + + try: + from worker_app.db import SessionLocal # type: ignore + except Exception: # noqa: BLE001 + from app.db import SessionLocal # type: ignore + + db: DBSession = SessionLocal() + try: + job = db.query(LipsyncJobModel).filter(LipsyncJobModel.id == job_id, LipsyncJobModel.user_id == user_id).first() + if job is None: + logger.warning("[lipsync_poll] Job not found: job_id=%s", job_id) + return + + # 已终态，不需要再轮询 + if job.status in ("completed", "failed", "cancelled"): + return + + if not job.mediakit_task_id: + logger.warning("[lipsync_poll] Job has no mediakit_task_id: job_id=%s status=%s", job_id, job.status) + return + + from app.services.lipsync_service import STATUS_COMPLETED as _SC, STATUS_FAILED as _SF + from app.services.lipsync_service import LipsyncService + from app.services.mediakit_client import MediaKitError, get_mediakit_client + + client = get_mediakit_client() + try: + status_data = client.get_task_status(job.mediakit_task_id) + except MediaKitError as exc: + logger.warning("[lipsync_poll] 拉取 MediaKit 状态失败，将重试: job_id=%s err=%s", job_id, exc) + raise self.retry(exc=exc) + + mk_status = status_data.get("status", "running") + + if mk_status in ("succeeded", _SC): + + svc = LipsyncService(db) + result = status_data.get("result", {}) + job.status = "completed" + output_url = result.get("video_url", "") + try: + job.output_video_url = svc._persist_output_video(output_url, job_id, user_id) + except Exception as exc: # noqa: BLE001 + logger.warning("[lipsync_poll] 转存 OSS 失败，保留临时 URL: job_id=%s err=%s", job_id, exc) + job.output_video_url = output_url + job.output_duration = result.get("duration", 0.0) + job.completed_at = datetime.now(timezone.utc) + job.updated_at = datetime.now(timezone.utc) + db.commit() + logger.info("[lipsync_poll] 任务完成: job_id=%s", job_id) + elif mk_status in ("failed", "error", _SF): + error = status_data.get("error", {}) + job.status = "failed" + job.error_message = error.get("message", "任务执行失败") + job.error_code = error.get("code", "TaskFailed") + job.completed_at = datetime.now(timezone.utc) + job.updated_at = datetime.now(timezone.utc) + db.commit() + logger.info("[lipsync_poll] 任务失败: job_id=%s err=%s", job_id, job.error_message) + else: + # 中间状态，更新时间戳，继续重试 + job.updated_at = datetime.now(timezone.utc) + if isinstance(mk_status, str) and mk_status: + job.status = mk_status + db.commit() + logger.debug("[lipsync_poll] 任务仍在 %s，继续轮询: job_id=%s", mk_status, job_id) + raise self.retry() + except Exception as exc: + logger.exception("[lipsync_poll] 未预期异常: job_id=%s", job_id) + try: + db.rollback() + except Exception: + pass + raise self.retry(exc=exc) + finally: + db.close() + + @shared_task( name="lipsync_tts.persist_output_video", max_retries=2, -- 2.54.0 From 425018ecc0bca1e34674dc47c3c752ec4767f7e6 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Mon, 14 Sep 2026 18:15:01 +0000 Subject: [PATCH 2/5] style: auto-format with black + isort + ruff + prettier [skip ci-format-check] --- apps/api/app/api/routes/lipsync.py | 3 ++- apps/api/app/tasks/lipsync_tts.py | 3 ++- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/apps/api/app/api/routes/lipsync.py b/apps/api/app/api/routes/lipsync.py index eb045be7a..dd4d4b125 100644 --- a/apps/api/app/api/routes/lipsync.py +++ b/apps/api/app/api/routes/lipsync.py @@ -195,7 +195,8 @@ def get_lipsync_job( if job.status not in ("completed", "failed"): # 三层防御 ①：如果距上次更新超过 30 秒，同步刷新一次（避免 background task # 静默失败导致前端永远看到 running）；否则挂后台异步刷新（避免阻塞轮询）。 - from datetime import datetime as _dt, timezone as _tz + from datetime import datetime as _dt + from datetime import timezone as _tz _now = _dt.now(_tz.utc) _stale = job.updated_at is None or (_now - job.updated_at).total_seconds() > 30 diff --git a/apps/api/app/tasks/lipsync_tts.py b/apps/api/app/tasks/lipsync_tts.py index 109899582..3e106fd60 100644 --- a/apps/api/app/tasks/lipsync_tts.py +++ b/apps/api/app/tasks/lipsync_tts.py @@ -347,7 +347,8 @@ def poll_mediakit_status(self, job_id: str, user_id: str): logger.warning("[lipsync_poll] Job has no mediakit_task_id: job_id=%s status=%s", job_id, job.status) return - from app.services.lipsync_service import STATUS_COMPLETED as _SC, STATUS_FAILED as _SF + from app.services.lipsync_service import STATUS_COMPLETED as _SC + from app.services.lipsync_service import STATUS_FAILED as _SF from app.services.lipsync_service import LipsyncService from app.services.mediakit_client import MediaKitError, get_mediakit_client -- 2.54.0 From 4c4d4b76c59b776a72884170fd220ee6ddc73ad0 Mon Sep 17 00:00:00 2001 From: xiaoxia-agent Date: Tue, 15 Sep 2026 02:57:57 +0800 Subject: [PATCH 3/5] =?UTF-8?q?test(#1867):=20=E8=A1=A5=E5=85=85=E4=B8=89?= =?UTF-8?q?=E5=B1=82=E9=98=B2=E5=BE=A1=E5=8D=95=E6=B5=8B=E6=8F=90=E8=A6=86?= =?UTF-8?q?=E7=9B=96=E7=8E=87=E5=88=B0=20CI=20=E9=98=88=E5=80=BC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Layer 1 (routes): stale>30s 同步刷新 / fresh 后台刷新 / completed 跳过 / 同步异常 fallback 到 background / 404 共 5 个分支 - Layer 2 (service): DB commit 异常 rollback + 中间态 commit + failed 写 error_info 共 3 个分支 - Layer 3 (worker): poll_mediakit_status 新任务 10 个分支 (not found / 终态跳过 / 无 mk_task_id / succeeded→completed+persist / persist 失败回退 temp_url / failed / error / running→retry) + tts 提交成功链式触发 poll 3 个分支 (成功 dispatch / 失败不 dispatch / dispatch 异常不影响主流程) - 共 21 个新测试，全量 15126 passed。 [skip ci-format-check] --- tests/unit/test_lipsync_routes.py | 129 +++++++++-- tests/unit/test_lipsync_tts.py | 348 ++++++++++++++++++++++++++++++ 2 files changed, 465 insertions(+), 12 deletions(-) diff --git a/tests/unit/test_lipsync_routes.py b/tests/unit/test_lipsync_routes.py index 9bfbcc735..2a8ad758b 100644 --- a/tests/unit/test_lipsync_routes.py +++ b/tests/unit/test_lipsync_routes.py @@ -227,9 +227,7 @@ class TestLipsyncServiceUnit: mock_repo = MagicMock() mock_repo.get.return_value = None # 预置音色，原样返回 voice_id - with patch( - "app.services.lipsync_service.tts_synthesize_and_submit" - ) as mock_task: + with patch("app.services.lipsync_service.tts_synthesize_and_submit") as mock_task: mock_task.apply_async.return_value = MagicMock(id="celery-task-123") svc = LipsyncService( @@ -275,9 +273,7 @@ class TestLipsyncServiceUnit: voice_clone_repo=mock_repo, ) - with patch( - "app.services.lipsync_service.tts_synthesize_and_submit" - ) as mock_task: + with patch("app.services.lipsync_service.tts_synthesize_and_submit") as mock_task: mock_task.apply_async.return_value = MagicMock(id="celery-task-456") job = svc.create_job( @@ -444,9 +440,7 @@ class TestLipsyncServiceUnit: mock_repo = MagicMock() mock_repo.get.return_value = None - with patch( - "app.services.lipsync_service.tts_synthesize_and_submit" - ) as mock_task: + with patch("app.services.lipsync_service.tts_synthesize_and_submit") as mock_task: mock_task.apply_async.return_value = MagicMock(id="celery-task-789") svc = LipsyncService( @@ -568,9 +562,7 @@ class TestErrorHandling: voice_clone_repo=mock_repo, ) - with patch( - "app.services.lipsync_service.tts_synthesize_and_submit" - ) as mock_task: + with patch("app.services.lipsync_service.tts_synthesize_and_submit") as mock_task: mock_task.apply_async.return_value = MagicMock(id="celery-task-789") # TTS 模式下 create_job 不再同步失败 @@ -718,3 +710,116 @@ class TestSignMediaUrl403Fix: kw = mock_mediakit.submit_lipsync.call_args.kwargs assert kw["video_url"] == signed_video assert kw["audio_url"] == signed_audio + + +class TestLipsyncRouteStaleRefresh: + """#1867 三层防御 ①：路由层 stale>30s 同步刷新 / 新鲜走后台 / 异常 fallback 单测. + + 直接调用路由函数 get_lipsync_job（不经过 TestClient/HTTP 栈），mock 掉 + current_user、svc、background 三个依赖，覆盖 stale/sync/exception/terminal/404 五个分支。 + """ + + def _call(self, job, svc, background=None): + from app.api.routes.lipsync import get_lipsync_job + + user = MagicMock() + user.user.id = "user-1" + bg = background if background is not None else MagicMock() + return get_lipsync_job(job_id="job-1", background=bg, current_user=user, svc=svc) + + def test_stale_job_triggers_sync_refresh(self, mock_mediakit, mock_cosyvoice): + """updated_at 超过 30 秒 → 同步调用 svc.refresh_job_status，不走 background.""" + from datetime import datetime, timedelta, timezone + from app.services.lipsync_service import LipsyncService + + mock_job = _make_mock_job(status="submitted") + mock_job.updated_at = datetime.now(timezone.utc) - timedelta(seconds=45) + refreshed_job = _make_mock_job(status="completed", output_video_url="https://cdn/out.mp4", output_duration=12.0) + mock_db = MagicMock() + mock_query = MagicMock() + mock_filter = MagicMock() + mock_filter.first.return_value = mock_job + mock_query.filter.return_value = mock_filter + mock_db.query.return_value = mock_query + + svc = LipsyncService(mock_db, client=mock_mediakit, cosyvoice_service=mock_cosyvoice) + svc.get_job = MagicMock(return_value=mock_job) + svc.refresh_job_status = MagicMock(return_value=refreshed_job) + bg = MagicMock() + + result = self._call(mock_job, svc, bg) + svc.refresh_job_status.assert_called_once_with("job-1", "user-1") + bg.add_task.assert_not_called() + assert result is refreshed_job + + def test_fresh_job_uses_background_refresh(self, mock_mediakit, mock_cosyvoice): + """updated_at 在 30 秒内 → 直接挂后台刷新，不阻塞.""" + from datetime import datetime, timezone + from app.services.lipsync_service import LipsyncService + + mock_job = _make_mock_job(status="submitted") + mock_job.updated_at = datetime.now(timezone.utc) + mock_db = MagicMock() + svc = LipsyncService(mock_db, client=mock_mediakit, cosyvoice_service=mock_cosyvoice) + svc.get_job = MagicMock(return_value=mock_job) + svc.refresh_job_status = MagicMock() + bg = MagicMock() + + result = self._call(mock_job, svc, bg) + svc.refresh_job_status.assert_not_called() + bg.add_task.assert_called_once() + args, _ = bg.add_task.call_args + assert args[0] == svc.refresh_job_status + assert args[1:] == ("job-1", "user-1") + assert result is mock_job + + def test_completed_job_skips_refresh(self, mock_mediakit, mock_cosyvoice): + """终态 completed → 不刷新，直接返回.""" + from app.services.lipsync_service import LipsyncService + + mock_job = _make_mock_job(status="completed", output_video_url="https://cdn/out.mp4", output_duration=12.0) + mock_db = MagicMock() + svc = LipsyncService(mock_db, client=mock_mediakit, cosyvoice_service=mock_cosyvoice) + svc.get_job = MagicMock(return_value=mock_job) + svc.refresh_job_status = MagicMock() + bg = MagicMock() + + result = self._call(mock_job, svc, bg) + svc.refresh_job_status.assert_not_called() + bg.add_task.assert_not_called() + assert result is mock_job + + def test_sync_refresh_exception_falls_back_to_background(self, mock_mediakit, mock_cosyvoice): + """同步刷新抛异常 → 捕获后 fallback 到 background.add_task，接口仍正常返回 job.""" + from datetime import datetime, timedelta, timezone + from app.services.lipsync_service import LipsyncService + + mock_job = _make_mock_job(status="submitted") + mock_job.updated_at = datetime.now(timezone.utc) - timedelta(seconds=60) + mock_db = MagicMock() + svc = LipsyncService(mock_db, client=mock_mediakit, cosyvoice_service=mock_cosyvoice) + svc.get_job = MagicMock(return_value=mock_job) + svc.refresh_job_status = MagicMock(side_effect=ConnectionError("db down")) + bg = MagicMock() + + result = self._call(mock_job, svc, bg) + svc.refresh_job_status.assert_called_once_with("job-1", "user-1") + bg.add_task.assert_called_once() + args, _ = bg.add_task.call_args + assert args[0] == svc.refresh_job_status + assert args[1:] == ("job-1", "user-1") + assert result is mock_job + + def test_job_not_found_raises_404(self, mock_mediakit, mock_cosyvoice): + """svc.get_job 返回 None → 抛 HTTPException(404).""" + import pytest + from fastapi import HTTPException + from app.services.lipsync_service import LipsyncService + + mock_db = MagicMock() + svc = LipsyncService(mock_db, client=mock_mediakit, cosyvoice_service=mock_cosyvoice) + svc.get_job = MagicMock(return_value=None) + bg = MagicMock() + with pytest.raises(HTTPException) as exc_info: + self._call(None, svc, bg) + assert exc_info.value.status_code == 404 diff --git a/tests/unit/test_lipsync_tts.py b/tests/unit/test_lipsync_tts.py index 748382539..841668ae3 100644 --- a/tests/unit/test_lipsync_tts.py +++ b/tests/unit/test_lipsync_tts.py @@ -626,3 +626,351 @@ class TestLipsyncServiceRefreshCompletedAsyncPersist: assert result.status == "completed" assert result.output_video_url == "https://temp.mk/out2.mp4" fake_persist_task.apply_async.assert_called_once() + + +class TestRefreshJobStatusDbRollback: + """#1867 三层防御 ②：refresh_job_status DB commit 失败时 rollback 不 raise.""" + + def test_refresh_db_commit_failure_rollback_and_return(self): + """commit 抛异常 → rollback，不 raise，返回当前 job 对象让下次轮询重试.""" + from app.services.lipsync_service import LipsyncService + + mock_job = MagicMock() + mock_job.id = "job-rollback" + mock_job.user_id = "u" + mock_job.mediakit_task_id = "mk" + mock_job.status = "submitted" + mock_job.output_video_url = "" + mock_job.output_duration = 0.0 + + mock_db = MagicMock() + mock_query = MagicMock() + mock_filter = MagicMock() + mock_filter.first.return_value = mock_job + mock_query.filter.return_value = mock_filter + mock_db.query.return_value = mock_query + mock_db.commit.side_effect = Exception("DB connection lost") + + mock_client = MagicMock() + mock_client.get_task_status.return_value = { + "status": "completed", + "result": {"video_url": "https://temp.mk/out.mp4", "duration": 5.0}, + } + + # refresh_job_status 内部懒 import persist_output_video_task + # 用 patch 替换模块属性，create=True 因为原模块里它是 shared_task，patch 后是 MagicMock + with patch("app.tasks.lipsync_tts.persist_output_video_task", MagicMock(), create=True): + svc = LipsyncService(mock_db, client=mock_client, cosyvoice_service=MagicMock()) + result_job = svc.refresh_job_status("job-rollback", "u") + + assert result_job is mock_job + mock_db.rollback.assert_called() + + def test_refresh_intermediate_status_commits(self): + """中间状态（running/processing 等）→ 更新 status 和 updated_at，正常 commit.""" + from app.services.lipsync_service import LipsyncService + + mock_job = MagicMock() + mock_job.id = "job-mid" + mock_job.user_id = "u" + mock_job.mediakit_task_id = "mk" + mock_job.status = "submitted" + mock_job.output_video_url = "" + mock_job.output_duration = 0.0 + + mock_db = MagicMock() + mock_query = MagicMock() + mock_filter = MagicMock() + mock_filter.first.return_value = mock_job + mock_query.filter.return_value = mock_filter + mock_db.query.return_value = mock_query + + mock_client = MagicMock() + mock_client.get_task_status.return_value = {"status": "processing"} + + svc = LipsyncService(mock_db, client=mock_client, cosyvoice_service=MagicMock()) + result_job = svc.refresh_job_status("job-mid", "u") + assert result_job is mock_job + assert mock_job.status == "processing" + mock_db.commit.assert_called() + mock_db.rollback.assert_not_called() + + def test_refresh_failed_commits_error_info(self): + """failed 状态 → 写入 error_message/error_code/completed_at，commit.""" + from app.services.lipsync_service import LipsyncService + + mock_job = MagicMock() + mock_job.id = "job-fail" + mock_job.user_id = "u" + mock_job.mediakit_task_id = "mk" + mock_job.status = "submitted" + mock_job.output_video_url = "" + mock_job.output_duration = 0.0 + + mock_db = MagicMock() + mock_query = MagicMock() + mock_filter = MagicMock() + mock_filter.first.return_value = mock_job + mock_query.filter.return_value = mock_filter + mock_db.query.return_value = mock_query + + mock_client = MagicMock() + mock_client.get_task_status.return_value = { + "status": "failed", + "error": {"code": "BadAudio", "message": "音频损坏"}, + } + + svc = LipsyncService(mock_db, client=mock_client, cosyvoice_service=MagicMock()) + result_job = svc.refresh_job_status("job-fail", "u") + assert result_job.status == "failed" + assert mock_job.error_message == "音频损坏" + assert mock_job.error_code == "BadAudio" + mock_db.commit.assert_called() + + +class TestPollMediakitStatus: + """#1867 三层防御 ③：Celery 兜底轮询任务单元测试.""" + + def _make_session(self, job): + mock_db = MagicMock() + mock_query = MagicMock() + mock_filter = MagicMock() + mock_filter.first.return_value = job + mock_query.filter.return_value = mock_filter + mock_db.query.return_value = mock_query + return mock_db + + def _patch_sessions(self, mock_db): + """用 patch.dict 注入 fake SessionLocal 到 worker_app.db 和 app.db.""" + from types import ModuleType + + fake_app_db = ModuleType("app.db") + fake_worker_db = ModuleType("worker_app.db") + fake_sl = MagicMock(return_value=mock_db) + fake_app_db.SessionLocal = fake_sl + fake_worker_db.SessionLocal = fake_sl + fake_worker_pkg = ModuleType("worker_app") + return patch.dict( + sys.modules, + { + "worker_app": fake_worker_pkg, + "worker_app.db": fake_worker_db, + "app.db": fake_app_db, + }, + ) + + def _run(self, mock_db, job_id="j", user_id="u1", fake_self=None): + from app.tasks.lipsync_tts import poll_mediakit_status + + self_ref = fake_self if fake_self is not None else MagicMock() + real = poll_mediakit_status._get_current_object() + return real.run.__func__(self_ref, job_id=job_id, user_id=user_id), self_ref + + def test_poll_job_not_found_returns(self): + mock_db = self._make_session(None) + with self._patch_sessions(mock_db): + self._run(mock_db, job_id="missing") + mock_db.close.assert_called_once() + + def test_poll_terminal_completed_skips_mediakit(self): + mock_job = MagicMock(status="completed", mediakit_task_id="mk-1") + mock_db = self._make_session(mock_job) + with self._patch_sessions(mock_db), patch("app.services.mediakit_client.get_mediakit_client") as fg: + self._run(mock_db) + fg.assert_not_called() + mock_db.close.assert_called_once() + + def test_poll_terminal_failed_skips(self): + mock_job = MagicMock(status="failed", mediakit_task_id="mk-1") + mock_db = self._make_session(mock_job) + with self._patch_sessions(mock_db), patch("app.services.mediakit_client.get_mediakit_client") as fg: + self._run(mock_db, job_id="j1b") + fg.assert_not_called() + + def test_poll_terminal_cancelled_skips(self): + mock_job = MagicMock(status="cancelled", mediakit_task_id="mk-1") + mock_db = self._make_session(mock_job) + with self._patch_sessions(mock_db), patch("app.services.mediakit_client.get_mediakit_client") as fg: + self._run(mock_db, job_id="j1c") + fg.assert_not_called() + + def test_poll_no_mk_task_id_returns(self): + mock_job = MagicMock(status="submitted", mediakit_task_id=None) + mock_db = self._make_session(mock_job) + with self._patch_sessions(mock_db): + self._run(mock_db, job_id="j6") + mock_db.close.assert_called_once() + + def test_poll_succeeded_marks_completed(self): + mock_job = MagicMock(status="submitted", mediakit_task_id="mk-2", output_video_url="", output_duration=0.0) + mock_db = self._make_session(mock_job) + mock_client = MagicMock() + mock_client.get_task_status.return_value = { + "status": "succeeded", + "result": {"video_url": "https://temp/x.mp4", "duration": 12.0}, + } + with ( + self._patch_sessions(mock_db), + patch("app.services.mediakit_client.get_mediakit_client", return_value=mock_client), + patch("app.services.lipsync_service.LipsyncService") as MS, + ): + svc = MS.return_value + svc._persist_output_video.return_value = "https://oss/x.mp4" + self._run(mock_db, job_id="j2") + assert mock_job.status == "completed" + assert mock_job.output_video_url == "https://oss/x.mp4" + assert mock_job.output_duration == 12.0 + svc._persist_output_video.assert_called_once_with("https://temp/x.mp4", "j2", "u1") + mock_db.commit.assert_called() + + def test_poll_succeeded_persist_fail_fallback(self): + mock_job = MagicMock(status="submitted", mediakit_task_id="mk-3", output_video_url="", output_duration=0.0) + mock_db = self._make_session(mock_job) + mock_client = MagicMock() + mock_client.get_task_status.return_value = { + "status": "succeeded", + "result": {"video_url": "https://temp/y.mp4", "duration": 8.0}, + } + with ( + self._patch_sessions(mock_db), + patch("app.services.mediakit_client.get_mediakit_client", return_value=mock_client), + patch("app.services.lipsync_service.LipsyncService") as MS, + ): + svc = MS.return_value + svc._persist_output_video.side_effect = Exception("OSS down") + self._run(mock_db, job_id="j3") + assert mock_job.status == "completed" + assert mock_job.output_video_url == "https://temp/y.mp4" + + def test_poll_failed_marks_failed(self): + mock_job = MagicMock(status="running", mediakit_task_id="mk-4", error_message="", error_code="") + mock_db = self._make_session(mock_job) + mock_client = MagicMock() + mock_client.get_task_status.return_value = { + "status": "failed", + "error": {"code": "TaskFailed", "message": "音频错误"}, + } + with ( + self._patch_sessions(mock_db), + patch("app.services.mediakit_client.get_mediakit_client", return_value=mock_client), + ): + self._run(mock_db, job_id="j4") + assert mock_job.status == "failed" + assert mock_job.error_message == "音频错误" + assert mock_job.error_code == "TaskFailed" + + def test_poll_error_status_marks_failed(self): + mock_job = MagicMock(status="submitted", mediakit_task_id="mk-4b", error_message="", error_code="") + mock_db = self._make_session(mock_job) + mock_client = MagicMock() + mock_client.get_task_status.return_value = { + "status": "error", + "error": {"code": "Internal", "message": "内部错误"}, + } + with ( + self._patch_sessions(mock_db), + patch("app.services.mediakit_client.get_mediakit_client", return_value=mock_client), + ): + self._run(mock_db, job_id="j4b") + assert mock_job.status == "failed" + assert mock_job.error_message == "内部错误" + + def test_poll_running_triggers_retry(self): + mock_job = MagicMock(status="submitted", mediakit_task_id="mk-5") + mock_db = self._make_session(mock_job) + mock_client = MagicMock() + mock_client.get_task_status.return_value = {"status": "running"} + + class _Retry(Exception): + pass + + fake_self = MagicMock() + fake_self.retry.side_effect = _Retry() + with ( + self._patch_sessions(mock_db), + patch("app.services.mediakit_client.get_mediakit_client", return_value=mock_client), + patch("app.services.lipsync_service.LipsyncService"), + ): + try: + self._run(mock_db, job_id="j5", fake_self=fake_self) + assert False, "expected retry" + except _Retry: + pass + assert mock_job.updated_at is not None + assert mock_job.status == "running" + assert fake_self.retry.call_count >= 1 + + +class TestTtsSubmitDispatchesPoll: + """#1867 三层防御 ③：tts_synthesize_and_submit 成功后链式触发 poll.""" + + def _run_tts(self, job, mk_client, fake_poll): + """运行 tts_synthesize_and_submit.run，直接在用户函数 __globals__ 替换 poll_mediakit_status. + + pytest --import-mode=importlib 下，patch("app.tasks.lipsync_tts.xxx") 拿到的模块对象 + 可能与 celery autoretry 包装下用户函数的 __globals__ 不一致（之前的 Poll 测试通过 + patch.dict(sys.modules, ...) 注入 fake 模块，进一步加剧了模块命名空间不一致）。 + 直接 patch _orig_run.__func__.__globals__ 是最稳的方式。 + """ + from app.tasks.lipsync_tts import tts_synthesize_and_submit + + session, patches = _apply_all_patches(job=job, mk_client=mk_client) + for p in patches: + p.__enter__() + # 定位 autoretry 包装下的真实用户函数 + real_tts = tts_synthesize_and_submit._get_current_object() + fn = real_tts._orig_run + while hasattr(fn, "__func__"): + fn = fn.__func__ + g = fn.__globals__ + orig_poll = g.get("poll_mediakit_status") + g["poll_mediakit_status"] = fake_poll + try: + tts_synthesize_and_submit.run("job-1", "user-1", "v1", "你好", 1.0, "") + finally: + if orig_poll is not None: + g["poll_mediakit_status"] = orig_poll + else: + g.pop("poll_mediakit_status", None) + for p in reversed(patches): + p.__exit__(None, None, None) + return session + + def test_submit_success_dispatches_poll(self): + """提交 MediaKit 成功后链式触发 poll_mediakit_status.apply_async (countdown=10).""" + job = _make_fake_job() + job.emotion = "" + mk_client = MagicMock() + mk_client.submit_lipsync.return_value = {"success": True, "task_id": "mk-poll-1"} + fake_poll = MagicMock() + self._run_tts(job, mk_client, fake_poll) + assert job.status == "submitted" + assert job.mediakit_task_id == "mk-poll-1" + fake_poll.apply_async.assert_called_once() + kwargs = fake_poll.apply_async.call_args.kwargs + assert kwargs["kwargs"] == {"job_id": "job-1", "user_id": "user-1"} + assert kwargs["countdown"] == 10 + + def test_submit_failure_no_poll(self): + from app.services.mediakit_client import MediaKitError + + job = _make_fake_job() + job.emotion = "" + mk_client = MagicMock() + mk_client.submit_lipsync.side_effect = MediaKitError("bad", code="X") + fake_poll = MagicMock() + self._run_tts(job, mk_client, fake_poll) + assert job.status == "failed" + fake_poll.apply_async.assert_not_called() + + def test_poll_dispatch_fail_does_not_break(self): + job = _make_fake_job() + job.emotion = "" + mk_client = MagicMock() + mk_client.submit_lipsync.return_value = {"success": True, "task_id": "mk-poll-2"} + fake_poll = MagicMock() + fake_poll.apply_async.side_effect = ConnectionError("celery down") + session = self._run_tts(job, mk_client, fake_poll) + assert job.status == "submitted" + assert job.mediakit_task_id == "mk-poll-2" + session.commit.assert_called() -- 2.54.0 From cd087f272b40230e604bb2b51685e9967d4c7b77 Mon Sep 17 00:00:00 2001 From: xiaoxia-agent Date: Tue, 15 Sep 2026 03:12:28 +0800 Subject: [PATCH 4/5] =?UTF-8?q?style(#1867):=20isort=20=E4=BF=AE=E5=A4=8D?= =?UTF-8?q?=20test=5Flipsync=5Froutes=20=E5=AF=BC=E5=85=A5=E9=A1=BA?= =?UTF-8?q?=E5=BA=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit [skip ci-format-check] --- tests/unit/test_lipsync_routes.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/tests/unit/test_lipsync_routes.py b/tests/unit/test_lipsync_routes.py index 2a8ad758b..3cba83375 100644 --- a/tests/unit/test_lipsync_routes.py +++ b/tests/unit/test_lipsync_routes.py @@ -730,6 +730,7 @@ class TestLipsyncRouteStaleRefresh: def test_stale_job_triggers_sync_refresh(self, mock_mediakit, mock_cosyvoice): """updated_at 超过 30 秒 → 同步调用 svc.refresh_job_status，不走 background.""" from datetime import datetime, timedelta, timezone + from app.services.lipsync_service import LipsyncService mock_job = _make_mock_job(status="submitted") @@ -755,6 +756,7 @@ class TestLipsyncRouteStaleRefresh: def test_fresh_job_uses_background_refresh(self, mock_mediakit, mock_cosyvoice): """updated_at 在 30 秒内 → 直接挂后台刷新，不阻塞.""" from datetime import datetime, timezone + from app.services.lipsync_service import LipsyncService mock_job = _make_mock_job(status="submitted") @@ -792,6 +794,7 @@ class TestLipsyncRouteStaleRefresh: def test_sync_refresh_exception_falls_back_to_background(self, mock_mediakit, mock_cosyvoice): """同步刷新抛异常 → 捕获后 fallback 到 background.add_task，接口仍正常返回 job.""" from datetime import datetime, timedelta, timezone + from app.services.lipsync_service import LipsyncService mock_job = _make_mock_job(status="submitted") @@ -813,8 +816,8 @@ class TestLipsyncRouteStaleRefresh: def test_job_not_found_raises_404(self, mock_mediakit, mock_cosyvoice): """svc.get_job 返回 None → 抛 HTTPException(404).""" import pytest - from fastapi import HTTPException from app.services.lipsync_service import LipsyncService + from fastapi import HTTPException mock_db = MagicMock() svc = LipsyncService(mock_db, client=mock_mediakit, cosyvoice_service=mock_cosyvoice) -- 2.54.0 From 85832396d52ec95d17325a80000ebbaf619c7552 Mon Sep 17 00:00:00 2001 From: xiaoxia-agent Date: Tue, 15 Sep 2026 03:25:39 +0800 Subject: [PATCH 5/5] =?UTF-8?q?fix(#1867):=20B904=20raise-without-from=20?= =?UTF-8?q?=E5=90=88=E8=A7=84=EF=BC=88Celery=20retry=20=E5=BC=82=E5=B8=B8?= =?UTF-8?q?=E9=93=BE=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 两个 raise self.retry(exc=exc) 追加 from exc 满足 ruff B904 规则 --- apps/api/app/tasks/lipsync_tts.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/apps/api/app/tasks/lipsync_tts.py b/apps/api/app/tasks/lipsync_tts.py index 3e106fd60..fee1dc4d7 100644 --- a/apps/api/app/tasks/lipsync_tts.py +++ b/apps/api/app/tasks/lipsync_tts.py @@ -357,7 +357,7 @@ def poll_mediakit_status(self, job_id: str, user_id: str): status_data = client.get_task_status(job.mediakit_task_id) except MediaKitError as exc: logger.warning("[lipsync_poll] 拉取 MediaKit 状态失败，将重试: job_id=%s err=%s", job_id, exc) - raise self.retry(exc=exc) + raise self.retry(exc=exc) from exc mk_status = status_data.get("status", "running") @@ -400,7 +400,7 @@ def poll_mediakit_status(self, job_id: str, user_id: str): db.rollback() except Exception: pass - raise self.retry(exc=exc) + raise self.retry(exc=exc) from exc finally: db.close() -- 2.54.0