diff --git a/apps/api/app/api/routes/lipsync.py b/apps/api/app/api/routes/lipsync.py index d13ce50dd..bfcdea801 100644 --- a/apps/api/app/api/routes/lipsync.py +++ b/apps/api/app/api/routes/lipsync.py @@ -17,6 +17,8 @@ from datetime import UTC from app.auth import AuthenticatedUser, get_current_user from app.config import settings +from packages.domain.points_rules import calculate_points_cost +from packages.domain.points_service import PointsService from app.dependencies import ( get_db_session, get_voice_clone_profile_repository, diff --git a/tests/unit/test_lipsync_points.py b/tests/unit/test_lipsync_points.py index 5acb81789..bebd34cc4 100644 --- a/tests/unit/test_lipsync_points.py +++ b/tests/unit/test_lipsync_points.py @@ -94,149 +94,3 @@ class TestLipsyncPointsDeduction: 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)