Backport some test for the lock_manager app

apps/lock_manager/tests.py

@@ -1,9 +1,68 @@
+from __future__ import absolute_import
+
+import time
+
 from django.test import TestCase
 
+from .exceptions import LockError#, StaleLock
+from .models import Lock
 
-class SimpleTest(TestCase):
-    def test_basic_addition(self):
-        """
-        Tests that 1 + 1 always equals 2.
-        """
-        self.assertEqual(1 + 1, 2)
+# Notice: StaleLock exception and tests are not available until more changes are
+# backported.
+
+
+class LockTestCase(TestCase):
+    def test_exclusive(self):
+        lock_1 = Lock.objects.acquire_lock(name='test_lock_1')
+        with self.assertRaises(LockError):
+            Lock.objects.acquire_lock(name='test_lock_1')
+
+        # Cleanup
+        lock_1.release()
+
+    def test_release(self):
+        lock_1 = Lock.objects.acquire_lock(name='test_lock_1')
+        lock_1.release()
+        lock_2 = Lock.objects.acquire_lock(name='test_lock_1')
+
+        # Cleanup
+        lock_2.release()
+
+    def test_timeout_expired(self):
+        lock_1 = Lock.objects.acquire_lock(name='test_lock_1', timeout=1)
+
+        # lock_1 not release and not expired, should raise LockError
+        with self.assertRaises(LockError):
+            Lock.objects.acquire_lock(name='test_lock_1')
+
+        time.sleep(2)
+        # lock_1 not release but has expired, should not raise LockError
+        lock_2 = Lock.objects.acquire_lock(name='test_lock_1')
+
+        # Cleanup
+        lock_2.release()
+
+    def test_double_release(self):
+        lock_1 = Lock.objects.acquire_lock(name='test_lock_1')
+        lock_1.release()
+        #with self.assertRaises(StaleLock):
+        #    lock_1.release()
+
+    def test_release_expired(self):
+        lock_1 = Lock.objects.acquire_lock(name='test_lock_1', timeout=1)
+        time.sleep(2)
+        lock_1.release()
+        # No exception is raised even though the lock has expired.
+        # The logic is that checking for expired locks during release is
+        # not necesary as any attempt by someone else to aquire the lock
+        # would be successfull, even after an extended lapse of time
+
+    def test_release_expired_reaquired(self):
+        lock_1 = Lock.objects.acquire_lock(name='test_lock_1', timeout=1)
+        time.sleep(2)
+        lock_2 = Lock.objects.acquire_lock(name='test_lock_1', timeout=1)
+        #with self.assertRaises(StaleLock):
+        #    lock_1.release()
+
+        # Cleanup
+        lock_2.release()