3 # © Copyright 2021-2022, Scott Gasch
5 """zookeeper unittest."""
14 import unittest_utils # Needed for --unittests_ignore_perf flag
16 logger = logging.getLogger(__name__)
19 class TestZookeeper(unittest.TestCase):
20 @zookeeper.obtain_lease(also_pass_lease=True, duration=datetime.timedelta(minutes=1))
21 def test_release_lease(self, lease: zookeeper.RenewableReleasableLease):
22 self.assertTrue(lease)
24 self.assertFalse(lease)
26 self.assertFalse(lease)
28 @zookeeper.obtain_lease(also_pass_lease=True, duration=datetime.timedelta(minutes=1))
29 def test_renew_lease(self, lease: zookeeper.RenewableReleasableLease):
30 self.assertTrue(lease)
31 self.assertTrue(lease.try_renew(datetime.timedelta(minutes=2)))
32 self.assertTrue(lease)
35 @zookeeper.obtain_lease(
37 duration=datetime.timedelta(minutes=1),
39 def test_cant_renew_lease_after_released(self, lease: zookeeper.RenewableReleasableLease):
40 self.assertTrue(lease)
42 self.assertFalse(lease)
43 self.assertFalse(lease.try_renew(datetime.timedelta(minutes=2)))
45 @zookeeper.obtain_lease(also_pass_lease=True, duration=datetime.timedelta(seconds=5))
46 def test_lease_expiration(self, lease: zookeeper.RenewableReleasableLease):
47 self.assertTrue(lease)
49 self.assertFalse(lease)
51 def test_leases_are_exclusive(self):
52 @zookeeper.obtain_lease(
53 contender_id='second',
54 duration=datetime.timedelta(seconds=10),
56 def i_will_fail_to_get_the_lease():
57 logger.debug("I seem to have gotten the lease, wtf?!?!")
58 self.fail("I should not have gotten the lease?!")
60 @zookeeper.obtain_lease(
62 duration=datetime.timedelta(seconds=10),
64 def i_will_hold_the_lease():
65 logger.debug("I have the lease.")
67 self.assertFalse(i_will_fail_to_get_the_lease())
69 i_will_hold_the_lease()
72 if __name__ == '__main__':