test: cover lipsync_gpu_process_async task + GPU fallback branches

CI Unit Tests failed at incremental diff coverage (23% < 40% gate). - Direct tests for Celery task: done path, timeout/failed fallback, mediakit fallback (success/MediaKitError/exception), job guards, unexpected-exception handler, _sign_media_url branches - Service tests: _submit_to_gpu_create None, _submit_to_gpu_wait timeout/failed, refresh_job_status stale GPU timeout (30min)
2026-09-20 11:51:06 +08:00
parent 60cacdf280
commit 04083e1f12
2 changed files with 337 additions and 0 deletions
@@ -0,0 +1,250 @@
+"""Celery 任务 lipsync_gpu_process_async 直接单测 (#1978 异步化).
+
+覆盖 apps/api/app/tasks/lipsync_gpu.py 的全部主路径：
+- 成功：wait_for_result 返回 done → 签名 URL → completed
+- GPU 超时/失败 → MediaKit 兜底（成功/MediaKitError/其他异常）
+- job 不存在 / 状态异常提前返回
+- 主流程异常 → job 标 failed
+- _sign_media_url 各分支
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import app.tasks.lipsync_gpu as task_mod
+import pytest
+
+
+def _make_job(status="processing"):
+    job = MagicMock()
+    job.id = "job-1"
+    job.user_id = "u1"
+    job.status = status
+    job.video_url = "videos/v.mp4"
+    job.audio_url = "audios/a.wav"
+    job.enable_video_loop = True
+    return job
+
+
+def _make_gpu_task(status="done", result_url="gpu-lipsync/results/t1.mp4", result_duration=11.2):
+    t = MagicMock()
+    t.status = status
+    t.result_url = result_url
+    t.result_duration = result_duration
+    return t
+
+
+@pytest.fixture()
+def db_patch():
+    """patch _get_db_session 返回 MagicMock，并在任务结束后断言 close."""
+    fake_db = MagicMock()
+    with patch.object(task_mod, "_get_db_session", return_value=fake_db):
+        yield fake_db
+
+
+def _patch_gpu_service(final_task):
+    fake_svc = MagicMock()
+    fake_svc.wait_for_result.return_value = final_task
+    return patch(
+        "app.services.gpu_lipsync_service.GpuLipsyncService",
+        return_value=fake_svc,
+    )
+
+
+def _run_task():
+    # @shared_task bind=True：直接调用任务对象会自动注入 self
+    task_mod.lipsync_gpu_process_async("job-1", "u1", "gpu-task-1")
+
+
+class TestHappyPath:
+    def test_gpu_done_marks_completed(self, db_patch):
+        job = _make_job()
+        db_patch.query.return_value.filter_by.return_value.first.return_value = job
+        gpu_task = _make_gpu_task()
+        storage = MagicMock()
+        storage.get_download_url.return_value = "https://signed.example.com/r1.mp4?sig=x"
+        with (
+            _patch_gpu_service(gpu_task),
+            patch.object(task_mod, "get_shared_storage_service", return_value=storage),
+        ):
+            _run_task()
+        assert job.status == "completed"
+        assert job.output_video_url == "https://signed.example.com/r1.mp4?sig=x"
+        assert job.output_duration == 11.2
+        assert job.completed_at is not None
+        db_patch.commit.assert_called_once()
+        db_patch.close.assert_called_once()
+
+    def test_gpu_done_empty_signed_url_keeps_original(self, db_patch):
+        job = _make_job()
+        db_patch.query.return_value.filter_by.return_value.first.return_value = job
+        gpu_task = _make_gpu_task(result_url="gpu/r2.mp4")
+        storage = MagicMock()
+        storage.get_download_url.return_value = ""
+        with (
+            _patch_gpu_service(gpu_task),
+            patch.object(task_mod, "get_shared_storage_service", return_value=storage),
+        ):
+            _run_task()
+        assert job.status == "completed"
+        assert job.output_video_url == "gpu/r2.mp4"
+
+    def test_gpu_done_result_duration_none_defaults_zero(self, db_patch):
+        job = _make_job()
+        db_patch.query.return_value.filter_by.return_value.first.return_value = job
+        gpu_task = _make_gpu_task(result_duration=None)
+        storage = MagicMock()
+        with (
+            _patch_gpu_service(gpu_task),
+            patch.object(task_mod, "get_shared_storage_service", return_value=storage),
+        ):
+            _run_task()
+        assert job.output_duration == 0.0
+
+    def test_sign_failure_uses_original_url(self, db_patch):
+        job = _make_job()
+        db_patch.query.return_value.filter_by.return_value.first.return_value = job
+        gpu_task = _make_gpu_task(result_url="gpu/r3.mp4")
+        with (
+            _patch_gpu_service(gpu_task),
+            patch.object(task_mod, "get_shared_storage_service", side_effect=RuntimeError("oss down")),
+        ):
+            _run_task()
+        assert job.status == "completed"
+        assert job.output_video_url == "gpu/r3.mp4"
+
+
+class TestJobGuards:
+    def test_job_not_found_returns(self, db_patch):
+        db_patch.query.return_value.filter_by.return_value.first.return_value = None
+        _run_task()
+        db_patch.commit.assert_not_called()
+        db_patch.close.assert_called_once()
+
+    def test_job_wrong_status_skipped(self, db_patch):
+        job = _make_job(status="completed")
+        db_patch.query.return_value.filter_by.return_value.first.return_value = job
+        _run_task()
+        db_patch.commit.assert_not_called()
+
+
+class TestGpuFailureFallback:
+    def test_gpu_timeout_falls_back_mediakit_success(self, db_patch):
+        job = _make_job()
+        db_patch.query.return_value.filter_by.return_value.first.return_value = job
+        with (
+            _patch_gpu_service(None),
+            patch.object(task_mod, "_fallback_to_mediakit") as fb,
+        ):
+            _run_task()
+        fb.assert_called_once_with(db_patch, job)
+
+    def test_gpu_failed_status_falls_back(self, db_patch):
+        job = _make_job()
+        db_patch.query.return_value.filter_by.return_value.first.return_value = job
+        gpu_task = _make_gpu_task(status="failed")
+        with (
+            _patch_gpu_service(gpu_task),
+            patch.object(task_mod, "_fallback_to_mediakit") as fb,
+        ):
+            _run_task()
+        fb.assert_called_once_with(db_patch, job)
+
+
+class TestFallbackToMediaKit:
+    def test_mediakit_success_marks_submitted(self, db_patch):
+        job = _make_job()
+        fake_client = MagicMock()
+        fake_client.submit_lipsync.return_value = {"task_id": "mk-99"}
+        with (
+            patch("app.services.mediakit_client.get_mediakit_client", return_value=fake_client),
+            patch.object(task_mod, "_sign_media_url", side_effect=lambda u: u + "?s"),
+        ):
+            task_mod._fallback_to_mediakit(db_patch, job)
+        fake_client.submit_lipsync.assert_called_once()
+        kwargs = fake_client.submit_lipsync.call_args.kwargs
+        assert kwargs["enable_video_loop"] is True
+        assert kwargs["client_token"] == "job-1"
+        assert job.status == "submitted"
+        assert job.mediakit_task_id == "mk-99"
+        db_patch.commit.assert_called_once()
+
+    def test_mediakit_error_marks_failed(self, db_patch):
+        from app.services.mediakit_client import MediaKitError
+
+        job = _make_job()
+        fake_client = MagicMock()
+        fake_client.submit_lipsync.side_effect = MediaKitError("api reject", code="MkReject")
+        with (
+            patch("app.services.mediakit_client.get_mediakit_client", return_value=fake_client),
+            patch.object(task_mod, "_sign_media_url", side_effect=lambda u: u),
+        ):
+            task_mod._fallback_to_mediakit(db_patch, job)
+        assert job.status == "failed"
+        assert job.error_code == "MkReject"
+        db_patch.commit.assert_called_once()
+
+    def test_other_exception_marks_failed(self, db_patch):
+        job = _make_job()
+        with (
+            patch("app.services.mediakit_client.get_mediakit_client", side_effect=RuntimeError("boom")),
+            patch.object(task_mod, "_sign_media_url", side_effect=lambda u: u),
+        ):
+            task_mod._fallback_to_mediakit(db_patch, job)
+        assert job.status == "failed"
+        assert job.error_code == "FallbackFailed"
+        db_patch.commit.assert_called_once()
+
+
+class TestTaskException:
+    def test_unexpected_exception_marks_job_failed(self, db_patch):
+        job = _make_job()
+        # query 第一次返回 job，异常路径里再次 query 也返回 job
+        db_patch.query.return_value.filter_by.return_value.first.return_value = job
+        with patch(
+            "app.services.gpu_lipsync_service.GpuLipsyncService",
+            side_effect=RuntimeError("svc ctor fail"),
+        ):
+            _run_task()
+        assert job.status == "failed"
+        assert job.error_code == "GpuAsyncError"
+
+    def test_exception_handler_failure_swallowed(self, db_patch):
+        # 主流程异常，且异常处理中的 query 也抛异常 → 不应再抛
+        db_patch.query.side_effect = RuntimeError("db totally broken")
+        _run_task()
+        db_patch.close.assert_called_once()
+
+
+class TestSignMediaUrl:
+    def test_empty_url_returned_as_is(self):
+        assert task_mod._sign_media_url("") == ""
+
+    def test_non_own_host_returned_as_is(self):
+        storage = MagicMock()
+        storage.public_url = "https://own-bucket.oss-cn-beijing.aliyuncs.com"
+        with patch.object(task_mod, "get_shared_storage_service", return_value=storage):
+            url = "https://other.example.com/a.wav"
+            assert task_mod._sign_media_url(url) == url
+
+    def test_own_host_signed(self):
+        storage = MagicMock()
+        storage.public_url = "https://own-bucket.oss-cn-beijing.aliyuncs.com"
+        storage.get_download_url.return_value = "https://own-bucket.oss-cn-beijing.aliyuncs.com/a?sig=1"
+        with patch.object(task_mod, "get_shared_storage_service", return_value=storage):
+            out = task_mod._sign_media_url("https://own-bucket.oss-cn-beijing.aliyuncs.com/a.wav")
+        assert out.endswith("?sig=1")
+        storage.get_download_url.assert_called_once()
+
+    def test_missing_public_url_returns_original(self):
+        storage = MagicMock()
+        storage.public_url = ""
+        with patch.object(task_mod, "get_shared_storage_service", return_value=storage):
+            url = "https://own-bucket.oss-cn-beijing.aliyuncs.com/a.wav"
+            assert task_mod._sign_media_url(url) == url
+
+    def test_exception_returns_original(self):
+        with patch.object(task_mod, "get_shared_storage_service", side_effect=RuntimeError("x")):
+            url = "https://own-bucket.oss-cn-beijing.aliyuncs.com/a.wav"
+            assert task_mod._sign_media_url(url) == url