# Path: aredis/exceptions.py
# class LockError(RedisError, ValueError):
#     """Errors acquiring or releasing a lock"""
#     # NOTE: For backwards compatability, this class derives from ValueError.
#     # This was originally chosen to behave like threading.Lock.
#     pass
# 
# Path: aredis/lock.py
# class ClusterLock(LuaLock):
#     """
#     Cluster lock is supposed to solve lock problem in redis cluster.
#     Since high availability is provided by redis cluster using master-slave model,
#     the kind of lock aims to solve the fail-over problem referred in distributed lock
#     post given by redis official.
# 
#     Why not use Redlock algorithm provided by official directly?
#     It is impossible to make a key hashed to different nodes
#     in a redis cluster and hard to generate keys
#     in a specific rule and make sure they do not migrated in cluster.
#     In the worst situation, all key slots may exists in one node.
#     Then the availability will be the same as one key in one node.
#     For more discussion please see:
#     https://github.com/NoneGG/aredis/issues/55
# 
#     My solution is to take advantage of `READONLY` mode of slaves to ensure
#     the lock key is synced from master to N/2 + 1 of its slaves to avoid the fail-over problem.
#     Since it is a single-key solution, the migration problem also do no matter.
# 
#     Please read these article below before using this cluster lock in your app.
#     https://redis.io/topics/distlock
#     http://martin.kleppmann.com/2016/02/08/how-to-do-distributed-locking.html
#     http://antirez.com/news/101
#     """
# 
#     def __init__(self, *args, **kwargs):
#         super(ClusterLock, self).__init__(*args, **kwargs)
#         if not self.timeout:
#             raise LockError('timeout must be provided for cluster lock')
# 
#     async def check_lock_in_slaves(self, token):
#         node_manager = self.redis.connection_pool.nodes
#         slot = node_manager.keyslot(self.name)
#         master_node_id = node_manager.node_from_slot(slot)['node_id']
#         slave_nodes = await self.redis.cluster_slaves(master_node_id)
#         count, quorum = 0, (len(slave_nodes) // 2) + 1
#         conn_kwargs = self.redis.connection_pool.connection_kwargs
#         conn_kwargs['readonly'] = True
#         for node in slave_nodes:
#             try:
#                 # todo: a little bit dirty here, try to reuse StrictRedis later
#                 # todo: it may be optimized by using a new connection pool
#                 conn = ClusterConnection(host=node['host'], port=node['port'], **conn_kwargs)
#                 await conn.send_command('get', self.name)
#                 res = await conn.read_response()
#                 if res == token:
#                     count += 1
#                 if count >= quorum:
#                     return True
#             except Exception as exc:
#                 warnings.warn('error {} during check lock {} status in slave nodes'.format(exc, self.name))
#         return False
# 
#     async def acquire(self, blocking=None, blocking_timeout=None):
#         """
#         Use Redis to hold a shared, distributed lock named ``name``.
#         Returns True once the lock is acquired.
# 
#         If ``blocking`` is False, always return immediately. If the lock
#         was acquired, return True, otherwise return False.
# 
#         ``blocking_timeout`` specifies the maximum number of seconds to
#         wait trying to acquire the lock. It should not be greater than
#         expire time of the lock
#         """
#         sleep = self.sleep
#         token = b(uuid.uuid1().hex)
#         if blocking is None:
#             blocking = self.blocking
#         if blocking_timeout is None:
#             blocking_timeout = self.blocking_timeout
#         blocking_timeout = blocking_timeout or self.timeout
#         stop_trying_at = mod_time.time() + min(blocking_timeout, self.timeout)
# 
#         while True:
#             if await self.do_acquire(token):
#                 lock_acquired_at = mod_time.time()
#                 if await self.check_lock_in_slaves(token):
#                     check_finished_at = mod_time.time()
#                     # if time expends on acquiring lock is greater than given time
#                     # the lock should be released manually
#                     if check_finished_at > stop_trying_at:
#                         await self.do_release(token)
#                         return False
#                     self.local.set(token)
#                     # validity time is considered to be the
#                     # initial validity time minus the time elapsed during check
#                     await self.do_extend(lock_acquired_at - check_finished_at)
#                     return True
#                 else:
#                     await self.do_release(token)
#                     return False
#             if not blocking or mod_time.time() > stop_trying_at:
#                 return False
#             await asyncio.sleep(sleep, loop=self.redis.connection_pool.loop)
# 
#     async def do_release(self, expected_token):
#         await super(ClusterLock, self).do_release(expected_token)
#         if await self.check_lock_in_slaves(expected_token):
#             raise LockError('Lock is released in master but not in slave yet')


import pytest
import time
from aredis.exceptions import LockError
from aredis.lock import ClusterLock
        assert 8 < await r.pttl('foo') <= 9500
        await lock.release()

    @pytest.mark.asyncio()
    async def test_blocking_timeout(self, r):
        lock1 = self.get_lock(r, 'foo', timeout=3)
        assert await lock1.acquire(blocking=False)
        lock2 = self.get_lock(r, 'foo', timeout=3, blocking_timeout=0.2)
        start = time.time()
        assert not await lock2.acquire()
        assert (time.time() - start) > 0.2
        await lock1.release()

    @pytest.mark.asyncio()
    async def test_context_manager(self, r):
        # blocking_timeout prevents a deadlock if the lock can't be acquired
        # for some reason
        async with self.get_lock(r, 'foo', timeout=3, blocking_timeout=0.2) as lock:
            assert await r.get('foo') == lock.local.get()
        assert await r.get('foo') is None

    @pytest.mark.asyncio()
    async 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(r, 'foo', timeout=1, sleep=2)

    @pytest.mark.asyncio()
    async def test_releasing_unlocked_lock_raises_error(self, r):
        lock = self.get_lock(r, 'foo', timeout=3)

# Path: aredis/exceptions.py
# class LockError(RedisError, ValueError):
#     """Errors acquiring or releasing a lock"""
#     # NOTE: For backwards compatability, this class derives from ValueError.
#     # This was originally chosen to behave like threading.Lock.
#     pass
# 
# Path: aredis/lock.py
# class ClusterLock(LuaLock):
#     """
#     Cluster lock is supposed to solve lock problem in redis cluster.
#     Since high availability is provided by redis cluster using master-slave model,
#     the kind of lock aims to solve the fail-over problem referred in distributed lock
#     post given by redis official.
# 
#     Why not use Redlock algorithm provided by official directly?
#     It is impossible to make a key hashed to different nodes
#     in a redis cluster and hard to generate keys
#     in a specific rule and make sure they do not migrated in cluster.
#     In the worst situation, all key slots may exists in one node.
#     Then the availability will be the same as one key in one node.
#     For more discussion please see:
#     https://github.com/NoneGG/aredis/issues/55
# 
#     My solution is to take advantage of `READONLY` mode of slaves to ensure
#     the lock key is synced from master to N/2 + 1 of its slaves to avoid the fail-over problem.
#     Since it is a single-key solution, the migration problem also do no matter.
# 
#     Please read these article below before using this cluster lock in your app.
#     https://redis.io/topics/distlock
#     http://martin.kleppmann.com/2016/02/08/how-to-do-distributed-locking.html
#     http://antirez.com/news/101
#     """
# 
#     def __init__(self, *args, **kwargs):
#         super(ClusterLock, self).__init__(*args, **kwargs)
#         if not self.timeout:
#             raise LockError('timeout must be provided for cluster lock')
# 
#     async def check_lock_in_slaves(self, token):
#         node_manager = self.redis.connection_pool.nodes
#         slot = node_manager.keyslot(self.name)
#         master_node_id = node_manager.node_from_slot(slot)['node_id']
#         slave_nodes = await self.redis.cluster_slaves(master_node_id)
#         count, quorum = 0, (len(slave_nodes) // 2) + 1
#         conn_kwargs = self.redis.connection_pool.connection_kwargs
#         conn_kwargs['readonly'] = True
#         for node in slave_nodes:
#             try:
#                 # todo: a little bit dirty here, try to reuse StrictRedis later
#                 # todo: it may be optimized by using a new connection pool
#                 conn = ClusterConnection(host=node['host'], port=node['port'], **conn_kwargs)
#                 await conn.send_command('get', self.name)
#                 res = await conn.read_response()
#                 if res == token:
#                     count += 1
#                 if count >= quorum:
#                     return True
#             except Exception as exc:
#                 warnings.warn('error {} during check lock {} status in slave nodes'.format(exc, self.name))
#         return False
# 
#     async def acquire(self, blocking=None, blocking_timeout=None):
#         """
#         Use Redis to hold a shared, distributed lock named ``name``.
#         Returns True once the lock is acquired.
# 
#         If ``blocking`` is False, always return immediately. If the lock
#         was acquired, return True, otherwise return False.
# 
#         ``blocking_timeout`` specifies the maximum number of seconds to
#         wait trying to acquire the lock. It should not be greater than
#         expire time of the lock
#         """
#         sleep = self.sleep
#         token = b(uuid.uuid1().hex)
#         if blocking is None:
#             blocking = self.blocking
#         if blocking_timeout is None:
#             blocking_timeout = self.blocking_timeout
#         blocking_timeout = blocking_timeout or self.timeout
#         stop_trying_at = mod_time.time() + min(blocking_timeout, self.timeout)
# 
#         while True:
#             if await self.do_acquire(token):
#                 lock_acquired_at = mod_time.time()
#                 if await self.check_lock_in_slaves(token):
#                     check_finished_at = mod_time.time()
#                     # if time expends on acquiring lock is greater than given time
#                     # the lock should be released manually
#                     if check_finished_at > stop_trying_at:
#                         await self.do_release(token)
#                         return False
#                     self.local.set(token)
#                     # validity time is considered to be the
#                     # initial validity time minus the time elapsed during check
#                     await self.do_extend(lock_acquired_at - check_finished_at)
#                     return True
#                 else:
#                     await self.do_release(token)
#                     return False
#             if not blocking or mod_time.time() > stop_trying_at:
#                 return False
#             await asyncio.sleep(sleep, loop=self.redis.connection_pool.loop)
# 
#     async def do_release(self, expected_token):
#         await super(ClusterLock, self).do_release(expected_token)
#         if await self.check_lock_in_slaves(expected_token):
#             raise LockError('Lock is released in master but not in slave yet')


# Path: tests/cluster/test_lock.py
import pytest
import time
from aredis.exceptions import LockError
from aredis.lock import ClusterLock

        assert 8 < await r.pttl('foo') <= 9500
        await lock.release()

    @pytest.mark.asyncio()
    async def test_blocking_timeout(self, r):
        lock1 = self.get_lock(r, 'foo', timeout=3)
        assert await lock1.acquire(blocking=False)
        lock2 = self.get_lock(r, 'foo', timeout=3, blocking_timeout=0.2)
        start = time.time()
        assert not await lock2.acquire()
        assert (time.time() - start) > 0.2
        await lock1.release()

    @pytest.mark.asyncio()
    async def test_context_manager(self, r):
        # blocking_timeout prevents a deadlock if the lock can't be acquired
        # for some reason
        async with self.get_lock(r, 'foo', timeout=3, blocking_timeout=0.2) as lock:
            assert await r.get('foo') == lock.local.get()
        assert await r.get('foo') is None

    @pytest.mark.asyncio()
    async 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(r, 'foo', timeout=1, sleep=2)

    @pytest.mark.asyncio()
    async def test_releasing_unlocked_lock_raises_error(self, r):
        lock = self.get_lock(r, 'foo', timeout=3)

        with pytest.raises(LockError):