1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
|
#!/usr/bin/env python3
import threading
import time
import unittest
import thread_utils
import unittest_utils
class TestThreadUtils(unittest.TestCase):
invocation_count = 0
@thread_utils.background_thread
def background_thread(self, a: int, b: str, stop_event: threading.Event) -> None:
while not stop_event.is_set():
self.assertEqual(123, a)
self.assertEqual('abc', b)
time.sleep(0.1)
def test_background_thread(self):
(thread, event) = self.background_thread(123, 'abc')
self.assertTrue(thread.is_alive())
time.sleep(1.0)
event.set()
thread.join()
self.assertFalse(thread.is_alive())
@thread_utils.periodically_invoke(period_sec=0.3, stop_after=3)
def periodic_invocation_target(self, a: int, b: str):
self.assertEqual(123, a)
self.assertEqual('abc', b)
TestThreadUtils.invocation_count += 1
def test_periodically_invoke_with_limit(self):
TestThreadUtils.invocation_count = 0
(thread, event) = self.periodic_invocation_target(123, 'abc')
self.assertTrue(thread.is_alive())
time.sleep(1.0)
self.assertEqual(3, TestThreadUtils.invocation_count)
self.assertFalse(thread.is_alive())
@thread_utils.periodically_invoke(period_sec=0.1, stop_after=None)
def forever_periodic_invocation_target(self, a: int, b: str):
self.assertEqual(123, a)
self.assertEqual('abc', b)
TestThreadUtils.invocation_count += 1
def test_periodically_invoke_runs_forever(self):
TestThreadUtils.invocation_count = 0
(thread, event) = self.forever_periodic_invocation_target(123, 'abc')
self.assertTrue(thread.is_alive())
time.sleep(1.0)
self.assertTrue(thread.is_alive())
time.sleep(1.0)
event.set()
thread.join()
self.assertFalse(thread.is_alive())
self.assertTrue(TestThreadUtils.invocation_count >= 19)
if __name__ == '__main__':
unittest.main()
|
