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 from pyutils import unittest_utils, zookeeper
  13
  14 logger = logging.getLogger(__name__)
  15
  16
  17 class TestZookeeper(unittest.TestCase):
  18     @zookeeper.obtain_lease(
  19         also_pass_lease=True, duration=datetime.timedelta(minutes=1)
  20     )
  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(
  29         also_pass_lease=True, duration=datetime.timedelta(minutes=1)
  30     )
  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())
  36
  37     @zookeeper.obtain_lease(
  38         also_pass_lease=True,
  39         duration=datetime.timedelta(minutes=1),
  40     )
  41     def test_cant_renew_lease_after_released(
  42         self, lease: zookeeper.RenewableReleasableLease
  43     ):
  44         self.assertTrue(lease)
  45         self.assertTrue(lease.release())
  46         self.assertFalse(lease)
  47         self.assertFalse(lease.try_renew(datetime.timedelta(minutes=2)))
  48
  49     @zookeeper.obtain_lease(
  50         also_pass_lease=True, duration=datetime.timedelta(seconds=5)
  51     )
  52     def test_lease_expiration(self, lease: zookeeper.RenewableReleasableLease):
  53         self.assertTrue(lease)
  54         time.sleep(7)
  55         self.assertFalse(lease)
  56
  57     def test_leases_are_exclusive(self):
  58         @zookeeper.obtain_lease(
  59             contender_id='second',
  60             duration=datetime.timedelta(seconds=10),
  61         )
  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?!")
  65
  66         @zookeeper.obtain_lease(
  67             contender_id='first',
  68             duration=datetime.timedelta(seconds=10),
  69         )
  70         def i_will_hold_the_lease():
  71             logger.debug("I have the lease.")
  72             time.sleep(1)
  73             self.assertFalse(i_will_fail_to_get_the_lease())
  74
  75         i_will_hold_the_lease()
  76
  77
  78 if __name__ == '__main__':
  79     unittest.main()