refactor: 移除TestRenderEngineResolver测试类
This commit is contained in:
CI Bot
@@ -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 环境） ───────────────────────
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user