xiaoxia-saas/tests/unit/test_lipsync_tts_preview.py

"""AI 数字人 对口型 TTS 预合成接口（#1845）单元测试 — 覆盖 LipsyncService.preview_tts 成功/失败路径.

直接调用 LipsyncService.preview_tts()，mock CosyVoiceService / safe_download_bytes / ffprobe，
验证返回结构、错误码、与共享 sentence_timings 工具的协作。
"""

import os
from unittest.mock import MagicMock, patch

import pytest

os.environ.setdefault("JWT_SECRET_KEY", "dev-secret-key-for-testing")


def _make_service(
    *,
    cosyvoice=None,
    download_bytes=b"FAKE_MP3_DATA",
    download_error=None,
    ffprobe_duration=5.0,
    timings_result=None,
):
    """构造 LipsyncService 并把 CosyVoiceService/safe_download_bytes/probe/compute 全部 mock 掉。"""
    from app.services.lipsync_service import LipsyncService

    db = MagicMock()

    # 构造唯一的 cosyvoice mock 实例，便于断言
    _cosy_inst = MagicMock()
    if cosyvoice is None:
        _cosy_inst.submit_synthesize_task.return_value = {"audio_url": "https://cosy.example.com/tts.mp3"}
    elif isinstance(cosyvoice, Exception):
        _cosy_inst.submit_synthesize_task.side_effect = cosyvoice
    else:
        _cosy_inst.submit_synthesize_task.return_value = cosyvoice

    def _fake_get_cosyvoice(self):  # noqa: ARG001
        return _cosy_inst

    def _fake_resolve_voice_id(self, voice_id, user_id):  # noqa: ARG001
        return voice_id

    svc = LipsyncService(db=db, client=MagicMock(), voice_clone_repo=MagicMock())
    svc._cosyvoice = _cosy_inst
    patch.object(LipsyncService, "_get_cosyvoice", _fake_get_cosyvoice).start()
    patch.object(LipsyncService, "_resolve_voice_id", _fake_resolve_voice_id).start()

    # mock safe_download_bytes
    if download_error is not None:
        patch(
            "app.services.lipsync_service.safe_download_bytes",
            side_effect=download_error,
        ).start()
    else:
        patch(
            "app.services.lipsync_service.safe_download_bytes",
            return_value=download_bytes,
        ).start()

    # mock probe_audio_duration（patch 到 lipsync_service 模块的命名空间）
    patch(
        "app.services.lipsync_service.probe_audio_duration",
        return_value=ffprobe_duration,
    ).start()

    # mock compute_sentence_timings
    default_timings = [
        {"index": 0, "text": "你好", "start_time": 0.0, "end_time": 1.5},
        {"index": 1, "text": "世界", "start_time": 1.5, "end_time": 5.0},
    ]
    patch(
        "app.services.lipsync_service.compute_sentence_timings",
        return_value=timings_result if timings_result is not None else default_timings,
    ).start()

    svc.__dict__["_test_cosy"] = _cosy_inst
    return svc


def test_preview_tts_success():
    """正常路径：TTS 合成成功 → 下载 → ffprobe → 计算 timings，返回完整结构。"""
    svc = _make_service(ffprobe_duration=5.0)
    try:
        result = svc.preview_tts(
            user_id="user-1",
            voice_id="longxiaochun",
            script_text="你好，世界",
            speed=1.0,
            emotion="natural",
        )
        assert result["audio_url"] == "https://cosy.example.com/tts.mp3"
        assert result["duration"] == 5.0
        assert isinstance(result["sentence_timings"], list)
        assert len(result["sentence_timings"]) == 2
        assert result["sentence_timings"][0]["text"] == "你好"
        cosy = svc.__dict__["_test_cosy"]
        cosy.submit_synthesize_task.assert_called_once()
        kwargs = cosy.submit_synthesize_task.call_args.kwargs
        assert kwargs["text"] == "你好，世界"
        assert kwargs["voice_id"] == "longxiaochun"
    finally:
        patch.stopall()


def test_preview_tts_cosyvoice_error():
    """CosyVoice 抛错：应该包装成 MediaKitError 抛出。"""
    from app.services.mediakit_client import MediaKitError

    from packages.application.cosyvoice_service import CosyVoiceError

    svc = _make_service(cosyvoice=CosyVoiceError("cosyvoice down"))
    try:
        with pytest.raises(MediaKitError):
            svc.preview_tts(
                user_id="user-1",
                voice_id="longxiaochun",
                script_text="你好",
            )
    finally:
        patch.stopall()


def test_preview_tts_download_fail_still_returns_url():
    """音频下载失败：不抛错，返回 audio_url + 空 timings，前端仍能继续（降级）。"""
    svc = _make_service(download_error=RuntimeError("network down"))
    try:
        result = svc.preview_tts(
            user_id="user-1",
            voice_id="longxiaochun",
            script_text="你好，世界",
        )
        assert result["audio_url"] == "https://cosy.example.com/tts.mp3"
        assert result["duration"] == 0.0
        assert result["sentence_timings"] == []
    finally:
        patch.stopall()


def test_preview_tts_ffprobe_zero_duration():
    """ffprobe 返回 0：timings 为空，不抛错。"""
    svc = _make_service(ffprobe_duration=0.0)
    try:
        result = svc.preview_tts(
            user_id="user-1",
            voice_id="longxiaochun",
            script_text="你好",
        )
        assert result["audio_url"]
        assert result["duration"] == 0.0
        assert result["sentence_timings"] == []
    finally:
        patch.stopall()


def test_preview_tts_no_audio_url_in_response():
    """CosyVoice 返回无 audio_url：抛 MediaKitError TTSNoAudio。"""
    from app.services.mediakit_client import MediaKitError

    svc = _make_service(cosyvoice={"audio_url": ""})
    try:
        with pytest.raises(MediaKitError) as exc_info:
            svc.preview_tts(
                user_id="user-1",
                voice_id="longxiaochun",
                script_text="你好",
            )
        assert exc_info.value.code == "TTSNoAudio"
    finally:
        patch.stopall()