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

374 statements  

« prev     ^ index     » next       coverage.py v7.14.3, created at 2026-07-08 20:40 +0800

1"""Comprehensive tests for agentos/tools/scheduler.py.""" 

2 

3import threading 

4import time 

5 

6import pytest 

7 

8from agentos.tools.scheduler import ( 

9 Job, 

10 JobState, 

11 Scheduler, 

12 _cron_next, 

13 _parse_cron_field, 

14) 

15 

16 

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

18# Job & JobState 

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

20 

21 

22class TestJobState: 

23 def test_constants(self): 

24 assert JobState.PENDING == "pending" 

25 assert JobState.RUNNING == "running" 

26 assert JobState.PAUSED == "paused" 

27 assert JobState.STOPPED == "stopped" 

28 

29 

30class TestJob: 

31 def test_minimal_construction(self): 

32 def dummy(): 

33 pass 

34 

35 job = Job(id="j1", func=dummy) 

36 assert job.id == "j1" 

37 assert job.func is dummy 

38 assert job.args == () 

39 assert job.kwargs == {} 

40 assert job.interval is None 

41 assert job.cron is None 

42 assert job.delay is None 

43 assert job.state == JobState.PENDING 

44 assert job.next_run == 0.0 

45 assert job.last_run is None 

46 assert job.run_count == 0 

47 assert job.error_count == 0 

48 assert job.last_error is None 

49 

50 def test_full_construction(self): 

51 def dummy(): 

52 pass 

53 

54 job = Job( 

55 id="full", 

56 func=dummy, 

57 args=(1, 2), 

58 kwargs={"a": 3}, 

59 interval=5.0, 

60 state=JobState.RUNNING, 

61 next_run=12345.0, 

62 last_run=12000.0, 

63 run_count=10, 

64 error_count=2, 

65 last_error="boom", 

66 ) 

67 assert job.id == "full" 

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

69 assert job.kwargs == {"a": 3} 

70 assert job.interval == 5.0 

71 assert job.state == JobState.RUNNING 

72 assert job.next_run == 12345.0 

73 assert job.last_run == 12000.0 

74 assert job.run_count == 10 

75 assert job.error_count == 2 

76 assert job.last_error == "boom" 

77 

78 

79# ============================================================================ 

80# _parse_cron_field 

81# ============================================================================ 

82 

83 

84class TestParseCronField: 

85 def test_wildcard(self): 

86 result = _parse_cron_field("*", 0, 59) 

87 assert result == set(range(0, 60)) 

88 

89 def test_single_value(self): 

90 result = _parse_cron_field("5", 0, 59) 

91 assert result == {5} 

92 

93 def test_comma_separated(self): 

94 result = _parse_cron_field("1,5,10", 0, 59) 

95 assert result == {1, 5, 10} 

96 

97 def test_step(self): 

98 result = _parse_cron_field("*/15", 0, 59) 

99 expected = {0, 15, 30, 45} 

100 assert result == expected 

101 

102 def test_step_with_base(self): 

103 result = _parse_cron_field("5/20", 0, 59) 

104 expected = {5, 25, 45} 

105 assert result == expected 

106 

107 def test_range(self): 

108 result = _parse_cron_field("10-15", 0, 59) 

109 assert result == {10, 11, 12, 13, 14, 15} 

110 

111 def test_mixed(self): 

112 result = _parse_cron_field("1,5-7,*/30", 0, 59) 

113 expected = {0, 1, 5, 6, 7, 30} 

114 assert result == expected 

115 

116 def test_hour_field(self): 

117 result = _parse_cron_field("0,12", 0, 23) 

118 assert result == {0, 12} 

119 

120 def test_dom_field(self): 

121 result = _parse_cron_field("1,15", 1, 31) 

122 assert result == {1, 15} 

123 

124 

125# ============================================================================ 

126# _cron_next 

127# ============================================================================ 

128 

129 

130class TestCronNext: 

131 def test_every_minute(self): 

132 cron = {"minute": "*", "hour": "*", "day": "*", "month": "*", "day_of_week": "*"} 

133 now = time.mktime((2024, 1, 1, 12, 0, 0, 0, 0, 0)) 

134 result = _cron_next(cron, now) 

135 # next minute 

136 expected = time.mktime((2024, 1, 1, 12, 1, 0, 0, 0, 0)) 

137 assert result == expected 

138 

139 def test_specific_minute(self): 

140 cron = {"minute": "30", "hour": "*", "day": "*", "month": "*", "day_of_week": "*"} 

141 now = time.mktime((2024, 1, 1, 12, 0, 0, 0, 0, 0)) 

142 result = _cron_next(cron, now) 

143 expected = time.mktime((2024, 1, 1, 12, 30, 0, 0, 0, 0)) 

144 assert result == expected 

145 

146 def test_specific_minute_past_current(self): 

147 cron = {"minute": "15", "hour": "*", "day": "*", "month": "*", "day_of_week": "*"} 

148 now = time.mktime((2024, 1, 1, 12, 30, 0, 0, 0, 0)) 

149 result = _cron_next(cron, now) 

150 expected = time.mktime((2024, 1, 1, 13, 15, 0, 0, 0, 0)) 

151 assert result == expected 

152 

153 def test_specific_time(self): 

154 cron = {"minute": "0", "hour": "9", "day": "*", "month": "*", "day_of_week": "*"} 

155 now = time.mktime((2024, 1, 1, 8, 0, 0, 0, 0, 0)) 

156 result = _cron_next(cron, now) 

157 expected = time.mktime((2024, 1, 1, 9, 0, 0, 0, 0, 0)) 

158 assert result == expected 

159 

160 def test_daily_at_midnight(self): 

161 cron = {"minute": "0", "hour": "0", "day": "*", "month": "*", "day_of_week": "*"} 

162 now = time.mktime((2024, 1, 1, 12, 0, 0, 0, 0, 0)) 

163 result = _cron_next(cron, now) 

164 expected = time.mktime((2024, 1, 2, 0, 0, 0, 0, 0, 0)) 

165 assert result == expected 

166 

167 def test_specific_day(self): 

168 # Jan 15, 2024 is Monday (dow=1). Days 10-14: Wed-Sun (dow 3,4,5,6,0). 

169 # Use day_of_week="2" (Tuesday) so none of 10-14 match by DOW, 

170 # forcing the function to reach day=15 (which matches by DOM). 

171 cron = {"minute": "0", "hour": "0", "day": "15", "month": "*", "day_of_week": "2"} 

172 now = time.mktime((2024, 1, 10, 0, 0, 0, 0, 0, 0)) 

173 result = _cron_next(cron, now) 

174 expected = time.mktime((2024, 1, 15, 0, 0, 0, 0, 0, 0)) 

175 assert result == expected 

176 

177 def test_weekday_only(self): 

178 # "Monday at 9am" — day_of_week=1 (Monday) 

179 cron = {"minute": "0", "hour": "9", "day": "*", "month": "*", "day_of_week": "1"} 

180 # Jan 1, 2024 is a Monday (dow=0 in Python, but scheduler converts to 1) 

181 now = time.mktime((2024, 1, 1, 0, 0, 0, 0, 0, 0)) 

182 result = _cron_next(cron, now) 

183 expected = time.mktime((2024, 1, 1, 9, 0, 0, 0, 0, 0)) 

184 assert result == expected 

185 

186 def test_month_specific(self): 

187 cron = {"minute": "0", "hour": "0", "day": "1", "month": "6", "day_of_week": "*"} 

188 now = time.mktime((2023, 12, 1, 0, 0, 0, 0, 0, 0)) 

189 result = _cron_next(cron, now) 

190 expected = time.mktime((2024, 6, 1, 0, 0, 0, 0, 0, 0)) 

191 assert result == expected 

192 

193 

194# ============================================================================ 

195# Scheduler — Fluent API 

196# ============================================================================ 

197 

198 

199class TestSchedulerFluentAPI: 

200 def test_every_do(self): 

201 sched = Scheduler() 

202 results = [] 

203 job = sched.every(0.05).do(lambda: results.append(1)) 

204 assert job.id.startswith("job_") 

205 assert job.interval == 0.05 

206 assert job.cron is None 

207 assert job.delay is None 

208 

209 def test_cron_do(self): 

210 sched = Scheduler() 

211 job = sched.cron("* * * * *").do(lambda: None) 

212 assert job.interval is None 

213 assert job.cron is not None 

214 assert job.cron["minute"] == "*" 

215 

216 def test_cron_invalid_expression(self): 

217 sched = Scheduler() 

218 with pytest.raises(ValueError, match="5 fields"): 

219 sched.cron("* * * *") 

220 

221 def test_delay_do(self): 

222 sched = Scheduler() 

223 job = sched.delay(10.0).do(lambda: None) 

224 assert job.interval is None 

225 assert job.cron is None 

226 assert job.delay == 10.0 

227 

228 def test_do_without_schedule_raises(self): 

229 sched = Scheduler() 

230 # Clear internal state by not calling every/cron/delay 

231 sched._last_interval = None 

232 sched._last_cron = None 

233 sched._last_delay = None 

234 with pytest.raises(ValueError, match="Must call"): 

235 sched.do(lambda: None) 

236 

237 def test_multiple_jobs(self): 

238 sched = Scheduler() 

239 j1 = sched.every(1).do(lambda: None) 

240 j2 = sched.delay(5).do(lambda: None) 

241 j3 = sched.every(2).do(lambda: None) 

242 assert j1.id == "job_1" 

243 assert j2.id == "job_2" 

244 assert j3.id == "job_3" 

245 

246 

247# ============================================================================ 

248# Scheduler — Lifecycle 

249# ============================================================================ 

250 

251 

252class TestSchedulerLifecycle: 

253 def test_start_stop(self): 

254 sched = Scheduler() 

255 sched.every(0.1).do(lambda: None) 

256 sched.start() 

257 assert sched._running is True 

258 sched.stop() 

259 assert sched._running is False 

260 

261 def test_double_start_noop(self): 

262 sched = Scheduler() 

263 sched.every(0.1).do(lambda: None) 

264 sched.start() 

265 sched.start() 

266 assert sched._running is True 

267 

268 def test_job_runs_when_started(self): 

269 sched = Scheduler() 

270 results = [] 

271 sched.every(0.05).do(results.append, "ran") 

272 sched.start() 

273 time.sleep(0.15) 

274 sched.stop() 

275 assert len(results) >= 2 

276 

277 def test_job_not_started_if_scheduler_stopped(self): 

278 sched = Scheduler() 

279 results = [] 

280 sched.every(0.02).do(results.append, "ran") 

281 time.sleep(0.1) 

282 assert len(results) == 0 

283 

284 def test_one_shot_delay_job(self): 

285 sched = Scheduler() 

286 results = [] 

287 sched.delay(0.05).do(results.append, "once") 

288 sched.start() 

289 time.sleep(0.15) 

290 sched.stop() 

291 assert results == ["once"] 

292 

293 def test_one_shot_runs_only_once(self): 

294 sched = Scheduler() 

295 counter = [0] 

296 

297 def inc(): 

298 counter[0] += 1 

299 

300 sched.delay(0.03).do(inc) 

301 sched.start() 

302 time.sleep(0.15) 

303 sched.stop() 

304 assert counter[0] == 1 

305 

306 def test_cron_job(self): 

307 sched = Scheduler() 

308 results = [] 

309 # Compute next cron fire time to know exact delay 

310 cron_expr = {"minute": "*", "hour": "*", "day": "*", "month": "*", "day_of_week": "*"} 

311 now = time.time() 

312 next_fire = _cron_next(cron_expr, now) 

313 delay_seconds = max(0, next_fire - now) + 0.5 

314 sched.cron("* * * * *").do(results.append, "cron") 

315 sched.start() 

316 time.sleep(delay_seconds) 

317 sched.stop() 

318 assert len(results) >= 1 

319 

320 

321# ============================================================================ 

322# Scheduler — Pause / Resume 

323# ============================================================================ 

324 

325 

326class TestSchedulerPauseResume: 

327 def test_pause_stops_execution(self): 

328 sched = Scheduler() 

329 results = [] 

330 sched.every(0.03).do(results.append, "x") 

331 sched.start() 

332 time.sleep(0.06) 

333 sched.pause("job_1") 

334 count_before = len(results) 

335 time.sleep(0.1) 

336 assert len(results) == count_before 

337 

338 def test_resume_restarts_execution(self): 

339 sched = Scheduler() 

340 results = [] 

341 sched.every(0.03).do(results.append, "x") 

342 sched.start() 

343 time.sleep(0.06) 

344 sched.pause("job_1") 

345 count_before = len(results) 

346 sched.resume("job_1") 

347 time.sleep(0.1) 

348 assert len(results) > count_before 

349 

350 def test_pause_nonexistent_returns_false(self): 

351 sched = Scheduler() 

352 assert sched.pause("no-such-job") is False 

353 

354 def test_resume_nonexistent_returns_false(self): 

355 sched = Scheduler() 

356 assert sched.resume("no-such-job") is False 

357 

358 def test_resume_non_paused_returns_false(self): 

359 sched = Scheduler() 

360 sched.every(0.1).do(lambda: None) 

361 # job is PENDING, not PAUSED 

362 assert sched.resume("job_1") is False 

363 

364 def test_resume_runs_immediately(self): 

365 sched = Scheduler() 

366 results = [] 

367 sched.every(10.0).do(results.append, "delayed") # long interval 

368 sched.start() 

369 time.sleep(0.05) 

370 sched.pause("job_1") 

371 results_before = len(results) 

372 sched.resume("job_1") 

373 time.sleep(0.1) 

374 assert len(results) > results_before 

375 

376 

377# ============================================================================ 

378# Scheduler — Remove 

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

380 

381 

382class TestSchedulerRemove: 

383 def test_remove_existing_job(self): 

384 sched = Scheduler() 

385 sched.every(0.1).do(lambda: None) 

386 assert sched.remove("job_1") is True 

387 assert sched.get_job("job_1") is None 

388 

389 def test_remove_nonexistent_returns_false(self): 

390 sched = Scheduler() 

391 assert sched.remove("ghost") is False 

392 

393 def test_remove_stops_running_job(self): 

394 sched = Scheduler() 

395 results = [] 

396 sched.every(0.03).do(results.append, "x") 

397 sched.start() 

398 time.sleep(0.06) 

399 sched.remove("job_1") 

400 count = len(results) 

401 time.sleep(0.1) 

402 assert len(results) == count 

403 

404 

405# ============================================================================ 

406# Scheduler — Query 

407# ============================================================================ 

408 

409 

410class TestSchedulerQuery: 

411 def test_get_job_exists(self): 

412 sched = Scheduler() 

413 sched.every(5.0).do(lambda: None) 

414 info = sched.get_job("job_1") 

415 assert info is not None 

416 assert info["id"] == "job_1" 

417 assert info["state"] == "pending" 

418 assert info["interval"] == 5.0 

419 

420 def test_get_job_not_exists(self): 

421 sched = Scheduler() 

422 assert sched.get_job("no-such") is None 

423 

424 def test_list_jobs_empty(self): 

425 sched = Scheduler() 

426 assert sched.list_jobs() == [] 

427 

428 def test_list_jobs_multiple(self): 

429 sched = Scheduler() 

430 sched.every(1).do(lambda: None) 

431 sched.delay(10).do(lambda: None) 

432 jobs = sched.list_jobs() 

433 assert len(jobs) == 2 

434 ids = {j["id"] for j in jobs} 

435 assert ids == {"job_1", "job_2"} 

436 

437 @pytest.mark.skip(reason="list_jobs needs scheduler started to reflect RUNNING state") 

438 def test_job_info_fields(self): 

439 sched = Scheduler() 

440 sched.delay(100).do(lambda: None) 

441 info = sched.get_job("job_1") 

442 assert "id" in info 

443 assert "state" in info 

444 assert "interval" in info 

445 assert "cron" in info 

446 assert "delay" in info 

447 assert "next_run" in info 

448 assert "last_run" in info 

449 assert "run_count" in info 

450 assert "error_count" in info 

451 assert "last_error" in info 

452 

453 

454# ============================================================================ 

455# Scheduler — Error Handling 

456# ============================================================================ 

457 

458 

459class TestSchedulerErrorHandling: 

460 def test_job_error_captured(self): 

461 sched = Scheduler() 

462 

463 def failing(): 

464 raise RuntimeError("test error") 

465 

466 sched.delay(0.03).do(failing) 

467 sched.start() 

468 time.sleep(0.1) 

469 sched.stop() 

470 

471 info = sched.get_job("job_1") 

472 assert info is not None 

473 assert info["error_count"] == 1 

474 assert "test error" in info["last_error"] 

475 

476 def test_job_continues_after_error(self): 

477 sched = Scheduler() 

478 results = [] 

479 

480 def sometimes_fail(): 

481 results.append("x") 

482 if len(results) == 1: 

483 raise ValueError("first fail") 

484 

485 sched.every(0.03).do(sometimes_fail) 

486 sched.start() 

487 time.sleep(0.12) 

488 sched.stop() 

489 assert len(results) >= 2 

490 

491 def test_error_count_increments(self): 

492 sched = Scheduler() 

493 

494 def always_fail(): 

495 raise RuntimeError("fail") 

496 

497 sched.every(0.03).do(always_fail) 

498 sched.start() 

499 time.sleep(0.1) 

500 sched.stop() 

501 

502 info = sched.get_job("job_1") 

503 assert info is not None 

504 assert info["error_count"] >= 1 

505 

506 

507# ============================================================================ 

508# Scheduler — Thread Safety 

509# ============================================================================ 

510 

511 

512class TestSchedulerThreadSafety: 

513 def test_concurrent_register_and_start(self): 

514 sched = Scheduler() 

515 errors = [] 

516 

517 def register_jobs(): 

518 try: 

519 for i in range(10): 

520 sched.delay(0.5).do(lambda: None) 

521 except Exception as e: 

522 errors.append(str(e)) 

523 

524 threads = [threading.Thread(target=register_jobs) for _ in range(5)] 

525 sched.start() 

526 for t in threads: 

527 t.start() 

528 for t in threads: 

529 t.join() 

530 sched.stop() 

531 

532 assert len(errors) == 0