test(wave124): quota配额系统单测 +66

覆盖Registry/Checker/告警级别/边界用例

tests/unit/test_quota.py

@@ -0,0 +1,397 @@
+"""Quota 配额系统单测 — 全维度覆盖."""
+
+from __future__ import annotations
+
+import math
+
+import pytest
+
+from packages.domain.quota import (
+    QuotaCheckResult,
+    QuotaChecker,
+    QuotaDimension,
+    QuotaRegistry,
+    QuotaTier,
+    QuotaWarningLevel,
+    QUOTA_TIERS,
+    get_warning_level,
+    quota_checker,
+    quota_registry,
+)
+
+
+# ── 枚举与常量 ──────────────────────────────────────────────────────────────
+
+
+class TestQuotaDimension:
+    def test_storage_gb(self):
+        assert QuotaDimension.STORAGE_GB == "storage_gb"
+
+    def test_videos_per_month(self):
+        assert QuotaDimension.VIDEOS_PER_MONTH == "videos_per_month"
+
+    def test_max_concurrent(self):
+        assert QuotaDimension.MAX_CONCURRENT == "max_concurrent"
+
+    def test_max_templates(self):
+        assert QuotaDimension.MAX_TEMPLATES == "max_templates"
+
+    def test_ai_voice_enabled(self):
+        assert QuotaDimension.AI_VOICE_ENABLED == "ai_voice_enabled"
+
+    def test_all_dimensions_count(self):
+        # 至少包含内置的几个核心维度
+        dims = list(QuotaDimension)
+        assert len(dims) >= 7
+
+    def test_from_string(self):
+        assert QuotaDimension("storage_gb") == QuotaDimension.STORAGE_GB
+
+
+class TestQuotaWarningLevel:
+    def test_normal(self):
+        assert QuotaWarningLevel.NORMAL == "normal"
+
+    def test_warning(self):
+        assert QuotaWarningLevel.WARNING == "warning"
+
+    def test_critical(self):
+        assert QuotaWarningLevel.CRITICAL == "critical"
+
+    def test_exceeded(self):
+        assert QuotaWarningLevel.EXCEEDED == "exceeded"
+
+
+# ── QuotaTier ──────────────────────────────────────────────────────────────
+
+
+class TestQuotaTier:
+    def test_get_limit_defined(self):
+        tier = QuotaTier(name="test", limits={"storage_gb": 100})
+        assert tier.get_limit("storage_gb") == 100
+
+    def test_get_limit_undefined_returns_zero(self):
+        tier = QuotaTier(name="test", limits={})
+        assert tier.get_limit("unknown_dim") == 0
+
+    def test_is_unlimited_true(self):
+        tier = QuotaTier(name="test", limits={"templates": float("inf")})
+        assert tier.is_unlimited("templates") is True
+
+    def test_is_unlimited_false(self):
+        tier = QuotaTier(name="test", limits={"storage_gb": 100})
+        assert tier.is_unlimited("storage_gb") is False
+
+    def test_is_unlimited_undefined_defaults_true(self):
+        # 未定义的维度，get 默认为 inf → is_unlimited 返回 True
+        tier = QuotaTier(name="test", limits={})
+        assert tier.is_unlimited("unknown") is True
+
+
+class TestQuotaTiers:
+    def test_free_tier_exists(self):
+        assert "free" in QUOTA_TIERS
+        free = QUOTA_TIERS["free"]
+        assert free.name == "free"
+        assert free.get_limit("storage_gb") == 2
+        assert free.get_limit("videos_per_month") == 5
+
+    def test_basic_tier_exists(self):
+        assert "basic" in QUOTA_TIERS
+        basic = QUOTA_TIERS["basic"]
+        assert basic.get_limit("storage_gb") == 20
+        assert basic.get_limit("videos_per_month") == 30
+        assert basic.get_limit("ai_voice_enabled") == 1
+
+    def test_premium_tier_exists(self):
+        assert "premium" in QUOTA_TIERS
+        premium = QUOTA_TIERS["premium"]
+        assert premium.get_limit("storage_gb") == 100
+        assert premium.get_limit("videos_per_month") == 100
+
+    def test_premium_templates_unlimited(self):
+        premium = QUOTA_TIERS["premium"]
+        assert premium.is_unlimited("max_templates") is True
+
+    def test_free_ai_voice_disabled(self):
+        free = QUOTA_TIERS["free"]
+        assert free.get_limit("ai_voice_enabled") == 0
+
+    def test_basic_ai_voice_enabled(self):
+        basic = QUOTA_TIERS["basic"]
+        assert basic.get_limit("ai_voice_enabled") == 1
+
+    def test_storage_increases_with_tier(self):
+        free = QUOTA_TIERS["free"].get_limit("storage_gb")
+        basic = QUOTA_TIERS["basic"].get_limit("storage_gb")
+        premium = QUOTA_TIERS["premium"].get_limit("storage_gb")
+        assert free < basic < premium
+
+
+# ── QuotaCheckResult ───────────────────────────────────────────────────────
+
+
+class TestQuotaCheckResult:
+    def test_usage_percent_normal(self):
+        result = QuotaCheckResult(
+            allowed=True, dimension="storage_gb",
+            limit=100, used=50, remaining=50,
+            warning_level=QuotaWarningLevel.NORMAL,
+        )
+        assert result.usage_percent == 50.0
+
+    def test_usage_percent_zero_usage(self):
+        result = QuotaCheckResult(
+            allowed=True, dimension="storage_gb",
+            limit=100, used=0, remaining=100,
+            warning_level=QuotaWarningLevel.NORMAL,
+        )
+        assert result.usage_percent == 0.0
+
+    def test_usage_percent_exceeded_capped(self):
+        result = QuotaCheckResult(
+            allowed=False, dimension="storage_gb",
+            limit=100, used=150, remaining=0,
+            warning_level=QuotaWarningLevel.EXCEEDED,
+        )
+        assert result.usage_percent == 100.0  # capped at 100
+
+    def test_usage_percent_zero_limit_with_usage(self):
+        result = QuotaCheckResult(
+            allowed=False, dimension="storage_gb",
+            limit=0, used=10, remaining=0,
+            warning_level=QuotaWarningLevel.EXCEEDED,
+        )
+        assert result.usage_percent == 100.0
+
+    def test_usage_percent_zero_limit_no_usage(self):
+        result = QuotaCheckResult(
+            allowed=True, dimension="storage_gb",
+            limit=0, used=0, remaining=0,
+            warning_level=QuotaWarningLevel.NORMAL,
+        )
+        assert result.usage_percent == 0.0
+
+    def test_usage_percent_unlimited(self):
+        result = QuotaCheckResult(
+            allowed=True, dimension="max_templates",
+            limit=float("inf"), used=100, remaining=float("inf"),
+            warning_level=QuotaWarningLevel.NORMAL,
+        )
+        assert result.usage_percent == 0.0
+
+
+# ── QuotaRegistry ──────────────────────────────────────────────────────────
+
+
+class TestQuotaRegistry:
+    def test_init_has_builtin_dimensions(self):
+        reg = QuotaRegistry()
+        dims = reg.list_dimensions()
+        assert "storage_gb" in dims
+        assert "videos_per_month" in dims
+        assert "max_concurrent" in dims
+
+    def test_init_has_three_tiers(self):
+        reg = QuotaRegistry()
+        tiers = reg.list_tiers()
+        assert "free" in tiers
+        assert "basic" in tiers
+        assert "premium" in tiers
+
+    def test_get_limit_free_storage(self):
+        reg = QuotaRegistry()
+        assert reg.get_limit("free", "storage_gb") == 2
+
+    def test_get_limit_unknown_plan_returns_zero(self):
+        reg = QuotaRegistry()
+        assert reg.get_limit("unknown_plan", "storage_gb") == 0
+
+    def test_get_tier_free(self):
+        reg = QuotaRegistry()
+        tier = reg.get_tier("free")
+        assert tier is not None
+        assert tier.name == "free"
+
+    def test_get_tier_unknown_returns_none(self):
+        reg = QuotaRegistry()
+        assert reg.get_tier("nonexistent") is None
+
+    def test_register_new_dimension(self):
+        reg = QuotaRegistry()
+        reg.register_dimension("custom_dim", "自定义维度", default_limits={"free": 5, "basic": 20})
+        assert "custom_dim" in reg.list_dimensions()
+        assert reg.get_limit("free", "custom_dim") == 5
+        assert reg.get_limit("basic", "custom_dim") == 20
+
+    def test_register_dimension_idempotent(self):
+        reg = QuotaRegistry()
+        reg.register_dimension("custom_dim", "v1", default_limits={"free": 5})
+        reg.register_dimension("custom_dim", "v2", default_limits={"free": 99})
+        # 幂等：第二次注册不改变
+        assert reg.list_dimensions()["custom_dim"] == "v1"
+        assert reg.get_limit("free", "custom_dim") == 5
+
+    def test_register_dimension_no_defaults(self):
+        reg = QuotaRegistry()
+        reg.register_dimension("new_dim", "新维度")
+        # 默认所有套餐都是 0
+        assert reg.get_limit("free", "new_dim") == 0
+        assert reg.get_limit("basic", "new_dim") == 0
+        assert reg.get_limit("premium", "new_dim") == 0
+
+    def test_list_dimensions_returns_copy(self):
+        reg = QuotaRegistry()
+        dims = reg.list_dimensions()
+        dims["fake"] = "test"
+        # 修改返回值不影响内部
+        assert "fake" not in reg.list_dimensions()
+
+
+# ── QuotaChecker ───────────────────────────────────────────────────────────
+
+
+class TestQuotaChecker:
+    def test_check_within_limit(self):
+        checker = QuotaChecker()
+        result = checker.check("free", "storage_gb", 1.0)
+        assert result.allowed is True
+        assert result.limit == 2
+        assert result.used == 1.0
+        assert result.remaining == 1.0
+        assert result.dimension == "storage_gb"
+
+    def test_check_exceeds_limit(self):
+        checker = QuotaChecker()
+        result = checker.check("free", "storage_gb", 3.0)
+        assert result.allowed is False
+        assert result.remaining == 0
+
+    def test_check_exactly_at_limit(self):
+        # used == limit 时 allowed 为 False（必须严格小于）
+        checker = QuotaChecker()
+        result = checker.check("free", "storage_gb", 2.0)
+        assert result.allowed is False
+
+    def test_check_unlimited(self):
+        checker = QuotaChecker()
+        result = checker.check("premium", "max_templates", 1000.0)
+        assert result.allowed is True
+        assert math.isinf(result.remaining)
+        assert result.warning_level == QuotaWarningLevel.NORMAL
+
+    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
+
+    def test_check_multiple(self):
+        checker = QuotaChecker()
+        results = checker.check_multiple(
+            "free",
+            {"storage_gb": 1.0, "videos_per_month": 2},
+        )
+        assert len(results) == 2
+        assert results[0].dimension == "storage_gb"
+        assert results[1].dimension == "videos_per_month"
+        assert all(r.allowed for r in results)
+
+    def test_warning_level_normal(self):
+        checker = QuotaChecker()
+        result = checker.check("free", "storage_gb", 1.0)  # 50%
+        assert result.warning_level == QuotaWarningLevel.NORMAL
+
+    def test_warning_level_warning(self):
+        checker = QuotaChecker()
+        # 80% < 95% → warning
+        result = checker.check("free", "storage_gb", 1.7)  # 85%
+        assert result.warning_level == QuotaWarningLevel.WARNING
+
+    def test_warning_level_critical(self):
+        checker = QuotaChecker()
+        # 95% <= < 100% → critical
+        result = checker.check("free", "storage_gb", 1.95)  # 97.5%
+        assert result.warning_level == QuotaWarningLevel.CRITICAL
+
+    def test_warning_level_exceeded(self):
+        checker = QuotaChecker()
+        result = checker.check("free", "storage_gb", 2.5)  # 125%
+        assert result.warning_level == QuotaWarningLevel.EXCEEDED
+
+    def test_warning_level_zero_limit_with_usage(self):
+        checker = QuotaChecker()
+        result = checker.check("free", "ai_voice_enabled", 1)  # limit=0, used=1
+        assert result.warning_level == QuotaWarningLevel.EXCEEDED
+
+    def test_warning_level_zero_limit_no_usage(self):
+        checker = QuotaChecker()
+        # limit=0, used=0 → 特殊处理为 normal
+        # 但 allowed 是 False（0 < 0 不成立）
+        result = checker.check("free", "ai_voice_enabled", 0)
+        # 0 < 0 是 False → not allowed
+        assert result.allowed is False
+
+    def test_checker_uses_provided_registry(self):
+        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
+
+
+# ── get_warning_level 便捷函数 ────────────────────────────────────────────
+
+
+class TestGetWarningLevel:
+    def test_normal_low_usage(self):
+        assert get_warning_level(50, 100) == QuotaWarningLevel.NORMAL
+
+    def test_normal_zero_usage(self):
+        assert get_warning_level(0, 100) == QuotaWarningLevel.NORMAL
+
+    def test_warning_threshold(self):
+        assert get_warning_level(80, 100) == QuotaWarningLevel.WARNING
+
+    def test_warning_between_80_and_95(self):
+        assert get_warning_level(90, 100) == QuotaWarningLevel.WARNING
+
+    def test_critical_threshold(self):
+        assert get_warning_level(95, 100) == QuotaWarningLevel.CRITICAL
+
+    def test_critical_between_95_and_100(self):
+        assert get_warning_level(99, 100) == QuotaWarningLevel.CRITICAL
+
+    def test_exceeded_at_100(self):
+        assert get_warning_level(100, 100) == QuotaWarningLevel.EXCEEDED
+
+    def test_exceeded_over_100(self):
+        assert get_warning_level(150, 100) == QuotaWarningLevel.EXCEEDED
+
+    def test_unlimited_always_normal(self):
+        assert get_warning_level(9999, float("inf")) == QuotaWarningLevel.NORMAL
+
+    def test_zero_limit_with_usage_exceeded(self):
+        assert get_warning_level(1, 0) == QuotaWarningLevel.EXCEEDED
+
+    def test_zero_limit_no_usage_normal(self):
+        assert get_warning_level(0, 0) == QuotaWarningLevel.NORMAL
+
+
+# ── 全局单例 ───────────────────────────────────────────────────────────────
+
+
+class TestGlobalSingletons:
+    def test_quota_registry_exists(self):
+        assert quota_registry is not None
+        assert isinstance(quota_registry, QuotaRegistry)
+
+    def test_quota_checker_exists(self):
+        assert quota_checker is not None
+        assert isinstance(quota_checker, QuotaChecker)
+
+    def test_global_checker_works(self):
+        result = quota_checker.check("free", "storage_gb", 1.0)
+        assert result.allowed is True
+        assert result.limit == 2