"""AI 数字人口型 TTS Celery 异步任务 — 单元测试. 覆盖 lipsync_tts.py 的全部主要分支: - Job 不存在/cancelled/正常/异常路径 - TTS 合成、音频下载、OSS 上传、MediaKit 提交 - CosyVoiceError/ValueError/MediaKitError/顶层异常等错误码 """ import os import sys import types from types import ModuleType from unittest.mock import MagicMock, patch os.environ.setdefault("JWT_SECRET_KEY", "dev-secret-key-for-testing") class _FakeQuery: """模拟 SQLAlchemy query.filter().first() 链式调用.""" def __init__(self, job): self._job = job def filter(self, *args, **kwargs): return self def first(self): return self._job def _make_fake_job(**kwargs): """构造可 setattr 的 job 记录.""" job = MagicMock() job.id = kwargs.get("job_id", "job-1") job.user_id = kwargs.get("user_id", "user-1") job.status = kwargs.get("status", "tts_processing") job.audio_url = kwargs.get("audio_url", "") job.video_url = kwargs.get("video_url", "https://oss/video.mp4") job.mediakit_task_id = kwargs.get("mediakit_task_id", "") job.enable_video_loop = kwargs.get("enable_video_loop", False) job.error_code = "" job.error_message = "" job.submitted_at = None job.updated_at = None return job def _build_session(job): """构造 mock DB session + factory. 返回 (session, factory).""" session = MagicMock() session.query.return_value = _FakeQuery(job) session.commit = MagicMock() session.close = MagicMock() factory = MagicMock(return_value=session) return session, factory def _apply_all_patches( *, job=None, cosyvoice_service=None, cosyvoice_side_effect=None, cosyvoice_error=None, download_bytes=b"AUDIO", download_error=None, storage=None, mk_client=None, mk_submit_return=None, mk_submit_error=None, ): """统一构造测试需要的 patch 列表. lipsync_tts.run() 在函数体内部懒 import 多个模块,通过 sys.modules 注入 伪造包路径避免真实导入;对存在的模块用 patch() 替换返回值/side_effect。 """ # SessionLocal 通过懒探测获取(Worker 用 worker_app.db,API 用 app.db), # 测试环境里两个模块都能被真实导入,必须同时 mock 保证用的是 fake session。 fake_app_db = ModuleType("app.db") fake_worker_db = ModuleType("worker_app.db") session, factory = _build_session(job) fake_app_db.SessionLocal = factory fake_worker_db.SessionLocal = factory patches = [ patch.dict(sys.modules, {"app.db": fake_app_db, "worker_app.db": fake_worker_db}), patch( "app.tasks.lipsync_tts._sign_media_url", side_effect=lambda url: url + "?signed" if url else url, ), ] # CosyVoice if cosyvoice_service is not None: cosy_instance = cosyvoice_service else: cosy_instance = MagicMock() if cosyvoice_side_effect is not None: cosy_instance.submit_synthesize_task.side_effect = cosyvoice_side_effect elif cosyvoice_error is not None: cosy_instance.submit_synthesize_task.side_effect = cosyvoice_error else: cosy_instance.submit_synthesize_task.return_value = {"audio_url": "https://tts/raw.mp3"} patches.append(patch("packages.application.cosyvoice_service.CosyVoiceService", return_value=cosy_instance)) # safe_download_bytes if download_error is not None: patches.append(patch("packages.shared.url_security.safe_download_bytes", side_effect=download_error)) else: patches.append(patch("packages.shared.url_security.safe_download_bytes", return_value=download_bytes)) # Storage if storage is None: storage = MagicMock() storage.public_url = "https://oss.example.com" storage.upload_file.return_value = "https://oss.example.com/tts.mp3" patches.append(patch("packages.shared.storage.get_shared_storage_service", return_value=storage)) # MediaKit client if mk_client is not None: patches.append(patch("app.services.mediakit_client.get_mediakit_client", return_value=mk_client)) else: client = MagicMock() if mk_submit_error is not None: client.submit_lipsync.side_effect = mk_submit_error else: client.submit_lipsync.return_value = mk_submit_return or {"task_id": "mk-1"} patches.append(patch("app.services.mediakit_client.get_mediakit_client", return_value=client)) return session, patches class TestTtsSynthesizeAndSubmit: """测试 Celery 任务 tts_synthesize_and_submit.run 的所有分支.""" def test_job_not_found_returns_early(self): """Job 不存在 → 日志报错直接返回,不抛异常.""" from app.tasks.lipsync_tts import tts_synthesize_and_submit session, patches = _apply_all_patches(job=None) entered = [p.__enter__() for p in patches] try: tts_synthesize_and_submit.run("missing-job", "user-1", "v1", "你好", 1.0, "") finally: for p in reversed(patches): p.__exit__(None, None, None) session.commit.assert_not_called() session.close.assert_called_once() def test_cancelled_job_skipped(self): """Job 已 cancelled → 跳过不处理,不调用 TTS/MediaKit.""" from app.tasks.lipsync_tts import tts_synthesize_and_submit job = _make_fake_job(status="cancelled") session, patches = _apply_all_patches(job=job) entered = [p.__enter__() for p in patches] try: tts_synthesize_and_submit.run("job-1", "user-1", "v1", "你好", 1.0, "") finally: for p in reversed(patches): p.__exit__(None, None, None) # cancelled 不应 commit,不应触发 TTS/MediaKit session.commit.assert_not_called() session.close.assert_called_once() def test_happy_path_tts_to_mediakit(self): """完整正常流程:TTS 合成 → OSS 上传 → 签名 → 提交 MediaKit → submitted.""" from app.tasks.lipsync_tts import tts_synthesize_and_submit job = _make_fake_job() mk_client = MagicMock() mk_client.submit_lipsync.return_value = {"task_id": "mk-999"} session, patches = _apply_all_patches(job=job, mk_client=mk_client) entered = [p.__enter__() for p in patches] try: tts_synthesize_and_submit.run("job-1", "user-1", "v1", "你好世界", 1.0, "happy") finally: for p in reversed(patches): p.__exit__(None, None, None) assert job.status == "submitted" assert job.mediakit_task_id == "mk-999" assert job.error_code == "" mk_client.submit_lipsync.assert_called_once() call_kwargs = mk_client.submit_lipsync.call_args.kwargs assert call_kwargs["client_token"] == "job-1" assert call_kwargs["audio_url"].endswith("?signed") session.commit.assert_called() session.close.assert_called_once() def test_cosyvoice_error_marks_tts_synthesis_failed(self): """CosyVoiceError → failed, error_code=TTSSynthesisFailed.""" from app.tasks.lipsync_tts import tts_synthesize_and_submit from packages.application.cosyvoice_service import CosyVoiceError job = _make_fake_job() session, patches = _apply_all_patches(job=job, cosyvoice_error=CosyVoiceError("tts boom")) entered = [p.__enter__() for p in patches] try: tts_synthesize_and_submit.run("job-1", "user-1", "v1", "你好", 1.0, "") finally: for p in reversed(patches): p.__exit__(None, None, None) assert job.status == "failed" assert job.error_code == "TTSSynthesisFailed" session.close.assert_called_once() def test_value_error_marks_tts_invalid_param(self): """ValueError(参数错误)→ failed, error_code=TTSInvalidParam.""" from app.tasks.lipsync_tts import tts_synthesize_and_submit job = _make_fake_job() session, patches = _apply_all_patches(job=job, cosyvoice_side_effect=ValueError("bad param")) entered = [p.__enter__() for p in patches] try: tts_synthesize_and_submit.run("job-1", "user-1", "v1", "你好", -1.0, "") finally: for p in reversed(patches): p.__exit__(None, None, None) assert job.status == "failed" assert job.error_code == "TTSInvalidParam" session.close.assert_called_once() def test_no_audio_url_marks_tts_no_audio(self): """TTS 返回空 audio_url → failed, error_code=TTSNoAudio.""" from app.tasks.lipsync_tts import tts_synthesize_and_submit job = _make_fake_job() cosy = MagicMock() cosy.submit_synthesize_task.return_value = {"audio_url": ""} session, patches = _apply_all_patches(job=job, cosyvoice_service=cosy) entered = [p.__enter__() for p in patches] try: tts_synthesize_and_submit.run("job-1", "user-1", "v1", "你好", 1.0, "") finally: for p in reversed(patches): p.__exit__(None, None, None) assert job.status == "failed" assert job.error_code == "TTSNoAudio" session.close.assert_called_once() def test_oss_upload_failure_falls_back_to_temp_url(self): """OSS 上传失败 → 回退临时 URL,仍然 submitted.""" from app.tasks.lipsync_tts import tts_synthesize_and_submit job = _make_fake_job() storage = MagicMock() storage.public_url = "https://oss.example.com" storage.upload_file.side_effect = RuntimeError("oss down") mk_client = MagicMock() mk_client.submit_lipsync.return_value = {"task_id": "mk-7"} session, patches = _apply_all_patches(job=job, storage=storage, mk_client=mk_client) entered = [p.__enter__() for p in patches] try: tts_synthesize_and_submit.run("job-1", "user-1", "v1", "你好", 1.0, "") finally: for p in reversed(patches): p.__exit__(None, None, None) # 上传失败后 audio_url 回退为临时 TTS URL,仍继续提交到 MediaKit assert job.audio_url == "https://tts/raw.mp3" assert job.status == "submitted" assert job.mediakit_task_id == "mk-7" mk_client.submit_lipsync.assert_called_once() session.close.assert_called_once() def test_mediakit_error_marks_mediakit_unavailable(self): """MediaKit 提交失败 → failed, error_code=MediaKitUnavailable.""" from app.services.mediakit_client import MediaKitError from app.tasks.lipsync_tts import tts_synthesize_and_submit job = _make_fake_job() mk_err = MediaKitError("mk down", code="MediaKitUnavailable") session, patches = _apply_all_patches(job=job, mk_submit_error=mk_err) entered = [p.__enter__() for p in patches] try: tts_synthesize_and_submit.run("job-1", "user-1", "v1", "你好", 1.0, "") finally: for p in reversed(patches): p.__exit__(None, None, None) assert job.status == "failed" assert job.error_code == "MediaKitUnavailable" session.close.assert_called_once() def test_top_level_exception_marks_async_task_error(self): """顶层未预期异常 → failed, error_code=AsyncTaskError.""" from app.tasks.lipsync_tts import tts_synthesize_and_submit job = _make_fake_job() fake_app_db = ModuleType("app.db") fake_worker_db = ModuleType("worker_app.db") session, factory = _build_session(job) fake_app_db.SessionLocal = factory fake_worker_db.SessionLocal = factory # CosyVoiceService 在 __init__ 抛 RuntimeError(非 CosyVoiceError/ValueError) fake_cosy_mod = ModuleType("packages.application.cosyvoice_service") class _CosyVoiceErrorForTest(Exception): pass class _BoomService: def __init__(self): raise RuntimeError("top-level boom") fake_cosy_mod.CosyVoiceError = _CosyVoiceErrorForTest fake_cosy_mod.CosyVoiceService = _BoomService with patch.dict( sys.modules, { "app.db": fake_app_db, "worker_app.db": fake_worker_db, "packages.application.cosyvoice_service": fake_cosy_mod, }, ): tts_synthesize_and_submit.run("job-1", "user-1", "v1", "你好", 1.0, "") assert job.status == "failed" assert job.error_code == "AsyncTaskError" session.close.assert_called() class TestSignMediaUrl: """覆盖模块内 _sign_media_url 的所有分支(CI 增量覆盖率需要).""" def test_empty_url_returns_empty(self): from app.tasks.lipsync_tts import _sign_media_url assert _sign_media_url("") == "" assert _sign_media_url(None) is None def test_own_oss_url_signed(self): """自家 OSS URL → 调用 storage.get_download_url 签名.""" from app.tasks.lipsync_tts import _sign_media_url fake_storage = MagicMock() fake_storage.public_url = "https://oss.example.com/" fake_storage.get_download_url.return_value = "https://oss.example.com/a?sig=xyz" with patch("packages.shared.storage.get_shared_storage_service", return_value=fake_storage): result = _sign_media_url("https://oss.example.com/lipsync/a.mp3") assert result == "https://oss.example.com/a?sig=xyz" fake_storage.get_download_url.assert_called_once() def test_external_url_passthrough(self): """外部 URL(不是自家 OSS host)→ 原样透传,不签名.""" from app.tasks.lipsync_tts import _sign_media_url fake_storage = MagicMock() fake_storage.public_url = "https://oss.example.com/" with patch("packages.shared.storage.get_shared_storage_service", return_value=fake_storage): result = _sign_media_url("https://tts.example.com/raw.mp3") assert result == "https://tts.example.com/raw.mp3" fake_storage.get_download_url.assert_not_called() def test_storage_exception_falls_back(self): """storage 调用异常 → 降级原样返回,不抛错.""" from app.tasks.lipsync_tts import _sign_media_url with patch( "packages.shared.storage.get_shared_storage_service", side_effect=RuntimeError("storage down"), ): result = _sign_media_url("https://oss.example.com/a.mp3") assert result == "https://oss.example.com/a.mp3" def test_no_public_url_passthrough(self): """storage.public_url 为空 → 原样透传.""" from app.tasks.lipsync_tts import _sign_media_url fake_storage = MagicMock() fake_storage.public_url = "" with patch("packages.shared.storage.get_shared_storage_service", return_value=fake_storage): result = _sign_media_url("https://anything.example.com/a.mp3") assert result == "https://anything.example.com/a.mp3" fake_storage.get_download_url.assert_not_called()