xiaoxia-saas/tests/unit/test_lipsync_points.py

"""lipsync 积分扣点单元测试 (#1895 P2 step 2.2)"""

from __future__ import annotations

import math
from unittest.mock import MagicMock

import pytest
from fastapi import HTTPException


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


class TestLipsyncDurationEstimate:
    @pytest.mark.parametrize(
        "text,expected",
        [
            ("你好", 1.0),
            ("你" * 240, 1.0),
            ("你" * 241, 2.0),
            ("你" * 1000, 5.0),
        ],
    )
    def test_text_estimate(self, text, expected):
        est = max(1.0, math.ceil(len(text) / 240))
        assert est == expected

    @pytest.mark.parametrize(
        "seconds,expected",
        [
            (30, 1.0),
            (60, 1.0),
            (61, 2.0),
            (120, 2.0),
            (180, 3.0),
        ],
    )
    def test_audio_duration_estimate(self, seconds, expected):
        est = max(1.0, math.ceil(seconds / 60.0))
        assert est == expected


class TestLipsyncPointsDeduction:
    def _deduct(self, text="你好", audio_duration=None, enabled=True, success=True, balance=100, **cu_kw):
        from packages.domain.points_rules import calculate_points_cost

        svc = MagicMock() if enabled else None
        cu = _make_cu(**cu_kw)
        if svc is None:
            return 0, cu
        if audio_duration and audio_duration > 0:
            est = max(1.0, math.ceil(audio_duration / 60.0))
        elif text:
            est = max(1.0, math.ceil(len(text) / 240))
        else:
            est = 1.0
        cost = calculate_points_cost(
            "ai_digital_human",
            is_member=getattr(cu.user, "is_member", False),
            duration_minutes=est,
            member_type=getattr(cu.user, "member_type", None),
        )
        svc.deduct_points.return_value = {"success": success, "balance": balance}
        res = svc.deduct_points(cu.user.id, cost, "ai_digital_human", MagicMock())
        if not res["success"]:
            raise HTTPException(status_code=402, detail={"code": "INSUFFICIENT_POINTS"})
        return cost, cu

    def test_disabled(self):
        cost, _ = self._deduct(enabled=False)
        assert cost == 0

    def test_short_text_min_1min(self):
        cost, _ = self._deduct(text="你好")
        assert cost >= 15  # 15 base/min for free user × 1.15

    def test_audio_duration_used(self):
        cost_long, _ = self._deduct(audio_duration=180)  # 3min
        cost_short, _ = self._deduct(audio_duration=30)  # 1min
        assert cost_long > cost_short

    def test_insufficient_402(self):
        with pytest.raises(HTTPException) as ei:
            self._deduct(text="你" * 500, success=False, balance=0)
        assert ei.value.status_code == 402

    def test_member_cheaper(self):
        cm, _ = self._deduct(text="你" * 500, is_member=True, member_type="yearly")
        cf, _ = self._deduct(text="你" * 500, is_member=False)
        assert cm < cf


# ── 直接调用 create_lipsync_job 覆盖扣点/402/退费分支 ──
import importlib
from types import SimpleNamespace
from unittest.mock import MagicMock, patch

import packages.middleware.points_gate as _pg_module


# Ensure the enable-gate fixture for lipsync also covers @points_gate (if any)
# (the existing autouse _enable is below; importlib to avoid duplicate)
def _do_enable(monkeypatch):
    monkeypatch.setattr(_pg_module, "_points_gate_enabled", lambda: True)


def _body(**kw):
    b = MagicMock()
    defaults = dict(
        video_url="http://x/v.mp4", audio_url=None, audio_duration=None,
        sentence_timings=None, voice_id=None, script_text="你好世界",
        speed=1.0, emotion="", enable_video_loop=False, project_id=None,
    )
    defaults.update(kw)
    for k, v in defaults.items():
        setattr(b, k, v)
    return b


def _cu(user_id="u1", 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


class TestLipsyncEndpointPoints:
    def test_insufficient_raises_402(self, monkeypatch):
        _do_enable(monkeypatch)
        from app.api.routes.lipsync import create_lipsync_job
        db = MagicMock()
        svc = MagicMock()
        ps = MagicMock()
        ps.deduct_points.return_value = {"success": False, "balance": 0}
        fs = MagicMock(points_enabled=True)
        with patch("app.api.routes.lipsync.PointsService", return_value=ps), \
             patch("app.api.routes.lipsync.settings", fs):
            with pytest.raises(HTTPException) as ei:
                create_lipsync_job(body=_body(script_text="你" * 500), current_user=_cu(), db=db, svc=svc)
            assert ei.value.status_code == 402

    def test_value_error_refunds(self, monkeypatch):
        _do_enable(monkeypatch)
        from app.api.routes.lipsync import create_lipsync_job
        db = MagicMock()
        svc = MagicMock()
        svc.create_job.side_effect = ValueError("bad input")
        ps = MagicMock()
        ps.deduct_points.return_value = {"success": True, "balance": 99}
        fs = MagicMock(points_enabled=True)
        with patch("app.api.routes.lipsync.PointsService", return_value=ps), \
             patch("app.api.routes.lipsync.settings", fs):
            with pytest.raises(HTTPException) as ei:
                create_lipsync_job(body=_body(), current_user=_cu(), db=db, svc=svc)
            assert ei.value.status_code == 400
            assert ps.refund_points.called

    def test_mediakit_error_refunds(self, monkeypatch):
        _do_enable(monkeypatch)
        from app.api.routes.lipsync import create_lipsync_job
        from app.services.mediakit_client import MediaKitError
        db = MagicMock()
        svc = MagicMock()
        svc.create_job.side_effect = MediaKitError("fail", code="InvalidInput")
        ps = MagicMock()
        ps.deduct_points.return_value = {"success": True, "balance": 99}
        fs = MagicMock(points_enabled=True)
        with patch("app.api.routes.lipsync.PointsService", return_value=ps), \
             patch("app.api.routes.lipsync.settings", fs):
            with pytest.raises(HTTPException) as ei:
                create_lipsync_job(body=_body(), current_user=_cu(), db=db, svc=svc)
            assert ei.value.status_code == 400
            assert ps.refund_points.called

    def test_generic_exception_refunds(self, monkeypatch):
        _do_enable(monkeypatch)
        from app.api.routes.lipsync import create_lipsync_job
        db = MagicMock()
        svc = MagicMock()
        svc.create_job.side_effect = RuntimeError("boom")
        ps = MagicMock()
        ps.deduct_points.return_value = {"success": True, "balance": 99}
        fs = MagicMock(points_enabled=True)
        with patch("app.api.routes.lipsync.PointsService", return_value=ps), \
             patch("app.api.routes.lipsync.settings", fs):
            with pytest.raises(HTTPException) as ei:
                create_lipsync_job(body=_body(), current_user=_cu(), db=db, svc=svc)
            assert ei.value.status_code == 400
            assert ps.refund_points.called

    def test_audio_duration_estimation(self, monkeypatch):
        _do_enable(monkeypatch)
        from app.api.routes.lipsync import create_lipsync_job
        from packages.domain.points_rules import calculate_points_cost
        db = MagicMock()
        svc = MagicMock()
        job = SimpleNamespace(id="job-1", status="queued")
        svc.create_job.return_value = job
        ps = MagicMock()
        ps.deduct_points.return_value = {"success": True, "balance": 99}
        fs = MagicMock(points_enabled=True)
        with patch("app.api.routes.lipsync.PointsService", return_value=ps), \
             patch("app.api.routes.lipsync.settings", fs):
            create_lipsync_job(
                body=_body(audio_url="http://x/a.mp3", audio_duration=180, script_text=None),
                current_user=_cu(), db=db, svc=svc,
            )
            # 180 seconds -> 3 minutes; assert deduct called with cost >= 15*3
            args = ps.deduct_points.call_args[0]
            assert args[1] >= calculate_points_cost("ai_digital_human", is_member=False, duration_minutes=3)