summaryrefslogtreecommitdiff
path: root/tests/test_lock.py
diff options
context:
space:
mode:
Diffstat (limited to 'tests/test_lock.py')
-rw-r--r--tests/test_lock.py124
1 files changed, 62 insertions, 62 deletions
diff --git a/tests/test_lock.py b/tests/test_lock.py
index b4f9a5d..9e0e2fe 100644
--- a/tests/test_lock.py
+++ b/tests/test_lock.py
@@ -12,17 +12,17 @@ class TestLock(object):
kwargs['lock_class'] = self.lock_class
return redis.lock(*args, **kwargs)
- def test_lock(self, sr):
- lock = self.get_lock(sr, 'foo')
+ def test_lock(self, r):
+ lock = self.get_lock(r, 'foo')
assert lock.acquire(blocking=False)
- assert sr.get('foo') == lock.local.token
- assert sr.ttl('foo') == -1
+ assert r.get('foo') == lock.local.token
+ assert r.ttl('foo') == -1
lock.release()
- assert sr.get('foo') is None
+ assert r.get('foo') is None
- def test_competing_locks(self, sr):
- lock1 = self.get_lock(sr, 'foo')
- lock2 = self.get_lock(sr, 'foo')
+ def test_competing_locks(self, r):
+ lock1 = self.get_lock(r, 'foo')
+ lock2 = self.get_lock(r, 'foo')
assert lock1.acquire(blocking=False)
assert not lock2.acquire(blocking=False)
lock1.release()
@@ -30,86 +30,86 @@ class TestLock(object):
assert not lock1.acquire(blocking=False)
lock2.release()
- def test_timeout(self, sr):
- lock = self.get_lock(sr, 'foo', timeout=10)
+ def test_timeout(self, r):
+ lock = self.get_lock(r, 'foo', timeout=10)
assert lock.acquire(blocking=False)
- assert 8 < sr.ttl('foo') <= 10
+ assert 8 < r.ttl('foo') <= 10
lock.release()
- def test_float_timeout(self, sr):
- lock = self.get_lock(sr, 'foo', timeout=9.5)
+ def test_float_timeout(self, r):
+ lock = self.get_lock(r, 'foo', timeout=9.5)
assert lock.acquire(blocking=False)
- assert 8 < sr.pttl('foo') <= 9500
+ assert 8 < r.pttl('foo') <= 9500
lock.release()
- def test_blocking_timeout(self, sr):
- lock1 = self.get_lock(sr, 'foo')
+ def test_blocking_timeout(self, r):
+ lock1 = self.get_lock(r, 'foo')
assert lock1.acquire(blocking=False)
- lock2 = self.get_lock(sr, 'foo', blocking_timeout=0.2)
+ lock2 = self.get_lock(r, 'foo', blocking_timeout=0.2)
start = time.time()
assert not lock2.acquire()
assert (time.time() - start) > 0.2
lock1.release()
- def test_context_manager(self, sr):
+ def test_context_manager(self, r):
# blocking_timeout prevents a deadlock if the lock can't be acquired
# for some reason
- with self.get_lock(sr, 'foo', blocking_timeout=0.2) as lock:
- assert sr.get('foo') == lock.local.token
- assert sr.get('foo') is None
+ with self.get_lock(r, 'foo', blocking_timeout=0.2) as lock:
+ assert r.get('foo') == lock.local.token
+ assert r.get('foo') is None
- def test_high_sleep_raises_error(self, sr):
+ def test_high_sleep_raises_error(self, r):
"If sleep is higher than timeout, it should raise an error"
with pytest.raises(LockError):
- self.get_lock(sr, 'foo', timeout=1, sleep=2)
+ self.get_lock(r, 'foo', timeout=1, sleep=2)
- def test_releasing_unlocked_lock_raises_error(self, sr):
- lock = self.get_lock(sr, 'foo')
+ def test_releasing_unlocked_lock_raises_error(self, r):
+ lock = self.get_lock(r, 'foo')
with pytest.raises(LockError):
lock.release()
- def test_releasing_lock_no_longer_owned_raises_error(self, sr):
- lock = self.get_lock(sr, 'foo')
+ def test_releasing_lock_no_longer_owned_raises_error(self, r):
+ lock = self.get_lock(r, 'foo')
lock.acquire(blocking=False)
# manually change the token
- sr.set('foo', 'a')
+ r.set('foo', 'a')
with pytest.raises(LockError):
lock.release()
# even though we errored, the token is still cleared
assert lock.local.token is None
- def test_extend_lock(self, sr):
- lock = self.get_lock(sr, 'foo', timeout=10)
+ def test_extend_lock(self, r):
+ lock = self.get_lock(r, 'foo', timeout=10)
assert lock.acquire(blocking=False)
- assert 8000 < sr.pttl('foo') <= 10000
+ assert 8000 < r.pttl('foo') <= 10000
assert lock.extend(10)
- assert 16000 < sr.pttl('foo') <= 20000
+ assert 16000 < r.pttl('foo') <= 20000
lock.release()
- def test_extend_lock_float(self, sr):
- lock = self.get_lock(sr, 'foo', timeout=10.0)
+ def test_extend_lock_float(self, r):
+ lock = self.get_lock(r, 'foo', timeout=10.0)
assert lock.acquire(blocking=False)
- assert 8000 < sr.pttl('foo') <= 10000
+ assert 8000 < r.pttl('foo') <= 10000
assert lock.extend(10.0)
- assert 16000 < sr.pttl('foo') <= 20000
+ assert 16000 < r.pttl('foo') <= 20000
lock.release()
- def test_extending_unlocked_lock_raises_error(self, sr):
- lock = self.get_lock(sr, 'foo', timeout=10)
+ def test_extending_unlocked_lock_raises_error(self, r):
+ lock = self.get_lock(r, 'foo', timeout=10)
with pytest.raises(LockError):
lock.extend(10)
- def test_extending_lock_with_no_timeout_raises_error(self, sr):
- lock = self.get_lock(sr, 'foo')
+ def test_extending_lock_with_no_timeout_raises_error(self, r):
+ lock = self.get_lock(r, 'foo')
assert lock.acquire(blocking=False)
with pytest.raises(LockError):
lock.extend(10)
lock.release()
- def test_extending_lock_no_longer_owned_raises_error(self, sr):
- lock = self.get_lock(sr, 'foo')
+ def test_extending_lock_no_longer_owned_raises_error(self, r):
+ lock = self.get_lock(r, 'foo')
assert lock.acquire(blocking=False)
- sr.set('foo', 'a')
+ r.set('foo', 'a')
with pytest.raises(LockError):
lock.extend(10)
@@ -119,48 +119,48 @@ class TestLuaLock(TestLock):
class TestLockClassSelection(object):
- def test_lock_class_argument(self, sr):
- lock = sr.lock('foo', lock_class=Lock)
+ def test_lock_class_argument(self, r):
+ lock = r.lock('foo', lock_class=Lock)
assert type(lock) == Lock
- lock = sr.lock('foo', lock_class=LuaLock)
+ lock = r.lock('foo', lock_class=LuaLock)
assert type(lock) == LuaLock
- def test_cached_lualock_flag(self, sr):
+ def test_cached_lualock_flag(self, r):
try:
- sr._use_lua_lock = True
- lock = sr.lock('foo')
+ r._use_lua_lock = True
+ lock = r.lock('foo')
assert type(lock) == LuaLock
finally:
- sr._use_lua_lock = None
+ r._use_lua_lock = None
- def test_cached_lock_flag(self, sr):
+ def test_cached_lock_flag(self, r):
try:
- sr._use_lua_lock = False
- lock = sr.lock('foo')
+ r._use_lua_lock = False
+ lock = r.lock('foo')
assert type(lock) == Lock
finally:
- sr._use_lua_lock = None
+ r._use_lua_lock = None
- def test_lua_compatible_server(self, sr, monkeypatch):
+ def test_lua_compatible_server(self, r, monkeypatch):
@classmethod
def mock_register(cls, redis):
return
monkeypatch.setattr(LuaLock, 'register_scripts', mock_register)
try:
- lock = sr.lock('foo')
+ lock = r.lock('foo')
assert type(lock) == LuaLock
- assert sr._use_lua_lock is True
+ assert r._use_lua_lock is True
finally:
- sr._use_lua_lock = None
+ r._use_lua_lock = None
- def test_lua_unavailable(self, sr, monkeypatch):
+ def test_lua_unavailable(self, r, monkeypatch):
@classmethod
def mock_register(cls, redis):
raise ResponseError()
monkeypatch.setattr(LuaLock, 'register_scripts', mock_register)
try:
- lock = sr.lock('foo')
+ lock = r.lock('foo')
assert type(lock) == Lock
- assert sr._use_lua_lock is False
+ assert r._use_lua_lock is False
finally:
- sr._use_lua_lock = None
+ r._use_lua_lock = None
View Lorry Controller Status