tests/zookeeper_test.py

   1 #!/usr/bin/env python3
   2
   3 # © Copyright 2021-2022, Scott Gasch
   4
   5 """zookeeper unittest."""
   6
   7 import datetime
   8 import logging
   9 import time
  10 import unittest
  11
  12 import zookeeper
  13
  14 import unittest_utils  # Needed for --unittests_ignore_perf flag
  15
  16 logger = logging.getLogger(__name__)
  17
  18
  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)
  23         self.assertTrue(lease.release())
  24         self.assertFalse(lease)
  25         self.assertFalse(lease.release())
  26         self.assertFalse(lease)
  27
  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)
  33         self.assertTrue(lease.release())
  34
  35     @zookeeper.obtain_lease(
  36         also_pass_lease=True,
  37         duration=datetime.timedelta(minutes=1),
  38     )
  39     def test_cant_renew_lease_after_released(self, lease: zookeeper.RenewableReleasableLease):
  40         self.assertTrue(lease)
  41         self.assertTrue(lease.release())
  42         self.assertFalse(lease)
  43         self.assertFalse(lease.try_renew(datetime.timedelta(minutes=2)))
  44
  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)
  48         time.sleep(7)
  49         self.assertFalse(lease)
  50
  51     def test_leases_are_exclusive(self):
  52         @zookeeper.obtain_lease(
  53             contender_id='second',
  54             duration=datetime.timedelta(seconds=10),
  55         )
  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?!")
  59
  60         @zookeeper.obtain_lease(
  61             contender_id='first',
  62             duration=datetime.timedelta(seconds=10),
  63         )
  64         def i_will_hold_the_lease():
  65             logger.debug("I have the lease.")
  66             time.sleep(1)
  67             self.assertFalse(i_will_fail_to_get_the_lease())
  68
  69         i_will_hold_the_lease()
  70
  71
  72 if __name__ == '__main__':
  73     unittest.main()