3 # © Copyright 2021-2022, Scott Gasch
5 """thread_utils unittest."""
15 class TestThreadUtils(unittest.TestCase):
18 @thread_utils.background_thread
19 def background_thread(self, a: int, b: str, stop_event: threading.Event) -> None:
20 while not stop_event.is_set():
21 self.assertEqual(123, a)
22 self.assertEqual('abc', b)
25 def test_background_thread(self):
26 (thread, event) = self.background_thread(123, 'abc')
27 self.assertTrue(thread.is_alive())
31 self.assertFalse(thread.is_alive())
33 @thread_utils.periodically_invoke(period_sec=0.3, stop_after=3)
34 def periodic_invocation_target(self, a: int, b: str):
35 self.assertEqual(123, a)
36 self.assertEqual('abc', b)
37 TestThreadUtils.invocation_count += 1
39 def test_periodically_invoke_with_limit(self):
40 TestThreadUtils.invocation_count = 0
41 (thread, event) = self.periodic_invocation_target(123, 'abc')
42 self.assertTrue(thread.is_alive())
44 self.assertEqual(3, TestThreadUtils.invocation_count)
45 self.assertFalse(thread.is_alive())
47 @thread_utils.periodically_invoke(period_sec=0.1, stop_after=None)
48 def forever_periodic_invocation_target(self, a: int, b: str):
49 self.assertEqual(123, a)
50 self.assertEqual('abc', b)
51 TestThreadUtils.invocation_count += 1
53 def test_periodically_invoke_runs_forever(self):
54 TestThreadUtils.invocation_count = 0
55 (thread, event) = self.forever_periodic_invocation_target(123, 'abc')
56 self.assertTrue(thread.is_alive())
58 self.assertTrue(thread.is_alive())
62 self.assertFalse(thread.is_alive())
63 self.assertTrue(TestThreadUtils.invocation_count >= 19)
66 if __name__ == '__main__':