tests/thread_utils_test.py

   1 #!/usr/bin/env python3
   2
   3 import threading
   4 import time
   5 import unittest
   6
   7 import thread_utils
   8 import unittest_utils
   9
  10
  11 class TestThreadUtils(unittest.TestCase):
  12     invocation_count = 0
  13
  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)
  19             time.sleep(0.1)
  20
  21     def test_background_thread(self):
  22         (thread, event) = self.background_thread(123, 'abc')
  23         self.assertTrue(thread.is_alive())
  24         time.sleep(1.0)
  25         event.set()
  26         thread.join()
  27         self.assertFalse(thread.is_alive())
  28
  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
  34
  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())
  39         time.sleep(1.0)
  40         self.assertEqual(3, TestThreadUtils.invocation_count)
  41         self.assertFalse(thread.is_alive())
  42
  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
  48
  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())
  53         time.sleep(1.0)
  54         self.assertTrue(thread.is_alive())
  55         time.sleep(1.0)
  56         event.set()
  57         thread.join()
  58         self.assertFalse(thread.is_alive())
  59         self.assertTrue(TestThreadUtils.invocation_count >= 19)
  60
  61
  62 if __name__ == '__main__':
  63     unittest.main()