"""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 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)