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

# ── 环境变量 & 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 unittest.mock import MagicMock

from app.api.routes.voice_clones import router
from app.auth import AuthenticatedUser, get_current_user
from app.core.storage import get_storage_service
from app.dependencies import (
    get_asset_repository,
    get_audio_url_signer,
    get_cosyvoice_service,
    get_project_repository,
    get_voice_clone_profile_repository,
)

from packages.domain.entities import Asset, AssetStatus, Project, 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=True)  # 异步模式，匹配真实 CosyVoice API 行为


def _make_voice_asset(
    *,
    asset_id: str = "asset-voice-1",
    project_id: str = "proj-1",
    user_id: str = "user-test-001",
    file_type: str = "audio",
    storage_key: str = "uploads/voice/sample.m4a",
) -> Asset:
    """构造一个音频素材（可改 file_type 模拟非音频）。"""
    mime = {"audio": "audio/mp4", "video": "video/mp4", "image": "image/jpeg"}[file_type]
    return Asset(
        id=asset_id,
        project_id=project_id,
        library_id="lib-1",
        name="配音素材.m4a",
        storage_key=storage_key,
        mime_type=mime,
        file_size=12345,
        duration=10.0,
        status=AssetStatus.READY,
        uploaded_by_user_id=user_id,
    )


class InMemoryAssetRepository:
    """最小内存 Asset 仓储（find_by_id 即可）。"""

    def __init__(self, assets: list[Asset] | None = None):
        self._items = {a.id: a for a in (assets or [])}

    def find_by_id(self, asset_id: str):
        return self._items.get(asset_id)


class InMemoryProjectRepository:
    """最小内存 Project 仓储（find_by_id 即可）。"""

    def __init__(self, projects: list[Project] | None = None):
        self._items = {p.id: p for p in (projects or [])}

    def find_by_id(self, project_id: str):
        return self._items.get(project_id)


class MockStorageService:
    """Mock 存储服务：get_download_url 返回固定签名 URL。"""

    def __init__(self):
        self.signed_keys: list[str] = []

    def get_download_url(self, storage_key: str, expires_seconds: int = 3600) -> str:
        self.signed_keys.append(storage_key)
        return f"https://oss.example.com/{storage_key}?signed=1&exp={expires_seconds}"


@pytest.fixture
def asset_repo():
    return InMemoryAssetRepository()


@pytest.fixture
def project_repo():
    return InMemoryProjectRepository()


@pytest.fixture
def storage_service():
    return MockStorageService()


@pytest.fixture
def client(clone_repo, cosyvoice_service, asset_repo, project_repo, storage_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
    test_app.dependency_overrides[get_audio_url_signer] = lambda: (lambda url: url)
    test_app.dependency_overrides[get_asset_repository] = lambda: asset_repo
    test_app.dependency_overrides[get_project_repository] = lambda: project_repo
    test_app.dependency_overrides[get_storage_service] = lambda: storage_service

    yield TestClient(test_app)

    test_app.dependency_overrides.clear()


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


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

    def test_create_with_source_audio(self, client, cosyvoice_service):
        """提供源音频时创建克隆，异步提交后状态为 processing。"""
        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

        # 异步模式下提交后状态为 processing，voice_id 为空
        assert data["status"] == "processing"
        assert data["voice_id"] == ""
        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):
        """重试失败的克隆，重新提交后期望 processing。"""
        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()
        # 异步模式下重试后状态为 processing，等待 CosyVoice 完成
        assert data["status"] == "processing"
        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):
        """重试后重试次数增加。"""
        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"] == "processing"

        # 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):
        """失败 → 重试 → processing（等待异步完成） 流程。"""
        # 创建一个失败的克隆
        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"

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

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


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


# ---------------------------------------------------------------------------
# 6. POST /?asset_id=xxx — 从配音素材创建克隆
# ---------------------------------------------------------------------------


class TestCreateVoiceCloneFromAsset:
    """asset_id 参数测试：用配音素材库音频发起克隆。"""

    def test_create_from_asset_success(self, client, clone_repo, asset_repo, project_repo, storage_service):
        """正常路径：asset 归属本人且为音频 → 201，source_audio_url 为签名 URL。"""
        project = Project.create(owner_user_id="user-test-001", name="默认项目")
        project_repo._items[project.id] = project
        asset = _make_voice_asset(asset_id="asset-ok", project_id=project.id)
        asset_repo._items[asset.id] = asset

        resp = client.post(
            "/voice-clones",
            json={"name": "素材克隆", "asset_id": asset.id},
        )
        assert resp.status_code == 201, resp.text
        data = resp.json()
        assert data["name"] == "素材克隆"
        # 用 OSS storage_key 生成了服务端签名 URL
        assert any("uploads/voice/sample.m4a" in k for k in storage_service.signed_keys)
        assert data["source_audio_url"].startswith("https://oss.example.com/uploads/voice/sample.m4a")

    def test_create_from_asset_not_found_returns_404(self, client, asset_repo, project_repo):
        """asset 不存在 → 404。"""
        resp = client.post("/voice-clones", json={"name": "克隆", "asset_id": "no-such-asset"})
        assert resp.status_code == 404, resp.text

    def test_create_from_asset_other_user_returns_403(self, client, asset_repo, project_repo):
        """asset 属于他人项目 → 403。"""
        project = Project.create(owner_user_id="other-user", name="他人项目")
        project_repo._items[project.id] = project
        asset = _make_voice_asset(asset_id="asset-other", project_id=project.id, user_id="other-user")
        asset_repo._items[asset.id] = asset

        resp = client.post("/voice-clones", json={"name": "克隆", "asset_id": asset.id})
        assert resp.status_code == 403, resp.text

    def test_create_from_non_audio_asset_returns_400(self, client, asset_repo, project_repo):
        """asset 是视频 → 400。"""
        project = Project.create(owner_user_id="user-test-001", name="默认项目")
        project_repo._items[project.id] = project
        asset = _make_voice_asset(
            asset_id="asset-video",
            project_id=project.id,
            file_type="video",
            storage_key="uploads/v/test.mp4",
        )
        asset_repo._items[asset.id] = asset

        resp = client.post("/voice-clones", json={"name": "克隆", "asset_id": asset.id})
        assert resp.status_code == 400, resp.text

    def test_create_with_both_asset_id_and_url_returns_400(self, client, asset_repo, project_repo):
        """asset_id 与 source_audio_url 同时传 → 400。"""
        project = Project.create(owner_user_id="user-test-001", name="默认项目")
        project_repo._items[project.id] = project
        asset = _make_voice_asset(asset_id="asset-both", project_id=project.id)
        asset_repo._items[asset.id] = asset

        resp = client.post(
            "/voice-clones",
            json={
                "name": "克隆",
                "asset_id": asset.id,
                "source_audio_url": "https://example.com/a.wav",
            },
        )
        assert resp.status_code == 400, resp.text