xiaoxia-saas/tests/unit/test_tts_synthesize_clone_voice.py

"""POST /tts/synthesize 克隆音色解析的单元测试（路由级）。

覆盖：voice_id 传克隆 profile UUID / 显式 voice_clone_profile_id /
预置音色+profile 同传时以克隆为准 / 403 / 400 未完成 / 404 / 预置透传。
"""

from __future__ import annotations

from datetime import datetime, timezone
from unittest.mock import MagicMock, patch

import pytest

from packages.domain.tts_job import TTSJob, TTSJobStatus


class FakeCloneProfile:
    def __init__(self, pid, user_id="user-1", voice_id="clone-voice-001"):
        self.id = pid
        self.user_id = user_id
        self.voice_id = voice_id


class FakeCloneRepo:
    def __init__(self, profiles):
        self._profiles = {p.id: p for p in profiles}

    def get(self, pid):
        return self._profiles.get(pid)


def _make_job(voice_id, voice_clone_profile_id=""):
    return TTSJob(
        id="job-new",
        user_id="user-1",
        input_text="x",
        voice_id=voice_id,
        voice_model="",
        project_id="",
        voice_clone_profile_id=voice_clone_profile_id,
        status=TTSJobStatus.PROCESSING,
        output_audio_url="",
        output_audio_key="",
        duration=0.0,
        file_size=0,
        sample_rate=22050,
        format="mp3",
        error_message="",
        retry_count=0,
        max_retries=3,
        metadata={},
        started_at=None,
        completed_at=None,
        created_at=datetime.now(timezone.utc),
        updated_at=datetime.now(timezone.utc),
    )


def _build_app(clone_repo):
    from fastapi import FastAPI

    app = FastAPI()
    from app.api.routes.tts import _get_repository, router
    from app.auth import get_current_user
    from app.dependencies import get_cosyvoice_service, get_voice_clone_profile_repository

    app.include_router(router, prefix="/tts")

    fake_user = MagicMock()
    fake_user.user.id = "user-1"
    app.dependency_overrides[get_current_user] = lambda: fake_user
    app.dependency_overrides[get_voice_clone_profile_repository] = lambda: clone_repo
    app.dependency_overrides[get_cosyvoice_service] = lambda: MagicMock()
    app.dependency_overrides[_get_repository] = lambda: MagicMock()
    return app


class TestSynthesizeCloneVoiceResolution:
    def _post(self, client, payload):
        captured = {}

        def fake_execute(*args, **kwargs):
            captured.update(kwargs)
            return _make_job(kwargs["voice_id"], kwargs.get("voice_clone_profile_id", ""))

        def fake_start(*args, **kwargs):
            return _make_job(captured.get("voice_id", ""), captured.get("voice_clone_profile_id", ""))

        with (
            patch(
                "packages.application.tts_job.use_cases.CreateTTSJobUseCase.execute",
                side_effect=fake_execute,
            ),
            patch(
                "packages.application.tts_job.workflow.TTSWorkflowService.start_synthesis",
                side_effect=fake_start,
            ),
            patch("app.api.routes.tts.celery_app.send_task"),
        ):
            resp = client.post("/tts/synthesize", json=payload)
        return resp, captured

    def test_voice_id_as_profile_uuid_resolves_to_clone_voice(self):
        from fastapi.testclient import TestClient

        profile = FakeCloneProfile("profile-uuid-1")
        client = TestClient(_build_app(FakeCloneRepo([profile])))
        resp, captured = self._post(client, {"text": "克隆音色", "voice_id": "profile-uuid-1"})
        assert resp.status_code == 201, resp.text
        assert captured["voice_id"] == "clone-voice-001"
        assert captured["voice_clone_profile_id"] == "profile-uuid-1"

    def test_explicit_profile_id_resolves(self):
        from fastapi.testclient import TestClient

        profile = FakeCloneProfile("profile-uuid-2")
        client = TestClient(_build_app(FakeCloneRepo([profile])))
        resp, captured = self._post(client, {"text": "显式", "voice_clone_profile_id": "profile-uuid-2"})
        assert resp.status_code == 201, resp.text
        assert captured["voice_id"] == "clone-voice-001"

    def test_preset_voice_id_with_explicit_profile_uses_clone(self):
        """voice_id 传预置音色 + 显式克隆 profile（不一致）→ 必须以克隆 profile 为准。"""
        from fastapi.testclient import TestClient

        profile = FakeCloneProfile("profile-uuid-3")
        client = TestClient(_build_app(FakeCloneRepo([profile])))
        resp, captured = self._post(
            client,
            {"text": "混合", "voice_id": "longxiaochun_v2", "voice_clone_profile_id": "profile-uuid-3"},
        )
        assert resp.status_code == 201, resp.text
        assert captured["voice_id"] == "clone-voice-001"
        assert captured["voice_clone_profile_id"] == "profile-uuid-3"

    def test_other_user_profile_returns_403(self):
        from fastapi.testclient import TestClient

        profile = FakeCloneProfile("profile-other", user_id="other-user")
        client = TestClient(_build_app(FakeCloneRepo([profile])))
        resp, _ = self._post(client, {"text": "越权", "voice_id": "profile-other"})
        assert resp.status_code == 403

    def test_unfinished_clone_returns_400(self):
        from fastapi.testclient import TestClient

        profile = FakeCloneProfile("profile-pending", voice_id="")
        client = TestClient(_build_app(FakeCloneRepo([profile])))
        resp, _ = self._post(client, {"text": "未完成", "voice_id": "profile-pending"})
        assert resp.status_code == 400
        assert "克隆尚未完成" in resp.json()["detail"]

    def test_nonexistent_profile_returns_404(self):
        from fastapi.testclient import TestClient

        client = TestClient(_build_app(FakeCloneRepo([])))
        resp, _ = self._post(client, {"text": "不存在", "voice_clone_profile_id": "no-such-profile"})
        assert resp.status_code == 404

    def test_preset_voice_passes_through(self):
        """预置音色（不命中任何 profile）→ voice_id 原样透传。"""
        from fastapi.testclient import TestClient

        client = TestClient(_build_app(FakeCloneRepo([])))
        resp, captured = self._post(client, {"text": "预置", "voice_id": "longxiaochun_v2"})
        assert resp.status_code == 201, resp.text
        assert captured["voice_id"] == "longxiaochun_v2"
        assert captured["voice_clone_profile_id"] == ""