Coverage for agentos/tests/test

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

3import threading

4import time

6import pytest

8from agentos.tools.scheduler import (

9 Job,

10 JobState,

11 Scheduler,

12 _cron_next,

13 _parse_cron_field,

14)

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

18# Job & JobState

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

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"

30class TestJob:

31 def test_minimal_construction(self):

32 def dummy():

33 pass

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

50 def test_full_construction(self):

51 def dummy():

52 pass

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"

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

80# _parse_cron_field

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

84class TestParseCronField:

85 def test_wildcard(self):

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

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

89 def test_single_value(self):

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

91 assert result == {5}

93 def test_comma_separated(self):

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

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

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

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

374 statements