3 # © Copyright 2021-2022, Scott Gasch
5 """zookeeper unittest."""
12 from pyutils import unittest_utils, zookeeper
14 logger = logging.getLogger(__name__)
17 class TestZookeeper(unittest.TestCase):
18 @zookeeper.obtain_lease(
19 also_pass_lease=True, duration=datetime.timedelta(minutes=1)
21 def test_release_lease(self, lease: zookeeper.RenewableReleasableLease):
22 self.assertTrue(lease)
23 self.assertTrue(lease.release())
24 self.assertFalse(lease)
25 self.assertFalse(lease.release())
26 self.assertFalse(lease)
28 @zookeeper.obtain_lease(
29 also_pass_lease=True, duration=datetime.timedelta(minutes=1)
31 def test_renew_lease(self, lease: zookeeper.RenewableReleasableLease):
32 self.assertTrue(lease)
33 self.assertTrue(lease.try_renew(datetime.timedelta(minutes=2)))
34 self.assertTrue(lease)
35 self.assertTrue(lease.release())
37 @zookeeper.obtain_lease(
39 duration=datetime.timedelta(minutes=1),
41 def test_cant_renew_lease_after_released(
42 self, lease: zookeeper.RenewableReleasableLease
44 self.assertTrue(lease)
45 self.assertTrue(lease.release())
46 self.assertFalse(lease)
47 self.assertFalse(lease.try_renew(datetime.timedelta(minutes=2)))
49 @zookeeper.obtain_lease(
50 also_pass_lease=True, duration=datetime.timedelta(seconds=5)
52 def test_lease_expiration(self, lease: zookeeper.RenewableReleasableLease):
53 self.assertTrue(lease)
55 self.assertFalse(lease)
57 def test_leases_are_exclusive(self):
58 @zookeeper.obtain_lease(
59 contender_id='second',
60 duration=datetime.timedelta(seconds=10),
62 def i_will_fail_to_get_the_lease():
63 logger.debug("I seem to have gotten the lease, wtf?!?!")
64 self.fail("I should not have gotten the lease?!")
66 @zookeeper.obtain_lease(
68 duration=datetime.timedelta(seconds=10),
70 def i_will_hold_the_lease():
71 logger.debug("I have the lease.")
73 self.assertFalse(i_will_fail_to_get_the_lease())
75 i_will_hold_the_lease()
78 if __name__ == '__main__':