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

1"""Tests for agentos.core.secrets — SecretsManager, backends, caching, fail-open.""" 

2 

3import json 

4import os 

5import tempfile 

6 

7import pytest 

8 

9from agentos.core.secrets import ( 

10 BackendUnavailableError, 

11 CompositeSecretsBackend, 

12 EncryptedFileBackend, 

13 EnvSecretsBackend, 

14 SecretNotFoundError, 

15 SecretsConfig, 

16 SecretsManager, 

17 VaultSecretsBackend, 

18 create_secrets_manager, 

19) 

20 

21# ============================================================================ 

22# SecretsConfig 

23# ============================================================================ 

24 

25 

26class TestSecretsConfig: 

27 def test_defaults(self): 

28 cfg = SecretsConfig() 

29 assert cfg.cache_ttl == 300.0 

30 assert cfg.max_cache_size == 1000 

31 assert cfg.fail_open is False 

32 assert cfg.allow_environment_fallback is True 

33 

34 def test_custom(self): 

35 cfg = SecretsConfig(cache_ttl=60, max_cache_size=100, fail_open=True) 

36 assert cfg.cache_ttl == 60 

37 assert cfg.fail_open is True 

38 

39 

40# ============================================================================ 

41# EnvSecretsBackend 

42# ============================================================================ 

43 

44 

45class TestEnvSecretsBackend: 

46 @pytest.mark.asyncio 

47 async def test_get(self): 

48 os.environ["TEST_SECRET_X"] = "hello" 

49 be = EnvSecretsBackend(prefix="TEST_SECRET_") 

50 assert await be.get("X") == "hello" 

51 del os.environ["TEST_SECRET_X"] 

52 

53 @pytest.mark.asyncio 

54 async def test_get_missing(self): 

55 be = EnvSecretsBackend() 

56 assert await be.get("NONEXISTENT") is None 

57 

58 @pytest.mark.asyncio 

59 async def test_get_all(self): 

60 os.environ["TS_A"] = "1" 

61 os.environ["TS_B"] = "2" 

62 os.environ["OTHER"] = "3" 

63 be = EnvSecretsBackend(prefix="TS_") 

64 result = await be.get_all("A") 

65 assert result == {"A": "1"} 

66 del os.environ["TS_A"] 

67 del os.environ["TS_B"] 

68 del os.environ["OTHER"] 

69 

70 @pytest.mark.asyncio 

71 async def test_get_all_no_prefix(self): 

72 os.environ["TS_X"] = "v" 

73 be = EnvSecretsBackend(prefix="TS_") 

74 result = await be.get_all() 

75 assert "X" in result 

76 assert result["X"] == "v" 

77 del os.environ["TS_X"] 

78 

79 @pytest.mark.asyncio 

80 async def test_health_check(self): 

81 be = EnvSecretsBackend() 

82 assert await be.health_check() is True 

83 

84 

85# ============================================================================ 

86# EncryptedFileBackend 

87# ============================================================================ 

88 

89 

90class TestEncryptedFileBackend: 

91 def _make_encrypted_file(self, data: dict) -> str: 

92 from cryptography.fernet import Fernet 

93 

94 key = Fernet.generate_key() 

95 fernet = Fernet(key) 

96 encrypted = fernet.encrypt(json.dumps(data).encode()) 

97 

98 with tempfile.NamedTemporaryFile(delete=False, suffix=".enc") as f: 

99 f.write(encrypted) 

100 self._cleanup_files = getattr(self, "_cleanup_files", []) 

101 self._cleanup_files.append(f.name) 

102 

103 self._test_key = key.decode() 

104 return f.name 

105 

106 def teardown_method(self): 

107 for fp in getattr(self, "_cleanup_files", []): 

108 try: 

109 os.unlink(fp) 

110 except OSError: 

111 pass 

112 

113 @pytest.mark.asyncio 

114 async def test_get(self): 

115 path = self._make_encrypted_file({"API_KEY": "abc123"}) 

116 be = EncryptedFileBackend(file_path=path, encryption_key=self._test_key) 

117 assert await be.get("API_KEY") == "abc123" 

118 

119 @pytest.mark.asyncio 

120 async def test_get_missing(self): 

121 path = self._make_encrypted_file({"API_KEY": "abc123"}) 

122 be = EncryptedFileBackend(file_path=path, encryption_key=self._test_key) 

123 assert await be.get("MISSING") is None 

124 

125 @pytest.mark.asyncio 

126 async def test_get_all(self): 

127 path = self._make_encrypted_file({"A": "1", "B": "2"}) 

128 be = EncryptedFileBackend(file_path=path, encryption_key=self._test_key) 

129 result = await be.get_all() 

130 assert result == {"A": "1", "B": "2"} 

131 

132 @pytest.mark.asyncio 

133 async def test_get_all_prefix(self): 

134 path = self._make_encrypted_file({"DB_HOST": "x", "DB_PORT": "y", "API_KEY": "z"}) 

135 be = EncryptedFileBackend(file_path=path, encryption_key=self._test_key) 

136 result = await be.get_all(prefix="DB_") 

137 assert result == {"DB_HOST": "x", "DB_PORT": "y"} 

138 

139 @pytest.mark.asyncio 

140 async def test_file_not_found(self): 

141 be = EncryptedFileBackend(file_path="/nonexistent/path.enc", encryption_key="dummy") 

142 assert await be.get("X") is None 

143 

144 @pytest.mark.asyncio 

145 async def test_health_check_ok(self): 

146 path = self._make_encrypted_file({"X": "1"}) 

147 be = EncryptedFileBackend(file_path=path, encryption_key=self._test_key) 

148 assert await be.health_check() is True 

149 

150 @pytest.mark.asyncio 

151 async def test_health_check_bad_key(self): 

152 path = self._make_encrypted_file({"X": "1"}) 

153 from cryptography.fernet import Fernet 

154 bad_key = Fernet.generate_key().decode() 

155 be = EncryptedFileBackend(file_path=path, encryption_key=bad_key) 

156 assert await be.health_check() is False 

157 

158 

159# ============================================================================ 

160# CompositeSecretsBackend 

161# ============================================================================ 

162 

163 

164class TestCompositeSecretsBackend: 

165 @pytest.mark.asyncio 

166 async def test_priority_order(self): 

167 be1 = EnvSecretsBackend() 

168 os.environ["KEY"] = "from_env" 

169 be2 = EnvSecretsBackend() 

170 os.environ["KEY2"] = "also_env" 

171 

172 comp = CompositeSecretsBackend([be1, be2]) 

173 # env backends both read from same os.environ, so this is degenerate 

174 # but verifies first hit wins 

175 assert await comp.get("NONEXISTENT") is None 

176 del os.environ["KEY"] 

177 del os.environ["KEY2"] 

178 

179 @pytest.mark.asyncio 

180 async def test_get_all_merged(self): 

181 # Create a mock backend for deterministic testing 

182 class MockBackend(EnvSecretsBackend): 

183 def __init__(self, data: dict): 

184 self._data = data 

185 

186 async def get(self, key: str) -> str | None: 

187 return self._data.get(key) 

188 

189 async def get_all(self, prefix: str = "") -> dict[str, str]: 

190 if prefix: 

191 return {k: v for k, v in self._data.items() if k.startswith(prefix)} 

192 return dict(self._data) 

193 

194 be1 = MockBackend({"A": "1"}) 

195 be2 = MockBackend({"B": "2"}) 

196 comp = CompositeSecretsBackend([be1, be2]) 

197 result = await comp.get_all() 

198 assert "A" in result 

199 assert "B" in result 

200 

201 @pytest.mark.asyncio 

202 async def test_health_check_any(self): 

203 class DeadBackend: 

204 async def health_check(self): 

205 return False 

206 

207 class AliveBackend: 

208 async def health_check(self): 

209 return True 

210 

211 comp = CompositeSecretsBackend([DeadBackend(), AliveBackend()]) 

212 assert await comp.health_check() is True 

213 

214 @pytest.mark.asyncio 

215 async def test_health_check_all_dead(self): 

216 class DeadBackend: 

217 async def health_check(self): 

218 return False 

219 

220 comp = CompositeSecretsBackend([DeadBackend(), DeadBackend()]) 

221 assert await comp.health_check() is False 

222 

223 

224# ============================================================================ 

225# SecretsManager 

226# ============================================================================ 

227 

228 

229class TestSecretsManager: 

230 @pytest.mark.asyncio 

231 async def test_get_from_backend(self): 

232 class MockBackend: 

233 async def get(self, key): 

234 return "val" if key == "KEY" else None 

235 

236 async def get_all(self, prefix=""): 

237 return {"KEY": "val"} 

238 

239 async def health_check(self): 

240 return True 

241 

242 sm = SecretsManager(MockBackend()) 

243 assert await sm.get("KEY") == "val" 

244 assert await sm.get("MISSING") is None 

245 

246 @pytest.mark.asyncio 

247 async def test_require(self): 

248 class MockBackend: 

249 async def get(self, key): 

250 return "val" if key == "KEY" else None 

251 

252 async def get_all(self, prefix=""): 

253 return {} 

254 

255 async def health_check(self): 

256 return True 

257 

258 sm = SecretsManager(MockBackend()) 

259 assert await sm.require("KEY") == "val" 

260 

261 @pytest.mark.asyncio 

262 async def test_require_raises(self): 

263 class MockBackend: 

264 async def get(self, key): 

265 return None 

266 

267 async def get_all(self, prefix=""): 

268 return {} 

269 

270 async def health_check(self): 

271 return True 

272 

273 sm = SecretsManager(MockBackend()) 

274 with pytest.raises(SecretNotFoundError): 

275 await sm.require("MISSING") 

276 

277 @pytest.mark.asyncio 

278 async def test_cache_reuse(self): 

279 call_count = 0 

280 

281 class CountingBackend: 

282 async def get(self, key): 

283 nonlocal call_count 

284 call_count += 1 

285 return "cached" 

286 

287 async def get_all(self, prefix=""): 

288 return {} 

289 

290 async def health_check(self): 

291 return True 

292 

293 sm = SecretsManager(CountingBackend(), SecretsConfig(cache_ttl=300)) 

294 await sm.get("KEY") 

295 await sm.get("KEY") 

296 assert call_count == 1 

297 

298 @pytest.mark.asyncio 

299 async def test_env_fallback(self): 

300 os.environ["FALLBACK_KEY"] = "from_env" 

301 

302 class EmptyBackend: 

303 async def get(self, key): 

304 return None 

305 

306 async def get_all(self, prefix=""): 

307 return {} 

308 

309 async def health_check(self): 

310 return True 

311 

312 sm = SecretsManager(EmptyBackend(), SecretsConfig(allow_environment_fallback=True)) 

313 assert await sm.get("FALLBACK_KEY") == "from_env" 

314 del os.environ["FALLBACK_KEY"] 

315 

316 @pytest.mark.asyncio 

317 async def test_no_env_fallback(self): 

318 os.environ["FALLBACK_KEY"] = "from_env" 

319 

320 class EmptyBackend: 

321 async def get(self, key): 

322 return None 

323 

324 async def get_all(self, prefix=""): 

325 return {} 

326 

327 async def health_check(self): 

328 return True 

329 

330 sm = SecretsManager(EmptyBackend(), SecretsConfig(allow_environment_fallback=False)) 

331 assert await sm.get("FALLBACK_KEY") is None 

332 del os.environ["FALLBACK_KEY"] 

333 

334 @pytest.mark.asyncio 

335 async def test_fail_open(self): 

336 class FailingBackend: 

337 async def get(self, key): 

338 raise BackendUnavailableError("down") 

339 

340 async def get_all(self, prefix=""): 

341 raise BackendUnavailableError("down") 

342 

343 async def health_check(self): 

344 return False 

345 

346 sm = SecretsManager(FailingBackend(), SecretsConfig(fail_open=True)) 

347 assert await sm.get("KEY") is None 

348 

349 @pytest.mark.asyncio 

350 async def test_fail_closed(self): 

351 class FailingBackend: 

352 async def get(self, key): 

353 raise BackendUnavailableError("down") 

354 

355 async def get_all(self, prefix=""): 

356 raise BackendUnavailableError("down") 

357 

358 async def health_check(self): 

359 return False 

360 

361 sm = SecretsManager(FailingBackend(), SecretsConfig(fail_open=False)) 

362 with pytest.raises(BackendUnavailableError): 

363 await sm.get("KEY") 

364 

365 @pytest.mark.asyncio 

366 async def test_get_all(self): 

367 class MockBackend: 

368 async def get(self, key): 

369 return "x" 

370 

371 async def get_all(self, prefix=""): 

372 return {"A": "1", "B": "2"} 

373 

374 async def health_check(self): 

375 return True 

376 

377 sm = SecretsManager(MockBackend()) 

378 assert await sm.get_all() == {"A": "1", "B": "2"} 

379 

380 @pytest.mark.asyncio 

381 async def test_health_check(self): 

382 class MockBackend: 

383 async def get(self, key): 

384 return None 

385 

386 async def get_all(self, prefix=""): 

387 return {} 

388 

389 async def health_check(self): 

390 return True 

391 

392 sm = SecretsManager(MockBackend()) 

393 assert await sm.health_check() is True 

394 

395 @pytest.mark.asyncio 

396 async def test_invalidate_single_key(self): 

397 call_count = 0 

398 

399 class CountingBackend: 

400 async def get(self, key): 

401 nonlocal call_count 

402 call_count += 1 

403 return "val" 

404 

405 async def get_all(self, prefix=""): 

406 return {} 

407 

408 async def health_check(self): 

409 return True 

410 

411 sm = SecretsManager(CountingBackend()) 

412 await sm.get("K1") 

413 await sm.get("K1") 

414 assert call_count == 1 

415 sm.invalidate_cache("K1") 

416 await sm.get("K1") 

417 assert call_count == 2 

418 

419 @pytest.mark.asyncio 

420 async def test_invalidate_all(self): 

421 call_count = 0 

422 

423 class CountingBackend: 

424 async def get(self, key): 

425 nonlocal call_count 

426 call_count += 1 

427 return "val" 

428 

429 async def get_all(self, prefix=""): 

430 return {} 

431 

432 async def health_check(self): 

433 return True 

434 

435 sm = SecretsManager(CountingBackend()) 

436 await sm.get("K1") 

437 await sm.get("K2") 

438 assert call_count == 2 

439 sm.invalidate_cache() 

440 await sm.get("K1") 

441 await sm.get("K2") 

442 assert call_count == 4 

443 

444 

445# ============================================================================ 

446# create_secrets_manager factory 

447# ============================================================================ 

448 

449 

450class TestCreateSecretsManager: 

451 def test_env_backend(self): 

452 sm = create_secrets_manager("env", prefix="P_") 

453 assert isinstance(sm, SecretsManager) 

454 

455 def test_unknown_backend(self): 

456 with pytest.raises(ValueError, match="Unknown backend"): 

457 create_secrets_manager("magic") 

458 

459 

460# ============================================================================ 

461# VaultSecretsBackend (mocked) 

462# ============================================================================ 

463 

464 

465class TestVaultSecretsBackend: 

466 @pytest.mark.asyncio 

467 async def test_health_check_false_when_unreachable(self): 

468 vb = VaultSecretsBackend(url="http://127.0.0.1:19999", token="fake") 

469 assert await vb.health_check() is False