xiaoxia-saas/tests/unit/test_tts_voice_clone_points.py

"""TTS (免费) + voice_clone 预览 (扣点) 单测 (#1895 P2 step 2.1)

v1.6.2: TTS 合成/预览（ai_voice）已免费，不再扣点；voice_clone 预览（voice_clone_synth）仍保持 1积分/分钟扣点。
"""

from __future__ import annotations

import math
from types import SimpleNamespace
from unittest.mock import MagicMock, patch

import pytest
from fastapi import HTTPException

import packages.middleware.points_gate as _pg_module


@pytest.fixture(autouse=True)
def _enable_gate(monkeypatch):
    monkeypatch.setattr(_pg_module, "_points_gate_enabled", lambda: True)
    yield


def _make_cu(user_id="user-1", is_member=False, member_type=None):
    cu = MagicMock()
    cu.user.id = user_id
    cu.user.is_member = is_member
    cu.user.member_type = member_type
    return cu


def _make_request(text="你好世界", voice_id="v1", **kw):
    r = MagicMock()
    r.text = text
    r.voice_id = voice_id
    r.voice_clone_profile_id = None
    r.speed = 1.0
    r.emotion = ""
    r.language = "zh-CN"
    r.metadata_ = {}
    r.voice_model = None
    for k, v in kw.items():
        setattr(r, k, v)
    return r


class TestTtsSynthesizeFree:
    """TTS synthesize/preview 已移除手动扣点，不再实例化 PointsService。"""

    def _setup(self, start_synth_raises=None):
        db = MagicMock()
        cu = _make_cu()
        repo = MagicMock()
        import enum

        class _S(enum.Enum):
            processing = "processing"

        job = SimpleNamespace(id="job-1", status=_S.processing, metadata={})
        uc = MagicMock()
        uc.execute.return_value = job
        wf = MagicMock()
        wf.start_synthesis.return_value = job
        wf.process_synthesis_failure.return_value = job
        if start_synth_raises:
            wf.start_synthesis.side_effect = start_synth_raises
        vc_repo = MagicMock()
        vc_repo.get.return_value = None
        return db, cu, repo, uc, wf, vc_repo, job

    def test_module_has_no_points_imports(self):
        import inspect

        from app.api.routes import tts as mod

        src = inspect.getsource(mod)
        assert "PointsService" not in src
        assert "calculate_points_cost" not in src
        assert "_points_deducted" not in src
        assert "import math" not in src

    def test_success_returns_job_without_points(self):
        db, cu, repo, uc, wf, vc_repo, job = self._setup()
        from app.api.routes.tts import synthesize

        with (
            patch("app.api.routes.tts.CreateTTSJobUseCase", return_value=uc),
            patch("app.api.routes.tts.TTSWorkflowService", return_value=wf),
            patch("app.api.routes.tts.celery_app.send_task"),
        ):
            resp = synthesize(
                request=_make_request(text="测试"),
                authenticated_user=cu,
                db=db,
                repository=repo,
                cosyvoice_service=MagicMock(),
                voice_clone_repo=vc_repo,
            )
            assert resp.job_id == job.id

    def test_ai_voice_cost_zero(self):
        from packages.domain.points_rules import calculate_points_cost

        assert calculate_points_cost("ai_voice", is_member=False, duration_minutes=10) == 0


class TestVoiceClonePreviewPoints:
    """voice_clone 预览（voice_clone_synth）保持 1 积分/分钟扣点。"""

    def _setup(self, text="你好", deduct_success=True, balance=0, synth_raises=None):
        db = MagicMock()
        cu = _make_cu()
        repo = MagicMock()
        profile = SimpleNamespace(is_ready=True, voice_id="vc-1", user_id=cu.user.id)
        uc = MagicMock()
        uc.execute.return_value = profile
        cosy = MagicMock()
        r = SimpleNamespace(audio_url="http://x/a.mp3", duration=1.2, file_size=1000)
        cosy.synthesize_speech.return_value = r
        if synth_raises:
            cosy.synthesize_speech.side_effect = synth_raises
        svc = MagicMock()
        svc.deduct_points.return_value = {"success": deduct_success, "balance": balance}
        fs = MagicMock(points_enabled=True)
        return db, cu, repo, uc, cosy, svc, fs

    def test_insufficient_raises_402(self):
        db, cu, repo, uc, cosy, svc, fs = self._setup(deduct_success=False, balance=0)
        from app.api.routes.voice_clones import get_voice_clone_preview

        with (
            patch("app.api.routes.voice_clones.GetVoiceCloneUseCase", return_value=uc),
            patch("app.api.routes.voice_clones.PointsService", return_value=svc),
            patch("app.api.routes.voice_clones.settings", fs),
            patch("app.api.routes.voice_clones._clone_preview_cache", {}),
        ):
            with pytest.raises(HTTPException) as ei:
                get_voice_clone_preview(
                    clone_id="c1",
                    text="你好",
                    speed=1.0,
                    emotion="",
                    authenticated_user=cu,
                    db=db,
                    repository=repo,
                    cosyvoice=cosy,
                )
            assert ei.value.status_code == 402

    def test_synth_cosyvoice_error_refunds_and_raises_502(self):
        from packages.application.cosyvoice_service import CosyVoiceError

        db, cu, repo, uc, cosy, svc, fs = self._setup(synth_raises=CosyVoiceError("fail"))
        from app.api.routes.voice_clones import get_voice_clone_preview

        with (
            patch("app.api.routes.voice_clones.GetVoiceCloneUseCase", return_value=uc),
            patch("app.api.routes.voice_clones.PointsService", return_value=svc),
            patch("app.api.routes.voice_clones.settings", fs),
            patch("app.api.routes.voice_clones._clone_preview_cache", {}),
        ):
            with pytest.raises(HTTPException) as ei:
                get_voice_clone_preview(
                    clone_id="c1",
                    text="你好",
                    speed=1.0,
                    emotion="",
                    authenticated_user=cu,
                    db=db,
                    repository=repo,
                    cosyvoice=cosy,
                )
            assert ei.value.status_code == 502
            assert svc.refund_points.called

    def test_success_returns_audio(self):
        db, cu, repo, uc, cosy, svc, fs = self._setup()
        from app.api.routes.voice_clones import get_voice_clone_preview

        with (
            patch("app.api.routes.voice_clones.GetVoiceCloneUseCase", return_value=uc),
            patch("app.api.routes.voice_clones.PointsService", return_value=svc),
            patch("app.api.routes.voice_clones.settings", fs),
            patch("app.api.routes.voice_clones._clone_preview_cache", {}),
        ):
            resp = get_voice_clone_preview(
                clone_id="c1",
                text="你好",
                speed=1.0,
                emotion="",
                authenticated_user=cu,
                db=db,
                repository=repo,
                cosyvoice=cosy,
            )
            svc.deduct_points.assert_called_once()
            assert resp.audio_url.startswith("http")

    def test_member_cheaper(self):
        from packages.domain.points_rules import calculate_points_cost

        cf = calculate_points_cost("voice_clone_synth", is_member=False, duration_minutes=2)
        cm = calculate_points_cost("voice_clone_synth", is_member=True, member_type="monthly", duration_minutes=2)
        assert cm < cf