diff --git a/tests/unit/test_feature_flag.py b/tests/unit/test_feature_flag.py index 8bab8c04e..99cd76850 100644 --- a/tests/unit/test_feature_flag.py +++ b/tests/unit/test_feature_flag.py @@ -1,6 +1,6 @@ """Feature Flag 单元测试。 -测试 FeatureFlagConfig、InMemoryFeatureFlagStore、RenderEngineResolver 的核心逻辑。 +测试 FeatureFlagConfig、InMemoryFeatureFlagStore、RedisFeatureFlagStore 的核心逻辑。 """ from __future__ import annotations @@ -184,148 +184,6 @@ class TestInMemoryFeatureFlagStore: assert store.is_active("nonexistent") is False -# ── RenderEngineResolver 测试 ─────────────────────────────────────────────── - - -class TestRenderEngineResolver: - """渲染引擎选择器测试。""" - - def test_default_legacy_when_flag_disabled(self): - """flag 关闭时使用默认引擎(legacy)。""" - store = InMemoryFeatureFlagStore() - resolver = self._make_resolver(store=store, default="legacy") - assert resolver.get_engine() == "legacy" - assert resolver.get_engine("user1") == "legacy" - - def test_default_unified_when_flag_disabled(self): - """flag 关闭但默认值是 unified 时返回 unified。""" - store = InMemoryFeatureFlagStore() - resolver = self._make_resolver(store=store, default="unified") - assert resolver.get_engine() == "unified" - - def test_whitelist_user_uses_unified(self): - """白名单用户走新引擎。""" - store = InMemoryFeatureFlagStore() - store.set( - FeatureFlagConfig( - name="render_engine", - enabled=True, - percentage=0, - whitelist={"beta_tester"}, - ) - ) - resolver = self._make_resolver(store=store, default="legacy") - assert resolver.get_engine("beta_tester") == "unified" - assert resolver.get_engine("normal_user") == "legacy" - - def test_100_percent_all_unified(self): - """100% 时所有用户走新引擎。""" - store = InMemoryFeatureFlagStore() - store.set(FeatureFlagConfig(name="render_engine", enabled=True, percentage=100)) - resolver = self._make_resolver(store=store, default="legacy") - for i in range(50): - assert resolver.get_engine(f"user_{i}") == "unified" - - def test_invalid_default_engine_fallback(self): - """无效默认值回退到 legacy。""" - store = InMemoryFeatureFlagStore() - resolver = self._make_resolver(store=store, default="invalid_value") - assert resolver.get_engine() == "legacy" - - def test_should_use_unified_helper(self): - """should_use_unified 便捷方法。""" - store = InMemoryFeatureFlagStore() - store.set( - FeatureFlagConfig( - name="render_engine", - enabled=True, - percentage=0, - whitelist={"user_a"}, - ) - ) - resolver = self._make_resolver(store=store) - assert resolver.should_use_unified("user_a") is True - assert resolver.should_use_unified("user_b") is False - - def test_config_snapshot(self): - """配置快照。""" - store = InMemoryFeatureFlagStore() - store.set( - FeatureFlagConfig( - name="render_engine", - enabled=True, - percentage=30, - whitelist={"u1", "u2"}, - ) - ) - resolver = self._make_resolver(store=store) - snapshot = resolver.get_config_snapshot() - assert snapshot["flag_name"] == "render_engine" - assert snapshot["enabled"] is True - assert snapshot["percentage"] == 30 - assert snapshot["whitelist"] == ["u1", "u2"] - - def test_set_flag_updates_config(self): - """通过 set_flag 修改后立即生效。""" - store = InMemoryFeatureFlagStore() - resolver = self._make_resolver(store=store, default="legacy") - - # 初始:关闭 - assert resolver.get_engine("user1") == "legacy" - - # 开启 100% - resolver.set_flag(FeatureFlagConfig(name="render_engine", enabled=True, percentage=100)) - assert resolver.get_engine("user1") == "unified" - - # 关闭 - resolver.set_flag(FeatureFlagConfig(name="render_engine", enabled=False)) - assert resolver.get_engine("user1") == "legacy" - - def test_force_refresh(self): - """强制刷新不报错。""" - store = InMemoryFeatureFlagStore() - resolver = self._make_resolver(store=store) - resolver.force_refresh() # 不抛异常即可 - - def test_does_not_affect_in_flight_tasks(self): - """ - 热更新不影响在途任务验证: - 任务开始时确定引擎,中途配置变更不改变当前任务的引擎选择。 - (这是通过"每次调用 get_engine 时读取当前配置"来保证的, - 任务开始时调用一次拿到结果,之后不再变化) - """ - store = InMemoryFeatureFlagStore() - store.set(FeatureFlagConfig(name="render_engine", enabled=True, percentage=100)) - resolver = self._make_resolver(store=store, default="legacy") - - # 模拟任务开始时获取引擎 - engine_at_start = resolver.get_engine("user1") - assert engine_at_start == "unified" - - # 任务进行中关闭 flag - store.set(FeatureFlagConfig(name="render_engine", enabled=False)) - resolver.force_refresh() - - # 在途任务持有的 engine_at_start 仍然是 unified(不随配置变化) - assert engine_at_start == "unified" - # 新任务会拿到 legacy - assert resolver.get_engine("user1") == "legacy" - - # ── 辅助方法 ── - - @staticmethod - def _make_resolver(store=None, default="legacy"): - from apps.worker.video_processing.render_engine_resolver import ( - RenderEngineResolver, - ) - - return RenderEngineResolver( - default_engine=default, - store=store or InMemoryFeatureFlagStore(), - refresh_interval=9999, # 测试时禁用自动刷新 - ) - - # ── RedisFeatureFlagStore 降级测试(无 Redis 环境) ───────────────────────