fix(#1895): /subscription/plans 路由挂载位置修复(解决 404) #1955
@@ -9,14 +9,12 @@ from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from typing import Any, Optional
|
||||
from typing import Optional
|
||||
|
||||
from app.auth import AuthenticatedUser, get_current_user
|
||||
from app.dependencies import get_db_session
|
||||
from app.schemas.points import (
|
||||
DailyUsageResponse,
|
||||
MembershipPlanItem,
|
||||
MembershipPlansResponse,
|
||||
MembershipStatusResponse,
|
||||
PointRuleItem,
|
||||
PointsBalanceResponse,
|
||||
@@ -38,7 +36,6 @@ from sqlalchemy.orm import Session
|
||||
from packages.domain.points_rules import (
|
||||
FREE_USER_MULTIPLIER,
|
||||
MEMBER_DISCOUNT,
|
||||
MEMBERSHIP_PRICES,
|
||||
POINTS_PACKAGES,
|
||||
POINTS_SCENES,
|
||||
calculate_points_cost,
|
||||
@@ -302,36 +299,6 @@ def create_recharge_order(
|
||||
return PointsOrderResponse(**order)
|
||||
|
||||
|
||||
@points_router.get("/subscription/plans", response_model=MembershipPlansResponse)
|
||||
def list_membership_plans(
|
||||
_current_user: AuthenticatedUser = Depends(get_current_user),
|
||||
):
|
||||
"""查询所有会员档位(名称/价格/权益),供前端会员购买页展示。"""
|
||||
plans: list[MembershipPlanItem] = []
|
||||
for plan_id, info in MEMBERSHIP_PRICES.items():
|
||||
days = info["duration_days"]
|
||||
monthly_cents = round(info["price_cents"] * 30 / days)
|
||||
features: dict[str, Any] = {"max_resolution": "1080p"}
|
||||
if plan_id == "monthly":
|
||||
features.update({"free_clips_daily": 2})
|
||||
elif plan_id == "quarterly":
|
||||
features.update({"free_clips_daily": 5})
|
||||
elif plan_id == "yearly":
|
||||
features.update({"free_clips_daily": "unlimited"})
|
||||
plans.append(
|
||||
MembershipPlanItem(
|
||||
plan_id=plan_id,
|
||||
name=info["name"],
|
||||
price_cents=info["price_cents"],
|
||||
monthly_price_cents=monthly_cents,
|
||||
duration_days=days,
|
||||
points_discount=MEMBER_DISCOUNT.get(plan_id, 1.0),
|
||||
features=features,
|
||||
)
|
||||
)
|
||||
return MembershipPlansResponse(plans=plans)
|
||||
|
||||
|
||||
@points_router.get("/subscription/membership", response_model=MembershipStatusResponse)
|
||||
def get_membership_status(
|
||||
current_user: AuthenticatedUser = Depends(get_current_user),
|
||||
|
||||
@@ -5,6 +5,7 @@ from __future__ import annotations
|
||||
import logging
|
||||
from dataclasses import replace
|
||||
from datetime import UTC, datetime
|
||||
from typing import Any
|
||||
|
||||
from app.auth import AuthenticatedUser, get_current_user
|
||||
from app.dependencies import get_user_repository
|
||||
@@ -99,6 +100,39 @@ async def get_current_subscription(
|
||||
return _build_subscription_info(current_user)
|
||||
|
||||
|
||||
@router.get("/plans")
|
||||
def list_membership_plans(
|
||||
current_user: AuthenticatedUser = Depends(get_current_user),
|
||||
) -> dict[str, list[dict[str, Any]]]:
|
||||
"""查询所有会员档位(供前端会员购买页展示)。
|
||||
|
||||
返回 points 积分体系下的会员档位(月卡/季卡/年卡),含价格、时长、积分折扣等信息。
|
||||
"""
|
||||
from packages.domain.points_rules import MEMBER_DISCOUNT, MEMBERSHIP_PRICES
|
||||
|
||||
plans: list[dict[str, Any]] = []
|
||||
for plan_id, info in MEMBERSHIP_PRICES.items():
|
||||
days = info["duration_days"]
|
||||
monthly_cents = round(info["price_cents"] * 30 / days)
|
||||
features: dict[str, Any] = {"max_resolution": "1080p"}
|
||||
if plan_id == "monthly":
|
||||
features.update({"free_clips_daily": 2})
|
||||
elif plan_id == "quarterly":
|
||||
features.update({"free_clips_daily": 5})
|
||||
elif plan_id == "yearly":
|
||||
features.update({"free_clips_daily": "unlimited"})
|
||||
plans.append({
|
||||
"plan_id": plan_id,
|
||||
"name": info["name"],
|
||||
"price_cents": info["price_cents"],
|
||||
"monthly_price_cents": monthly_cents,
|
||||
"duration_days": days,
|
||||
"points_discount": MEMBER_DISCOUNT.get(plan_id, 1.0),
|
||||
"features": features,
|
||||
})
|
||||
return {"plans": plans}
|
||||
|
||||
|
||||
@router.get("/billing-records", response_model=list[BillingRecord])
|
||||
async def get_billing_records(
|
||||
current_user: AuthenticatedUser = Depends(get_current_user),
|
||||
|
||||
@@ -10,6 +10,7 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
@@ -143,28 +144,48 @@ class TestPointsRulesDescription:
|
||||
|
||||
|
||||
class TestSubscriptionPlans:
|
||||
def test_plans_endpoint_returns_three_tiers(self):
|
||||
from app.api.routes.points import list_membership_plans
|
||||
@staticmethod
|
||||
def _import_plans_fn():
|
||||
"""Import from the real file to avoid sys.modules shadowing by integration fixtures."""
|
||||
import importlib.util
|
||||
_route_path = os.path.join(
|
||||
os.path.dirname(os.path.abspath(__file__)),
|
||||
"..", "..", "apps", "api", "app", "api", "routes", "subscription.py",
|
||||
)
|
||||
_spec = importlib.util.spec_from_file_location(
|
||||
"_real_subscription_routes", os.path.abspath(_route_path)
|
||||
)
|
||||
_mod = importlib.util.module_from_spec(_spec)
|
||||
# inject settings before exec
|
||||
import os as _os
|
||||
_os.environ.setdefault("JWT_SECRET_KEY", "test-secret")
|
||||
_spec.loader.exec_module(_mod)
|
||||
return _mod.list_membership_plans
|
||||
|
||||
resp = list_membership_plans(_current_user=_make_cu())
|
||||
plan_ids = {p.plan_id for p in resp.plans}
|
||||
def test_plans_endpoint_returns_three_tiers(self):
|
||||
import os # noqa: F401 (used by _import_plans_fn)
|
||||
list_membership_plans = self._import_plans_fn()
|
||||
resp = list_membership_plans(current_user=_make_cu())
|
||||
plans = resp["plans"]
|
||||
plan_ids = {p["plan_id"] for p in plans}
|
||||
assert plan_ids == {"monthly", "quarterly", "yearly"}
|
||||
for p in resp.plans:
|
||||
assert p.price_cents > 0
|
||||
assert p.duration_days in (30, 90, 365)
|
||||
assert 0 < p.points_discount <= 1.0
|
||||
assert "max_resolution" in p.features
|
||||
for p in plans:
|
||||
assert p["price_cents"] > 0
|
||||
assert p["duration_days"] in (30, 90, 365)
|
||||
assert 0 < p["points_discount"] <= 1.0
|
||||
assert "max_resolution" in p["features"]
|
||||
|
||||
def test_longer_plans_cheaper_per_month(self):
|
||||
from app.api.routes.points import list_membership_plans
|
||||
|
||||
resp = list_membership_plans(_current_user=_make_cu())
|
||||
monthly = next(p for p in resp.plans if p.plan_id == "monthly")
|
||||
quarterly = next(p for p in resp.plans if p.plan_id == "quarterly")
|
||||
yearly = next(p for p in resp.plans if p.plan_id == "yearly")
|
||||
assert monthly.monthly_price_cents == 1990
|
||||
assert quarterly.monthly_price_cents < monthly.monthly_price_cents
|
||||
assert yearly.monthly_price_cents < quarterly.monthly_price_cents
|
||||
import os # noqa: F401
|
||||
list_membership_plans = self._import_plans_fn()
|
||||
resp = list_membership_plans(current_user=_make_cu())
|
||||
plans = resp["plans"]
|
||||
monthly = next(p for p in plans if p["plan_id"] == "monthly")
|
||||
quarterly = next(p for p in plans if p["plan_id"] == "quarterly")
|
||||
yearly = next(p for p in plans if p["plan_id"] == "yearly")
|
||||
assert monthly["monthly_price_cents"] == 1990
|
||||
assert quarterly["monthly_price_cents"] < monthly["monthly_price_cents"]
|
||||
assert yearly["monthly_price_cents"] < quarterly["monthly_price_cents"]
|
||||
|
||||
|
||||
# ── P1-7: multiplier consistency ──────────────────────────────────────
|
||||
|
||||
Reference in New Issue
Block a user