Coverage for agentos/tests/test_feature_flags.py: 0%

1"""Tests for agentos.core.feature_flags — FeatureFlagManager, FlagRule, FlagContext.""" 

2 

3import pytest 

4from datetime import datetime, timedelta, UTC 

5from agentos.core.feature_flags import ( 

6 FeatureFlagManager, 

7 FlagContext, 

8 FlagEvaluation, 

9 FlagRule, 

10 FlagStore, 

11 FlagType, 

12 InMemoryFlagStore, 

13 create_flag_manager, 

14) 

15 

16 

17# ============================================================================ 

18# FlagType 

19# ============================================================================ 

20 

21class TestFlagType: 

22 def test_values(self): 

23 assert FlagType.BOOLEAN == "boolean" 

24 assert FlagType.PERCENTAGE == "percentage" 

25 assert FlagType.VARIANT == "variant" 

26 assert FlagType.SCHEDULED == "scheduled" 

27 

28 

29# ============================================================================ 

30# FlagRule — to_dict 

31# ============================================================================ 

32 

33class TestFlagRule: 

34 def test_defaults(self): 

35 rule = FlagRule() 

36 assert rule.flag_type == FlagType.BOOLEAN 

37 assert rule.enabled is False 

38 assert rule.rollout_percentage == 0 

39 assert rule.variants == {} 

40 assert rule.allowlist_users == set() 

41 assert rule.start_time is None 

42 

43 def test_to_dict_basic(self): 

44 rule = FlagRule(flag_type=FlagType.BOOLEAN, enabled=True) 

45 d = rule.to_dict() 

46 assert d["flag_type"] == "boolean" 

47 assert d["enabled"] is True 

48 assert d["rollout_percentage"] == 0 

49 

50 def test_to_dict_with_times(self): 

51 t1 = datetime(2025, 1, 1, tzinfo=UTC) 

52 t2 = datetime(2025, 12, 31, tzinfo=UTC) 

53 rule = FlagRule(start_time=t1, end_time=t2) 

54 d = rule.to_dict() 

55 assert "2025-01-01" in d["start_time"] 

56 assert "2025-12-31" in d["end_time"] 

57 

58 

59# ============================================================================ 

60# FlagContext 

61# ============================================================================ 

62 

63class TestFlagContext: 

64 def test_defaults(self): 

65 ctx = FlagContext() 

66 assert ctx.user_id is None 

67 assert ctx.tenant_id is None 

68 

69 def test_custom(self): 

70 ctx = FlagContext(user_id="u1", tenant_id="t1", attributes={"role": "admin"}) 

71 assert ctx.user_id == "u1" 

72 assert ctx.attributes == {"role": "admin"} 

73 

74 

75# ============================================================================ 

76# FlagEvaluation 

77# ============================================================================ 

78 

79class TestFlagEvaluation: 

80 def test_enabled(self): 

81 e = FlagEvaluation(flag_name="f1", enabled=True, reason="ok") 

82 assert e.enabled 

83 assert e.flag_name == "f1" 

84 

85 def test_disabled(self): 

86 e = FlagEvaluation(flag_name="f1", enabled=False, reason="nope") 

87 assert not e.enabled 

88 

89 

90# ============================================================================ 

91# InMemoryFlagStore 

92# ============================================================================ 

93 

94class TestInMemoryFlagStore: 

95 @pytest.mark.asyncio 

96 async def test_set_get(self): 

97 store = InMemoryFlagStore() 

98 rule = FlagRule(enabled=True) 

99 await store.set("f1", rule) 

100 assert await store.get("f1") is rule 

101 

102 @pytest.mark.asyncio 

103 async def test_get_missing(self): 

104 store = InMemoryFlagStore() 

105 assert await store.get("missing") is None 

106 

107 @pytest.mark.asyncio 

108 async def test_delete_existing(self): 

109 store = InMemoryFlagStore() 

110 await store.set("f1", FlagRule()) 

111 assert await store.delete("f1") is True 

112 assert await store.get("f1") is None 

113 

114 @pytest.mark.asyncio 

115 async def test_delete_missing(self): 

116 store = InMemoryFlagStore() 

117 assert await store.delete("missing") is False 

118 

119 @pytest.mark.asyncio 

120 async def test_list(self): 

121 store = InMemoryFlagStore() 

122 await store.set("a", FlagRule()) 

123 await store.set("b", FlagRule()) 

124 names = await store.list() 

125 assert sorted(names) == ["a", "b"] 

126 

127 

128# ============================================================================ 

129# FeatureFlagManager — Flag Management 

130# ============================================================================ 

131 

132class TestFeatureFlagManagerManagement: 

133 @pytest.mark.asyncio 

134 async def test_set_and_get_flag(self): 

135 mgr = FeatureFlagManager(InMemoryFlagStore()) 

136 rule = FlagRule(enabled=True) 

137 await mgr.set_flag("f1", rule) 

138 assert await mgr.get_flag("f1") is rule 

139 

140 @pytest.mark.asyncio 

141 async def test_delete_flag(self): 

142 mgr = FeatureFlagManager(InMemoryFlagStore()) 

143 await mgr.set_flag("f1", FlagRule()) 

144 assert await mgr.delete_flag("f1") is True 

145 assert await mgr.get_flag("f1") is None 

146 

147 @pytest.mark.asyncio 

148 async def test_list_flags(self): 

149 mgr = FeatureFlagManager(InMemoryFlagStore()) 

150 await mgr.set_flag("a", FlagRule()) 

151 await mgr.set_flag("b", FlagRule()) 

152 assert sorted(await mgr.list_flags()) == ["a", "b"] 

153 

154 @pytest.mark.asyncio 

155 async def test_kill_switch(self): 

156 mgr = FeatureFlagManager(InMemoryFlagStore()) 

157 await mgr.set_flag("f1", FlagRule(enabled=True)) 

158 await mgr.kill_switch("f1") 

159 rule = await mgr.get_flag("f1") 

160 assert rule.enabled is False 

161 

162 @pytest.mark.asyncio 

163 async def test_kill_switch_missing(self): 

164 mgr = FeatureFlagManager(InMemoryFlagStore()) 

165 await mgr.kill_switch("missing") # should not raise 

166 

167 

168# ============================================================================ 

169# FeatureFlagManager — is_enabled / evaluate 

170# ============================================================================ 

171 

172class TestFeatureFlagManagerEvaluation: 

173 @pytest.mark.asyncio 

174 async def test_flag_not_found(self): 

175 mgr = FeatureFlagManager(InMemoryFlagStore()) 

176 assert not await mgr.is_enabled("missing") 

177 

178 @pytest.mark.asyncio 

179 async def test_boolean_disabled(self): 

180 mgr = FeatureFlagManager(InMemoryFlagStore()) 

181 await mgr.set_flag("f1", FlagRule(flag_type=FlagType.BOOLEAN, enabled=False)) 

182 assert not await mgr.is_enabled("f1") 

183 

184 @pytest.mark.asyncio 

185 async def test_boolean_enabled(self): 

186 mgr = FeatureFlagManager(InMemoryFlagStore()) 

187 await mgr.set_flag("f1", FlagRule(flag_type=FlagType.BOOLEAN, enabled=True)) 

188 assert await mgr.is_enabled("f1") 

189 

190 @pytest.mark.asyncio 

191 async def test_blocklist_user(self): 

192 mgr = FeatureFlagManager(InMemoryFlagStore()) 

193 await mgr.set_flag("f1", FlagRule( 

194 flag_type=FlagType.BOOLEAN, enabled=True, 

195 blocklist_users={"u_bad"}, 

196 )) 

197 ctx = FlagContext(user_id="u_bad") 

198 assert not await mgr.is_enabled("f1", ctx) 

199 

200 @pytest.mark.asyncio 

201 async def test_blocklist_tenant(self): 

202 mgr = FeatureFlagManager(InMemoryFlagStore()) 

203 await mgr.set_flag("f1", FlagRule( 

204 flag_type=FlagType.BOOLEAN, enabled=True, 

205 blocklist_tenants={"t_bad"}, 

206 )) 

207 ctx = FlagContext(tenant_id="t_bad") 

208 assert not await mgr.is_enabled("f1", ctx) 

209 

210 @pytest.mark.asyncio 

211 async def test_allowlist_user_granted(self): 

212 mgr = FeatureFlagManager(InMemoryFlagStore()) 

213 await mgr.set_flag("f1", FlagRule( 

214 flag_type=FlagType.BOOLEAN, enabled=True, 

215 allowlist_users={"u_good"}, 

216 )) 

217 ctx = FlagContext(user_id="u_good") 

218 assert await mgr.is_enabled("f1", ctx) 

219 

220 @pytest.mark.asyncio 

221 async def test_allowlist_user_not_in(self): 

222 mgr = FeatureFlagManager(InMemoryFlagStore()) 

223 await mgr.set_flag("f1", FlagRule( 

224 flag_type=FlagType.BOOLEAN, enabled=True, 

225 allowlist_users={"u_good"}, 

226 )) 

227 ctx = FlagContext(user_id="u_other") 

228 assert not await mgr.is_enabled("f1", ctx) 

229 

230 @pytest.mark.asyncio 

231 async def test_allowlist_tenant_not_in(self): 

232 mgr = FeatureFlagManager(InMemoryFlagStore()) 

233 await mgr.set_flag("f1", FlagRule( 

234 flag_type=FlagType.BOOLEAN, enabled=True, 

235 allowlist_tenants={"t_good"}, 

236 )) 

237 ctx = FlagContext(tenant_id="t_other") 

238 assert not await mgr.is_enabled("f1", ctx) 

239 

240 @pytest.mark.asyncio 

241 async def test_start_time_future(self): 

242 mgr = FeatureFlagManager(InMemoryFlagStore()) 

243 future = datetime.now(UTC) + timedelta(days=30) 

244 await mgr.set_flag("f1", FlagRule( 

245 flag_type=FlagType.SCHEDULED, enabled=True, 

246 start_time=future, 

247 )) 

248 assert not await mgr.is_enabled("f1") 

249 

250 @pytest.mark.asyncio 

251 async def test_end_time_past(self): 

252 mgr = FeatureFlagManager(InMemoryFlagStore()) 

253 past = datetime.now(UTC) - timedelta(days=30) 

254 await mgr.set_flag("f1", FlagRule( 

255 flag_type=FlagType.SCHEDULED, enabled=True, 

256 end_time=past, 

257 )) 

258 assert not await mgr.is_enabled("f1") 

259 

260 @pytest.mark.asyncio 

261 async def test_scheduled_within_window(self): 

262 mgr = FeatureFlagManager(InMemoryFlagStore()) 

263 past = datetime.now(UTC) - timedelta(days=1) 

264 future = datetime.now(UTC) + timedelta(days=1) 

265 await mgr.set_flag("f1", FlagRule( 

266 flag_type=FlagType.SCHEDULED, enabled=True, 

267 start_time=past, end_time=future, 

268 )) 

269 assert await mgr.is_enabled("f1") 

270 

271 

272# ============================================================================ 

273# FeatureFlagManager — Percentage rollout 

274# ============================================================================ 

275 

276class TestFeatureFlagManagerPercentage: 

277 @pytest.mark.asyncio 

278 async def test_percentage_zero(self): 

279 mgr = FeatureFlagManager(InMemoryFlagStore()) 

280 await mgr.set_flag("f1", FlagRule( 

281 flag_type=FlagType.PERCENTAGE, enabled=True, 

282 rollout_percentage=0, 

283 )) 

284 ctx = FlagContext(user_id="u1") 

285 assert not await mgr.is_enabled("f1", ctx) 

286 

287 @pytest.mark.asyncio 

288 async def test_percentage_hundred(self): 

289 mgr = FeatureFlagManager(InMemoryFlagStore()) 

290 await mgr.set_flag("f1", FlagRule( 

291 flag_type=FlagType.PERCENTAGE, enabled=True, 

292 rollout_percentage=100, 

293 )) 

294 ctx = FlagContext(user_id="u1") 

295 assert await mgr.is_enabled("f1", ctx) 

296 

297 @pytest.mark.asyncio 

298 async def test_percentage_deterministic(self): 

299 """Same user should always get same result.""" 

300 mgr = FeatureFlagManager(InMemoryFlagStore()) 

301 await mgr.set_flag("f1", FlagRule( 

302 flag_type=FlagType.PERCENTAGE, enabled=True, 

303 rollout_percentage=50, 

304 )) 

305 ctx = FlagContext(user_id="u1") 

306 r1 = await mgr.is_enabled("f1", ctx) 

307 r2 = await mgr.is_enabled("f1", ctx) 

308 assert r1 == r2 

309 

310 

311# ============================================================================ 

312# FeatureFlagManager — Variant (A/B) 

313# ============================================================================ 

314 

315class TestFeatureFlagManagerVariant: 

316 @pytest.mark.asyncio 

317 async def test_get_variant(self): 

318 mgr = FeatureFlagManager(InMemoryFlagStore()) 

319 await mgr.set_flag("ab_test", FlagRule( 

320 flag_type=FlagType.VARIANT, enabled=True, 

321 variants={"control": 50, "treatment": 50}, 

322 )) 

323 ctx = FlagContext(user_id="u1") 

324 variant = await mgr.get_variant("ab_test", ctx) 

325 assert variant in ("control", "treatment") 

326 

327 @pytest.mark.asyncio 

328 async def test_variant_deterministic(self): 

329 mgr = FeatureFlagManager(InMemoryFlagStore()) 

330 await mgr.set_flag("ab_test", FlagRule( 

331 flag_type=FlagType.VARIANT, enabled=True, 

332 variants={"a": 30, "b": 70}, 

333 )) 

334 ctx = FlagContext(user_id="u1") 

335 v1 = await mgr.get_variant("ab_test", ctx) 

336 v2 = await mgr.get_variant("ab_test", ctx) 

337 assert v1 == v2 

338 

339 @pytest.mark.asyncio 

340 async def test_variant_no_variants(self): 

341 mgr = FeatureFlagManager(InMemoryFlagStore()) 

342 await mgr.set_flag("f1", FlagRule( 

343 flag_type=FlagType.VARIANT, enabled=True, 

344 )) 

345 ctx = FlagContext(user_id="u1") 

346 assert await mgr.get_variant("f1", ctx) is None 

347 

348 

349# ============================================================================ 

350# FeatureFlagManager — Audit log 

351# ============================================================================ 

352 

353class TestFeatureFlagManagerAudit: 

354 @pytest.mark.asyncio 

355 async def test_evaluation_log(self): 

356 mgr = FeatureFlagManager(InMemoryFlagStore()) 

357 await mgr.set_flag("f1", FlagRule(flag_type=FlagType.BOOLEAN, enabled=True)) 

358 await mgr.is_enabled("f1") 

359 log = mgr.get_evaluation_log() 

360 assert len(log) == 1 

361 assert log[0].enabled 

362 

363 @pytest.mark.asyncio 

364 async def test_log_limit(self): 

365 mgr = FeatureFlagManager(InMemoryFlagStore()) 

366 await mgr.set_flag("f1", FlagRule(flag_type=FlagType.BOOLEAN, enabled=True)) 

367 for _ in range(5): 

368 await mgr.is_enabled("f1") 

369 assert len(mgr.get_evaluation_log(limit=3)) == 3 

370 

371 @pytest.mark.asyncio 

372 async def test_clear_log(self): 

373 mgr = FeatureFlagManager(InMemoryFlagStore()) 

374 await mgr.set_flag("f1", FlagRule(enabled=True)) 

375 await mgr.is_enabled("f1") 

376 mgr.clear_evaluation_log() 

377 assert len(mgr.get_evaluation_log()) == 0 

378 

379 

380# ============================================================================ 

381# FeatureFlagManager — Edge cases 

382# ============================================================================ 

383 

384class TestFeatureFlagManagerEdgeCases: 

385 @pytest.mark.asyncio 

386 async def test_null_context(self): 

387 mgr = FeatureFlagManager(InMemoryFlagStore()) 

388 await mgr.set_flag("f1", FlagRule(flag_type=FlagType.BOOLEAN, enabled=True)) 

389 assert await mgr.is_enabled("f1") # no context passed 

390 

391 @pytest.mark.asyncio 

392 async def test_evaluate_returns_flag_evaluation(self): 

393 mgr = FeatureFlagManager(InMemoryFlagStore()) 

394 await mgr.set_flag("f1", FlagRule(flag_type=FlagType.BOOLEAN, enabled=True)) 

395 result = await mgr.evaluate("f1") 

396 assert isinstance(result, FlagEvaluation) 

397 assert result.enabled 

398 

399 @pytest.mark.asyncio 

400 async def test_allowlist_exclusive_behavior(self): 

401 """Non-empty allowlist + user/tenant NOT in it → disabled.""" 

402 mgr = FeatureFlagManager(InMemoryFlagStore()) 

403 await mgr.set_flag("f1", FlagRule( 

404 flag_type=FlagType.BOOLEAN, enabled=True, 

405 allowlist_users={"specific"}, 

406 allowlist_tenants={"t1"}, 

407 )) 

408 # No user/tenant context 

409 ctx = FlagContext() 

410 assert not await mgr.is_enabled("f1", ctx) 

411 

412 

413# ============================================================================ 

414# create_flag_manager convenience 

415# ============================================================================ 

416 

417class TestCreateFlagManager: 

418 @pytest.mark.asyncio 

419 async def test_creates_functional_manager(self): 

420 mgr = create_flag_manager() 

421 await mgr.set_flag("test", FlagRule(enabled=True)) 

422 assert await mgr.is_enabled("test")