xiaoxia-saas/tests/unit/test_tts_save_to_library.py

"""最后一公里：TTS 合成结果保存到配音库 单元测试。

覆盖：
- 正常保存已完成 TTS job 到配音库
- 自动携带元信息（音色名、时长、语速等）
- 自定义名称
- TTS job 不存在 → 404
- TTS job 未完成 → 400
- 配音库配额已满 → 429
"""

from __future__ import annotations

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

import pytest

from packages.domain.tts_job import TTSJob, TTSJobStatus
from packages.domain.voice_library import VoiceLibraryItem


def _make_completed_job(**kwargs) -> TTSJob:
    """构造一个已完成的 TTSJob。"""
    defaults = {
        "id": "tts_job_001",
        "user_id": "user_001",
        "input_text": "你好世界",
        "voice_id": "voice_001",
        "voice_model": "CosyVoice-v1",
        "project_id": "proj_001",
        "voice_clone_profile_id": "",
        "status": TTSJobStatus.COMPLETED,
        "output_audio_url": "https://oss.example.com/audio.mp3",
        "output_audio_key": "tts-outputs/user_001/tts_job_001.mp3",
        "duration": 5.5,
        "file_size": 88000,
        "sample_rate": 22050,
        "format": "mp3",
        "error_message": "",
        "retry_count": 0,
        "max_retries": 3,
        "metadata": {"speed": 1.0, "language": "zh-CN"},
        "started_at": datetime(2026, 7, 7, 10, 0, 0, tzinfo=timezone.utc),
        "completed_at": datetime(2026, 7, 7, 10, 0, 5, tzinfo=timezone.utc),
        "created_at": datetime(2026, 7, 7, 10, 0, 0, tzinfo=timezone.utc),
        "updated_at": datetime(2026, 7, 7, 10, 0, 5, tzinfo=timezone.utc),
    }
    defaults.update(kwargs)
    return TTSJob(**defaults)


def _make_voice_library_item(**kwargs) -> VoiceLibraryItem:
    """构造一个配音库条目。"""
    defaults = {
        "id": "voice_lib_001",
        "user_id": "user_001",
        "name": "TTS-tts_job_",
        "text": "你好世界",
        "voice_provider": "cosyvoice",
        "voice_id": "voice_001",
        "voice_name": "CosyVoice-v1",
        "audio_url": "https://oss.example.com/audio.mp3",
        "duration": 5.5,
        "file_size": 88000,
        "status": "completed",
        "project_id": "proj_001",
        "tags": [],
        "metadata_": {
            "source": "tts_job",
            "tts_job_id": "tts_job_001",
            "format": "mp3",
            "sample_rate": 22050,
            "speed": 1.0,
            "language": "zh-CN",
        },
        "created_at": datetime(2026, 7, 7, 10, 1, 0, tzinfo=timezone.utc),
        "updated_at": datetime(2026, 7, 7, 10, 1, 0, tzinfo=timezone.utc),
    }
    defaults.update(kwargs)
    return VoiceLibraryItem(**defaults)


class TestSaveToLibraryMapping:
    """测试 TTSJob → VoiceLibraryItem 字段映射。"""

    def test_completed_job_maps_correctly(self) -> None:
        """已完成的 TTS job 字段正确映射到配音库条目。"""
        job = _make_completed_job()

        # 验证 is_completed 属性
        assert job.is_completed is True

        # 验证关键字段映射
        assert job.output_audio_url == "https://oss.example.com/audio.mp3"
        assert job.duration == 5.5
        assert job.file_size == 88000
        assert job.voice_id == "voice_001"
        assert job.voice_model == "CosyVoice-v1"
        assert job.input_text == "你好世界"
        assert job.format == "mp3"
        assert job.sample_rate == 22050

    def test_metadata_carries_speed_and_format(self) -> None:
        """元信息携带语速、格式等。"""
        job = _make_completed_job()

        metadata = {
            "source": "tts_job",
            "tts_job_id": job.id,
            "format": job.format,
            "sample_rate": job.sample_rate,
        }
        if job.metadata:
            for key in ("speed", "language"):
                if key in job.metadata:
                    metadata[key] = job.metadata[key]

        assert metadata["format"] == "mp3"
        assert metadata["sample_rate"] == 22050
        assert metadata["speed"] == 1.0
        assert metadata["language"] == "zh-CN"
        assert metadata["source"] == "tts_job"

    def test_name_auto_generated_when_empty(self) -> None:
        """未提供名称时自动生成。"""
        job = _make_completed_job()
        name = None  # 模拟未提供名称
        generated_name = name or f"TTS-{job.id[:8]}"
        assert generated_name == "TTS-tts_job_"

    def test_name_uses_custom_when_provided(self) -> None:
        """提供自定义名称时使用自定义名称。"""
        custom_name = "我的配音"
        generated_name = custom_name or "TTS-fallback"
        assert generated_name == "我的配音"


class TestSaveToLibraryNotCompleted:
    """测试未完成 job 不能保存。"""

    def test_pending_job_not_completed(self) -> None:
        """pending 状态的 job 不能保存。"""
        job = _make_completed_job(status=TTSJobStatus.PENDING)
        assert job.is_completed is False

    def test_processing_job_not_completed(self) -> None:
        """processing 状态的 job 不能保存。"""
        job = _make_completed_job(status=TTSJobStatus.PROCESSING)
        assert job.is_completed is False

    def test_failed_job_not_completed(self) -> None:
        """failed 状态的 job 不能保存。"""
        job = _make_completed_job(status=TTSJobStatus.FAILED)
        assert job.is_completed is False

    def test_completed_without_url_not_completed(self) -> None:
        """status=completed 但没有 audio_url 的 job 不算完成。"""
        job = _make_completed_job(
            status=TTSJobStatus.COMPLETED,
            output_audio_url="",
        )
        assert job.is_completed is False


class TestSaveToLibraryQuota:
    """测试配额检查。"""

    def test_quota_exceeded_raises(self) -> None:
        """配音库配额已满时抛出 QuotaExceededError。"""
        from packages.application.voice_library.use_cases import QuotaExceededError

        error = QuotaExceededError(dimension="max_voiceovers", limit=10, used=10)
        assert "10/10" in str(error)

    def test_quota_under_limit_passes(self) -> None:
        """配额未满时不报错。"""
        from packages.domain.quota import QuotaDimension, quota_checker

        result = quota_checker.check("free", QuotaDimension.MAX_VOICEOVERS.value, 5)
        assert result.allowed is True


class TestSaveToLibraryCreateCommand:
    """测试 CreateVoiceLibraryCommand 构建。"""

    def test_command_fields_from_tts_job(self) -> None:
        """从 TTSJob 构建的 Command 字段正确。"""
        from packages.application.voice_library.commands import CreateVoiceLibraryCommand

        job = _make_completed_job()
        metadata_ = {
            "source": "tts_job",
            "tts_job_id": job.id,
            "format": job.format,
            "sample_rate": job.sample_rate,
            "speed": 1.0,
            "language": "zh-CN",
        }

        command = CreateVoiceLibraryCommand(
            user_id=job.user_id,
            name=f"TTS-{job.id[:8]}",
            text=job.input_text,
            voice_provider="cosyvoice",
            voice_id=job.voice_id,
            voice_name=job.voice_model,
            audio_url=job.output_audio_url,
            duration=job.duration,
            file_size=job.file_size,
            status="completed",
            project_id=job.project_id,
            tags=[],
            metadata_=metadata_,
        )

        assert command.user_id == "user_001"
        assert command.name == "TTS-tts_job_"
        assert command.text == "你好世界"
        assert command.voice_provider == "cosyvoice"
        assert command.voice_id == "voice_001"
        assert command.voice_name == "CosyVoice-v1"
        assert command.audio_url == "https://oss.example.com/audio.mp3"
        assert command.duration == 5.5
        assert command.file_size == 88000
        assert command.status == "completed"
        assert command.project_id == "proj_001"
        assert command.metadata_["source"] == "tts_job"

    def test_command_with_empty_project_id(self) -> None:
        """project_id 为空时传空字符串。"""
        from packages.application.voice_library.commands import CreateVoiceLibraryCommand

        job = _make_completed_job(project_id="")

        command = CreateVoiceLibraryCommand(
            user_id=job.user_id,
            name="test",
            text=job.input_text,
            voice_provider="cosyvoice",
            voice_id=job.voice_id,
            voice_name="",
            audio_url=job.output_audio_url,
            duration=job.duration,
            file_size=job.file_size,
            status="completed",
            project_id=job.project_id or "",
            tags=[],
            metadata_={},
        )

        assert command.project_id == ""

    def test_command_voice_name_fallback(self) -> None:
        """voice_model 为空时 voice_name 回退为空字符串。"""
        from packages.application.voice_library.commands import CreateVoiceLibraryCommand

        job = _make_completed_job(voice_model="")

        command = CreateVoiceLibraryCommand(
            user_id=job.user_id,
            name="test",
            text=job.input_text,
            voice_provider="cosyvoice",
            voice_id=job.voice_id,
            voice_name=job.voice_model or "",
            audio_url=job.output_audio_url,
            duration=job.duration,
            file_size=job.file_size,
            status="completed",
            project_id="",
            tags=[],
            metadata_={},
        )

        assert command.voice_name == ""


class TestSaveToLibraryUseCase:
    """测试 CreateVoiceLibraryUseCase 调用。"""

    def test_use_case_creates_item(self) -> None:
        """UseCase 正确创建配音库条目。"""
        from packages.application.voice_library.commands import CreateVoiceLibraryCommand
        from packages.application.voice_library.use_cases import CreateVoiceLibraryUseCase

        repo = MagicMock()
        repo.count_by_user.return_value = 0  # 配额未满

        repo.create.side_effect = lambda item: item

        use_case = CreateVoiceLibraryUseCase(repo)
        command = CreateVoiceLibraryCommand(
            user_id="user_001",
            name="test",
            text="你好",
            voice_provider="cosyvoice",
            voice_id="v1",
            voice_name="Voice1",
            audio_url="https://example.com/audio.mp3",
            duration=3.0,
            file_size=5000,
            status="completed",
            project_id="",
            tags=[],
            metadata_={},
        )

        item = use_case.execute(command, plan_name="free")

        repo.create.assert_called_once()
        assert item is not None

    def test_use_case_quota_exceeded(self) -> None:
        """UseCase 配额已满时抛出 QuotaExceededError。"""
        from packages.application.voice_library.commands import CreateVoiceLibraryCommand
        from packages.application.voice_library.use_cases import (
            CreateVoiceLibraryUseCase,
            QuotaExceededError,
        )

        repo = MagicMock()
        repo.count_by_user.return_value = 100  # 超过 premium 配额

        use_case = CreateVoiceLibraryUseCase(repo)
        command = CreateVoiceLibraryCommand(
            user_id="user_001",
            name="test",
            text="你好",
            voice_provider="cosyvoice",
            voice_id="v1",
            voice_name="Voice1",
            audio_url="https://example.com/audio.mp3",
            duration=3.0,
            file_size=5000,
            status="completed",
            project_id="",
            tags=[],
            metadata_={},
        )

        with pytest.raises(QuotaExceededError):
            use_case.execute(command, plan_name="premium")