From: Scott Gasch <scott@gasch.org>
Date: Sun, 21 Aug 2022 00:22:33 +0000 (-0700)
Subject: Add a unittest for this zookeeper stuff.
X-Git-Url: https://wannabe.guru.org/gitweb/?a=commitdiff_plain;h=a36aa2f6f29ee07198e07b8d2beadc2183d4ea16;p=python_utils.git

Add a unittest for this zookeeper stuff.
---

diff --git a/tests/zookeeper_test.py b/tests/zookeeper_test.py
new file mode 100644
index 0000000..6c00314
--- /dev/null
+++ b/tests/zookeeper_test.py
@@ -0,0 +1,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)
+        lease.release()
+        self.assertFalse(lease)
+        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)
+        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)
+        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()