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
64
65
66
67
68
69
70
71
72
73
|
#!/usr/bin/env python3
# © Copyright 2021-2022, Scott Gasch
"""zookeeper unittest."""
import datetime
import logging
import time
import unittest
import zookeeper
import unittest_utils # Needed for --unittests_ignore_perf flag
logger = logging.getLogger(__name__)
class TestZookeeper(unittest.TestCase):
@zookeeper.obtain_lease(also_pass_lease=True, duration=datetime.timedelta(minutes=1))
def test_release_lease(self, lease: zookeeper.RenewableReleasableLease):
self.assertTrue(lease)
self.assertTrue(lease.release())
self.assertFalse(lease)
self.assertFalse(lease.release())
self.assertFalse(lease)
@zookeeper.obtain_lease(also_pass_lease=True, duration=datetime.timedelta(minutes=1))
def test_renew_lease(self, lease: zookeeper.RenewableReleasableLease):
self.assertTrue(lease)
self.assertTrue(lease.try_renew(datetime.timedelta(minutes=2)))
self.assertTrue(lease)
self.assertTrue(lease.release())
@zookeeper.obtain_lease(
also_pass_lease=True,
duration=datetime.timedelta(minutes=1),
)
def test_cant_renew_lease_after_released(self, lease: zookeeper.RenewableReleasableLease):
self.assertTrue(lease)
self.assertTrue(lease.release())
self.assertFalse(lease)
self.assertFalse(lease.try_renew(datetime.timedelta(minutes=2)))
@zookeeper.obtain_lease(also_pass_lease=True, duration=datetime.timedelta(seconds=5))
def test_lease_expiration(self, lease: zookeeper.RenewableReleasableLease):
self.assertTrue(lease)
time.sleep(7)
self.assertFalse(lease)
def test_leases_are_exclusive(self):
@zookeeper.obtain_lease(
contender_id='second',
duration=datetime.timedelta(seconds=10),
)
def i_will_fail_to_get_the_lease():
logger.debug("I seem to have gotten the lease, wtf?!?!")
self.fail("I should not have gotten the lease?!")
@zookeeper.obtain_lease(
contender_id='first',
duration=datetime.timedelta(seconds=10),
)
def i_will_hold_the_lease():
logger.debug("I have the lease.")
time.sleep(1)
self.assertFalse(i_will_fail_to_get_the_lease())
i_will_hold_the_lease()
if __name__ == '__main__':
unittest.main()
|
