xiaoxia-saas/tests/unit/test_quota.py

"""Quota 配额系统单测 — #1894 档位清理后版本 (free/monthly/quarterly/yearly)."""

from __future__ import annotations

import math

import pytest

from packages.domain.quota import (
    QUOTA_TIERS,
    QuotaChecker,
    QuotaCheckResult,
    QuotaDimension,
    QuotaRegistry,
    QuotaTier,
    QuotaWarningLevel,
    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):
        tier = QuotaTier(name="test", limits={})
        assert tier.is_unlimited("unknown") is True


# ── QuotaTiers — #1894 新档位: free / monthly / quarterly / yearly ─────────


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_monthly_tier_exists(self):
        assert "monthly" in QUOTA_TIERS
        monthly = QUOTA_TIERS["monthly"]
        assert monthly.get_limit("storage_gb") == 20
        assert monthly.get_limit("videos_per_month") == 30
        assert monthly.get_limit("ai_voice_enabled") == 1

    def test_quarterly_tier_exists(self):
        assert "quarterly" in QUOTA_TIERS
        quarterly = QUOTA_TIERS["quarterly"]
        assert quarterly.get_limit("storage_gb") == 100
        assert quarterly.get_limit("videos_per_month") == 100

    def test_yearly_tier_exists(self):
        assert "yearly" in QUOTA_TIERS
        yearly = QUOTA_TIERS["yearly"]
        assert yearly.get_limit("storage_gb") == 100
        assert yearly.is_unlimited("videos_per_month") is True
        assert yearly.get_limit("ai_voice_credits") == 2000

    def test_quarterly_templates_unlimited(self):
        quarterly = QUOTA_TIERS["quarterly"]
        assert quarterly.is_unlimited("max_templates") is True

    def test_yearly_templates_unlimited(self):
        yearly = QUOTA_TIERS["yearly"]
        assert yearly.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_monthly_ai_voice_enabled(self):
        monthly = QUOTA_TIERS["monthly"]
        assert monthly.get_limit("ai_voice_enabled") == 1

    def test_storage_increases_with_tier(self):
        free = QUOTA_TIERS["free"].get_limit("storage_gb")
        monthly = QUOTA_TIERS["monthly"].get_limit("storage_gb")
        quarterly = QUOTA_TIERS["quarterly"].get_limit("storage_gb")
        assert free < monthly <= quarterly

    def test_legacy_tiers_are_aliases(self):
        """#1894: old standard/pro/enterprise/basic/premium 保留为别名以兼容历史数据。
        basic/standard → monthly; premium/pro/enterprise → quarterly."""
        assert QUOTA_TIERS["basic"] is QUOTA_TIERS["monthly"]
        assert QUOTA_TIERS["standard"] is QUOTA_TIERS["monthly"]
        assert QUOTA_TIERS["premium"] is QUOTA_TIERS["quarterly"]
        assert QUOTA_TIERS["pro"] is QUOTA_TIERS["quarterly"]
        assert QUOTA_TIERS["enterprise"] is QUOTA_TIERS["quarterly"]


# ── 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

    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_core_four_tiers(self):
        reg = QuotaRegistry()
        tiers = reg.list_tiers()
        assert "free" in tiers
        assert "monthly" in tiers
        assert "quarterly" in tiers
        assert "yearly" in tiers

    def test_legacy_tiers_are_aliases(self):
        """旧档位作为别名注册以兼容历史数据."""
        reg = QuotaRegistry()
        tiers = reg.list_tiers()
        for legacy in ("standard", "pro", "enterprise", "basic", "premium"):
            assert legacy in tiers
        assert reg.get_tier("basic") is reg.get_tier("monthly")
        assert reg.get_tier("pro") is reg.get_tier("quarterly")

    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, "monthly": 20, "quarterly": 50, "yearly": 100}
        )
        assert "custom_dim" in reg.list_dimensions()
        assert reg.get_limit("free", "custom_dim") == 5
        assert reg.get_limit("monthly", "custom_dim") == 20
        assert reg.get_limit("yearly", "custom_dim") == 100

    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", "新维度")
        assert reg.get_limit("free", "new_dim") == 0
        assert reg.get_limit("monthly", "new_dim") == 0
        assert reg.get_limit("yearly", "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

    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):
        checker = QuotaChecker()
        result = checker.check("free", "storage_gb", 2.0)
        assert result.allowed is False

    def test_check_unlimited_quarterly_templates(self):
        checker = QuotaChecker()
        result = checker.check("quarterly", "max_templates", 1000.0)
        assert result.allowed is True
        assert math.isinf(result.remaining)

    def test_check_unlimited_yearly_videos(self):
        checker = QuotaChecker()
        result = checker.check("yearly", "videos_per_month", 9999.0)
        assert result.allowed is True
        assert math.isinf(result.remaining)

    def test_check_unknown_plan(self):
        checker = QuotaChecker()
        result = checker.check("unknown", "storage_gb", 1.0)
        assert result.limit == 0
        assert not result.allowed

    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 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()
        result = checker.check("free", "storage_gb", 1.7)  # 85%
        assert result.warning_level == QuotaWarningLevel.WARNING

    def test_warning_level_critical(self):
        checker = QuotaChecker()
        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)
        assert result.warning_level == QuotaWarningLevel.EXCEEDED

    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_warning_threshold(self):
        assert get_warning_level(80, 100) == QuotaWarningLevel.WARNING

    def test_critical_threshold(self):
        assert get_warning_level(95, 100) == QuotaWarningLevel.CRITICAL

    def test_exceeded_at_100(self):
        assert get_warning_level(100, 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