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

1"""Tests for agentos.core.lifecycle — LifecycleManager, hooks, probes, reports.""" 

2 

3import asyncio 

4 

5import pytest 

6 

7from agentos.core.lifecycle import ( 

8 ComponentHealth, 

9 ComponentStatus, 

10 LifecycleHook, 

11 LifecycleManager, 

12 LifecyclePhase, 

13 LifecycleReport, 

14 get_lifecycle, 

15) 

16 

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

18# LifecycleHook 

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

20 

21 

22class TestLifecycleHook: 

23 def test_defaults(self): 

24 hook = LifecycleHook(name="test", phase=LifecyclePhase.SERVICES, fn=lambda: None) 

25 assert hook.name == "test" 

26 assert hook.phase == LifecyclePhase.SERVICES 

27 assert hook.timeout_seconds == 30.0 

28 assert hook.critical is True 

29 assert hook.weight == 50 

30 assert hook.retries == 0 

31 

32 def test_custom(self): 

33 hook = LifecycleHook( 

34 name="test", phase=LifecyclePhase.API, fn=lambda: None, 

35 critical=False, weight=10, retries=3, 

36 ) 

37 assert hook.critical is False 

38 assert hook.weight == 10 

39 assert hook.retries == 3 

40 

41 

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

43# ComponentHealth 

44# ============================================================================ 

45 

46 

47class TestComponentHealth: 

48 def test_defaults(self): 

49 ch = ComponentHealth(name="db") 

50 assert ch.name == "db" 

51 assert ch.status == ComponentStatus.UNINITIALIZED 

52 assert ch.phase is None 

53 assert ch.message == "" 

54 

55 def test_custom(self): 

56 ch = ComponentHealth( 

57 name="db", status=ComponentStatus.HEALTHY, 

58 phase=LifecyclePhase.INFRA, message="ok", 

59 ) 

60 assert ch.status == ComponentStatus.HEALTHY 

61 

62 

63# ============================================================================ 

64# LifecycleReport 

65# ============================================================================ 

66 

67 

68class TestLifecycleReport: 

69 def test_is_healthy(self): 

70 r = LifecycleReport(overall_status=ComponentStatus.HEALTHY) 

71 assert r.is_healthy is True 

72 

73 def test_not_healthy(self): 

74 r = LifecycleReport(overall_status=ComponentStatus.UNHEALTHY) 

75 assert r.is_healthy is False 

76 

77 def test_is_ready_healthy(self): 

78 r = LifecycleReport(overall_status=ComponentStatus.HEALTHY) 

79 assert r.is_ready is True 

80 

81 def test_is_ready_degraded(self): 

82 r = LifecycleReport(overall_status=ComponentStatus.DEGRADED) 

83 assert r.is_ready is True 

84 

85 def test_not_ready(self): 

86 r = LifecycleReport(overall_status=ComponentStatus.UNINITIALIZED) 

87 assert r.is_ready is False 

88 

89 

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

91# LifecycleManager 

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

93 

94 

95class TestLifecycleManagerCore: 

96 @pytest.mark.asyncio 

97 async def test_single_startup_hook(self): 

98 lm = LifecycleManager() 

99 ran = [] 

100 

101 @lm.on_startup(name="s1") 

102 async def startup(): 

103 ran.append(1) 

104 

105 report = await lm.start() 

106 assert ran == [1] 

107 assert report.overall_status == ComponentStatus.HEALTHY 

108 

109 @pytest.mark.asyncio 

110 async def test_sync_startup_hook(self): 

111 lm = LifecycleManager() 

112 ran = [] 

113 

114 @lm.on_startup(name="s1") 

115 def startup(): 

116 ran.append(1) 

117 

118 report = await lm.start() 

119 assert ran == [1] 

120 assert report.overall_status == ComponentStatus.HEALTHY 

121 

122 @pytest.mark.asyncio 

123 async def test_phase_ordering(self): 

124 lm = LifecycleManager() 

125 order = [] 

126 

127 @lm.on_startup(phase=LifecyclePhase.API) 

128 async def api(): 

129 order.append("api") 

130 

131 @lm.on_startup(phase=LifecyclePhase.CONFIG) 

132 async def cfg(): 

133 order.append("config") 

134 

135 @lm.on_startup(phase=LifecyclePhase.SECURITY) 

136 async def sec(): 

137 order.append("security") 

138 

139 await lm.start() 

140 assert order == ["config", "security", "api"] 

141 

142 @pytest.mark.asyncio 

143 async def test_weight_ordering_same_phase(self): 

144 lm = LifecycleManager() 

145 order = [] 

146 

147 @lm.on_startup(phase=LifecyclePhase.SERVICES, weight=90) 

148 async def s3(): 

149 order.append("s3") 

150 

151 @lm.on_startup(phase=LifecyclePhase.SERVICES, weight=10) 

152 async def s1(): 

153 order.append("s1") 

154 

155 @lm.on_startup(phase=LifecyclePhase.SERVICES, weight=50) 

156 async def s2(): 

157 order.append("s2") 

158 

159 await lm.start() 

160 assert order == ["s1", "s2", "s3"] 

161 

162 @pytest.mark.asyncio 

163 async def test_critical_hook_failure(self): 

164 lm = LifecycleManager() 

165 

166 @lm.on_startup(name="bad", critical=True) 

167 async def bad(): 

168 raise RuntimeError("boom") 

169 

170 @lm.on_startup(name="after", phase=LifecyclePhase.API) 

171 async def after(): 

172 pass 

173 

174 report = await lm.start() 

175 assert report.overall_status == ComponentStatus.UNHEALTHY 

176 assert report.components["bad"].status == ComponentStatus.UNHEALTHY 

177 

178 @pytest.mark.asyncio 

179 async def test_noncritical_hook_failure(self): 

180 lm = LifecycleManager() 

181 ran = [] 

182 

183 @lm.on_startup(name="bad", critical=False) 

184 async def bad(): 

185 raise RuntimeError("boom") 

186 

187 @lm.on_startup(name="after", phase=LifecyclePhase.API) 

188 async def after(): 

189 ran.append("after") 

190 

191 report = await lm.start() 

192 assert ran == ["after"] 

193 assert report.overall_status == ComponentStatus.HEALTHY 

194 

195 @pytest.mark.asyncio 

196 async def test_hook_timeout(self): 

197 lm = LifecycleManager() 

198 

199 @lm.on_startup(name="slow", timeout_seconds=0.01) 

200 async def slow(): 

201 await asyncio.sleep(99) 

202 

203 report = await lm.start() 

204 assert report.overall_status == ComponentStatus.UNHEALTHY 

205 

206 @pytest.mark.asyncio 

207 async def test_hook_retries_succeed(self): 

208 lm = LifecycleManager() 

209 attempts = [] 

210 

211 @lm.on_startup(name="retry", retries=2, retry_delay=0.01) 

212 async def retry(): 

213 attempts.append(1) 

214 if len(attempts) < 3: 

215 raise RuntimeError("fail") 

216 

217 report = await lm.start() 

218 assert len(attempts) == 3 

219 assert report.overall_status == ComponentStatus.HEALTHY 

220 

221 @pytest.mark.asyncio 

222 async def test_hook_retries_exhausted(self): 

223 lm = LifecycleManager() 

224 

225 @lm.on_startup(name="retry", retries=1, retry_delay=0.01, critical=True) 

226 async def retry(): 

227 raise RuntimeError("always fail") 

228 

229 report = await lm.start() 

230 assert report.overall_status == ComponentStatus.UNHEALTHY 

231 

232 

233class TestLifecycleManagerShutdown: 

234 @pytest.mark.asyncio 

235 async def test_shutdown_reverse_order(self): 

236 lm = LifecycleManager() 

237 order = [] 

238 

239 @lm.on_shutdown(name="a") 

240 async def a(): 

241 order.append("a") 

242 

243 @lm.on_shutdown(name="b") 

244 async def b(): 

245 order.append("b") 

246 

247 @lm.on_shutdown(name="c") 

248 async def c(): 

249 order.append("c") 

250 

251 report = await lm.shutdown() 

252 assert order == ["c", "b", "a"] 

253 assert report.overall_status == ComponentStatus.STOPPED 

254 

255 @pytest.mark.asyncio 

256 async def test_shutdown_sync_hook(self): 

257 lm = LifecycleManager() 

258 ran = [] 

259 

260 @lm.on_shutdown(name="sync") 

261 def sync(): 

262 ran.append(1) 

263 

264 await lm.shutdown() 

265 assert ran == [1] 

266 

267 @pytest.mark.asyncio 

268 async def test_shutdown_idempotent(self): 

269 lm = LifecycleManager() 

270 count = 0 

271 

272 @lm.on_shutdown(name="x") 

273 async def x(): 

274 nonlocal count 

275 count += 1 

276 

277 await lm.shutdown() 

278 await lm.shutdown() 

279 assert count == 1 

280 

281 @pytest.mark.asyncio 

282 async def test_shutdown_after_start_stops(self): 

283 lm = LifecycleManager() 

284 

285 @lm.on_shutdown(name="close") 

286 async def close(): 

287 pass 

288 

289 await lm.start() 

290 report = await lm.shutdown() 

291 assert report.overall_status == ComponentStatus.STOPPED 

292 assert lm.is_live() is False 

293 

294 

295class TestLifecycleManagerProbes: 

296 def test_initial_probes(self): 

297 lm = LifecycleManager() 

298 assert lm.is_ready() is False 

299 assert lm.is_live() is True 

300 

301 @pytest.mark.asyncio 

302 async def test_ready_after_start(self): 

303 lm = LifecycleManager() 

304 await lm.start() 

305 assert lm.is_ready() is True 

306 assert lm.is_live() is True 

307 

308 @pytest.mark.asyncio 

309 async def test_not_ready_before_start(self): 

310 lm = LifecycleManager() 

311 

312 @lm.on_startup(name="s1") 

313 async def s1(): 

314 pass 

315 

316 assert lm.is_ready() is False 

317 

318 @pytest.mark.asyncio 

319 async def test_live_after_startup_failure(self): 

320 lm = LifecycleManager() 

321 

322 @lm.on_startup(name="bad", critical=True) 

323 async def bad(): 

324 raise RuntimeError("fail") 

325 

326 await lm.start() 

327 assert lm.is_ready() is False 

328 assert lm.is_live() is False 

329 

330 

331class TestLifecycleManagerContext: 

332 @pytest.mark.asyncio 

333 async def test_context_manager(self): 

334 ran_start = False 

335 ran_stop = False 

336 

337 lm = LifecycleManager() 

338 

339 @lm.on_startup(name="s") 

340 async def s(): 

341 nonlocal ran_start 

342 ran_start = True 

343 

344 @lm.on_shutdown(name="close") 

345 async def close(): 

346 nonlocal ran_stop 

347 ran_stop = True 

348 

349 async with lm: 

350 assert ran_start is True 

351 assert ran_stop is False 

352 

353 assert ran_stop is True 

354 

355 

356class TestLifecycleManagerReport: 

357 @pytest.mark.asyncio 

358 async def test_report_after_start(self): 

359 lm = LifecycleManager() 

360 

361 @lm.on_startup(name="db") 

362 async def db(): 

363 pass 

364 

365 report = await lm.start() 

366 assert report.overall_status == ComponentStatus.HEALTHY 

367 assert "db" in report.components 

368 assert report.components["db"].status == ComponentStatus.HEALTHY 

369 assert report.startup_duration_ms >= 0 

370 

371 @pytest.mark.asyncio 

372 async def test_report_after_shutdown(self): 

373 lm = LifecycleManager() 

374 report = await lm.shutdown() 

375 assert report.overall_status == ComponentStatus.STOPPED 

376 assert report.components == {} 

377 

378 

379# ============================================================================ 

380# get_lifecycle singleton 

381# ============================================================================ 

382 

383 

384class TestGetLifecycle: 

385 def test_singleton(self): 

386 lm1 = get_lifecycle() 

387 lm2 = get_lifecycle() 

388 assert lm1 is lm2 

389 

390 def test_initial_status(self): 

391 import agentos.core.lifecycle as lc 

392 lc._default_lifecycle = None 

393 lm = get_lifecycle() 

394 assert lm.is_ready() is False 

395 assert lm.is_live() is True