errands

small jobs the runtime does for the studio.
git clone https://owenrusk.dev/errands.git
errands / tests/test_jobs.py -rw-r--r-- · 1614 bytes
 1 import unittest
 2 from datetime import datetime, timedelta, timezone
 3 
 4 from errands import jobs
 5 
 6 CENTRAL = timezone(timedelta(hours=-6))
 7 
 8 
 9 def at(hour, minute=0, day=10):
10     return datetime(2024, 12, day, hour, minute, tzinfo=CENTRAL)
11 
12 
13 class DueTest(unittest.TestCase):
14     def test_every_n_minutes(self):
15         job = jobs.Job("x", lambda: None, every=15)
16         self.assertTrue(jobs.is_due(job, at(9), None))
17         self.assertFalse(jobs.is_due(job, at(9, 10), at(9)))
18         self.assertTrue(jobs.is_due(job, at(9, 15), at(9)))
19 
20     def test_once_a_day(self):
21         job = jobs.Job("x", lambda: None, at="09:00")
22         self.assertFalse(jobs.is_due(job, at(8, 59), None))
23         self.assertTrue(jobs.is_due(job, at(9), at(9, day=9)))
24         self.assertFalse(jobs.is_due(job, at(13), at(9)))
25         self.assertTrue(jobs.is_due(job, at(9, day=11), at(9)))
26 
27 
28 class TickTest(unittest.TestCase):
29     def test_runs_what_is_due_and_remembers(self):
30         calls = []
31         job = jobs.Job("x", lambda: calls.append(1), every=5)
32         last = {}
33         self.assertEqual(jobs.tick(at(9), last, [job]), ["x"])
34         self.assertEqual(jobs.tick(at(9, 1), last, [job]), [])
35         self.assertEqual(calls, [1])
36 
37     def test_a_failure_doesnt_stop_the_rest(self):
38         def broken():
39             raise RuntimeError("down")
40 
41         calls = []
42         jobs_ = [jobs.Job("broken", broken, every=5), jobs.Job("fine", lambda: calls.append(1), every=5)]
43         with self.assertLogs("errands", level="ERROR"):
44             self.assertEqual(jobs.tick(at(9), {}, jobs_), ["fine"])
45         self.assertEqual(calls, [1])