xiaoxia-saas/tests/unit/test_quota_domain.py

"""
Quota 配额系统单元测试
"""

import math

import pytest

from packages.domain.quota import (
    QuotaChecker,
    QuotaCheckResult,
    QuotaDimension,
    QuotaRegistry,
    QuotaTier,
    QuotaWarningLevel,
    get_warning_level,
    quota_checker,
    quota_registry,
)


class TestQuotaDimension:
    """配额维度枚举测试"""

    def test_builtin_dimensions_exist(self):
        """测试内置维度存在"""
        assert QuotaDimension.STORAGE_GB == "storage_gb"
        assert QuotaDimension.VIDEOS_PER_MONTH == "videos_per_month"
        assert QuotaDimension.MAX_CONCURRENT == "max_concurrent"
        assert QuotaDimension.MAX_TEMPLATES == "max_templates"
        assert QuotaDimension.MAX_TITLES == "max_titles"
        assert QuotaDimension.MAX_VOICEOVERS == "max_voiceovers"
        assert QuotaDimension.AI_VOICE_ENABLED == "ai_voice_enabled"

    def test_extended_dimensions_exist(self):
        """测试扩展维度存在"""
        assert QuotaDimension.AI_VOICE_CREDITS == "ai_voice_credits"
        assert QuotaDimension.BATCH_EXPORT_ENABLED == "batch_export_enabled"
        assert QuotaDimension.MULTI_PLATFORM_ENABLED == "multi_platform_enabled"
        assert QuotaDimension.DEDUP_REPORT_ENABLED == "dedup_report_enabled"

    def test_dimension_is_string(self):
        """测试枚举值是字符串"""
        assert isinstance(QuotaDimension.STORAGE_GB, str)
        assert QuotaDimension.STORAGE_GB == "storage_gb"


class TestQuotaTier:
    """配额等级测试"""

    def test_get_limit_defined(self):
        """测试获取已定义的配额限制"""
        tier = QuotaTier(name="test", limits={"storage_gb": 10, "videos_per_month": 50})
        assert tier.get_limit("storage_gb") == 10
        assert tier.get_limit("videos_per_month") == 50

    def test_get_limit_undefined_returns_zero(self):
        """测试未定义维度返回 0"""
        tier = QuotaTier(name="test", limits={"storage_gb": 10})
        assert tier.get_limit("unknown_dim") == 0

    def test_is_unlimited_with_inf(self):
        """测试不限量判断（inf）"""
        tier = QuotaTier(name="test", limits={"templates": float("inf")})
        assert tier.is_unlimited("templates") is True

    def test_is_unlimited_with_finite(self):
        """测试有限量判断"""
        tier = QuotaTier(name="test", limits={"storage_gb": 10})
        assert tier.is_unlimited("storage_gb") is False

    def test_is_unlimited_undefined(self):
        """测试未定义维度默认不限量（因为默认值是 inf）"""
        tier = QuotaTier(name="test", limits={})
        # is_unlimited 使用 limits.get(dim, float("inf")) == float("inf")
        # 未定义时默认是 inf，所以返回 True
        assert tier.is_unlimited("undefined") is True

    def test_default_limits_empty(self):
        """测试默认 limits 为空 dict"""
        tier = QuotaTier(name="test")
        assert tier.limits == {}


class TestQuotaTiers:
    """预定义配额等级测试"""

    def test_free_tier_limits(self):
        """测试 free 套餐限制"""
        from packages.domain.quota import 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
        assert free.get_limit("max_concurrent") == 3
        assert free.get_limit("max_templates") == 3
        assert free.get_limit("max_titles") == 50
        assert free.get_limit("max_voiceovers") == 10
        assert free.get_limit("ai_voice_enabled") == 0
        assert free.get_limit("ai_voice_credits") == 0

    def test_basic_tier_limits(self):
        """测试 basic 套餐限制"""
        from packages.domain.quota import QUOTA_TIERS

        basic = QUOTA_TIERS["basic"]
        assert basic.name == "basic"
        assert basic.get_limit("storage_gb") == 20
        assert basic.get_limit("videos_per_month") == 30
        assert basic.get_limit("max_concurrent") == 10
        assert basic.get_limit("max_templates") == 15
        assert basic.get_limit("max_titles") == 500
        assert basic.get_limit("max_voiceovers") == 100
        assert basic.get_limit("ai_voice_enabled") == 1
        assert basic.get_limit("ai_voice_credits") == 100
        assert basic.get_limit("batch_export_enabled") == 1

    def test_premium_tier_limits(self):
        """测试 premium 套餐限制"""
        from packages.domain.quota import QUOTA_TIERS

        premium = QUOTA_TIERS["premium"]
        assert premium.name == "premium"
        assert premium.get_limit("storage_gb") == 100
        assert premium.get_limit("videos_per_month") == 100
        assert premium.get_limit("max_concurrent") == 20
        assert premium.is_unlimited("max_templates") is True
        assert premium.get_limit("max_titles") == 500
        assert premium.get_limit("max_voiceovers") == 100
        assert premium.get_limit("ai_voice_enabled") == 1
        assert premium.get_limit("ai_voice_credits") == 500
        assert premium.get_limit("batch_export_enabled") == 1
        assert premium.get_limit("multi_platform_enabled") == 1
        assert premium.get_limit("dedup_report_enabled") == 1

    def test_three_tiers_exist(self):
        """测试三个套餐等级都存在"""
        from packages.domain.quota import QUOTA_TIERS

        assert "free" in QUOTA_TIERS
        assert "basic" in QUOTA_TIERS
        assert "premium" in QUOTA_TIERS


class TestQuotaWarningLevel:
    """告警级别测试"""

    def test_warning_level_values(self):
        """测试告警级别常量值"""
        assert QuotaWarningLevel.NORMAL == "normal"
        assert QuotaWarningLevel.WARNING == "warning"
        assert QuotaWarningLevel.CRITICAL == "critical"
        assert QuotaWarningLevel.EXCEEDED == "exceeded"


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_over_limit(self):
        """测试超出限制时 capped at 100%"""
        result = QuotaCheckResult(
            allowed=False,
            dimension="storage_gb",
            limit=100,
            used=150,
            remaining=0,
            warning_level=QuotaWarningLevel.EXCEEDED,
        )
        assert result.usage_percent == 100.0

    def test_usage_percent_zero_limit_with_usage(self):
        """测试限制为 0 但有使用量时返回 100%"""
        result = QuotaCheckResult(
            allowed=False,
            dimension="ai_voice",
            limit=0,
            used=5,
            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

    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_exactly_100(self):
        """测试刚好 100% 使用"""
        result = QuotaCheckResult(
            allowed=False,
            dimension="storage_gb",
            limit=100,
            used=100,
            remaining=0,
            warning_level=QuotaWarningLevel.EXCEEDED,
        )
        assert result.usage_percent == 100.0


class TestQuotaRegistry:
    """配额注册表测试"""

    def test_initial_builtin_dimensions(self):
        """测试初始化后内置维度已注册"""
        registry = QuotaRegistry()
        dims = registry.list_dimensions()

        assert "storage_gb" in dims
        assert "videos_per_month" in dims
        assert "max_concurrent" in dims
        assert "max_templates" in dims
        assert "max_titles" in dims
        assert "max_voiceovers" in dims
        assert "ai_voice_enabled" in dims

    def test_register_new_dimension(self):
        """测试注册新维度"""
        registry = QuotaRegistry()
        registry.register_dimension("custom_dim", "自定义维度")

        dims = registry.list_dimensions()
        assert "custom_dim" in dims
        assert dims["custom_dim"] == "自定义维度"

    def test_register_dimension_with_default_limits(self):
        """测试注册带默认限制的新维度"""
        registry = QuotaRegistry()
        registry.register_dimension(
            "custom_feature",
            "自定义功能",
            default_limits={"free": 0, "basic": 1, "premium": 5},
        )

        assert registry.get_limit("free", "custom_feature") == 0
        assert registry.get_limit("basic", "custom_feature") == 1
        assert registry.get_limit("premium", "custom_feature") == 5

    def test_register_dimension_without_default_limits(self):
        """测试注册不带默认限制的新维度（所有套餐默认 0）"""
        registry = QuotaRegistry()
        registry.register_dimension("new_feature", "新功能")

        assert registry.get_limit("free", "new_feature") == 0
        assert registry.get_limit("basic", "new_feature") == 0
        assert registry.get_limit("premium", "new_feature") == 0

    def test_register_dimension_idempotent(self):
        """测试重复注册是幂等的"""
        registry = QuotaRegistry()
        registry.register_dimension("test_dim", "测试维度", default_limits={"free": 10})
        # 第二次注册不应该改变任何东西
        registry.register_dimension("test_dim", "另一个描述", default_limits={"free": 999})

        dims = registry.list_dimensions()
        assert dims["test_dim"] == "测试维度"  # 保留第一次的描述
        assert registry.get_limit("free", "test_dim") == 10  # 保留第一次的限制

    def test_register_unknown_plan_ignored(self):
        """测试未知套餐的默认限制被忽略"""
        registry = QuotaRegistry()
        registry.register_dimension(
            "test_dim",
            "测试",
            default_limits={"free": 1, "enterprise": 100},
        )

        assert registry.get_limit("free", "test_dim") == 1
        # enterprise 套餐不存在，不影响
        assert "enterprise" not in registry.list_tiers()

    def test_get_tier_existing(self):
        """测试获取存在的套餐"""
        registry = QuotaRegistry()
        tier = registry.get_tier("free")
        assert tier is not None
        assert tier.name == "free"

    def test_get_tier_nonexistent(self):
        """测试获取不存在的套餐返回 None"""
        registry = QuotaRegistry()
        assert registry.get_tier("nonexistent") is None

    def test_get_limit_nonexistent_plan(self):
        """测试不存在套餐的限制返回 0"""
        registry = QuotaRegistry()
        assert registry.get_limit("enterprise", "storage_gb") == 0

    def test_list_tiers(self):
        """测试列出所有套餐"""
        registry = QuotaRegistry()
        tiers = registry.list_tiers()
        assert "free" in tiers
        assert "basic" in tiers
        assert "premium" in tiers
        assert len(tiers) == 3

    def test_list_dimensions_returns_copy(self):
        """测试 list_dimensions 返回副本（修改不影响内部）"""
        registry = QuotaRegistry()
        dims = registry.list_dimensions()
        dims["fake_dim"] = "fake"

        # 原始注册表不应被修改
        assert "fake_dim" not in registry.list_dimensions()


class TestQuotaChecker:
    """配额检查器测试"""

    @pytest.fixture
    def checker(self):
        return QuotaChecker()

    # ===== 基础检查 =====

    def test_check_free_storage_under_limit(self, checker):
        """测试 free 套餐存储未超限"""
        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.warning_level == QuotaWarningLevel.NORMAL
        assert result.dimension == "storage_gb"

    def test_check_free_storage_over_limit(self, checker):
        """测试 free 套餐存储超限"""
        result = checker.check("free", "storage_gb", 3.0)

        assert result.allowed is False
        assert result.remaining == 0
        assert result.warning_level == QuotaWarningLevel.EXCEEDED

    def test_check_free_storage_exactly_at_limit(self, checker):
        """测试刚好达到限制（不允许）"""
        result = checker.check("free", "storage_gb", 2.0)

        # used < limit → 2 < 2 → False
        assert result.allowed is False
        assert result.warning_level == QuotaWarningLevel.EXCEEDED

    # ===== 告警级别 =====

    def test_warning_level_normal(self, checker):
        """测试正常级别（< 80%）"""
        result = checker.check("free", "storage_gb", 1.0)  # 50%
        assert result.warning_level == QuotaWarningLevel.NORMAL

    def test_warning_level_warning(self, checker):
        """测试警告级别（80% ~ 95%）"""
        result = checker.check("free", "storage_gb", 1.7)  # 85%
        assert result.warning_level == QuotaWarningLevel.WARNING

    def test_warning_level_critical(self, checker):
        """测试严重级别（95% ~ 100%）"""
        result = checker.check("free", "storage_gb", 1.95)  # 97.5%
        assert result.warning_level == QuotaWarningLevel.CRITICAL

    def test_warning_level_exceeded(self, checker):
        """测试超限级别（>= 100%）"""
        result = checker.check("free", "storage_gb", 2.0)  # 100%
        assert result.warning_level == QuotaWarningLevel.EXCEEDED

    # ===== 不限量 =====

    def test_check_unlimited_templates_premium(self, checker):
        """测试 premium 套餐模板不限量"""
        result = checker.check("premium", "max_templates", 9999)

        assert result.allowed is True
        assert result.limit == float("inf")
        assert result.remaining == float("inf")
        assert result.warning_level == QuotaWarningLevel.NORMAL

    # ===== 0 限制 =====

    def test_check_zero_limit_with_usage(self, checker):
        """测试限制为 0 但有使用量"""
        result = checker.check("free", "ai_voice_enabled", 1)

        assert result.allowed is False
        assert result.warning_level == QuotaWarningLevel.EXCEEDED

    def test_check_zero_limit_no_usage(self, checker):
        """测试限制为 0 且无使用量"""
        result = checker.check("free", "ai_voice_enabled", 0)

        # used < limit → 0 < 0 → False? 让我们看看...
        # 实际上 0 < 0 是 False，所以 allowed = False
        # 但 warning_level: limit <= 0 and used == 0 → NORMAL
        # 等一下，看看代码逻辑：
        # if limit <= 0: return EXCEEDED if used > 0 else NORMAL
        assert result.warning_level == QuotaWarningLevel.NORMAL

    # ===== 多维度检查 =====

    def test_check_multiple(self, checker):
        """测试批量检查多个维度"""
        usage = {
            "storage_gb": 1.0,
            "videos_per_month": 3,
            "max_concurrent": 2,
        }
        results = checker.check_multiple("free", usage)

        assert len(results) == 3
        dims = {r.dimension: r for r in results}
        assert dims["storage_gb"].allowed is True
        assert dims["videos_per_month"].allowed is True
        assert dims["max_concurrent"].allowed is True

    def test_check_multiple_some_exceeded(self, checker):
        """测试批量检查中有超限的"""
        usage = {
            "storage_gb": 5.0,  # 超限
            "videos_per_month": 3,  # 正常
        }
        results = checker.check_multiple("free", usage)

        dims = {r.dimension: r for r in results}
        assert dims["storage_gb"].allowed is False
        assert dims["videos_per_month"].allowed is True

    # ===== 自定义 registry =====

    def test_check_with_custom_registry(self):
        """测试使用自定义 registry"""
        registry = QuotaRegistry()
        registry.register_dimension(
            "custom_feature",
            "自定义",
            default_limits={"free": 5, "basic": 20},
        )
        checker = QuotaChecker(registry)

        result = checker.check("free", "custom_feature", 3)
        assert result.allowed is True
        assert result.limit == 5

        result = checker.check("basic", "custom_feature", 25)
        assert result.allowed is False

    def test_check_unknown_plan(self, checker):
        """测试未知套餐（限制为 0）"""
        result = checker.check("enterprise", "storage_gb", 1)
        assert result.allowed is False
        assert result.limit == 0


class TestGetWarningLevel:
    """便捷函数 get_warning_level 测试"""

    def test_normal(self):
        assert get_warning_level(50, 100) == QuotaWarningLevel.NORMAL

    def test_warning(self):
        assert get_warning_level(85, 100) == QuotaWarningLevel.WARNING

    def test_critical(self):
        assert get_warning_level(96, 100) == QuotaWarningLevel.CRITICAL

    def test_exceeded(self):
        assert get_warning_level(100, 100) == QuotaWarningLevel.EXCEEDED
        assert get_warning_level(150, 100) == QuotaWarningLevel.EXCEEDED

    def test_zero_limit_with_usage(self):
        assert get_warning_level(5, 0) == QuotaWarningLevel.EXCEEDED

    def test_zero_limit_no_usage(self):
        assert get_warning_level(0, 0) == QuotaWarningLevel.NORMAL

    def test_unlimited(self):
        assert get_warning_level(9999, float("inf")) == QuotaWarningLevel.NORMAL

    def test_boundary_79_percent(self):
        """测试 79% 仍是 normal"""
        assert get_warning_level(79, 100) == QuotaWarningLevel.NORMAL

    def test_boundary_80_percent(self):
        """测试 80% 是 warning"""
        assert get_warning_level(80, 100) == QuotaWarningLevel.WARNING

    def test_boundary_94_percent(self):
        """测试 94% 仍是 warning"""
        assert get_warning_level(94, 100) == QuotaWarningLevel.WARNING

    def test_boundary_95_percent(self):
        """测试 95% 是 critical"""
        assert get_warning_level(95, 100) == QuotaWarningLevel.CRITICAL

    def test_boundary_99_percent(self):
        """测试 99% 仍是 critical"""
        assert get_warning_level(99, 100) == QuotaWarningLevel.CRITICAL

    def test_zero_usage(self):
        """测试 0 使用量"""
        assert get_warning_level(0, 100) == QuotaWarningLevel.NORMAL


class TestGlobalSingletons:
    """全局单例测试"""

    def test_quota_registry_exists(self):
        """测试全局 quota_registry 存在"""
        assert quota_registry is not None
        assert isinstance(quota_registry, QuotaRegistry)
        assert "free" in quota_registry.list_tiers()

    def test_quota_checker_exists(self):
        """测试全局 quota_checker 存在"""
        assert quota_checker is not None
        assert isinstance(quota_checker, QuotaChecker)

    def test_global_checker_uses_global_registry(self):
        """测试全局 checker 使用全局 registry"""
        result = quota_checker.check("free", "storage_gb", 1.0)
        assert result.limit == 2