Compare commits

...

2 Commits

Author SHA1 Message Date
CI Bot d8f0bda439 style: auto-format with black + isort + prettier [skip ci-format-check]
CI/CD Pipeline / Check if frontend-only change (pull_request) Waiting to run
CI/CD Pipeline / Validate - Code Quality (pull_request) Waiting to run
CI/CD Pipeline / Validate - Type Check (mypy) (pull_request) Waiting to run
CI/CD Pipeline / Validate - Migration (alembic) (pull_request) Waiting to run
CI/CD Pipeline / Unit Tests (pull_request) Blocked by required conditions
CI/CD Pipeline / Integration Tests (pull_request) Blocked by required conditions
CI/CD Pipeline / Frontend Lint (pull_request) Waiting to run
CI/CD Pipeline / Frontend Unit Tests (pull_request) Blocked by required conditions
CI/CD Pipeline / PR Build API Image (pull_request) Waiting to run
CI/CD Pipeline / PR Build Web Image (pull_request) Waiting to run
CI/CD Pipeline / PR Build Worker Image (pull_request) Waiting to run
CI/CD Pipeline / Build Staging API Image (pull_request) Waiting to run
CI/CD Pipeline / Build Staging Web Image (pull_request) Waiting to run
CI/CD Pipeline / Build Staging Worker Image (pull_request) Waiting to run
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Blocked by required conditions
CI/CD Pipeline / Staging E2E Tests (pull_request) Blocked by required conditions
CI/CD Pipeline / Staging API Integration Tests (pull_request) Blocked by required conditions
CI/CD Pipeline / Build Production API Image (pull_request) Blocked by required conditions
CI/CD Pipeline / Build Production Web Image (pull_request) Blocked by required conditions
CI/CD Pipeline / Build Production Worker Image (pull_request) Blocked by required conditions
CI/CD Pipeline / Deploy Production (pull_request) Blocked by required conditions
CI/CD Pipeline / Production Browser E2E (pull_request) Blocked by required conditions
CI/CD Pipeline / ACR Image Cleanup (pull_request) Blocked by required conditions
CI/CD Pipeline / Canary Release to Production (pull_request) Blocked by required conditions
PR Automation / Auto Approve on CI Green (pull_request) Waiting to run
PR Automation / Auto Merge on CI Green + Approved (pull_request) Waiting to run
Preview Deploy / Deploy Preview Environment (pull_request) Waiting to run
CI/CD Pipeline / CI Gate (pull_request) 手动设置
AI Code Review / AI Code Review (pull_request) 手动设置
Preview Cleanup / Cleanup Preview Environment (pull_request) Waiting to run
ACR Cleanup / ACR Image Cleanup (pull_request_target) Has been cancelled
2026-07-28 16:12:00 +08:00
xiaoxia 745b500810 test(wave151): quota配额系统单测补全 +80
为 domain/quota.py 新增 80 个单测,纯逻辑 0 外部依赖:

- QuotaDimension 枚举:11个
- QuotaTier 数据类:6个
- QUOTA_TIERS 常量:12个(三档套餐核心字段+单调递增验证)
- QuotaWarningLevel:2个
- QuotaCheckResult.usage_percent:7个(正常/0/100%/超量/不限量/零限制)
- get_warning_level 告警级别:13个(0%/80%/95%/100%/超量/零限制/不限量/负数)
- QuotaRegistry 注册/查询:14个
- QuotaChecker 配额检查:13个
- 全局单例:3个
2026-07-28 16:12:00 +08:00
+575
View File
@@ -0,0 +1,575 @@
"""quota 单测.
domain 层配额系统纯逻辑模块,0 外部依赖。
覆盖:枚举常量、QuotaTier、QUOTA_TIERS常量、QuotaWarningLevel、
QuotaCheckResult、QuotaRegistry注册/查询、QuotaChecker检查/告警级别。
"""
from __future__ import annotations
import math
from packages.domain.quota import (
QUOTA_TIERS,
QuotaChecker,
QuotaCheckResult,
QuotaDimension,
QuotaRegistry,
QuotaTier,
QuotaWarningLevel,
get_warning_level,
quota_checker,
quota_registry,
)
class TestQuotaDimension:
"""QuotaDimension 枚举测试."""
def test_member_count(self):
"""内置维度数量."""
assert len(QuotaDimension) == 11
def test_storage_gb(self):
"""存储维度值."""
assert QuotaDimension.STORAGE_GB.value == "storage_gb"
def test_videos_per_month(self):
"""每月视频数维度."""
assert QuotaDimension.VIDEOS_PER_MONTH.value == "videos_per_month"
def test_max_concurrent(self):
"""并发数维度."""
assert QuotaDimension.MAX_CONCURRENT.value == "max_concurrent"
def test_max_templates(self):
"""模板数维度."""
assert QuotaDimension.MAX_TEMPLATES.value == "max_templates"
def test_max_titles(self):
"""标题库维度."""
assert QuotaDimension.MAX_TITLES.value == "max_titles"
def test_max_voiceovers(self):
"""配音库维度."""
assert QuotaDimension.MAX_VOICEOVERS.value == "max_voiceovers"
def test_ai_voice_enabled(self):
"""AI配音开关维度."""
assert QuotaDimension.AI_VOICE_ENABLED.value == "ai_voice_enabled"
def test_ai_voice_credits(self):
"""AI配音积分维度."""
assert QuotaDimension.AI_VOICE_CREDITS.value == "ai_voice_credits"
def test_all_values_are_strings(self):
"""所有枚举值都是字符串."""
for dim in QuotaDimension:
assert isinstance(dim.value, str)
assert len(dim.value) > 0
class TestQuotaTier:
"""QuotaTier 数据类测试."""
def test_create_empty(self):
"""创建空配额等级."""
tier = QuotaTier(name="test")
assert tier.name == "test"
assert tier.limits == {}
def test_create_with_limits(self):
"""创建带限制的配额等级."""
tier = QuotaTier(name="pro", limits={"storage": 100, "videos": 50})
assert tier.name == "pro"
assert tier.get_limit("storage") == 100
assert tier.get_limit("videos") == 50
def test_get_limit_undefined_returns_zero(self):
"""未定义的维度返回 0."""
tier = QuotaTier(name="test")
assert tier.get_limit("nonexistent") == 0
def test_is_unlimited_inf(self):
"""inf 视为不限量."""
tier = QuotaTier(name="test", limits={"templates": float("inf")})
assert tier.is_unlimited("templates") is True
def test_is_unlimited_finite(self):
"""有限值不是不限量."""
tier = QuotaTier(name="test", limits={"storage": 100})
assert tier.is_unlimited("storage") is False
def test_is_unlimited_undefined(self):
"""未定义的维度默认不限量."""
tier = QuotaTier(name="test")
# 未定义的 key 取默认值 inf,因此 is_unlimited 应该返回 True
assert tier.is_unlimited("unknown") is True
class TestQuotaTiers:
"""QUOTA_TIERS 常量测试."""
def test_three_tiers_exist(self):
"""三个套餐等级都存在."""
assert "free" in QUOTA_TIERS
assert "basic" in QUOTA_TIERS
assert "premium" in QUOTA_TIERS
def test_free_storage(self):
"""免费版 2GB 存储."""
assert QUOTA_TIERS["free"].get_limit(QuotaDimension.STORAGE_GB) == 2
def test_free_videos_per_month(self):
"""免费版 5个视频/月."""
assert QUOTA_TIERS["free"].get_limit(QuotaDimension.VIDEOS_PER_MONTH) == 5
def test_free_ai_voice_disabled(self):
"""免费版无AI配音."""
assert QUOTA_TIERS["free"].get_limit(QuotaDimension.AI_VOICE_ENABLED) == 0
def test_basic_storage(self):
"""基础版 20GB 存储."""
assert QUOTA_TIERS["basic"].get_limit(QuotaDimension.STORAGE_GB) == 20
def test_basic_videos(self):
"""基础版 30视频/月."""
assert QUOTA_TIERS["basic"].get_limit(QuotaDimension.VIDEOS_PER_MONTH) == 30
def test_basic_ai_voice_enabled(self):
"""基础版有AI配音."""
assert QUOTA_TIERS["basic"].get_limit(QuotaDimension.AI_VOICE_ENABLED) == 1
def test_basic_ai_voice_credits(self):
"""基础版 100 AI配音积分."""
assert QUOTA_TIERS["basic"].get_limit(QuotaDimension.AI_VOICE_CREDITS) == 100
def test_premium_storage(self):
"""高级版 100GB 存储."""
assert QUOTA_TIERS["premium"].get_limit(QuotaDimension.STORAGE_GB) == 100
def test_premium_videos(self):
"""高级版 100视频/月."""
assert QUOTA_TIERS["premium"].get_limit(QuotaDimension.VIDEOS_PER_MONTH) == 100
def test_premium_templates_unlimited(self):
"""高级版模板不限量."""
assert QUOTA_TIERS["premium"].is_unlimited(QuotaDimension.MAX_TEMPLATES)
def test_premium_multi_platform_enabled(self):
"""高级版多平台发布."""
assert QUOTA_TIERS["premium"].get_limit(QuotaDimension.MULTI_PLATFORM_ENABLED) == 1
def test_storage_monotonic(self):
"""存储量随套餐升级单调递增."""
free = QUOTA_TIERS["free"].get_limit(QuotaDimension.STORAGE_GB)
basic = QUOTA_TIERS["basic"].get_limit(QuotaDimension.STORAGE_GB)
premium = QUOTA_TIERS["premium"].get_limit(QuotaDimension.STORAGE_GB)
assert free < basic < premium
def test_videos_monotonic(self):
"""视频数随套餐升级单调递增."""
free = QUOTA_TIERS["free"].get_limit(QuotaDimension.VIDEOS_PER_MONTH)
basic = QUOTA_TIERS["basic"].get_limit(QuotaDimension.VIDEOS_PER_MONTH)
premium = QUOTA_TIERS["premium"].get_limit(QuotaDimension.VIDEOS_PER_MONTH)
assert free < basic < premium
class TestQuotaWarningLevel:
"""告警级别常量测试."""
def test_levels_defined(self):
"""四个级别都有定义."""
assert QuotaWarningLevel.NORMAL == "normal"
assert QuotaWarningLevel.WARNING == "warning"
assert QuotaWarningLevel.CRITICAL == "critical"
assert QuotaWarningLevel.EXCEEDED == "exceeded"
def test_four_distinct_levels(self):
"""四个级别各不相同."""
levels = {
QuotaWarningLevel.NORMAL,
QuotaWarningLevel.WARNING,
QuotaWarningLevel.CRITICAL,
QuotaWarningLevel.EXCEEDED,
}
assert len(levels) == 4
class TestQuotaCheckResult:
"""QuotaCheckResult 测试."""
def test_usage_percent_normal(self):
"""正常使用百分比."""
result = QuotaCheckResult(
allowed=True,
dimension="storage",
limit=100,
used=30,
remaining=70,
warning_level=QuotaWarningLevel.NORMAL,
)
assert result.usage_percent == 30.0
def test_usage_percent_zero_used(self):
"""使用量为 0."""
result = QuotaCheckResult(
allowed=True,
dimension="storage",
limit=100,
used=0,
remaining=100,
warning_level=QuotaWarningLevel.NORMAL,
)
assert result.usage_percent == 0.0
def test_usage_percent_exactly_100(self):
"""刚好用完."""
result = QuotaCheckResult(
allowed=False,
dimension="storage",
limit=100,
used=100,
remaining=0,
warning_level=QuotaWarningLevel.EXCEEDED,
)
assert result.usage_percent == 100.0
def test_usage_percent_over_limit_capped(self):
"""超出限制时封顶 100%."""
result = QuotaCheckResult(
allowed=False,
dimension="storage",
limit=100,
used=150,
remaining=0,
warning_level=QuotaWarningLevel.EXCEEDED,
)
assert result.usage_percent == 100.0
def test_usage_percent_unlimited(self):
"""不限量时使用率为 0."""
result = QuotaCheckResult(
allowed=True,
dimension="templates",
limit=float("inf"),
used=1000,
remaining=float("inf"),
warning_level=QuotaWarningLevel.NORMAL,
)
assert result.usage_percent == 0.0
def test_usage_percent_zero_limit_with_usage(self):
"""限制为 0 但有使用量,返回 100%."""
result = QuotaCheckResult(
allowed=False,
dimension="ai_voice",
limit=0,
used=1,
remaining=0,
warning_level=QuotaWarningLevel.EXCEEDED,
)
assert result.usage_percent == 100.0
def test_usage_percent_zero_limit_no_usage(self):
"""限制为 0 且无使用量,返回 0%."""
result = QuotaCheckResult(
allowed=True,
dimension="ai_voice",
limit=0,
used=0,
remaining=0,
warning_level=QuotaWarningLevel.NORMAL,
)
assert result.usage_percent == 0.0
class TestGetWarningLevel:
"""get_warning_level 便捷函数测试."""
def test_zero_usage(self):
"""0% 使用 - normal."""
assert get_warning_level(0, 100) == QuotaWarningLevel.NORMAL
def test_below_80_percent(self):
"""低于80% - normal."""
assert get_warning_level(50, 100) == QuotaWarningLevel.NORMAL
assert get_warning_level(79, 100) == QuotaWarningLevel.NORMAL
def test_at_80_percent(self):
"""刚好80% - warning."""
assert get_warning_level(80, 100) == QuotaWarningLevel.WARNING
def test_between_80_and_95(self):
"""80%-95%之间 - warning."""
assert get_warning_level(90, 100) == QuotaWarningLevel.WARNING
def test_at_95_percent(self):
"""刚好95% - critical."""
assert get_warning_level(95, 100) == QuotaWarningLevel.CRITICAL
def test_between_95_and_100(self):
"""95%-100%之间 - critical."""
assert get_warning_level(99, 100) == QuotaWarningLevel.CRITICAL
def test_at_100_percent(self):
"""刚好100% - exceeded."""
assert get_warning_level(100, 100) == QuotaWarningLevel.EXCEEDED
def test_over_100_percent(self):
"""超过100% - exceeded."""
assert get_warning_level(150, 100) == QuotaWarningLevel.EXCEEDED
def test_zero_limit_with_usage(self):
"""限制为0但有使用 - exceeded."""
assert get_warning_level(1, 0) == QuotaWarningLevel.EXCEEDED
def test_zero_limit_no_usage(self):
"""限制为0且无使用 - normal."""
assert get_warning_level(0, 0) == QuotaWarningLevel.NORMAL
def test_unlimited(self):
"""不限量 - 始终 normal."""
assert get_warning_level(0, float("inf")) == QuotaWarningLevel.NORMAL
assert get_warning_level(9999, float("inf")) == QuotaWarningLevel.NORMAL
def test_negative_usage(self):
"""负使用量 - normal."""
assert get_warning_level(-10, 100) == QuotaWarningLevel.NORMAL
class TestQuotaRegistry:
"""QuotaRegistry 测试."""
def test_initial_dimensions(self):
"""初始化后内置维度都在."""
reg = QuotaRegistry()
dims = reg.list_dimensions()
for dim in QuotaDimension:
assert dim.value in dims
def test_initial_dimensions_count(self):
"""初始维度数量等于枚举数量."""
reg = QuotaRegistry()
assert len(reg.list_dimensions()) == len(QuotaDimension)
def test_list_tiers(self):
"""三个套餐等级."""
reg = QuotaRegistry()
tiers = reg.list_tiers()
assert "free" in tiers
assert "basic" in tiers
assert "premium" in tiers
assert len(tiers) == 3
def test_get_tier_existing(self):
"""获取已有的套餐."""
reg = QuotaRegistry()
tier = reg.get_tier("free")
assert tier is not None
assert tier.name == "free"
def test_get_tier_nonexistent(self):
"""获取不存在的套餐返回 None."""
reg = QuotaRegistry()
assert reg.get_tier("enterprise") is None
def test_get_limit_existing(self):
"""获取已有限制."""
reg = QuotaRegistry()
assert reg.get_limit("free", QuotaDimension.STORAGE_GB) == 2
def test_get_limit_nonexistent_plan(self):
"""不存在的套餐返回 0."""
reg = QuotaRegistry()
assert reg.get_limit("enterprise", QuotaDimension.STORAGE_GB) == 0
def test_get_limit_unknown_dimension(self):
"""未知维度返回 0."""
reg = QuotaRegistry()
assert reg.get_limit("free", "unknown_dim") == 0
def test_register_dimension_new(self):
"""注册新维度."""
reg = QuotaRegistry()
count_before = len(reg.list_dimensions())
reg.register_dimension("custom_dim", "自定义维度")
dims = reg.list_dimensions()
assert "custom_dim" in dims
assert len(dims) == count_before + 1
def test_register_dimension_with_default_limits(self):
"""注册带默认限制的维度."""
reg = QuotaRegistry()
reg.register_dimension(
"api_calls",
"API调用次数",
default_limits={"free": 100, "basic": 1000, "premium": 10000},
)
assert reg.get_limit("free", "api_calls") == 100
assert reg.get_limit("basic", "api_calls") == 1000
assert reg.get_limit("premium", "api_calls") == 10000
def test_register_dimension_default_zero(self):
"""不带默认限制的维度,各套餐默认为 0."""
reg = QuotaRegistry()
reg.register_dimension("beta_feature", "测试功能")
assert reg.get_limit("free", "beta_feature") == 0
assert reg.get_limit("basic", "beta_feature") == 0
assert reg.get_limit("premium", "beta_feature") == 0
def test_register_dimension_idempotent(self):
"""重复注册幂等."""
reg = QuotaRegistry()
reg.register_dimension("dup", "重复测试", default_limits={"free": 10})
count_before = len(reg.list_dimensions())
# 第二次注册不同的限制,应该不生效
reg.register_dimension("dup", "重复测试2", default_limits={"free": 999})
count_after = len(reg.list_dimensions())
assert count_before == count_after
assert reg.get_limit("free", "dup") == 10 # 仍然是第一次的值
def test_list_dimensions_returns_copy(self):
"""list_dimensions 返回副本,修改不影响内部."""
reg = QuotaRegistry()
dims = reg.list_dimensions()
dims["hacked"] = "hack"
assert "hacked" not in reg.list_dimensions()
def test_register_partial_default_limits(self):
"""只给部分套餐设置默认限制."""
reg = QuotaRegistry()
reg.register_dimension(
"partial",
"部分套餐",
default_limits={"premium": 100},
)
assert reg.get_limit("free", "partial") == 0
assert reg.get_limit("basic", "partial") == 0
assert reg.get_limit("premium", "partial") == 100
class TestQuotaChecker:
"""QuotaChecker 测试."""
def test_check_within_limit(self):
"""在限制内,allowed=True."""
checker = QuotaChecker()
result = checker.check("free", QuotaDimension.STORAGE_GB, 1)
assert result.allowed is True
assert result.dimension == QuotaDimension.STORAGE_GB
assert result.limit == 2
assert result.used == 1
assert result.remaining == 1
assert result.warning_level == QuotaWarningLevel.NORMAL
def test_check_exceeded(self):
"""超出限制,allowed=False."""
checker = QuotaChecker()
result = checker.check("free", QuotaDimension.STORAGE_GB, 3)
assert result.allowed is False
assert result.remaining == 0
assert result.warning_level == QuotaWarningLevel.EXCEEDED
def test_check_exactly_at_limit(self):
"""刚好等于限制视为超出(used < limit 才允许)."""
checker = QuotaChecker()
result = checker.check("free", QuotaDimension.STORAGE_GB, 2)
assert result.allowed is False
def test_check_unlimited(self):
"""不限量维度."""
checker = QuotaChecker()
result = checker.check("premium", QuotaDimension.MAX_TEMPLATES, 999)
assert result.allowed is True
assert math.isinf(result.remaining)
assert result.warning_level == QuotaWarningLevel.NORMAL
def test_check_warning_level_boundaries(self):
"""各告警级别的边界值."""
checker = QuotaChecker()
# 79% - normal
assert checker.check("free", QuotaDimension.STORAGE_GB, 1.58).warning_level == QuotaWarningLevel.NORMAL
# 80% - warning
assert checker.check("free", QuotaDimension.STORAGE_GB, 1.6).warning_level == QuotaWarningLevel.WARNING
# 95% - critical
assert checker.check("free", QuotaDimension.STORAGE_GB, 1.9).warning_level == QuotaWarningLevel.CRITICAL
# 100% - exceeded
assert checker.check("free", QuotaDimension.STORAGE_GB, 2).warning_level == QuotaWarningLevel.EXCEEDED
def test_check_zero_usage(self):
"""0使用量."""
checker = QuotaChecker()
result = checker.check("basic", QuotaDimension.VIDEOS_PER_MONTH, 0)
assert result.allowed is True
assert result.remaining == 30
assert result.usage_percent == 0.0
def test_check_unknown_plan(self):
"""未知套餐,限制为0."""
checker = QuotaChecker()
result = checker.check("enterprise", QuotaDimension.STORAGE_GB, 0)
assert result.limit == 0
# used=0, limit=0 → 0 < 0 is False → allowed=False
assert result.allowed is False
assert result.warning_level == QuotaWarningLevel.NORMAL
def test_check_unknown_dimension(self):
"""未知维度,限制为0."""
checker = QuotaChecker()
result = checker.check("free", "unknown", 0)
assert result.limit == 0
def test_check_multiple(self):
"""批量检查多个维度."""
checker = QuotaChecker()
usage = {
QuotaDimension.STORAGE_GB: 1,
QuotaDimension.VIDEOS_PER_MONTH: 10,
}
results = checker.check_multiple("free", usage)
assert len(results) == 2
dims = {r.dimension for r in results}
assert QuotaDimension.STORAGE_GB in dims
assert QuotaDimension.VIDEOS_PER_MONTH in dims
def test_check_multiple_empty(self):
"""空字典返回空列表."""
checker = QuotaChecker()
results = checker.check_multiple("free", {})
assert results == []
def test_remaining_never_negative(self):
"""剩余量不为负."""
checker = QuotaChecker()
result = checker.check("free", QuotaDimension.STORAGE_GB, 100)
assert result.remaining >= 0
def test_custom_registry(self):
"""使用自定义 registry."""
reg = QuotaRegistry()
reg.register_dimension("custom", "自定义", default_limits={"free": 42})
checker = QuotaChecker(reg)
result = checker.check("free", "custom", 10)
assert result.limit == 42
assert result.allowed is True
class TestGlobalSingletons:
"""全局单例测试."""
def test_quota_registry_exists(self):
"""全局 registry 单例存在."""
assert quota_registry is not None
assert isinstance(quota_registry, QuotaRegistry)
def test_quota_checker_exists(self):
"""全局 checker 单例存在."""
assert quota_checker is not None
assert isinstance(quota_checker, QuotaChecker)
def test_global_checker_works(self):
"""全局 checker 能正常工作."""
result = quota_checker.check("free", QuotaDimension.STORAGE_GB, 1)
assert result.allowed is True
assert result.limit == 2