feat(points): P2 step 2.4 - wire generation_tasks points gate (ai_video scene)

apps/api/app/api/routes/generation_tasks.py

@@ -4,6 +4,7 @@ from typing import Any
 
 from app.api.routes._helpers import check_project_access
 from app.auth import AuthenticatedUser, get_current_user
+from packages.middleware.points_gate import points_gate
 from app.core.storage import OSSStorageService, get_storage_service
 from app.core.task_enqueue import (
     GLOBAL_PENDING_LIMIT,

tests/unit/test_generation_tasks.py

@@ -18,6 +18,16 @@ from packages.application.generation_tasks import (
 )
 from packages.domain import GenerationTask
 
+import packages.middleware.points_gate as _pg_module
+
+
+@pytest.fixture(autouse=True)
+def _disable_points_gate(monkeypatch):
+    """默认关闭积分闸门，避免影响既有用例。"""
+    monkeypatch.setattr(_pg_module, "_points_gate_enabled", lambda: False)
+    yield
+
+
 
 @pytest.fixture(autouse=True)
 def _disable_points_gate(monkeypatch):

tests/unit/test_generation_tasks_points.py

@@ -1,5 +1,4 @@
 """视频生成 积分扣点单元测试 (#1895 P2 step 2.4)"""
-
 from __future__ import annotations
 
 from unittest.mock import MagicMock, patch
@@ -18,21 +17,19 @@ def _enable(monkeypatch):
 class TestGenerationTasksPoints:
     def test_ai_video_base_cost(self):
         from packages.domain.points_rules import calculate_points_cost
-
         assert calculate_points_cost("ai_video", is_member=False) == 4
         assert calculate_points_cost("ai_video", is_member=True, member_type="monthly") == 2
 
     def test_ai_video_quantity_scales(self):
         from packages.domain.points_rules import calculate_points_cost
-
         c1 = calculate_points_cost("ai_video", is_member=False, quantity=1)
         c3 = calculate_points_cost("ai_video", is_member=False, quantity=3)
         assert c3 > c1
 
     def test_insufficient_raises_402(self):
+        from fastapi import HTTPException
         from app.api.routes.generation_tasks import create_generation_task
         from app.schemas.generation_task import CreateGenerationTaskRequest
-        from fastapi import HTTPException
 
         db = MagicMock()
         cu = MagicMock()
@@ -47,17 +44,12 @@ class TestGenerationTasksPoints:
             MS.return_value = svc
             with pytest.raises(HTTPException) as ei:
                 create_generation_task(
-                    request=req,
-                    authenticated_user=cu,
-                    db=db,
-                    generation_task_repository=MagicMock(),
-                    project_repository=MagicMock(),
-                    asset_library_repository=MagicMock(),
-                    asset_repository=MagicMock(),
+                    request=req, authenticated_user=cu, db=db,
+                    generation_task_repository=MagicMock(), project_repository=MagicMock(),
+                    asset_library_repository=MagicMock(), asset_repository=MagicMock(),
                 )
             assert ei.value.status_code == 402
 
     def test_decorator_attached(self):
         from app.api.routes.generation_tasks import create_generation_task
-
         assert hasattr(create_generation_task, "__wrapped__"), "missing @points_gate"