fix: 配额系统添加 pro 套餐并支持未知套餐 fallback 到 free
CI/CD Pipeline / Build Staging API Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI/CD Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 39s
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Successful in 1m0s
CI/CD Pipeline / Frontend Lint (pull_request) Has been skipped
CI/CD Pipeline / Frontend Unit Tests (pull_request) Has been skipped
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Successful in 1m7s
CI/CD Pipeline / PR Build Web Image (pull_request) Has been skipped
Preview Deploy / Deploy Preview Environment (pull_request) Successful in 1m31s
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / ACR Image Cleanup (pull_request) Has been skipped
PR Automation / Auto Merge on CI Green + Approved (pull_request) Successful in 1m44s
CI/CD Pipeline / Validate - Code Quality (pull_request) Failing after 2m8s
CI/CD Pipeline / Unit Tests (pull_request) Has been cancelled
CI/CD Pipeline / Integration Tests (pull_request) Has been cancelled
CI/CD Pipeline / PR Build API Image (pull_request) Has been cancelled
CI/CD Pipeline / PR Build Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production API Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Web Image (pull_request) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (pull_request) Has been cancelled
CI/CD Pipeline / Deploy Production (pull_request) Has been cancelled
CI/CD Pipeline / Production Browser E2E (pull_request) Has been cancelled
CI/CD Pipeline / Canary Release to Production (pull_request) Has been cancelled
CI/CD Pipeline / CI Gate (pull_request) Has been cancelled
AI Code Review / AI Code Review (pull_request) Has been cancelled
PR Automation / Auto Approve on CI Green (pull_request) Has been cancelled

- 在 QUOTA_TIERS 中添加 pro 套餐,配额与 premium 相同
- QuotaRegistry.get_tier() 对未知套餐 fallback 到 free 而非返回 None
- QuotaRegistry.get_limit() 同步更新 fallback 逻辑
- 更新所有相关单元测试以匹配新行为
- 新增 TestProTier 测试类验证 pro 套餐正确性

修复问题:用户 subscription_plan='pro' 时所有配额 limit=0 导致创建操作被拦截
This commit is contained in:
SaaS Frontend Agent
2026-08-07 14:57:30 +08:00
parent df8325542d
commit ffc56aa511
4 changed files with 73 additions and 25 deletions
+23 -8
View File
@@ -1,9 +1,10 @@
"""Quota system with registry pattern.
Three subscription tiers with different limits:
Four subscription tiers with different limits:
- free: 2GB storage, 5 videos/month, 3 concurrent, 3 templates, 50 titles, 10 voiceovers, no AI voice
- basic: 20GB storage, 30 videos/month, 10 concurrent, 15 templates, 500 titles, 100 voiceovers, AI voice
- premium: 100GB storage, 100 videos/month, 20 concurrent, unlimited templates, 500 titles, 100 voiceovers, AI voice
- pro: Same as premium (alias for premium tier)
Quota dimensions are registered by modules via the ModuleRegistry,
and checked against the user's subscription plan.
@@ -99,6 +100,22 @@ QUOTA_TIERS: Dict[str, QuotaTier] = {
QuotaDimension.DEDUP_REPORT_ENABLED: 1,
},
),
"pro": QuotaTier(
name="pro",
limits={
QuotaDimension.STORAGE_GB: 100,
QuotaDimension.VIDEOS_PER_MONTH: 100,
QuotaDimension.MAX_CONCURRENT: 20,
QuotaDimension.MAX_TEMPLATES: float("inf"), # 不限量
QuotaDimension.MAX_TITLES: 500,
QuotaDimension.MAX_VOICEOVERS: 100,
QuotaDimension.AI_VOICE_ENABLED: 1,
QuotaDimension.AI_VOICE_CREDITS: 500,
QuotaDimension.BATCH_EXPORT_ENABLED: 1,
QuotaDimension.MULTI_PLATFORM_ENABLED: 1,
QuotaDimension.DEDUP_REPORT_ENABLED: 1,
},
),
}
@@ -174,15 +191,13 @@ class QuotaRegistry:
for tier in self._tiers.values():
tier.limits.setdefault(dimension, 0)
def get_tier(self, plan_name: str) -> Optional[QuotaTier]:
"""获取套餐配额定义"""
return self._tiers.get(plan_name)
def get_tier(self, plan_name: str) -> QuotaTier:
"""获取套餐配额定义,未知套餐 fallback 到 free"""
return self._tiers.get(plan_name, self._tiers["free"])
def get_limit(self, plan_name: str, dimension: str) -> float:
"""获取指定套餐在指定维度的限制"""
tier = self._tiers.get(plan_name)
if tier is None:
return 0
"""获取指定套餐在指定维度的限制,未知套餐 fallback 到 free"""
tier = self._tiers.get(plan_name, self._tiers["free"])
return tier.get_limit(dimension)
def list_dimensions(self) -> Dict[str, str]:
+12 -10
View File
@@ -354,13 +354,13 @@ class TestQuotaRegistry:
assert len(reg.list_dimensions()) == len(QuotaDimension)
def test_list_tiers(self):
"""个套餐等级."""
"""个套餐等级."""
reg = QuotaRegistry()
tiers = reg.list_tiers()
assert "pro" in tiers
assert "free" in tiers
assert "basic" in tiers
assert "premium" in tiers
assert len(tiers) == 3
assert len(tiers) == 4
def test_get_tier_existing(self):
"""获取已有的套餐."""
@@ -370,9 +370,11 @@ class TestQuotaRegistry:
assert tier.name == "free"
def test_get_tier_nonexistent(self):
"""获取不存在的套餐返回 None."""
"""不存在的套餐 fallback 到 free"""
reg = QuotaRegistry()
assert reg.get_tier("enterprise") is None
tier = reg.get_tier("enterprise")
assert tier is not None
assert tier.name == "free"
def test_get_limit_existing(self):
"""获取已有限制."""
@@ -380,9 +382,9 @@ class TestQuotaRegistry:
assert reg.get_limit("free", QuotaDimension.STORAGE_GB) == 2
def test_get_limit_nonexistent_plan(self):
"""不存在的套餐返回 0."""
"""不存在的套餐 fallback 到 free 配额"""
reg = QuotaRegistry()
assert reg.get_limit("enterprise", QuotaDimension.STORAGE_GB) == 0
assert reg.get_limit("enterprise", QuotaDimension.STORAGE_GB) == 2 # fallback to free
def test_get_limit_unknown_dimension(self):
"""未知维度返回 0."""
@@ -506,12 +508,12 @@ class TestQuotaChecker:
assert result.usage_percent == 0.0
def test_check_unknown_plan(self):
"""未知套餐,限制为0."""
"""未知套餐 fallback 到 free"""
checker = QuotaChecker()
result = checker.check("enterprise", QuotaDimension.STORAGE_GB, 0)
assert result.limit == 0
assert result.limit == 2 # fallback to free tier, storage_gb=2
# used=0, limit=0 → 0 < 0 is False → allowed=False
assert result.allowed is False
assert result.allowed is True # used=0, limit=2, allowed
assert result.warning_level == QuotaWarningLevel.NORMAL
def test_check_unknown_dimension(self):
+32 -4
View File
@@ -222,7 +222,7 @@ class TestQuotaRegistry:
def test_get_limit_unknown_plan_returns_zero(self):
reg = QuotaRegistry()
assert reg.get_limit("unknown_plan", "storage_gb") == 0
assert reg.get_limit("unknown_plan", "storage_gb") == 2 # fallback to free
def test_get_tier_free(self):
reg = QuotaRegistry()
@@ -231,8 +231,11 @@ class TestQuotaRegistry:
assert tier.name == "free"
def test_get_tier_unknown_returns_none(self):
"""未知套餐 fallback 到 free"""
reg = QuotaRegistry()
assert reg.get_tier("nonexistent") is None
tier = reg.get_tier("nonexistent")
assert tier is not None
assert tier.name == "free"
def test_register_new_dimension(self):
reg = QuotaRegistry()
@@ -300,8 +303,8 @@ class TestQuotaChecker:
def test_check_unknown_plan(self):
checker = QuotaChecker()
result = checker.check("unknown", "storage_gb", 1.0)
assert result.allowed is False
assert result.limit == 0
assert result.allowed is True # fallback to free, limit=2, used=1.0 < 2
assert result.limit == 2 # fallback to free tier
def test_check_multiple(self):
checker = QuotaChecker()
@@ -412,3 +415,28 @@ class TestGlobalSingletons:
result = quota_checker.check("free", "storage_gb", 1.0)
assert result.allowed is True
assert result.limit == 2
class TestProTier:
"""Pro 套餐专项测试"""
def test_pro_tier_exists(self):
"""pro 套餐存在于 QUOTA_TIERS"""
from packages.domain.quota import QUOTA_TIERS
assert "pro" in QUOTA_TIERS
def test_pro_tier_same_as_premium(self):
"""pro 套餐配额与 premium 完全一致"""
from packages.domain.quota import QUOTA_TIERS
pro = QUOTA_TIERS["pro"]
premium = QUOTA_TIERS["premium"]
assert pro.limits == premium.limits
def test_pro_tier_get_limit(self):
"""pro 套餐各维度配额正确"""
reg = QuotaRegistry()
assert reg.get_limit("pro", "storage_gb") == 100
assert reg.get_limit("pro", "videos_per_month") == 100
assert reg.get_limit("pro", "max_concurrent") == 20
assert reg.get_limit("pro", "max_titles") == 500
assert reg.get_limit("pro", "ai_voice_enabled") == 1
+6 -3
View File
@@ -185,9 +185,10 @@ class TestQuotaRegistry:
reg = QuotaRegistry()
tiers = reg.list_tiers()
assert "free" in tiers
assert "pro" in tiers
assert "basic" in tiers
assert "premium" in tiers
assert len(tiers) == 3
assert len(tiers) == 4
def test_get_tier_existing(self):
reg = QuotaRegistry()
@@ -197,7 +198,9 @@ class TestQuotaRegistry:
def test_get_tier_unknown(self):
reg = QuotaRegistry()
assert reg.get_tier("unknown_plan") is None
tier = reg.get_tier("unknown_plan")
assert tier is not None
assert tier.name == "free"
def test_get_limit_known(self):
reg = QuotaRegistry()
@@ -206,7 +209,7 @@ class TestQuotaRegistry:
def test_get_limit_unknown_plan(self):
reg = QuotaRegistry()
assert reg.get_limit("unknown", "storage_gb") == 0
assert reg.get_limit("unknown", "storage_gb") == 2 # fallback to free tier
def test_register_new_dimension(self):
reg = QuotaRegistry()