From baf2e183006e93bcf2fcd34545c84de03460660f Mon Sep 17 00:00:00 2001
From: xiaoxia-agent <agent@xiaoxiajianji.com>
Date: Wed, 16 Sep 2026 01:28:18 +0800
Subject: [PATCH] test(points): add lipsync endpoint tests for
 insufficient/refund paths

---
 tests/unit/test_lipsync_points.py | 121 ++++++++++++++++++++++++++++++
 1 file changed, 121 insertions(+)

diff --git a/tests/unit/test_lipsync_points.py b/tests/unit/test_lipsync_points.py
index bebd34cc4..97d0c4338 100644
--- a/tests/unit/test_lipsync_points.py
+++ b/tests/unit/test_lipsync_points.py
@@ -94,3 +94,124 @@ 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 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)