xiaoxia-saas/tests/unit/test_lipsync_routes.py

"""对口型 API 路由 + Service 单元测试 — #1796, #1809 参数调整.

CI 增量映射: lipsync.py (route) + lipsync_service.py → test_lipsync_routes.py
"""

import os
from unittest.mock import MagicMock, patch

import pytest

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


@pytest.fixture
def mock_mediakit():
    """Mock MediaKit 客户端."""
    client = MagicMock()
    client.is_available = True
    client.submit_lipsync.return_value = {
        "success": True,
        "task_id": "mk-task-123",
        "request_id": "mk-req-456",
    }
    client.get_task_status.return_value = {
        "success": True,
        "task_id": "mk-task-123",
        "status": "completed",
        "result": {"video_url": "https://output.mp4", "duration": 30.0},
        "created_at": 1777291767,
        "finished_at": 1777291851,
        "expires_at": 1777464650,
    }
    return client


@pytest.fixture
def mock_cosyvoice():
    """Mock CosyVoice 服务（v3: service 内部走 submit_synthesize_task，返回 dict）."""
    service = MagicMock()
    service.submit_synthesize_task.return_value = {
        "audio_url": "https://oss.example.com/tts-output.mp3",
        "request_id": "tts-req-789",
    }
    # synthesize_speech 保留给直接同步调用场景
    service.synthesize_speech.return_value = MagicMock(audio_url="https://oss.example.com/tts-output.mp3")
    return service


def _make_mock_job(
    job_id="job-1",
    user_id="user-1",
    status="submitted",
    mediakit_task_id="mk-task-123",
    output_video_url="",
    output_duration=0.0,
    error_message="",
    error_code="",
):
    m = MagicMock()
    m.id = job_id
    m.user_id = user_id
    m.project_id = ""
    m.video_url = "https://example.com/video.mp4"
    m.audio_url = "https://oss.example.com/tts-output.mp3"
    m.enable_video_loop = False
    m.mediakit_task_id = mediakit_task_id
    m.status = status
    m.output_video_url = output_video_url
    m.output_duration = output_duration
    m.error_message = error_message
    m.error_code = error_code
    m.submitted_at = None
    m.completed_at = None
    m.created_at = None
    m.updated_at = None
    return m


class TestSchemaValidation:
    """Schema 验证测试 — #1809 新参数结构."""

    def test_valid_request(self):
        from app.schemas.lipsync import CreateLipsyncJobRequest

        req = CreateLipsyncJobRequest(
            video_url="https://example.com/video.mp4",
            voice_id="longxiaochun_v3",
            script_text="大家好，欢迎来到直播间",
        )
        assert req.video_url == "https://example.com/video.mp4"
        assert req.voice_id == "longxiaochun_v3"
        assert req.script_text == "大家好，欢迎来到直播间"

    def test_invalid_video_url_unsupported_ext(self):
        from app.schemas.lipsync import CreateLipsyncJobRequest

        # 不支持的扩展名（.txt）应报错
        with pytest.raises(ValueError, match="格式不支持"):
            CreateLipsyncJobRequest(
                video_url="https://example.com/video.txt",
                voice_id="longxiaochun_v3",
                script_text="测试文本",
            )

    def test_mov_video_url_accepted(self):
        from app.schemas.lipsync import CreateLipsyncJobRequest

        # .MOV 是 iPhone 拍摄的常见容器，h264 编码可直接被 MediaKit 处理
        req = CreateLipsyncJobRequest(
            video_url="https://example.com/video.mov",
            audio_url="https://example.com/audio.mp3",
        )
        assert req.video_url.endswith(".mov")

    def test_invalid_video_url_empty(self):
        from app.schemas.lipsync import CreateLipsyncJobRequest

        with pytest.raises(ValueError, match="不能为空"):
            CreateLipsyncJobRequest(
                video_url="  ",
                voice_id="longxiaochun_v3",
                script_text="测试文本",
            )

    def test_invalid_video_url_not_http(self):
        from app.schemas.lipsync import CreateLipsyncJobRequest

        with pytest.raises(ValueError, match="HTTP"):
            CreateLipsyncJobRequest(
                video_url="ftp://example.com/video.mp4",
                voice_id="longxiaochun_v3",
                script_text="测试文本",
            )

    def test_empty_voice_id_rejected(self):
        from app.schemas.lipsync import CreateLipsyncJobRequest

        with pytest.raises(ValueError, match="voice_id"):
            CreateLipsyncJobRequest(
                video_url="https://example.com/video.mp4",
                voice_id="  ",
                script_text="测试文本",
            )

    def test_empty_script_text_rejected(self):
        from app.schemas.lipsync import CreateLipsyncJobRequest

        with pytest.raises(ValueError, match="script_text"):
            CreateLipsyncJobRequest(
                video_url="https://example.com/video.mp4",
                voice_id="longxiaochun_v3",
                script_text="",
            )

    def test_script_text_too_long(self):
        from app.schemas.lipsync import CreateLipsyncJobRequest

        with pytest.raises(ValueError, match="5000"):
            CreateLipsyncJobRequest(
                video_url="https://example.com/video.mp4",
                voice_id="longxiaochun_v3",
                script_text="x" * 5001,
            )

    def test_enable_video_loop_default(self):
        from app.schemas.lipsync import CreateLipsyncJobRequest

        req = CreateLipsyncJobRequest(
            video_url="https://example.com/video.mp4",
            voice_id="longxiaochun_v3",
            script_text="测试文本",
        )
        # AI数字人场景文案长度不可控，默认开启视频循环，防止音频长于视频时被截断
        assert req.enable_video_loop is True

    def test_video_url_strip_query_params(self):
        """视频 URL 含查询参数时，扩展名检查应忽略 ? 后面的部分."""
        from app.schemas.lipsync import CreateLipsyncJobRequest

        req = CreateLipsyncJobRequest(
            video_url="https://example.com/video.mp4?token=abc",
            voice_id="longxiaochun_v3",
            script_text="测试文本",
        )
        assert "?token=" in req.video_url

    def test_dual_mode_fields_present(self):
        """v3 契约: 双模式——支持直接音频 audio_url，也支持 TTS 直生 voice_id+script_text."""
        from app.schemas.lipsync import CreateLipsyncJobRequest

        fields = CreateLipsyncJobRequest.model_fields.keys()
        # 直接音频模式
        assert "audio_url" in fields
        # TTS 直生模式
        assert "voice_id" in fields
        assert "script_text" in fields
        # 语速/情绪透传
        assert "speed" in fields
        assert "emotion" in fields

    def test_direct_audio_mode_accepted(self):
        """v3 契约: 只传 audio_url（直接音频模式）也合法，无需 voice_id/script_text."""
        from app.schemas.lipsync import CreateLipsyncJobRequest

        req = CreateLipsyncJobRequest(
            video_url="https://example.com/video.mp4",
            audio_url="https://example.com/audio.mp3",
        )
        assert req.audio_url == "https://example.com/audio.mp3"

    def test_neither_mode_rejected(self):
        """v3 契约: audio_url 与 voice_id+script_text 都缺时应报错."""
        from app.schemas.lipsync import CreateLipsyncJobRequest

        with pytest.raises(ValueError):
            CreateLipsyncJobRequest(video_url="https://example.com/video.mp4")


class TestLipsyncServiceUnit:
    """Service 层单元测试（纯 mock，不依赖数据库）— #1809 更新."""

    def test_create_job_success(self, mock_mediakit, mock_cosyvoice):
        """TTS 直生——v4 异步模式：create_job 只创建 DB 记录 + dispatch Celery 任务."""
        from app.services.lipsync_service import LipsyncService

        mock_db = MagicMock()
        mock_repo = MagicMock()
        mock_repo.get.return_value = None  # 预置音色，原样返回 voice_id

        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(
                mock_db,
                client=mock_mediakit,
                voice_clone_repo=mock_repo,
            )

            job = svc.create_job(
                user_id="user-1",
                video_url="https://example.com/video.mp4",
                voice_id="longxiaochun_v3",
                script_text="大家好，欢迎来到直播间",
                speed=1.2,
                emotion="兴奋",
            )

        # TTS 模式：异步返回，状态为 tts_processing
        assert job.status == "tts_processing"
        assert not job.audio_url  # TTS 音频尚未合成（默认空字符串）
        # 不直接调用 CosyVoice
        mock_cosyvoice.submit_synthesize_task.assert_not_called()
        # dispatch 了 Celery 任务
        mock_task.apply_async.assert_called_once()
        # job 记录透传字段
        assert job.speed == 1.2
        assert job.emotion == "excited"
        # MediaKit 尚未提交（由 Celery 任务处理）
        mock_mediakit.submit_lipsync.assert_not_called()

    def test_create_job_tts_failure(self, mock_mediakit):
        """v4: TTS 模式下 create_job 不再同步失败，而是 dispatch Celery 任务。
        TTS 合成失败由 Celery 任务内部处理（见 test_lipsync_speed_optimization.py）。"""
        from app.services.lipsync_service import LipsyncService

        mock_db = MagicMock()
        mock_repo = MagicMock()
        mock_repo.get.return_value = None

        svc = LipsyncService(
            mock_db,
            client=mock_mediakit,
            voice_clone_repo=mock_repo,
        )

        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(
                user_id="user-1",
                video_url="https://example.com/video.mp4",
                voice_id="longxiaochun_v3",
                script_text="测试文本",
            )

        # TTS 模式下 create_job 成功返回，状态为 tts_processing
        assert job.status == "tts_processing"
        mock_mediakit.submit_lipsync.assert_not_called()
        mock_task.apply_async.assert_called_once()

    def test_create_job_api_failure(self, mock_mediakit):
        """MediaKit 提交失败（直传音频模式同步触发）."""
        from app.services.lipsync_service import LipsyncService
        from app.services.mediakit_client import MediaKitError

        mock_mediakit.submit_lipsync.side_effect = MediaKitError("API 调用失败", code="SubmitFailed")

        mock_db = MagicMock()
        svc = LipsyncService(mock_db, client=mock_mediakit)

        with pytest.raises(MediaKitError, match="API 调用失败"):
            svc.create_job(
                user_id="user-1",
                video_url="https://example.com/video.mp4",
                audio_url="https://example.com/audio.mp3",
            )

    def test_get_job_delegates_to_db(self, mock_mediakit, mock_cosyvoice):
        from app.services.lipsync_service import LipsyncService

        mock_job = _make_mock_job()
        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)
        result = svc.get_job("job-1", "user-1")

        assert result is mock_job
        mock_db.query.assert_called_once()

    def test_get_job_not_found(self, mock_mediakit, mock_cosyvoice):
        from app.services.lipsync_service import LipsyncService

        mock_db = MagicMock()
        mock_query = MagicMock()
        mock_filter = MagicMock()
        mock_filter.first.return_value = None
        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)
        result = svc.get_job("nonexistent", "user-1")
        assert result is None

    def test_refresh_job_completed(self, mock_mediakit, mock_cosyvoice):
        from app.services.lipsync_service import LipsyncService

        mock_job = _make_mock_job(status="submitted")
        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)
        result = svc.refresh_job_status("job-1", "user-1")

        assert result.status == "completed"
        assert result.output_video_url == "https://output.mp4"
        assert result.output_duration == 30.0

    def test_refresh_job_failed(self, mock_mediakit, mock_cosyvoice):
        from app.services.lipsync_service import LipsyncService

        mock_mediakit.get_task_status.return_value = {
            "success": True,
            "task_id": "mk-task-123",
            "status": "failed",
            "error": {"code": "DownloadFailed", "message": "无法下载"},
            "created_at": 1777291767,
            "finished_at": 1777291851,
        }

        mock_job = _make_mock_job(status="submitted")
        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)
        result = svc.refresh_job_status("job-1", "user-1")

        assert result.status == "failed"
        assert result.error_code == "DownloadFailed"

    def test_refresh_job_already_completed(self, mock_mediakit, mock_cosyvoice):
        """已完成的任务不轮询."""
        from app.services.lipsync_service import LipsyncService

        mock_job = _make_mock_job(status="completed")
        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)
        result = svc.refresh_job_status("job-1", "user-1")

        # 不应调用 MediaKit
        mock_mediakit.get_task_status.assert_not_called()
        assert result.status == "completed"

    def test_cancel_job_pending(self, mock_mediakit, mock_cosyvoice):
        from app.services.lipsync_service import LipsyncService

        mock_job = _make_mock_job(status="pending")
        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)
        result = svc.cancel_job("job-1", "user-1")

        assert result.status == "cancelled"

    def test_cancel_job_completed_not_allowed(self, mock_mediakit, mock_cosyvoice):
        from app.services.lipsync_service import LipsyncService

        mock_job = _make_mock_job(status="completed")
        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)
        result = svc.cancel_job("job-1", "user-1")

        # 已完成不可取消
        assert result.status == "completed"

    def test_create_job_stores_tts_audio_url(self, mock_mediakit, mock_cosyvoice):
        """v4: TTS 模式下 create_job 返回 tts_processing 状态，audio_url 尚未设置（由 Celery 任务处理）."""
        from app.services.lipsync_service import LipsyncService

        mock_db = MagicMock()
        mock_repo = MagicMock()
        mock_repo.get.return_value = None

        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(
                mock_db,
                client=mock_mediakit,
                voice_clone_repo=mock_repo,
            )
            job = svc.create_job(
                user_id="user-1",
                video_url="https://example.com/video.mp4",
                voice_id="my-clone-voice",
                script_text="这是一段测试文本",
            )

        # TTS 模式下 create_job 返回 tts_processing 状态
        assert job.status == "tts_processing"
        # audio_url 尚未设置（由 Celery 任务异步处理），模型默认为空字符串
        assert not job.audio_url
        mock_task.apply_async.assert_called_once()

    def test_create_job_direct_audio_skips_tts(self, mock_mediakit, mock_cosyvoice):
        """v3: 直接音频模式（传 audio_url）不触发 TTS，原样把 audio_url 提交 MediaKit."""
        from app.services.lipsync_service import LipsyncService

        mock_db = MagicMock()
        svc = LipsyncService(
            mock_db,
            client=mock_mediakit,
            cosyvoice_service=mock_cosyvoice,
            voice_clone_repo=MagicMock(),
        )
        job = svc.create_job(
            user_id="user-1",
            video_url="https://example.com/video.mp4",
            audio_url="https://example.com/direct-audio.mp3",
        )

        mock_cosyvoice.submit_synthesize_task.assert_not_called()
        call_kwargs = mock_mediakit.submit_lipsync.call_args
        assert call_kwargs.kwargs["audio_url"] == "https://example.com/direct-audio.mp3"
        assert job.audio_url == "https://example.com/direct-audio.mp3"


class TestErrorHandling:
    """v3: 音色解析与错误码在 service 层处理，路由层做 HTTP 状态码映射."""

    def test_voice_id_resolve_forbidden(self, mock_mediakit, mock_cosyvoice):
        """v3: 克隆音色属于他人时 service._resolve_voice_id 抛 VoiceForbidden（路由映射 403）."""
        from app.services.lipsync_service import LipsyncService
        from app.services.mediakit_client import MediaKitError

        mock_db = MagicMock()
        mock_repo = MagicMock()
        other_profile = MagicMock()
        other_profile.user_id = "user-other"
        other_profile.voice_id = "cv-voice-1"
        mock_repo.get.return_value = other_profile

        svc = LipsyncService(
            mock_db,
            client=mock_mediakit,
            cosyvoice_service=mock_cosyvoice,
            voice_clone_repo=mock_repo,
        )

        with pytest.raises(MediaKitError) as exc_info:
            svc.create_job(
                user_id="user-1",
                video_url="https://example.com/video.mp4",
                voice_id="clone-profile-id",
                script_text="测试",
            )
        assert exc_info.value.code == "VoiceForbidden"
        mock_mediakit.submit_lipsync.assert_not_called()

    def test_voice_id_resolve_not_ready(self, mock_mediakit, mock_cosyvoice):
        """v3: 克隆音色尚未生成 voice_id 时抛 VoiceNotReady（路由映射 400）."""
        from app.services.lipsync_service import LipsyncService
        from app.services.mediakit_client import MediaKitError

        mock_db = MagicMock()
        mock_repo = MagicMock()
        profile = MagicMock()
        profile.user_id = "user-1"
        profile.voice_id = ""  # 克隆未完成
        mock_repo.get.return_value = profile

        svc = LipsyncService(
            mock_db,
            client=mock_mediakit,
            cosyvoice_service=mock_cosyvoice,
            voice_clone_repo=mock_repo,
        )

        with pytest.raises(MediaKitError) as exc_info:
            svc.create_job(
                user_id="user-1",
                video_url="https://example.com/video.mp4",
                voice_id="clone-profile-id",
                script_text="测试",
            )
        assert exc_info.value.code == "VoiceNotReady"

    def test_tts_value_error_mapped_to_invalid_param(self, mock_mediakit):
        """v4: TTS 模式下 create_job 不再同步调用 CosyVoice，
        而是 dispatch Celery 任务。ValueError 由 Celery 任务内部处理。"""
        from app.services.lipsync_service import LipsyncService

        mock_cosyvoice = MagicMock()
        mock_cosyvoice.submit_synthesize_task.side_effect = ValueError("voice_id 为空")

        mock_db = MagicMock()
        mock_repo = MagicMock()
        mock_repo.get.return_value = None

        svc = LipsyncService(
            mock_db,
            client=mock_mediakit,
            voice_clone_repo=mock_repo,
        )

        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 不再同步失败
            job = svc.create_job(
                user_id="user-1",
                video_url="https://example.com/video.mp4",
                voice_id="some-voice",
                script_text="test",
            )

        # 确认返回 tts_processing 状态
        assert job.status == "tts_processing"
        # TTS 合成由 Celery 任务处理，不直接调用 CosyVoice
        mock_cosyvoice.submit_synthesize_task.assert_not_called()
        mock_task.apply_async.assert_called_once()

    def test_missing_both_inputs_raises_invalid_input(self, mock_mediakit, mock_cosyvoice):
        """v3: 既无 audio_url 又无 voice_id+script_text 时抛 InvalidInput（路由映射 400）."""
        from app.services.lipsync_service import LipsyncService
        from app.services.mediakit_client import MediaKitError

        mock_db = MagicMock()
        svc = LipsyncService(
            mock_db,
            client=mock_mediakit,
            cosyvoice_service=mock_cosyvoice,
            voice_clone_repo=MagicMock(),
        )

        with pytest.raises(MediaKitError) as exc_info:
            svc.create_job(
                user_id="user-1",
                video_url="https://example.com/video.mp4",
            )
        assert exc_info.value.code == "InvalidInput"
        mock_cosyvoice.submit_synthesize_task.assert_not_called()
        mock_mediakit.submit_lipsync.assert_not_called()


class TestSignMediaUrl403Fix:
    """#1839 私有桶 OSS URL 重签：MediaKit GPU worker 拉取裸/过期 URL 会 403.

    - 自家 OSS 的裸 public_url / 已过期短预签名 → 重签 7 天长有效期
    - 外部临时 URL（CosyVoice/MediaKit）→ 原样透传
    - 签名异常 → 降级原 URL，不阻断
    """

    OSS_PUBLIC_BASE = "https://xiaoxia-autocut.oss-cn-hangzhou.aliyuncs.com"

    def _svc(self, mock_mediakit, mock_cosyvoice):
        from app.services.lipsync_service import LipsyncService

        return LipsyncService(
            MagicMock(),
            client=mock_mediakit,
            cosyvoice_service=mock_cosyvoice,
            voice_clone_repo=MagicMock(),
        )

    def test_own_oss_unsigned_url_gets_resigned(self, mock_mediakit, mock_cosyvoice):
        """裸 public_url（不带签名，私有桶匿名 403）必须被重签."""
        from app.services.lipsync_service import MEDIAKIT_URL_TTL_SECONDS

        svc = self._svc(mock_mediakit, mock_cosyvoice)
        raw = f"{self.OSS_PUBLIC_BASE}/lipsync-tts/user-1/audio.mp3"
        signed = raw + "?Expires=999&Signature=abc&OSSAccessKeyId=key"

        storage = MagicMock()
        storage.public_url = self.OSS_PUBLIC_BASE
        storage.get_download_url.return_value = signed
        with patch("app.services.lipsync_service.get_shared_storage_service", return_value=storage):
            out = svc._sign_media_url(raw)

        assert out == signed
        storage.get_download_url.assert_called_once_with(raw, expires_seconds=MEDIAKIT_URL_TTL_SECONDS)
        assert MEDIAKIT_URL_TTL_SECONDS == 7 * 24 * 3600

    def test_own_oss_expired_presign_gets_resigned(self, mock_mediakit, mock_cosyvoice):
        """已带过期签名的旧预签名 URL 也要抽 key 后重签（不把旧 query 带进新签名）."""
        svc = self._svc(mock_mediakit, mock_cosyvoice)
        old = f"{self.OSS_PUBLIC_BASE}/avatar/video.mp4?Expires=111&Signature=old"
        fresh = f"{self.OSS_PUBLIC_BASE}/avatar/video.mp4?Expires=999&Signature=fresh"

        storage = MagicMock()
        storage.public_url = self.OSS_PUBLIC_BASE
        storage.get_download_url.return_value = fresh
        with patch("app.services.lipsync_service.get_shared_storage_service", return_value=storage):
            out = svc._sign_media_url(old)

        assert out == fresh
        # 传给 get_download_url 的是原始完整 URL（内部抽 key），有效期 7 天
        called_url = storage.get_download_url.call_args.args[0]
        assert called_url == old

    def test_external_url_passthrough(self, mock_mediakit, mock_cosyvoice):
        """CosyVoice/MediaKit 外部临时链接不处理，原样透传."""
        svc = self._svc(mock_mediakit, mock_cosyvoice)
        external = "https://cv-tts.cosyvoice.aliyuncs.com/output/x.mp3"

        storage = MagicMock()
        storage.public_url = self.OSS_PUBLIC_BASE
        with patch("app.services.lipsync_service.get_shared_storage_service", return_value=storage):
            out = svc._sign_media_url(external)

        assert out == external
        storage.get_download_url.assert_not_called()

    def test_empty_url_returns_empty(self, mock_mediakit, mock_cosyvoice):
        svc = self._svc(mock_mediakit, mock_cosyvoice)
        assert svc._sign_media_url("") == ""

    def test_sign_error_falls_back_to_raw(self, mock_mediakit, mock_cosyvoice):
        """签名抛异常时降级返回原 URL，不阻断对口型提交."""
        svc = self._svc(mock_mediakit, mock_cosyvoice)
        raw = f"{self.OSS_PUBLIC_BASE}/lipsync-tts/u/a.mp3"

        storage = MagicMock()
        storage.public_url = self.OSS_PUBLIC_BASE
        storage.get_download_url.side_effect = RuntimeError("oss down")
        with patch("app.services.lipsync_service.get_shared_storage_service", return_value=storage):
            out = svc._sign_media_url(raw)

        assert out == raw

    def test_create_job_resigns_oss_urls_before_submit(self, mock_mediakit, mock_cosyvoice):
        """端到端：create_job 提交 MediaKit 前，自家 OSS 的 video_url 必须是重签后的长签名 URL."""
        svc = self._svc(mock_mediakit, mock_cosyvoice)
        raw_video = f"{self.OSS_PUBLIC_BASE}/avatar/person.mp4"
        signed_video = raw_video + "?Expires=999&Signature=fresh"
        raw_audio = f"{self.OSS_PUBLIC_BASE}/direct/audio.mp3"
        signed_audio = raw_audio + "?Expires=999&Signature=afresh"

        storage = MagicMock()
        storage.public_url = self.OSS_PUBLIC_BASE
        storage.get_download_url.side_effect = lambda u, expires_seconds=0: (
            signed_video if u == raw_video else signed_audio
        )
        with patch("app.services.lipsync_service.get_shared_storage_service", return_value=storage):
            svc.create_job(
                user_id="user-1",
                video_url=raw_video,
                audio_url=raw_audio,
            )

        kw = mock_mediakit.submit_lipsync.call_args.kwargs
        assert kw["video_url"] == signed_video
        assert kw["audio_url"] == signed_audio