xiaoxia-saas/tests/unit/test_feature_flag_store.py

"""
FeatureFlagStore 单元测试

覆盖：
- FeatureFlagConfig: to_dict / from_dict 序列化
- FeatureFlagConfig.is_active: 全局开关/白名单/百分比哈希
- InMemoryFeatureFlagStore: CRUD / is_active
"""

import pytest

from packages.adapters.redis.feature_flag_store import (
    FEATURE_FLAG_REDIS_PREFIX,
    FeatureFlagConfig,
    InMemoryFeatureFlagStore,
)

# ============================================================
# 常量
# ============================================================


class TestConstants:
    """常量验证"""

    def test_redis_prefix(self):
        assert FEATURE_FLAG_REDIS_PREFIX == "feature_flag:"


# ============================================================
# FeatureFlagConfig - 默认值 & 基础
# ============================================================


class TestFeatureFlagConfigDefaults:
    """FeatureFlagConfig 默认值"""

    def test_required_name(self):
        config = FeatureFlagConfig(name="test_flag")
        assert config.name == "test_flag"

    def test_default_disabled(self):
        config = FeatureFlagConfig(name="test_flag")
        assert config.enabled is False

    def test_default_percentage_zero(self):
        config = FeatureFlagConfig(name="test_flag")
        assert config.percentage == 0

    def test_default_whitelist_empty(self):
        config = FeatureFlagConfig(name="test_flag")
        assert config.whitelist == set()

    def test_full_config(self):
        config = FeatureFlagConfig(
            name="full_flag",
            enabled=True,
            percentage=50,
            whitelist={"user1", "user2"},
        )
        assert config.name == "full_flag"
        assert config.enabled is True
        assert config.percentage == 50
        assert config.whitelist == {"user1", "user2"}


# ============================================================
# FeatureFlagConfig - 序列化
# ============================================================


class TestFeatureFlagConfigSerialization:
    """to_dict / from_dict 序列化"""

    def test_to_dict_defaults(self):
        config = FeatureFlagConfig(name="test")
        d = config.to_dict()
        assert d["name"] == "test"
        assert d["enabled"] is False
        assert d["percentage"] == 0
        assert d["whitelist"] == []

    def test_to_dict_with_values(self):
        config = FeatureFlagConfig(
            name="test",
            enabled=True,
            percentage=75,
            whitelist={"a", "b", "c"},
        )
        d = config.to_dict()
        assert d["name"] == "test"
        assert d["enabled"] is True
        assert d["percentage"] == 75
        # whitelist 排序后输出
        assert sorted(d["whitelist"]) == ["a", "b", "c"]

    def test_from_dict_minimal(self):
        d = {"name": "test"}
        config = FeatureFlagConfig.from_dict(d)
        assert config.name == "test"
        assert config.enabled is False
        assert config.percentage == 0
        assert config.whitelist == set()

    def test_from_dict_full(self):
        d = {
            "name": "full",
            "enabled": True,
            "percentage": 30,
            "whitelist": ["u1", "u2"],
        }
        config = FeatureFlagConfig.from_dict(d)
        assert config.name == "full"
        assert config.enabled is True
        assert config.percentage == 30
        assert config.whitelist == {"u1", "u2"}

    def test_round_trip(self):
        original = FeatureFlagConfig(
            name="round_trip",
            enabled=True,
            percentage=42,
            whitelist={"alice", "bob", "charlie"},
        )
        d = original.to_dict()
        restored = FeatureFlagConfig.from_dict(d)
        assert restored.name == original.name
        assert restored.enabled == original.enabled
        assert restored.percentage == original.percentage
        assert restored.whitelist == original.whitelist

    def test_from_dict_coerces_types(self):
        """from_dict 应该做类型转换"""
        d = {
            "name": "coerce",
            "enabled": 1,  # int → bool
            "percentage": "50",  # str → int
            "whitelist": ("a", "b"),  # tuple → set
        }
        config = FeatureFlagConfig.from_dict(d)
        assert config.enabled is True
        assert config.percentage == 50
        assert config.whitelist == {"a", "b"}


# ============================================================
# FeatureFlagConfig.is_active - 全局开关
# ============================================================


class TestIsActiveGlobalSwitch:
    """is_active - 全局开关基础"""

    def test_disabled_returns_false(self):
        config = FeatureFlagConfig(name="test", enabled=False)
        assert config.is_active() is False

    def test_disabled_with_identifier_returns_false(self):
        config = FeatureFlagConfig(name="test", enabled=False)
        assert config.is_active(identifier="user1") is False

    def test_enabled_no_percentage_no_whitelist_returns_true(self):
        config = FeatureFlagConfig(name="test", enabled=True)
        # percentage=0, whitelist=空，但 enabled=True
        # 按逻辑：全局开了但百分比0且无白名单 → 其实应该是 False？
        # 让我看代码...
        # 代码里 percentage <= 0 时返回 False（没有白名单且百分比为0）
        assert config.is_active() is False

    def test_enabled_100_percent_returns_true(self):
        config = FeatureFlagConfig(name="test", enabled=True, percentage=100)
        assert config.is_active() is True


# ============================================================
# FeatureFlagConfig.is_active - 白名单
# ============================================================


class TestIsActiveWhitelist:
    """is_active - 白名单优先级"""

    def test_whitelist_match_returns_true(self):
        config = FeatureFlagConfig(
            name="test",
            enabled=True,
            whitelist={"user1", "user2"},
        )
        assert config.is_active(identifier="user1") is True

    def test_whitelist_no_match_falls_through(self):
        config = FeatureFlagConfig(
            name="test",
            enabled=True,
            percentage=0,
            whitelist={"user1"},
        )
        # 不在白名单，且百分比为0 → False
        assert config.is_active(identifier="user3") is False

    def test_whitelist_overrides_percentage_zero(self):
        """白名单优先级最高，即使百分比为0也能启用"""
        config = FeatureFlagConfig(
            name="test",
            enabled=True,
            percentage=0,
            whitelist={"vip_user"},
        )
        assert config.is_active(identifier="vip_user") is True

    def test_whitelist_overrides_partial_percentage(self):
        """白名单用户即使在百分比外也能启用"""
        config = FeatureFlagConfig(
            name="test",
            enabled=True,
            percentage=1,  # 只有1%的用户
            whitelist={"important_user"},
        )
        # 白名单用户直接通过
        assert config.is_active(identifier="important_user") is True

    def test_no_identifier_no_whitelist_check(self):
        """不传 identifier 时不做白名单检查"""
        config = FeatureFlagConfig(
            name="test",
            enabled=True,
            percentage=100,
            whitelist={"user1"},
        )
        # 无 identifier，直接看百分比（100%）
        assert config.is_active() is True


# ============================================================
# FeatureFlagConfig.is_active - 百分比边界值
# ============================================================


class TestIsActivePercentageBoundaries:
    """is_active - 百分比边界值"""

    def test_percentage_0_returns_false(self):
        config = FeatureFlagConfig(name="test", enabled=True, percentage=0)
        assert config.is_active(identifier="any_user") is False

    def test_percentage_100_returns_true(self):
        config = FeatureFlagConfig(name="test", enabled=True, percentage=100)
        assert config.is_active(identifier="any_user") is True

    def test_percentage_negative_treated_as_0(self):
        """percentage < 0 应该按 0 处理"""
        config = FeatureFlagConfig(name="test", enabled=True, percentage=-5)
        assert config.is_active(identifier="any_user") is False

    def test_percentage_over_100_treated_as_100(self):
        """percentage > 100 应该按 100 处理"""
        config = FeatureFlagConfig(name="test", enabled=True, percentage=150)
        assert config.is_active(identifier="any_user") is True


# ============================================================
# FeatureFlagConfig.is_active - 哈希一致性
# ============================================================


class TestIsActiveHashConsistency:
    """is_active - 哈希取模一致性验证"""

    def test_same_user_same_result_every_time(self):
        """同一用户多次调用结果一致（确定性哈希）"""
        config = FeatureFlagConfig(name="test", enabled=True, percentage=50)
        results = {config.is_active(identifier="user_xyz") for _ in range(100)}
        assert len(results) == 1  # 全部相同

    def test_different_flags_same_user_can_differ(self):
        """不同 flag 对同一用户可以有不同结果（因为 flag name 参与哈希）"""
        config_a = FeatureFlagConfig(name="flag_a", enabled=True, percentage=50)
        config_b = FeatureFlagConfig(name="flag_b", enabled=True, percentage=50)
        # 不保证一定不同，但大部分情况下应该不同
        # 这里只验证哈希输入包含了 flag name（通过机制保证）
        # 具体是否不同取决于哈希值

    def test_percentage_coverage_roughly_correct(self):
        """大量用户中，命中比例大致接近百分比"""
        config = FeatureFlagConfig(name="coverage_test", enabled=True, percentage=30)
        users = [f"user_{i}" for i in range(1000)]
        active_count = sum(1 for u in users if config.is_active(identifier=u))
        # 30% ± 10% 的容差
        assert 200 <= active_count <= 400

    def test_50_percent_roughly_half(self):
        config = FeatureFlagConfig(name="half_test", enabled=True, percentage=50)
        users = [f"user_{i}" for i in range(1000)]
        active_count = sum(1 for u in users if config.is_active(identifier=u))
        # 50% ± 10%
        assert 400 <= active_count <= 600

    def test_10_percent_roughly_tenth(self):
        config = FeatureFlagConfig(name="ten_pct", enabled=True, percentage=10)
        users = [f"user_{i}" for i in range(1000)]
        active_count = sum(1 for u in users if config.is_active(identifier=u))
        assert 50 <= active_count <= 150

    def test_empty_identifier_treated_as_no_identifier(self):
        """空字符串 identifier 应该如何处理？"""
        config = FeatureFlagConfig(name="test", enabled=True, percentage=50)
        # 空字符串是 falsy，走无 identifier 分支（随机）
        # 但白名单检查也会跳过
        # 验证不会崩溃
        result = config.is_active(identifier="")
        assert isinstance(result, bool)


# ============================================================
# InMemoryFeatureFlagStore - CRUD
# ============================================================


class TestInMemoryFeatureFlagStore:
    """InMemoryFeatureFlagStore 内存实现"""

    def test_get_nonexistent_returns_default_disabled(self):
        store = InMemoryFeatureFlagStore()
        config = store.get("nonexistent")
        assert config.name == "nonexistent"
        assert config.enabled is False
        assert config.percentage == 0

    def test_set_and_get(self):
        store = InMemoryFeatureFlagStore()
        original = FeatureFlagConfig(
            name="my_flag",
            enabled=True,
            percentage=50,
            whitelist={"admin"},
        )
        store.set(original)
        retrieved = store.get("my_flag")
        assert retrieved.name == "my_flag"
        assert retrieved.enabled is True
        assert retrieved.percentage == 50
        assert retrieved.whitelist == {"admin"}

    def test_set_overwrites_existing(self):
        store = InMemoryFeatureFlagStore()
        store.set(FeatureFlagConfig(name="flag", enabled=True, percentage=30))
        store.set(FeatureFlagConfig(name="flag", enabled=False, percentage=70))
        config = store.get("flag")
        assert config.enabled is False
        assert config.percentage == 70

    def test_delete_existing_returns_true(self):
        store = InMemoryFeatureFlagStore()
        store.set(FeatureFlagConfig(name="delete_me"))
        result = store.delete("delete_me")
        assert result is True
        # 删除后获取返回默认配置
        assert store.get("delete_me").enabled is False

    def test_delete_nonexistent_returns_false(self):
        store = InMemoryFeatureFlagStore()
        result = store.delete("no_such_flag")
        assert result is False

    def test_list_all_empty(self):
        store = InMemoryFeatureFlagStore()
        assert store.list_all() == {}

    def test_list_all_multiple(self):
        store = InMemoryFeatureFlagStore()
        store.set(FeatureFlagConfig(name="flag1", enabled=True))
        store.set(FeatureFlagConfig(name="flag2", percentage=50))
        store.set(FeatureFlagConfig(name="flag3"))

        all_flags = store.list_all()
        assert len(all_flags) == 3
        assert "flag1" in all_flags
        assert "flag2" in all_flags
        assert "flag3" in all_flags
        assert all_flags["flag1"].enabled is True
        assert all_flags["flag2"].percentage == 50

    def test_list_all_returns_copy(self):
        """返回的是副本，修改不影响内部状态"""
        store = InMemoryFeatureFlagStore()
        store.set(FeatureFlagConfig(name="flag1"))
        flags = store.list_all()
        flags["fake"] = FeatureFlagConfig(name="fake")
        assert "fake" not in store.list_all()


# ============================================================
# InMemoryFeatureFlagStore - is_active
# ============================================================


class TestInMemoryStoreIsActive:
    """store.is_active 便捷方法"""

    def test_is_active_enabled_flag(self):
        store = InMemoryFeatureFlagStore()
        store.set(FeatureFlagConfig(name="on", enabled=True, percentage=100))
        assert store.is_active("on") is True

    def test_is_active_disabled_flag(self):
        store = InMemoryFeatureFlagStore()
        store.set(FeatureFlagConfig(name="off", enabled=False))
        assert store.is_active("off") is False

    def test_is_active_nonexistent_flag(self):
        store = InMemoryFeatureFlagStore()
        assert store.is_active("unknown") is False

    def test_is_active_with_identifier_whitelist(self):
        store = InMemoryFeatureFlagStore()
        store.set(
            FeatureFlagConfig(
                name="beta",
                enabled=True,
                percentage=0,
                whitelist={"tester1"},
            )
        )
        assert store.is_active("beta", identifier="tester1") is True
        assert store.is_active("beta", identifier="other_user") is False