xiaoxia-saas/tests/unit/test_voice_clone_from_asset.py

"""POST /voice-clones 支持 asset_id 的单元测试（路由级）。

覆盖：asset_id 成功（签名 URL 传给 workflow）/ 与 source_audio_url 同传 400 /
素材不存在 404 / 他人素材 403 / 非音频 400 / 无 storage_key 400 /
原 source_audio_url 方式不受影响。
"""

from __future__ import annotations

from unittest.mock import MagicMock

import pytest


class FakeAsset:
    def __init__(self, aid, project_id="proj-1", file_type="audio", storage_key="uploads/voice/a.mp3"):
        self.id = aid
        self.project_id = project_id
        self.file_type = file_type
        self.storage_key = storage_key


class FakeAssetRepo:
    def __init__(self, assets):
        self._assets = {a.id: a for a in assets}

    def find_by_id(self, aid):
        return self._assets.get(aid)


class FakeProject:
    def __init__(self, pid="proj-1", owner="user-1"):
        self.id = pid
        self.owner_user_id = owner

    def can_access(self, user_id):
        return user_id == self.owner_user_id


class FakeProjectRepo:
    def __init__(self, projects):
        self._projects = {p.id: p for p in projects}

    def find_by_id(self, pid):
        return self._projects.get(pid)


class FakeStorage:
    def get_download_url(self, key, expires_seconds=3600):
        return f"https://oss.example.com/signed/{key}?expires={expires_seconds}"


def _build_app(*, assets, projects, current_user="user-1"):
    from fastapi import FastAPI

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

    app.include_router(router, prefix="/voice-clones")

    fake_user = MagicMock()
    fake_user.user.id = current_user
    app.dependency_overrides[get_current_user] = lambda: fake_user

    asset_repo = FakeAssetRepo(assets)
    project_repo = FakeProjectRepo(projects)
    storage = FakeStorage()
    app.dependency_overrides[get_asset_repository] = lambda: asset_repo
    app.dependency_overrides[get_project_repository] = lambda: project_repo
    app.dependency_overrides[get_storage_service] = lambda: storage
    app.dependency_overrides[get_voice_clone_profile_repository] = lambda: MagicMock()
    app.dependency_overrides[get_cosyvoice_service] = lambda: MagicMock()

    from packages.domain.voice_clone_profile import VoiceCloneProfile

    workflow = MagicMock()
    profile = VoiceCloneProfile.create(
        user_id=current_user,
        name="我的克隆",
        voice_model="cosyvoice-v2",
    )  # pending 状态，不触发 celery send_task
    workflow.start_clone.return_value = profile
    app.dependency_overrides[_get_workflow_service] = lambda: workflow
    return app, workflow


class TestCreateVoiceCloneFromAsset:
    def test_success_with_asset_id(self):
        from fastapi.testclient import TestClient

        asset = FakeAsset("asset-1")
        app, workflow = _build_app(assets=[asset], projects=[FakeProject()])
        client = TestClient(app)

        resp = client.post("/voice-clones", json={"name": "我的克隆", "asset_id": "asset-1"})
        assert resp.status_code in (200, 201), resp.text

        kwargs = workflow.start_clone.call_args.kwargs
        # worker 拿到的是服务端用 storage_key 生成的签名 URL，不依赖前端
        assert kwargs["source_audio_url"].startswith("https://oss.example.com/signed/uploads/voice/a.mp3")
        assert "expires=604800" in kwargs["source_audio_url"]
        assert kwargs["metadata"]["source_asset_id"] == "asset-1"

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

        app, workflow = _build_app(assets=[FakeAsset("asset-1")], projects=[FakeProject()])
        client = TestClient(app)
        resp = client.post(
            "/voice-clones",
            json={
                "name": "冲突",
                "asset_id": "asset-1",
                "source_audio_url": "https://example.com/a.mp3",
            },
        )
        assert resp.status_code == 400
        workflow.start_clone.assert_not_called()

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

        app, workflow = _build_app(assets=[], projects=[FakeProject()])
        client = TestClient(app)
        resp = client.post("/voice-clones", json={"name": "x", "asset_id": "no-such-asset"})
        assert resp.status_code == 404
        workflow.start_clone.assert_not_called()

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

        asset = FakeAsset("asset-2", project_id="proj-other")
        app, workflow = _build_app(
            assets=[asset],
            projects=[FakeProject(pid="proj-other", owner="other-user")],
        )
        client = TestClient(app)
        resp = client.post("/voice-clones", json={"name": "越权", "asset_id": "asset-2"})
        assert resp.status_code == 403
        workflow.start_clone.assert_not_called()

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

        asset = FakeAsset("asset-3", file_type="video")
        app, workflow = _build_app(assets=[asset], projects=[FakeProject()])
        client = TestClient(app)
        resp = client.post("/voice-clones", json={"name": "视频素材", "asset_id": "asset-3"})
        assert resp.status_code == 400
        assert "音频" in resp.json()["detail"]
        workflow.start_clone.assert_not_called()

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

        asset = FakeAsset("asset-4", storage_key="")
        app, workflow = _build_app(assets=[asset], projects=[FakeProject()])
        client = TestClient(app)
        resp = client.post("/voice-clones", json={"name": "空key", "asset_id": "asset-4"})
        assert resp.status_code == 400
        workflow.start_clone.assert_not_called()

    def test_legacy_source_audio_url_still_works(self):
        """不传 asset_id、只传 source_audio_url 的旧流程保持兼容。"""
        from fastapi.testclient import TestClient

        app, workflow = _build_app(assets=[], projects=[])
        client = TestClient(app)
        resp = client.post(
            "/voice-clones",
            json={"name": "旧流程", "source_audio_url": "https://example.com/voice.mp3"},
        )
        assert resp.status_code in (200, 201), resp.text
        kwargs = workflow.start_clone.call_args.kwargs
        assert kwargs["source_audio_url"] == "https://example.com/voice.mp3"
        assert "source_asset_id" not in kwargs["metadata"]