tests/thread_utils_test.py

   1 #!/usr/bin/env python3
   2
   3 # © Copyright 2021-2022, Scott Gasch
   4
   5 """thread_utils unittest."""
   6
   7 import threading
   8 import time
   9 import unittest
  10
  11 import thread_utils
  12 import unittest_utils
  13
  14
  15 class TestThreadUtils(unittest.TestCase):
  16     invocation_count = 0
  17
  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)
  23             time.sleep(0.1)
  24
  25     def test_background_thread(self):
  26         (thread, event) = self.background_thread(123, 'abc')
  27         self.assertTrue(thread.is_alive())
  28         time.sleep(1.0)
  29         event.set()
  30         thread.join()
  31         self.assertFalse(thread.is_alive())
  32
  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
  38
  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())
  43         time.sleep(1.0)
  44         self.assertEqual(3, TestThreadUtils.invocation_count)
  45         self.assertFalse(thread.is_alive())
  46
  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
  52
  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())
  57         time.sleep(1.0)
  58         self.assertTrue(thread.is_alive())
  59         time.sleep(1.0)
  60         event.set()
  61         thread.join()
  62         self.assertFalse(thread.is_alive())
  63         self.assertTrue(TestThreadUtils.invocation_count >= 19)
  64
  65
  66 if __name__ == '__main__':
  67     unittest.main()