3921a657e8
CI/CD Pipeline / Check if frontend-only change (push) Has been skipped
CI/CD Pipeline / PR Build API Image (push) Has been skipped
CI/CD Pipeline / PR Build Web Image (push) Has been skipped
CI/CD Pipeline / PR Build Worker Image (push) Has been skipped
CI/CD Pipeline / Frontend Lint (push) Successful in 52s
CI/CD Pipeline / Build Production API Image (push) Has been skipped
CI/CD Pipeline / Build Production Web Image (push) Has been skipped
CI/CD Pipeline / Build Production Worker Image (push) Has been skipped
CI/CD Pipeline / Validate - Type Check (mypy) (push) Successful in 1m56s
CI/CD Pipeline / Validate - Migration (alembic) (push) Successful in 2m7s
CI/CD Pipeline / Build Staging Web Image (push) Successful in 2m7s
CI/CD Pipeline / Frontend Unit Tests (push) Successful in 54s
CI/CD Pipeline / Validate - Code Quality (push) Failing after 3m20s
CI/CD Pipeline / Build Staging Worker Image (push) Successful in 6m27s
CI/CD Pipeline / Unit Tests (push) Failing after 8m24s
CI/CD Pipeline / Deploy Production (push) Has been skipped
CI/CD Pipeline / Build Staging API Image (push) Successful in 16m1s
CI/CD Pipeline / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Successful in 1m14s
CI/CD Pipeline / Integration Tests (push) Successful in 3m28s
CI/CD Pipeline / ACR Image Cleanup (push) Successful in 21s
CI/CD Pipeline / Staging E2E Tests (push) Failing after 1m26s
CI/CD Pipeline / Staging API Integration Tests (push) Successful in 3m34s
249 lines
10 KiB
Python
Executable File
249 lines
10 KiB
Python
Executable File
"""Feature Flags 单元测试"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import pytest
|
||
|
||
from packages.infrastructure.feature_flags import (
|
||
FeatureFlag,
|
||
FeatureFlags,
|
||
FeatureScope,
|
||
)
|
||
|
||
|
||
class TestFeatureFlag:
|
||
"""单个 FeatureFlag 测试"""
|
||
|
||
def test_default_enabled(self):
|
||
"""默认全局启用"""
|
||
flag = FeatureFlag(name="test_feature")
|
||
assert flag.is_enabled() is True
|
||
|
||
def test_global_disabled(self):
|
||
"""全局禁用"""
|
||
flag = FeatureFlag(name="test_feature", global_enabled=False)
|
||
assert flag.is_enabled() is False
|
||
|
||
def test_plan_override_free_disabled(self):
|
||
"""free 套餐被覆盖为禁用"""
|
||
flag = FeatureFlag(name="test_feature", global_enabled=True, plan_overrides={"free": False})
|
||
assert flag.is_enabled(user_plan="free") is False
|
||
assert flag.is_enabled(user_plan="basic") is True
|
||
assert flag.is_enabled(user_plan="premium") is True
|
||
|
||
def test_plan_override_premium_only(self):
|
||
"""仅 premium 可用"""
|
||
flag = FeatureFlag(
|
||
name="test_feature",
|
||
global_enabled=True,
|
||
plan_overrides={"free": False, "basic": False},
|
||
)
|
||
assert flag.is_enabled(user_plan="free") is False
|
||
assert flag.is_enabled(user_plan="basic") is False
|
||
assert flag.is_enabled(user_plan="premium") is True
|
||
|
||
def test_user_override_priority_higher_than_plan(self):
|
||
"""用户白名单优先级高于套餐"""
|
||
flag = FeatureFlag(
|
||
name="test_feature",
|
||
global_enabled=False,
|
||
plan_overrides={"free": False},
|
||
user_overrides={"user_001": True},
|
||
)
|
||
# 用户在白名单中,即使全局禁用+free套餐也启用
|
||
assert flag.is_enabled(user_plan="free", user_id="user_001") is True
|
||
|
||
def test_user_override_disable(self):
|
||
"""用户白名单可单独禁用"""
|
||
flag = FeatureFlag(
|
||
name="test_feature",
|
||
global_enabled=True,
|
||
user_overrides={"user_002": False},
|
||
)
|
||
assert flag.is_enabled(user_id="user_002") is False
|
||
assert flag.is_enabled(user_id="user_001") is True
|
||
|
||
def test_user_override_without_plan(self):
|
||
"""用户白名单无需套餐也生效"""
|
||
flag = FeatureFlag(name="test_feature", global_enabled=False, user_overrides={"u1": True})
|
||
assert flag.is_enabled(user_id="u1") is True
|
||
assert flag.is_enabled(user_id="u2") is False
|
||
|
||
def test_no_plan_uses_global(self):
|
||
"""不传 user_plan 时回退到全局开关"""
|
||
flag = FeatureFlag(name="test", global_enabled=True, plan_overrides={"free": False})
|
||
assert flag.is_enabled() is True
|
||
|
||
def test_default_values(self):
|
||
"""默认值正确"""
|
||
flag = FeatureFlag(name="test")
|
||
assert flag.name == "test"
|
||
assert flag.description == ""
|
||
assert flag.global_enabled is True
|
||
assert flag.plan_overrides == {}
|
||
assert flag.user_overrides == {}
|
||
|
||
|
||
class TestFeatureFlags:
|
||
"""FeatureFlags 管理器测试"""
|
||
|
||
def test_singleton_default_flags(self):
|
||
"""默认有 5 个 feature flags"""
|
||
ff = FeatureFlags()
|
||
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_is_enabled_existing_flag(self):
|
||
"""已存在的 flag 正常判断"""
|
||
ff = FeatureFlags()
|
||
assert ff.is_enabled(FeatureScope.AI_VOICE_GENERATION) is True
|
||
|
||
def test_is_enabled_nonexistent_flag(self):
|
||
"""不存在的 flag 默认禁用"""
|
||
ff = FeatureFlags()
|
||
assert ff.is_enabled("nonexistent_flag") is False
|
||
|
||
def test_is_enabled_with_plan(self):
|
||
"""按套餐判断"""
|
||
ff = FeatureFlags()
|
||
# free 套餐 AI 配音不可用
|
||
assert ff.is_enabled(FeatureScope.AI_VOICE_GENERATION, user_plan="free") is False
|
||
assert ff.is_enabled(FeatureScope.AI_VOICE_GENERATION, user_plan="basic") is True
|
||
assert ff.is_enabled(FeatureScope.AI_VOICE_GENERATION, user_plan="premium") is True
|
||
|
||
def test_is_enabled_premium_only_features(self):
|
||
"""仅 premium 可用的功能"""
|
||
ff = FeatureFlags()
|
||
for feat in [FeatureScope.DEDUPLICATION_REPORT, FeatureScope.MULTI_PLATFORM_OUTPUT]:
|
||
assert ff.is_enabled(feat, user_plan="free") is False
|
||
assert ff.is_enabled(feat, user_plan="basic") is False
|
||
assert ff.is_enabled(feat, user_plan="premium") is True
|
||
|
||
def test_register_new_flag(self):
|
||
"""注册新 flag"""
|
||
ff = FeatureFlags()
|
||
new_flag = FeatureFlag(name="new_feature", description="新功能", global_enabled=True)
|
||
ff.register(new_flag)
|
||
|
||
assert ff.is_enabled("new_feature") is True
|
||
assert ff.get("new_feature") is not None
|
||
assert ff.get("new_feature").description == "新功能"
|
||
|
||
def test_register_overwrites_existing(self):
|
||
"""注册同名 flag 覆盖旧的"""
|
||
ff = FeatureFlags()
|
||
original = ff.get(FeatureScope.AI_VOICE_GENERATION)
|
||
assert original.global_enabled is True
|
||
|
||
new_flag = FeatureFlag(name=FeatureScope.AI_VOICE_GENERATION, global_enabled=False)
|
||
ff.register(new_flag)
|
||
|
||
assert ff.is_enabled(FeatureScope.AI_VOICE_GENERATION) is False
|
||
|
||
def test_get_returns_none_for_missing(self):
|
||
"""获取不存在的 flag 返回 None"""
|
||
ff = FeatureFlags()
|
||
assert ff.get("no_such_flag") is None
|
||
|
||
def test_set_global(self):
|
||
"""设置全局开关"""
|
||
ff = FeatureFlags()
|
||
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_missing_raises(self):
|
||
"""设置不存在的 flag 抛 KeyError"""
|
||
ff = FeatureFlags()
|
||
with pytest.raises(KeyError):
|
||
ff.set_global("nonexistent", True)
|
||
|
||
def test_set_plan_override(self):
|
||
"""设置套餐级别覆盖"""
|
||
ff = FeatureFlags()
|
||
ff.set_plan_override(FeatureScope.BATCH_EXPORT, "basic", False)
|
||
assert ff.is_enabled(FeatureScope.BATCH_EXPORT, user_plan="basic") is False
|
||
assert ff.is_enabled(FeatureScope.BATCH_EXPORT, user_plan="premium") is True
|
||
|
||
def test_set_plan_override_missing_raises(self):
|
||
"""设置不存在的 flag 抛 KeyError"""
|
||
ff = FeatureFlags()
|
||
with pytest.raises(KeyError):
|
||
ff.set_plan_override("nonexistent", "free", False)
|
||
|
||
def test_set_user_override(self):
|
||
"""设置用户白名单"""
|
||
ff = FeatureFlags()
|
||
ff.set_user_override(FeatureScope.DEDUPLICATION_REPORT, "user_42", True)
|
||
assert ff.is_enabled(FeatureScope.DEDUPLICATION_REPORT, user_plan="free", user_id="user_42") is True
|
||
|
||
def test_set_user_override_disable(self):
|
||
"""用户白名单禁用"""
|
||
ff = FeatureFlags()
|
||
ff.set_user_override(FeatureScope.AI_VOICE_GENERATION, "user_99", False)
|
||
assert ff.is_enabled(FeatureScope.AI_VOICE_GENERATION, user_plan="premium", user_id="user_99") is False
|
||
|
||
def test_set_user_override_missing_raises(self):
|
||
"""设置不存在的 flag 抛 KeyError"""
|
||
ff = FeatureFlags()
|
||
with pytest.raises(KeyError):
|
||
ff.set_user_override("nonexistent", "u1", True)
|
||
|
||
def test_list_flags_returns_copy(self):
|
||
"""list_flags 返回副本,修改不影响内部"""
|
||
ff = FeatureFlags()
|
||
flags = ff.list_flags()
|
||
flags["new_one"] = FeatureFlag(name="new_one")
|
||
assert ff.get("new_one") is None
|
||
|
||
def test_get_enabled_for_plan_free(self):
|
||
"""获取 free 套餐下启用的功能"""
|
||
ff = FeatureFlags()
|
||
enabled = ff.get_enabled_for_plan("free")
|
||
# free 套餐只有 recipe_reuse 可用?不对,看看默认配置
|
||
# AI_VOICE_GENERATION: free=False
|
||
# DEDUPLICATION_REPORT: free=False, basic=False
|
||
# BATCH_EXPORT: free=False
|
||
# MULTI_PLATFORM_OUTPUT: free=False, basic=False
|
||
# RECIPE_REUSE: free=False
|
||
# 所以 free 套餐全部禁用?不,recipe_reuse free=False
|
||
# 等等,RECIPE_REUSE 的 plan_overrides 是 {"free": False},
|
||
# 那对于 free 套餐,返回 False;但全局是 True
|
||
# 所以 free 套餐没有任何启用的?不对...
|
||
# 让我重新看:global_enabled=True,plan_overrides={"free": False}
|
||
# 那么 free 套餐 is_enabled 是 False,其他套餐是 True
|
||
# 所以 free 套餐应该 0 个启用?不对,等等...
|
||
# 不,我需要重新检查每个 flag 的 plan_overrides
|
||
# AI_VOICE_GENERATION: free=False → free: False, basic/premium: True
|
||
# DEDUPLICATION_REPORT: free=False, basic=False → free/basic: False, premium: True
|
||
# BATCH_EXPORT: free=False → free: False, basic/premium: True
|
||
# MULTI_PLATFORM_OUTPUT: free=False, basic=False → free/basic: False, premium: True
|
||
# RECIPE_REUSE: free=False → free: False, basic/premium: True
|
||
assert len(enabled) == 0
|
||
|
||
def test_get_enabled_for_plan_basic(self):
|
||
"""获取 basic 套餐下启用的功能"""
|
||
ff = FeatureFlags()
|
||
enabled = ff.get_enabled_for_plan("basic")
|
||
# basic 套餐:AI_VOICE、BATCH_EXPORT、RECIPE_REUSE 可用
|
||
# DEDUPLICATION_REPORT、MULTI_PLATFORM_OUTPUT 不可用
|
||
assert FeatureScope.AI_VOICE_GENERATION in enabled
|
||
assert FeatureScope.BATCH_EXPORT in enabled
|
||
assert FeatureScope.RECIPE_REUSE in enabled
|
||
assert FeatureScope.DEDUPLICATION_REPORT not in enabled
|
||
assert FeatureScope.MULTI_PLATFORM_OUTPUT not in enabled
|
||
assert len(enabled) == 3
|
||
|
||
def test_get_enabled_for_plan_premium(self):
|
||
"""获取 premium 套餐下所有功能都启用"""
|
||
ff = FeatureFlags()
|
||
enabled = ff.get_enabled_for_plan("premium")
|
||
assert len(enabled) == 5
|