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

1"""Tests for agentos.tools.retry_queue — RetryQueue, RetryJob, BackoffStrategy.""" 

2 

3import pytest 

4import time 

5from agentos.tools.retry_queue import BackoffStrategy, RetryJob, RetryQueue 

6 

7 

8# ============================================================================ 

9# BackoffStrategy 

10# ============================================================================ 

11 

12class TestBackoffStrategy: 

13 def test_enum_values(self): 

14 assert BackoffStrategy.EXPONENTIAL.value == "exponential" 

15 assert BackoffStrategy.CONSTANT.value == "constant" 

16 assert BackoffStrategy.LINEAR.value == "linear" 

17 

18 def test_enum_membership(self): 

19 assert BackoffStrategy("exponential") == BackoffStrategy.EXPONENTIAL 

20 assert BackoffStrategy("constant") == BackoffStrategy.CONSTANT 

21 assert BackoffStrategy("linear") == BackoffStrategy.LINEAR 

22 

23 

24# ============================================================================ 

25# RetryJob 

26# ============================================================================ 

27 

28class TestRetryJob: 

29 def test_creation_defaults(self): 

30 job = RetryJob(id="j1", func=lambda: 42) 

31 assert job.id == "j1" 

32 assert job.args == () 

33 assert job.kwargs == {} 

34 assert job.attempts == 0 

35 assert job.last_error is None 

36 assert isinstance(job.created_at, float) 

37 

38 def test_creation_with_args(self): 

39 def f(a, b, c=3): 

40 return a + b + c 

41 

42 job = RetryJob(id="j2", func=f, args=(1, 2), kwargs={"c": 4}) 

43 assert job.args == (1, 2) 

44 assert job.kwargs == {"c": 4} 

45 

46 def test_execute(self): 

47 job = RetryJob(id="j3", func=lambda x: x * 2, args=(21,)) 

48 assert job.execute() == 42 

49 

50 def test_execute_raises(self): 

51 def fail(): 

52 raise ValueError("boom") 

53 

54 job = RetryJob(id="j4", func=fail) 

55 with pytest.raises(ValueError, match="boom"): 

56 job.execute() 

57 

58 def test_attempts_not_auto_incremented(self): 

59 job = RetryJob(id="j5", func=lambda: 1) 

60 job.execute() 

61 assert job.attempts == 0 # execute doesn't increment attempts 

62 

63 

64# ============================================================================ 

65# RetryQueue — Construction 

66# ============================================================================ 

67 

68class TestRetryQueueInit: 

69 def test_defaults(self): 

70 rq = RetryQueue() 

71 assert rq._max_attempts == 3 

72 assert rq._base_delay == 1.0 

73 assert rq._max_delay == 60.0 

74 assert rq._backoff == BackoffStrategy.EXPONENTIAL 

75 assert rq._jitter is True 

76 

77 def test_custom_values(self): 

78 rq = RetryQueue(max_attempts=5, base_delay=0.5, max_delay=10.0, 

79 backoff=BackoffStrategy.CONSTANT, jitter=False) 

80 assert rq._max_attempts == 5 

81 assert rq._base_delay == 0.5 

82 assert rq._max_delay == 10.0 

83 assert rq._backoff == BackoffStrategy.CONSTANT 

84 assert rq._jitter is False 

85 

86 def test_max_attempts_validation(self): 

87 with pytest.raises(ValueError, match="max_attempts must be at least 1"): 

88 RetryQueue(max_attempts=0) 

89 with pytest.raises(ValueError, match="max_attempts must be at least 1"): 

90 RetryQueue(max_attempts=-1) 

91 

92 def test_max_attempts_one_is_valid(self): 

93 rq = RetryQueue(max_attempts=1) 

94 assert rq._max_attempts == 1 

95 

96 

97# ============================================================================ 

98# RetryQueue — Successful execution 

99# ============================================================================ 

100 

101class TestRetryQueueSuccess: 

102 def test_submit_simple(self): 

103 rq = RetryQueue() 

104 result = rq.submit(lambda x, y: x + y, 3, 4) 

105 assert result == 7 

106 

107 def test_submit_no_args(self): 

108 rq = RetryQueue() 

109 result = rq.submit(lambda: 99) 

110 assert result == 99 

111 

112 def test_submit_keyword_args(self): 

113 rq = RetryQueue() 

114 result = rq.submit(lambda a, b: a * b, a=6, b=7) 

115 assert result == 42 

116 

117 def test_submit_mixed_args(self): 

118 rq = RetryQueue() 

119 result = rq.submit(lambda a, b, c=1: a + b + c, 2, 3, c=4) 

120 assert result == 9 

121 

122 def test_stats_after_success(self): 

123 rq = RetryQueue() 

124 rq.submit(lambda: 1) 

125 rq.submit(lambda: 2) 

126 s = rq.stats 

127 assert s["total_submitted"] == 2 

128 assert s["total_succeeded"] == 2 

129 assert s["total_failed"] == 0 

130 assert s["dead_letter_count"] == 0 

131 

132 def test_success_hook(self): 

133 rq = RetryQueue() 

134 hooks = [] 

135 rq.on_success(lambda job, result: hooks.append(("success", job.id, result))) 

136 rq.submit(lambda: 42) 

137 assert len(hooks) == 1 

138 assert hooks[0][0] == "success" 

139 assert hooks[0][2] == 42 

140 

141 

142# ============================================================================ 

143# RetryQueue — Retry & failure 

144# ============================================================================ 

145 

146class TestRetryQueueRetry: 

147 def test_retry_then_succeed(self): 

148 attempts = [] 

149 

150 def flaky(): 

151 attempts.append(1) 

152 if len(attempts) < 3: 

153 raise RuntimeError("fail") 

154 return "ok" 

155 

156 rq = RetryQueue(max_attempts=3, base_delay=0.01, jitter=False) 

157 result = rq.submit(flaky) 

158 assert result == "ok" 

159 assert len(attempts) == 3 

160 

161 def test_exhausted_attempts_raises(self): 

162 def always_fail(): 

163 raise RuntimeError("always") 

164 

165 rq = RetryQueue(max_attempts=2, base_delay=0.01, jitter=False) 

166 with pytest.raises(RuntimeError, match="always"): 

167 rq.submit(always_fail) 

168 

169 s = rq.stats 

170 assert s["total_submitted"] == 1 

171 assert s["total_succeeded"] == 0 

172 assert s["total_failed"] == 1 

173 assert s["dead_letter_count"] == 1 

174 

175 def test_dead_letters_populated(self): 

176 def fail(): 

177 raise RuntimeError("dead") 

178 

179 rq = RetryQueue(max_attempts=1, base_delay=0.01) 

180 try: 

181 rq.submit(fail) 

182 except RuntimeError: 

183 pass 

184 

185 assert len(rq.dead_letters) == 1 

186 job, error = rq.dead_letters[0] 

187 assert isinstance(error, RuntimeError) 

188 assert str(error) == "dead" 

189 assert job.attempts == 1 

190 

191 def test_retry_hook(self): 

192 hooks = [] 

193 

194 def fail_once(): 

195 if fail_once.calls == 0: 

196 fail_once.calls += 1 

197 raise RuntimeError("first") 

198 return "ok" 

199 

200 fail_once.calls = 0 

201 

202 rq = RetryQueue(max_attempts=3, base_delay=0.01, jitter=False) 

203 rq.on_retry(lambda job, err, attempt: hooks.append((job.id, attempt, str(err)))) 

204 rq.submit(fail_once) 

205 

206 assert len(hooks) == 1 

207 assert hooks[0][1] == 1 

208 assert "first" in hooks[0][2] 

209 

210 def test_multiple_retry_hooks(self): 

211 hooks = [] 

212 

213 def fail(): 

214 raise RuntimeError("x") 

215 

216 rq = RetryQueue(max_attempts=2, base_delay=0.01, jitter=False) 

217 rq.on_retry(lambda j, e, a: hooks.append(1)) 

218 rq.on_retry(lambda j, e, a: hooks.append(2)) 

219 try: 

220 rq.submit(fail) 

221 except RuntimeError: 

222 pass 

223 

224 assert len(hooks) == 2 # 1 hook fire x 1 retry 

225 

226 def test_failure_hook(self): 

227 hooks = [] 

228 

229 def fail(): 

230 raise RuntimeError("gone") 

231 

232 rq = RetryQueue(max_attempts=1, base_delay=0.01) 

233 rq.on_failure(lambda job, err: hooks.append(("fail", str(err)))) 

234 try: 

235 rq.submit(fail) 

236 except RuntimeError: 

237 pass 

238 

239 assert len(hooks) == 1 

240 assert hooks[0][0] == "fail" 

241 assert "gone" in hooks[0][1] 

242 

243 def test_hook_exceptions_do_not_propagate(self): 

244 def bad_hook(job, err): 

245 raise RuntimeError("hook broken") 

246 

247 rq = RetryQueue(max_attempts=1, base_delay=0.01) 

248 rq.on_failure(bad_hook) 

249 # Should not raise from hook 

250 with pytest.raises(RuntimeError, match="x"): 

251 rq.submit(lambda: (_ for _ in ()).throw(RuntimeError("x"))) 

252 

253 def test_retry_hook_exception_does_not_propagate(self): 

254 def bad_retry_hook(job, err, attempt): 

255 raise RuntimeError("hook broken") 

256 

257 def fail_once(): 

258 if fail_once.calls == 0: 

259 fail_once.calls += 1 

260 raise RuntimeError("first") 

261 return "ok" 

262 

263 fail_once.calls = 0 

264 

265 rq = RetryQueue(max_attempts=3, base_delay=0.01, jitter=False) 

266 rq.on_retry(bad_retry_hook) 

267 result = rq.submit(fail_once) 

268 assert result == "ok" 

269 

270 def test_success_hook_exception_does_not_propagate(self): 

271 def bad_success_hook(job, result): 

272 raise RuntimeError("hook broken") 

273 

274 rq = RetryQueue(max_attempts=1, base_delay=0.01) 

275 rq.on_success(bad_success_hook) 

276 result = rq.submit(lambda: 42) 

277 assert result == 42 

278 

279 

280# ============================================================================ 

281# RetryQueue — Backoff computation 

282# ============================================================================ 

283 

284class TestRetryQueueBackoff: 

285 def test_exponential_backoff(self): 

286 rq = RetryQueue(base_delay=1.0, max_delay=30.0, backoff=BackoffStrategy.EXPONENTIAL, jitter=False) 

287 # attempt 1: 1.0 * 2^0 = 1.0 

288 assert rq._compute_delay(1) == 1.0 

289 # attempt 2: 1.0 * 2^1 = 2.0 

290 assert rq._compute_delay(2) == 2.0 

291 # attempt 3: 1.0 * 2^2 = 4.0 

292 assert rq._compute_delay(3) == 4.0 

293 

294 def test_constant_backoff(self): 

295 rq = RetryQueue(base_delay=2.0, backoff=BackoffStrategy.CONSTANT, jitter=False) 

296 for i in range(1, 6): 

297 assert rq._compute_delay(i) == 2.0 

298 

299 def test_linear_backoff(self): 

300 rq = RetryQueue(base_delay=1.5, backoff=BackoffStrategy.LINEAR, jitter=False) 

301 assert rq._compute_delay(1) == 1.5 

302 assert rq._compute_delay(2) == 3.0 

303 assert rq._compute_delay(3) == 4.5 

304 

305 def test_max_delay_clamp(self): 

306 rq = RetryQueue(base_delay=10.0, max_delay=15.0, backoff=BackoffStrategy.EXPONENTIAL, jitter=False) 

307 # attempt 1: 10 * 2^0 = 10, under 15 

308 assert rq._compute_delay(1) == 10.0 

309 # attempt 2: 10 * 2^1 = 20, clamped to 15 

310 assert rq._compute_delay(2) == 15.0 

311 

312 def test_jitter_range(self): 

313 rq = RetryQueue(base_delay=10.0, jitter=True) 

314 for _ in range(20): 

315 d = rq._compute_delay(1) 

316 # 10 * 0.5 = 5, 10 * 1.0 = 10 

317 assert 5.0 <= d <= 10.0 

318 

319 

320# ============================================================================ 

321# RetryQueue — Dead letters 

322# ============================================================================ 

323 

324class TestRetryQueueDeadLetters: 

325 def test_clear_dead_letters(self): 

326 def fail(): 

327 raise RuntimeError("x") 

328 

329 rq = RetryQueue(max_attempts=1, base_delay=0.01) 

330 try: 

331 rq.submit(fail) 

332 except RuntimeError: 

333 pass 

334 

335 assert len(rq.dead_letters) == 1 

336 rq.clear_dead_letters() 

337 assert len(rq.dead_letters) == 0 

338 

339 def test_retry_dead_letter_success(self): 

340 calls = [] 

341 

342 def flaky(): 

343 calls.append(1) 

344 if len(calls) == 1: 

345 raise RuntimeError("first") 

346 return "recovered" 

347 

348 rq = RetryQueue(max_attempts=1, base_delay=0.01) 

349 try: 

350 rq.submit(flaky) 

351 except RuntimeError: 

352 pass 

353 

354 assert len(rq.dead_letters) == 1 

355 result = rq.retry_dead_letter(0) 

356 assert result == "recovered" 

357 assert len(rq.dead_letters) == 0 

358 assert rq.stats["total_succeeded"] == 1 

359 

360 def test_retry_dead_letter_index_error(self): 

361 rq = RetryQueue() 

362 with pytest.raises(IndexError, match="out of range"): 

363 rq.retry_dead_letter(0) 

364 

365 def test_retry_dead_letter_still_fails(self): 

366 def fail(): 

367 raise RuntimeError("again") 

368 

369 rq = RetryQueue(max_attempts=1, base_delay=0.01) 

370 try: 

371 rq.submit(fail) 

372 except RuntimeError: 

373 pass 

374 

375 assert len(rq.dead_letters) == 1 

376 with pytest.raises(RuntimeError, match="again"): 

377 rq.retry_dead_letter(0) 

378 # Dead letter re-added since retry also failed (max_attempts=1) 

379 assert len(rq.dead_letters) == 1 

380 

381 

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

383# RetryQueue — Stats 

384# ============================================================================ 

385 

386class TestRetryQueueStats: 

387 def test_default_stats(self): 

388 rq = RetryQueue(max_attempts=5, backoff=BackoffStrategy.LINEAR) 

389 s = rq.stats 

390 assert s["total_submitted"] == 0 

391 assert s["total_succeeded"] == 0 

392 assert s["total_failed"] == 0 

393 assert s["dead_letter_count"] == 0 

394 assert s["max_attempts"] == 5 

395 assert s["backoff"] == "linear" 

396 

397 def test_stats_reflect_state(self): 

398 rq = RetryQueue(max_attempts=2, base_delay=0.01, jitter=False) 

399 # succeed once 

400 rq.submit(lambda: 1) 

401 

402 # fail once 

403 try: 

404 rq.submit(lambda: (_ for _ in ()).throw(ValueError("x"))) 

405 except ValueError: 

406 pass 

407 

408 s = rq.stats 

409 assert s["total_submitted"] == 2 

410 assert s["total_succeeded"] == 1 

411 assert s["total_failed"] == 1 

412 assert s["dead_letter_count"] == 1 

413 

414 def test_stats_thread_safety(self): 

415 import threading 

416 

417 rq = RetryQueue(max_attempts=5, base_delay=0.01, jitter=False) 

418 errors = [] 

419 

420 def worker(): 

421 try: 

422 rq.submit(lambda x: x + 1, 1) 

423 rq.stats # concurrent read 

424 except Exception as e: 

425 errors.append(e) 

426 

427 threads = [threading.Thread(target=worker) for _ in range(5)] 

428 for t in threads: 

429 t.start() 

430 for t in threads: 

431 t.join() 

432 

433 assert len(errors) == 0 

434 assert rq.stats["total_submitted"] == 5 

435 assert rq.stats["total_succeeded"] == 5