matthias/mayan-edms
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Add plugable locking backend support. Add threadsafe file lock backend.

Browse Source
This commit is contained in:
Roberto Rosario
2016-11-13 03:50:09 -04:00
parent f5e3d5a8f2
commit 44531bd92a
12 changed files with 155 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mayan/apps/lock_manager/tests/test_models.py → mayan/apps/lock_manager/tests/test_backends.py
+21 -12
View File
@@ -3,48 +3,53 @@ from __future__ import unicode_literals
import time
from django.test import TestCase
from django.utils.module_loading import import_string
from ..exceptions import LockError
from ..models import Lock
class LockTestCase(TestCase):
class FileLockTestCase(TestCase):
backend_string = 'lock_manager.backends.file_lock.FileLock'
def setUp(self):
self.locking_backend = import_string(self.backend_string)
def test_exclusive(self):
lock_1 = Lock.objects.acquire_lock(name='test_lock_1')
lock_1 = self.locking_backend.acquire_lock(name='test_lock_1')
with self.assertRaises(LockError):
Lock.objects.acquire_lock(name='test_lock_1')
self.locking_backend.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 = self.locking_backend.acquire_lock(name='test_lock_1')
lock_1.release()
lock_2 = Lock.objects.acquire_lock(name='test_lock_1')
lock_2 = self.locking_backend.acquire_lock(name='test_lock_1')
# Cleanup
lock_2.release()
def test_timeout_expired(self):
Lock.objects.acquire_lock(name='test_lock_1', timeout=1)
self.locking_backend.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')
self.locking_backend.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')
lock_2 = self.locking_backend.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 = self.locking_backend.acquire_lock(name='test_lock_1')
lock_1.release()
def test_release_expired(self):
lock_1 = Lock.objects.acquire_lock(name='test_lock_1', timeout=1)
lock_1 = self.locking_backend.acquire_lock(name='test_lock_1', timeout=1)
time.sleep(2)
lock_1.release()
# No exception is raised even though the lock has expired.
@@ -54,7 +59,11 @@ class LockTestCase(TestCase):
def test_release_expired_reaquired(self):
time.sleep(2)
lock_2 = Lock.objects.acquire_lock(name='test_lock_1', timeout=1)
lock_2 = self.locking_backend.acquire_lock(name='test_lock_1', timeout=1)
# Cleanup
lock_2.release()
class ModelLockTestCase(FileLockTestCase):
backend_string = 'lock_manager.backends.model_lock.ModelLock'