xiaoxia-saas/tests/unit/test_feature_flags_infra.py

"""
Feature Flags 基础设施测试.
"""

from __future__ import annotations

import pytest

from packages.infrastructure.feature_flags import (
    FeatureFlag,
    FeatureFlags,
    FeatureScope,
)


class TestFeatureFlagDefaults:
    """FeatureFlag 默认值."""

    def test_default_enabled(self):
        flag = FeatureFlag(name="test")
        assert flag.global_enabled is True
        assert flag.plan_overrides == {}
        assert flag.user_overrides == {}
        assert flag.description == ""

    def test_custom_description(self):
        flag = FeatureFlag(name="test", description="测试功能")
        assert flag.description == "测试功能"

    def test_global_disabled(self):
        flag = FeatureFlag(name="test", global_enabled=False)
        assert flag.global_enabled is False


class TestFeatureFlagIsEnabled:
    """is_enabled 优先级逻辑."""

    def test_global_enabled_no_user_no_plan(self):
        flag = FeatureFlag(name="test", global_enabled=True)
        assert flag.is_enabled() is True

    def test_global_disabled_no_user_no_plan(self):
        flag = FeatureFlag(name="test", global_enabled=False)
        assert flag.is_enabled() is False

    def test_plan_override_free_disabled(self):
        flag = FeatureFlag(
            name="test",
            global_enabled=True,
            plan_overrides={"free": False},
        )
        assert flag.is_enabled(user_plan="free") is False
        assert flag.is_enabled(user_plan="premium") is True  # 走全局

    def test_plan_override_premium_enabled(self):
        flag = FeatureFlag(
            name="test",
            global_enabled=False,
            plan_overrides={"premium": True},
        )
        assert flag.is_enabled(user_plan="premium") is True
        assert flag.is_enabled(user_plan="free") is False  # 走全局

    def test_user_override_highest_priority(self):
        """用户白名单优先级最高"""
        flag = FeatureFlag(
            name="test",
            global_enabled=False,
            plan_overrides={"premium": True},
            user_overrides={"user_1": True},
        )
        # free 套餐全局禁用，但用户在白名单 → 启用
        assert flag.is_enabled(user_plan="free", user_id="user_1") is True

    def test_user_override_disable_overrides_plan(self):
        flag = FeatureFlag(
            name="test",
            global_enabled=True,
            plan_overrides={"premium": True},
            user_overrides={"user_1": False},
        )
        # premium 套餐应该启用，但用户在禁用名单 → 禁用
        assert flag.is_enabled(user_plan="premium", user_id="user_1") is False

    def test_user_not_in_overrides_falls_to_plan(self):
        flag = FeatureFlag(
            name="test",
            global_enabled=True,
            plan_overrides={"free": False},
            user_overrides={"user_x": True},
        )
        assert flag.is_enabled(user_plan="free", user_id="user_other") is False

    def test_none_user_id_skipped(self):
        flag = FeatureFlag(
            name="test",
            global_enabled=True,
            plan_overrides={"free": False},
            user_overrides={"None": False},
        )
        assert flag.is_enabled(user_plan="premium", user_id=None) is True

    def test_none_plan_skipped(self):
        flag = FeatureFlag(
            name="test",
            global_enabled=True,
            plan_overrides={"free": False},
        )
        assert flag.is_enabled(user_plan=None) is True

    def test_empty_user_id_skipped(self):
        flag = FeatureFlag(
            name="test",
            global_enabled=True,
            plan_overrides={"free": False},
        )
        assert flag.is_enabled(user_plan="premium", user_id="") is True

    def test_empty_plan_skipped(self):
        flag = FeatureFlag(
            name="test",
            global_enabled=True,
            plan_overrides={"free": False},
        )
        assert flag.is_enabled(user_plan="") is True


class TestFeatureScope:
    """FeatureScope 常量."""

    def test_constants_exist(self):
        assert FeatureScope.AI_VOICE_GENERATION == "ai_voice_generation"
        assert FeatureScope.DEDUPLICATION_REPORT == "deduplication_report"
        assert FeatureScope.BATCH_EXPORT == "batch_export"
        assert FeatureScope.MULTI_PLATFORM_OUTPUT == "multi_platform_output"
        assert FeatureScope.RECIPE_REUSE == "recipe_reuse"


class TestFeatureFlagsManager:
    """FeatureFlags 管理器."""

    @pytest.fixture
    def ff(self):
        return FeatureFlags()

    def test_default_flags_registered(self, ff):
        flags = ff.list_flags()
        assert len(flags) >= 5
        assert FeatureScope.AI_VOICE_GENERATION in flags
        assert FeatureScope.DEDUPLICATION_REPORT in flags
        assert FeatureScope.BATCH_EXPORT in flags
        assert FeatureScope.MULTI_PLATFORM_OUTPUT in flags
        assert FeatureScope.RECIPE_REUSE in flags

    def test_get_existing_flag(self, ff):
        flag = ff.get(FeatureScope.BATCH_EXPORT)
        assert flag is not None
        assert flag.name == FeatureScope.BATCH_EXPORT

    def test_get_nonexistent_flag(self, ff):
        assert ff.get("nonexistent") is None

    def test_is_enabled_global(self, ff):
        assert ff.is_enabled(FeatureScope.BATCH_EXPORT) is True

    def test_is_enabled_nonexistent_returns_false(self, ff):
        """未知 flag 默认禁用（安全保守）"""
        assert ff.is_enabled("unknown_feature") is False

    def test_free_plan_ai_voice_disabled(self, ff):
        """AI 配音 free 套餐不可用"""
        assert ff.is_enabled(FeatureScope.AI_VOICE_GENERATION, user_plan="free") is False

    def test_premium_plan_ai_voice_enabled(self, ff):
        assert ff.is_enabled(FeatureScope.AI_VOICE_GENERATION, user_plan="premium") is True

    def test_deduplication_only_premium(self, ff):
        """去重报告仅 premium 可用"""
        assert ff.is_enabled(FeatureScope.DEDUPLICATION_REPORT, user_plan="free") is False
        assert ff.is_enabled(FeatureScope.DEDUPLICATION_REPORT, user_plan="basic") is False
        assert ff.is_enabled(FeatureScope.DEDUPLICATION_REPORT, user_plan="premium") is True

    def test_set_global(self, ff):
        ff.set_global(FeatureScope.BATCH_EXPORT, False)
        assert ff.is_enabled(FeatureScope.BATCH_EXPORT) is False
        # 恢复
        ff.set_global(FeatureScope.BATCH_EXPORT, True)
        assert ff.is_enabled(FeatureScope.BATCH_EXPORT) is True

    def test_set_global_nonexistent_raises(self, ff):
        with pytest.raises(KeyError):
            ff.set_global("nonexistent", True)

    def test_set_plan_override(self, ff):
        ff.set_plan_override(FeatureScope.BATCH_EXPORT, "enterprise", False)
        assert ff.is_enabled(FeatureScope.BATCH_EXPORT, user_plan="enterprise") is False

    def test_set_plan_override_nonexistent_raises(self, ff):
        with pytest.raises(KeyError):
            ff.set_plan_override("nonexistent", "free", True)

    def test_set_user_override(self, ff):
        ff.set_user_override(FeatureScope.BATCH_EXPORT, "user_42", True)
        assert (
            ff.is_enabled(
                FeatureScope.BATCH_EXPORT,
                user_plan="free",
                user_id="user_42",
            )
            is True
        )

    def test_set_user_override_nonexistent_raises(self, ff):
        with pytest.raises(KeyError):
            ff.set_user_override("nonexistent", "user_1", True)

    def test_register_new_flag(self, ff):
        new_flag = FeatureFlag(name="new_feature", description="新功能")
        ff.register(new_flag)
        assert ff.get("new_feature") is not None
        assert ff.is_enabled("new_feature") is True

    def test_register_overwrites(self, ff):
        flag1 = FeatureFlag(name="test", global_enabled=True)
        flag2 = FeatureFlag(name="test", global_enabled=False)
        ff.register(flag1)
        ff.register(flag2)
        assert ff.is_enabled("test") is False

    def test_list_flags_returns_copy(self, ff):
        flags = ff.list_flags()
        assert isinstance(flags, dict)
        # 修改返回值不影响内部
        flags["new"] = FeatureFlag(name="new")
        assert ff.get("new") is None

    def test_get_enabled_for_plan(self, ff):
        free_features = ff.get_enabled_for_plan("free")
        premium_features = ff.get_enabled_for_plan("premium")
        assert len(premium_features) >= len(free_features)
        # free 套餐功能是 premium 的子集
        for f in free_features:
            assert f in premium_features