11 class TestThreadUtils(unittest.TestCase):
14 @thread_utils.background_thread
15 def background_thread(self, a: int, b: str, stop_event: threading.Event) -> None:
16 while not stop_event.is_set():
17 self.assertEqual(123, a)
18 self.assertEqual('abc', b)
21 def test_background_thread(self):
22 (thread, event) = self.background_thread(123, 'abc')
23 self.assertTrue(thread.is_alive())
27 self.assertFalse(thread.is_alive())
29 @thread_utils.periodically_invoke(period_sec=0.3, stop_after=3)
30 def periodic_invocation_target(self, a: int, b: str):
31 self.assertEqual(123, a)
32 self.assertEqual('abc', b)
33 TestThreadUtils.invocation_count += 1
35 def test_periodically_invoke_with_limit(self):
36 TestThreadUtils.invocation_count = 0
37 (thread, event) = self.periodic_invocation_target(123, 'abc')
38 self.assertTrue(thread.is_alive())
40 self.assertEqual(3, TestThreadUtils.invocation_count)
41 self.assertFalse(thread.is_alive())
43 @thread_utils.periodically_invoke(period_sec=0.1, stop_after=None)
44 def forever_periodic_invocation_target(self, a: int, b: str):
45 self.assertEqual(123, a)
46 self.assertEqual('abc', b)
47 TestThreadUtils.invocation_count += 1
49 def test_periodically_invoke_runs_forever(self):
50 TestThreadUtils.invocation_count = 0
51 (thread, event) = self.forever_periodic_invocation_target(123, 'abc')
52 self.assertTrue(thread.is_alive())
54 self.assertTrue(thread.is_alive())
58 self.assertFalse(thread.is_alive())
59 self.assertTrue(TestThreadUtils.invocation_count >= 19)
62 if __name__ == '__main__':