xiaoxia-saas/tests/integration/test_voice_clone_api.py

"""
声音克隆 API 集成测试。

覆盖端点：
- POST   /voice-clones — 创建声音克隆
- GET    /voice-clones — 列出声音克隆
- GET    /voice-clones/{clone_id} — 获取克隆详情
- GET    /voice-clones/{clone_id}/status — 获取克隆状态
- POST   /voice-clones/{clone_id}/retry — 重试克隆
- DELETE /voice-clones/{clone_id} — 删除克隆

使用 FastAPI TestClient + dependency_overrides 模式，
mock repository 和 CosyVoice 服务。
"""

from __future__ import annotations

import os
import sys
from datetime import datetime, timezone
from unittest.mock import MagicMock

# ── 环境变量 & sys.path（必须在导入 app.* 之前设置） ──────────────────────────
os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")

import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient

sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "apps", "api"))

from app.api.routes.voice_clones import router
from app.auth import AuthenticatedUser, get_current_user
from app.dependencies import get_cosyvoice_service, get_voice_clone_profile_repository

from packages.domain.entities import User
from packages.domain.voice_clone_profile import (
    VoiceCloneProfile,
    VoiceCloneStatus,
)

# ---------------------------------------------------------------------------
# 1. 内存 Repository
# ---------------------------------------------------------------------------


class InMemoryVoiceCloneProfileRepository:
    """内存中的音色克隆档案 Repository。"""

    def __init__(self):
        self._items: dict[str, VoiceCloneProfile] = {}

    def create(self, profile: VoiceCloneProfile) -> VoiceCloneProfile:
        self._items[profile.id] = profile
        return profile

    def get(self, profile_id: str) -> VoiceCloneProfile | None:
        return self._items.get(profile_id)

    def update(self, profile: VoiceCloneProfile) -> VoiceCloneProfile:
        self._items[profile.id] = profile
        return profile

    def delete(self, profile_id: str) -> bool:
        if profile_id in self._items:
            del self._items[profile_id]
            return True
        return False

    def list_by_user(
        self,
        user_id: str,
        *,
        status=None,
        limit: int = 50,
        offset: int = 0,
    ) -> list[VoiceCloneProfile]:
        items = [p for p in self._items.values() if p.user_id == user_id]
        if status:
            status_str = status.value if hasattr(status, "value") else str(status)
            items = [p for p in items if p.status.value == status_str]
        # 按 created_at 倒序
        items.sort(key=lambda p: p.created_at, reverse=True)
        return items[offset : offset + limit]

    def count_by_user(
        self,
        user_id: str,
        *,
        status=None,
    ) -> int:
        items = [p for p in self._items.values() if p.user_id == user_id]
        if status:
            status_str = status.value if hasattr(status, "value") else str(status)
            items = [p for p in items if p.status.value == status_str]
        return len(items)

    def find_by_voice_id(self, voice_id: str) -> VoiceCloneProfile | None:
        for p in self._items.values():
            if p.voice_id == voice_id:
                return p
        return None

    def find_profile_ids_by_voice_ids(self, voice_ids: list[str]) -> dict[str, str]:
        result = {}
        for p in self._items.values():
            if p.voice_id in voice_ids:
                result[p.voice_id] = p.id
        return result


# ---------------------------------------------------------------------------
# 2. Mock CosyVoice 服务
# ---------------------------------------------------------------------------


class MockCosyVoiceService:
    """Mock CosyVoice 服务，模拟克隆任务提交和状态查询。"""

    def __init__(self, *, fail_submit: bool = False, async_mode: bool = True):
        self.fail_submit = fail_submit
        self.async_mode = async_mode
        self.submit_called = False
        self.submit_args = None

    def submit_clone_task(self, *, audio_url: str, voice_name: str, language: str = "zh-CN") -> dict:
        self.submit_called = True
        self.submit_args = {"audio_url": audio_url, "voice_name": voice_name, "language": language}

        if self.fail_submit:
            from packages.application.cosyvoice_service import CosyVoiceError

            raise CosyVoiceError("模拟 CosyVoice 提交失败")

        if self.async_mode:
            # 异步模式：返回 task_id，需要轮询
            return {"task_id": "mock-task-123", "request_id": "req-456", "status": "processing"}
        else:
            # 同步模式：直接返回 voice_id
            return {"voice_id": "mock-voice-789", "status": "success"}

    def check_task_status(self, task_id: str) -> dict:
        return {"status": "completed", "voice_id": "mock-voice-789"}

    def list_preset_voices(self) -> list:
        return []

    def submit_synthesize_task(self, **kwargs) -> dict:
        return {"task_id": "synth-1", "status": "processing"}

    def synthesize_speech(self, **kwargs) -> dict:
        return {"audio_url": "https://example.com/audio.mp3", "duration": 5.0}

    def poll_synthesize_task(self, task_id: str, timeout: float = 120.0) -> dict:
        return {
            "status": "completed",
            "audio_url": "https://example.com/audio.mp3",
            "duration": 5.0,
        }


# ---------------------------------------------------------------------------
# 3. 辅助函数
# ---------------------------------------------------------------------------


def _make_user(**overrides) -> User:
    defaults = dict(
        id="user-test-001",
        email="test@example.com",
        display_name="Test User",
        username="testuser",
        subscription_plan="free",
        subscription_status="active",
        max_projects=3,
        max_storage_gb=10,
        created_at=datetime(2026, 1, 1, tzinfo=timezone.utc),
    )
    defaults.update(overrides)
    return User(**defaults)


def _make_clone_profile(
    name: str = "我的音色",
    user_id: str = "user-test-001",
    status: VoiceCloneStatus = VoiceCloneStatus.PENDING,
    source_audio_url: str = "https://example.com/source.wav",
    **kwargs,
) -> VoiceCloneProfile:
    profile = VoiceCloneProfile.create(
        user_id=user_id,
        name=name,
        source_audio_url=source_audio_url,
        voice_model=kwargs.get("voice_model", "cosyvoice-v2"),
        language=kwargs.get("language", "zh-CN"),
        gender=kwargs.get("gender", "female"),
        max_retries=kwargs.get("max_retries", 3),
        metadata=kwargs.get("metadata", None),
        description=kwargs.get("description", ""),
    )
    # 设置状态
    if status == VoiceCloneStatus.PROCESSING:
        profile.mark_processing()
        profile.metadata = {"cosyvoice_task_id": "task-123"}
    elif status == VoiceCloneStatus.READY:
        profile.mark_processing()
        profile.mark_ready("voice-ready-001")
    elif status == VoiceCloneStatus.FAILED:
        profile.mark_processing()
        profile.mark_failed("模拟失败")
    elif status == VoiceCloneStatus.DISABLED:
        profile.mark_disabled()
    return profile


# ---------------------------------------------------------------------------
# 4. Fixtures
# ---------------------------------------------------------------------------


@pytest.fixture
def clone_repo():
    return InMemoryVoiceCloneProfileRepository()


@pytest.fixture
def cosyvoice_service():
    return MockCosyVoiceService(async_mode=False)  # 同步模式，简化测试


@pytest.fixture
def client(clone_repo, cosyvoice_service):
    """创建带有依赖覆盖的 TestClient。"""
    test_app = FastAPI()
    test_app.include_router(router, prefix="/voice-clones")

    def _override_current_user():
        return AuthenticatedUser(user=_make_user())

    test_app.dependency_overrides[get_current_user] = _override_current_user
    test_app.dependency_overrides[get_voice_clone_profile_repository] = lambda: clone_repo
    test_app.dependency_overrides[get_cosyvoice_service] = lambda: cosyvoice_service

    yield TestClient(test_app)

    test_app.dependency_overrides.clear()


# ---------------------------------------------------------------------------
# 5. POST / — 创建声音克隆
# ---------------------------------------------------------------------------


class TestCreateVoiceClone:
    """创建声音克隆端点测试。"""

    def test_create_with_source_audio(self, client, cosyvoice_service):
        """提供源音频时创建克隆，同步模式下直接 ready。"""
        resp = client.post(
            "/voice-clones",
            json={
                "name": "我的专属音色",
                "source_audio_url": "https://example.com/voice.wav",
                "voice_model": "cosyvoice-v2",
                "language": "zh-CN",
                "gender": "female",
            },
        )
        assert resp.status_code == 201
        data = resp.json()
        assert data["name"] == "我的专属音色"
        assert data["source_audio_url"] == "https://example.com/voice.wav"
        assert data["voice_model"] == "cosyvoice-v2"
        assert data["language"] == "zh-CN"
        assert data["gender"] == "female"
        assert "id" in data
        assert len(data["id"]) > 0

        # 同步模式下应直接 ready
        assert data["status"] == "ready"
        assert data["voice_id"] == "mock-voice-789"
        assert data["error_message"] == ""

    def test_create_without_source_audio(self, client):
        """不提供源音频时创建，状态为 pending。"""
        resp = client.post(
            "/voice-clones",
            json={
                "name": "待上传音色",
                "description": "等待上传音频",
            },
        )
        assert resp.status_code == 201
        data = resp.json()
        assert data["name"] == "待上传音色"
        assert data["status"] == "pending"
        assert data["source_audio_url"] == ""
        assert data["voice_id"] == ""

    def test_create_persists_to_repository(self, client, clone_repo):
        """创建后档案保存到 repository。"""
        resp = client.post("/voice-clones", json={"name": "持久化测试"})
        profile_id = resp.json()["id"]

        saved = clone_repo.get(profile_id)
        assert saved is not None
        assert saved.name == "持久化测试"
        assert saved.user_id == "user-test-001"

    def test_create_missing_name_returns_422(self, client):
        """缺少 name 返回 422。"""
        resp = client.post("/voice-clones", json={})
        assert resp.status_code == 422

    def test_create_empty_name_returns_422(self, client):
        """空 name 返回 422。"""
        resp = client.post("/voice-clones", json={"name": ""})
        assert resp.status_code == 422

    def test_create_name_too_long_returns_422(self, client):
        """名称超长返回 422。"""
        long_name = "a" * 101
        resp = client.post("/voice-clones", json={"name": long_name})
        assert resp.status_code == 422

    def test_create_with_metadata(self, client, cosyvoice_service):
        """支持自定义 metadata。"""
        resp = client.post(
            "/voice-clones",
            json={
                "name": "带元数据的克隆",
                "source_audio_url": "https://example.com/v.wav",
                "metadata": {"source": "mobile_app", "version": "1.0"},
            },
        )
        assert resp.status_code == 201
        data = resp.json()
        assert data["metadata"]["source"] == "mobile_app"
        assert data["metadata"]["version"] == "1.0"

    def test_create_cosyvoice_failure_returns_failed(self, client, clone_repo, cosyvoice_service):
        """CosyVoice 提交失败时返回 201 + failed 状态（不抛 500）。"""
        cosyvoice_service.fail_submit = True
        cosyvoice_service.async_mode = True  # 异步模式才会调用 submit_clone_task

        resp = client.post(
            "/voice-clones",
            json={
                "name": "会失败的克隆",
                "source_audio_url": "https://example.com/bad.wav",
            },
        )
        # 不抛 500，返回 201 + failed 状态
        assert resp.status_code == 201
        data = resp.json()
        assert data["status"] == "failed"
        assert data["error_message"] != ""


# ---------------------------------------------------------------------------
# 6. GET / — 列出声音克隆
# ---------------------------------------------------------------------------


class TestListVoiceClones:
    """列出声音克隆端点测试。"""

    def test_empty_list(self, client):
        """无克隆时返回空列表。"""
        resp = client.get("/voice-clones")
        assert resp.status_code == 200
        data = resp.json()
        assert data["items"] == []
        assert data["total"] == 0

    def test_list_user_clones(self, client, clone_repo):
        """只返回当前用户的克隆。"""
        p1 = _make_clone_profile("音色1", "user-test-001")
        p2 = _make_clone_profile("音色2", "user-test-001")
        p3 = _make_clone_profile("他人音色", "other-user")
        clone_repo.create(p1)
        clone_repo.create(p2)
        clone_repo.create(p3)

        resp = client.get("/voice-clones")
        assert resp.status_code == 200
        data = resp.json()
        assert data["total"] == 2
        assert len(data["items"]) == 2
        names = {item["name"] for item in data["items"]}
        assert names == {"音色1", "音色2"}

    def test_filter_by_status(self, client, clone_repo):
        """按状态筛选。"""
        ready = _make_clone_profile("已就绪", status=VoiceCloneStatus.READY)
        failed = _make_clone_profile("已失败", status=VoiceCloneStatus.FAILED)
        clone_repo.create(ready)
        clone_repo.create(failed)

        resp = client.get("/voice-clones?status=ready")
        assert resp.status_code == 200
        data = resp.json()
        assert data["total"] == 1
        assert data["items"][0]["name"] == "已就绪"

    def test_filter_by_failed_status(self, client, clone_repo):
        """筛选失败状态。"""
        failed = _make_clone_profile("失败的", status=VoiceCloneStatus.FAILED)
        ready = _make_clone_profile("成功的", status=VoiceCloneStatus.READY)
        clone_repo.create(failed)
        clone_repo.create(ready)

        resp = client.get("/voice-clones?status=failed")
        assert resp.status_code == 200
        assert resp.json()["total"] == 1
        assert resp.json()["items"][0]["name"] == "失败的"

    def test_list_response_fields(self, client, clone_repo):
        """列表响应包含所有必需字段。"""
        p = _make_clone_profile("字段测试")
        clone_repo.create(p)

        resp = client.get("/voice-clones")
        item = resp.json()["items"][0]
        for field in [
            "id",
            "user_id",
            "name",
            "description",
            "source_audio_url",
            "voice_id",
            "voice_model",
            "language",
            "gender",
            "status",
            "error_message",
            "retry_count",
            "max_retries",
            "created_at",
            "updated_at",
        ]:
            assert field in item, f"缺少字段: {field}"


# ---------------------------------------------------------------------------
# 7. GET /{clone_id} — 获取克隆详情
# ---------------------------------------------------------------------------


class TestGetVoiceClone:
    """获取克隆详情端点测试。"""

    def test_get_existing_clone(self, client, clone_repo):
        """获取存在的克隆返回详情。"""
        p = _make_clone_profile("详情测试", description="这是一段描述")
        clone_repo.create(p)

        resp = client.get(f"/voice-clones/{p.id}")
        assert resp.status_code == 200
        data = resp.json()
        assert data["id"] == p.id
        assert data["name"] == "详情测试"
        assert data["description"] == "这是一段描述"

    def test_get_nonexistent_returns_404(self, client):
        """获取不存在的克隆返回 404。"""
        resp = client.get("/voice-clones/nonexistent-clone-id")
        assert resp.status_code == 404
        assert "not found" in resp.json()["detail"].lower()

    def test_get_other_user_clone_returns_404(self, client, clone_repo):
        """获取其他用户的克隆返回 404（安全隔离）。"""
        p = _make_clone_profile("他人音色", user_id="other-user")
        clone_repo.create(p)

        resp = client.get(f"/voice-clones/{p.id}")
        assert resp.status_code == 404

    def test_get_ready_clone_has_voice_id(self, client, clone_repo):
        """就绪状态的克隆有 voice_id。"""
        p = _make_clone_profile("就绪音色", status=VoiceCloneStatus.READY)
        clone_repo.create(p)

        resp = client.get(f"/voice-clones/{p.id}")
        data = resp.json()
        assert data["status"] == "ready"
        assert data["voice_id"] == "voice-ready-001"

    def test_get_failed_clone_has_error_message(self, client, clone_repo):
        """失败状态的克隆有错误信息。"""
        p = _make_clone_profile("失败音色", status=VoiceCloneStatus.FAILED)
        clone_repo.create(p)

        resp = client.get(f"/voice-clones/{p.id}")
        data = resp.json()
        assert data["status"] == "failed"
        assert "模拟失败" in data["error_message"]


# ---------------------------------------------------------------------------
# 8. GET /{clone_id}/status — 获取克隆状态
# ---------------------------------------------------------------------------


class TestGetVoiceCloneStatus:
    """获取克隆状态端点测试。"""

    def test_status_pending(self, client, clone_repo):
        """pending 状态。"""
        p = _make_clone_profile("pending", status=VoiceCloneStatus.PENDING)
        clone_repo.create(p)

        resp = client.get(f"/voice-clones/{p.id}/status")
        assert resp.status_code == 200
        data = resp.json()
        assert data["id"] == p.id
        assert data["status"] == "pending"
        assert data["retry_count"] == 0

    def test_status_ready(self, client, clone_repo):
        """ready 状态包含 voice_id。"""
        p = _make_clone_profile("ready", status=VoiceCloneStatus.READY)
        clone_repo.create(p)

        resp = client.get(f"/voice-clones/{p.id}/status")
        data = resp.json()
        assert data["status"] == "ready"
        assert data["voice_id"] == "voice-ready-001"

    def test_status_failed(self, client, clone_repo):
        """failed 状态包含错误信息。"""
        p = _make_clone_profile("failed", status=VoiceCloneStatus.FAILED)
        clone_repo.create(p)

        resp = client.get(f"/voice-clones/{p.id}/status")
        data = resp.json()
        assert data["status"] == "failed"
        assert data["error_message"] != ""
        assert data["retry_count"] == 0  # mark_failed 不增加 retry_count，只有重试时才增加

    def test_status_nonexistent_returns_404(self, client):
        """获取不存在克隆的状态返回 404。"""
        resp = client.get("/voice-clones/nonexistent/status")
        assert resp.status_code == 404


# ---------------------------------------------------------------------------
# 9. POST /{clone_id}/retry — 重试克隆
# ---------------------------------------------------------------------------


class TestRetryVoiceClone:
    """重试克隆端点测试。"""

    def test_retry_failed_clone(self, client, clone_repo, cosyvoice_service):
        """重试失败的克隆应成功。"""
        cosyvoice_service.async_mode = False
        p = _make_clone_profile("重试测试", status=VoiceCloneStatus.FAILED)
        clone_repo.create(p)

        resp = client.post(f"/voice-clones/{p.id}/retry")
        assert resp.status_code == 200
        data = resp.json()
        # 同步模式下重试后应变为 ready
        assert data["status"] == "ready"
        assert data["retry_count"] >= 1

    def test_retry_nonexistent_returns_404(self, client):
        """重试不存在的克隆返回 404。"""
        resp = client.post("/voice-clones/nonexistent/retry")
        assert resp.status_code == 404

    def test_retry_ready_clone_returns_400(self, client, clone_repo):
        """重试已就绪的克隆返回 400（不可重试）。"""
        p = _make_clone_profile("已就绪", status=VoiceCloneStatus.READY)
        clone_repo.create(p)

        resp = client.post(f"/voice-clones/{p.id}/retry")
        assert resp.status_code == 400
        assert "retryable" in resp.json()["detail"].lower() or "not" in resp.json()["detail"].lower()

    def test_retry_processing_clone_returns_400(self, client, clone_repo):
        """重试处理中的克隆返回 400。"""
        p = _make_clone_profile("处理中", status=VoiceCloneStatus.PROCESSING)
        clone_repo.create(p)

        resp = client.post(f"/voice-clones/{p.id}/retry")
        assert resp.status_code == 400

    def test_retry_increments_retry_count(self, client, clone_repo, cosyvoice_service):
        """重试后重试次数增加。"""
        cosyvoice_service.async_mode = False
        p = _make_clone_profile("重试计数", status=VoiceCloneStatus.FAILED)
        clone_repo.create(p)

        before_count = p.retry_count
        resp = client.post(f"/voice-clones/{p.id}/retry")
        after_count = resp.json()["retry_count"]

        assert after_count > before_count


# ---------------------------------------------------------------------------
# 10. DELETE /{clone_id} — 删除克隆
# ---------------------------------------------------------------------------


class TestDeleteVoiceClone:
    """删除克隆端点测试。"""

    def test_delete_existing_clone(self, client, clone_repo):
        """删除存在的克隆返回 204。"""
        p = _make_clone_profile("待删除")
        clone_repo.create(p)

        resp = client.delete(f"/voice-clones/{p.id}")
        assert resp.status_code == 204

        # 验证已删除
        assert clone_repo.get(p.id) is None

    def test_delete_nonexistent_returns_404(self, client):
        """删除不存在的克隆返回 404。"""
        resp = client.delete("/voice-clones/nonexistent-clone-id")
        assert resp.status_code == 404

    def test_delete_other_user_clone_returns_404(self, client, clone_repo):
        """删除其他用户的克隆返回 404（安全隔离）。"""
        p = _make_clone_profile("他人音色", user_id="other-user")
        clone_repo.create(p)

        resp = client.delete(f"/voice-clones/{p.id}")
        assert resp.status_code == 404
        # 验证未被删除
        assert clone_repo.get(p.id) is not None

    def test_delete_idempotent(self, client, clone_repo):
        """删除后再次删除返回 404。"""
        p = _make_clone_profile("幂等测试")
        clone_repo.create(p)

        resp1 = client.delete(f"/voice-clones/{p.id}")
        assert resp1.status_code == 204

        resp2 = client.delete(f"/voice-clones/{p.id}")
        assert resp2.status_code == 404


# ---------------------------------------------------------------------------
# 11. 跨端点场景
# ---------------------------------------------------------------------------


class TestVoiceCloneLifecycle:
    """音色克隆完整生命周期测试。"""

    def test_full_lifecycle_create_list_get_delete(self, client, clone_repo, cosyvoice_service):
        """创建 → 列表 → 详情 → 删除 完整流程。"""
        # 1. 创建
        create_resp = client.post(
            "/voice-clones",
            json={
                "name": "生命周期测试",
                "source_audio_url": "https://example.com/voice.wav",
            },
        )
        assert create_resp.status_code == 201
        clone_id = create_resp.json()["id"]

        # 2. 列表
        list_resp = client.get("/voice-clones")
        assert list_resp.json()["total"] == 1

        # 3. 详情
        detail_resp = client.get(f"/voice-clones/{clone_id}")
        assert detail_resp.status_code == 200
        assert detail_resp.json()["name"] == "生命周期测试"

        # 4. 状态
        status_resp = client.get(f"/voice-clones/{clone_id}/status")
        assert status_resp.status_code == 200
        assert status_resp.json()["status"] == "ready"

        # 5. 删除
        del_resp = client.delete(f"/voice-clones/{clone_id}")
        assert del_resp.status_code == 204

        # 6. 删除后列表为空
        list_resp2 = client.get("/voice-clones")
        assert list_resp2.json()["total"] == 0

    def test_failed_retry_flow(self, client, clone_repo, cosyvoice_service):
        """失败 → 重试 → 成功 流程。"""
        # 创建一个失败的克隆
        p = _make_clone_profile("失败重试", status=VoiceCloneStatus.FAILED)
        clone_repo.create(p)

        # 确认状态
        status_resp = client.get(f"/voice-clones/{p.id}/status")
        assert status_resp.json()["status"] == "failed"

        # 重试
        cosyvoice_service.async_mode = False
        retry_resp = client.post(f"/voice-clones/{p.id}/retry")
        assert retry_resp.status_code == 200
        assert retry_resp.json()["status"] == "ready"

        # 再次确认状态
        status_resp2 = client.get(f"/voice-clones/{p.id}/status")
        assert status_resp2.json()["status"] == "ready"
        assert status_resp2.json()["voice_id"] != ""


if __name__ == "__main__":
    pytest.main([__file__, "-v"])