xiaoxia-agent
27677e7c82
- create_lipsync_job(): deduct ai_digital_human points (15/min), estimated from audio_duration (if provided) or script_text (chars/240), min 1 min; refund on ValueError/MediaKitError/Exception; refund on failed job - preview_tts(): deduct ai_digital_human points by script_text length; refund on MediaKitError/Exception - Both gated by settings.points_enabled (default false) - Raises HTTP 402 INSUFFICIENT_POINTS on insufficient balance - 14 new unit tests; 15468 total passed
97 lines
3.0 KiB
Python
97 lines
3.0 KiB
Python
"""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
|