def test_token(redis_server):
    conn = StrictRedis(unix_socket_path=UDS_PATH)
    lock = Lock(conn, "foobar-tok")
    tok = lock.id
    assert conn.get(lock._name) is None
    lock.acquire(blocking=False)
    assert conn.get(lock._name) == tok

def test_reset(redis_server):
    conn = StrictRedis(unix_socket_path=UDS_PATH)
    with Lock(conn, "foobar") as lock:
        lock.reset()
        new_lock = Lock(conn, "foobar")
        new_lock.acquire(blocking=False)
        new_lock.release()

def test_signal_cleanup_on_reset_all(conn):
    """After resetting all locks, no signal keys should not remain."""
    lock = Lock(conn, 'foo')
    lock.acquire()
    reset_all(conn)
    assert conn.llen('lock-signal:foo') == 0
    assert conn.exists('lock-signal:foo') == 0

def test_signal_cleanup_on_reset(conn):
    """After resetting a lock, the signal key should not remain."""
    lock = Lock(conn, 'foo')
    lock.acquire()
    lock.reset()
    assert conn.llen('lock-signal:foo') == 0
    assert conn.exists('lock-signal:foo') == 0

def test_signal_expiration(conn):
    """Signal keys expire within two seconds after releasing the lock."""
    lock = Lock(conn, 'signal_expiration')
    lock.acquire()
    lock.release()
    time.sleep(2)
    assert conn.llen('lock-signal:signal_expiration') == 0

def continuous_migration(skip_files=None):
    """Task to continuously migrate what is pushed up by Legacy."""
    if skip_files is None:
        skip_files = current_app.config.get(
            'RECORDS_MIGRATION_SKIP_FILES',
            False,
        )
    redis_url = current_app.config.get('CACHE_REDIS_URL')
    r = StrictRedis.from_url(redis_url)
    lock = Lock(r, 'continuous_migration', expire=120, auto_renewal=True)
    if lock.acquire(blocking=False):
        try:
            while r.llen('legacy_records'):
                raw_record = r.lrange('legacy_records', 0, 0)
                if raw_record:
                    migrate_and_insert_record(
                        zlib.decompress(raw_record[0]),
                        skip_files=skip_files,
                    )
                    db.session.commit()
                r.lpop('legacy_records')
        finally:
            lock.release()
    else:
        LOGGER.info("Continuous_migration already executed. Skipping.")

    def _execute_task_group(self, queue, tasks, all_task_ids):
        """
        Executes the given tasks in the queue. Updates the heartbeat for task
        IDs passed in all_task_ids. This internal method is only meant to be
        called from within _process_from_queue.
        """
        log = self.log.bind(queue=queue)

        locks = []
        # Keep track of the acquired locks: If two tasks in the list require
        # the same lock we only acquire it once.
        lock_ids = set()

        ready_tasks = []
        for task in tasks:
            if task.get('lock', False):
                if task.get('lock_key'):
                    kwargs = task.get('kwargs', {})
                    lock_id = gen_unique_id(
                        task['func'],
                        None,
                        {key: kwargs.get(key) for key in task['lock_key']},
                    )
                else:
                    lock_id = gen_unique_id(
                        task['func'],
                        task.get('args', []),
                        task.get('kwargs', {}),
                    )

                if lock_id not in lock_ids:
                    lock = Lock(self.connection, self._key('lock', lock_id), timeout=self.config['ACTIVE_TASK_UPDATE_TIMEOUT'])

                    acquired = lock.acquire(blocking=False)
                    if acquired:
                        lock_ids.add(lock_id)
                        locks.append(lock)
                    else:
                        log.info('could not acquire lock', task_id=task['id'])

                        # Reschedule the task
                        when = time.time() + self.config['LOCK_RETRY']
                        self._redis_move_task(queue, task['id'], ACTIVE, SCHEDULED, when)
                        # Make sure to remove it from this list so we don't re-add
                        # to the ACTIVE queue by updating the heartbeat.
                        all_task_ids.remove(task['id'])
                        continue

            ready_tasks.append(task)

        if not ready_tasks:
            return True, []

        success = self._execute(queue, ready_tasks, log, locks, all_task_ids)

        for lock in locks:
            lock.release()

        return success, ready_tasks

def lock_redis(app):
    redis_url = app.config.get('CACHE_REDIS_URL')
    redis = StrictRedis.from_url(redis_url)
    lock = Lock(redis, 'my_lock', expire=60)
    lock.acquire(blocking=False)

    yield

    lock.release()

    def workerfn(go, count_lock, count):
        redis_lock = Lock(make_conn(), 'lock')
        with count_lock:
            count.value += 1

        go.wait()

        if redis_lock.acquire(blocking=True):
            with count_lock:
                count.value += 1

def test_timeout_acquired(conn):
    with TestProcess(sys.executable, HELPER, 'test_timeout') as proc:
        with dump_on_error(proc.read):
            name = 'lock:foobar'
            wait_for_strings(
                proc.read, TIMEOUT,
                'Getting %r ...' % name,
                'Got lock for %r.' % name,
            )
            lock = Lock(conn, "foobar")
            assert lock.acquire(timeout=2)

 def wrapper(*args, **kwargs):
     client = get_redis_connection()
     lock = Lock(client, lock_name, expire=expire)
     try:
         if lock.acquire(blocking=False):
             return func(*args, **kwargs)
         else:
             logger.warning('another instance of %s is running',
                     func_path)
     finally:
         lock.release()

def test_auto_renewal(conn):
    lock = Lock(conn, 'lock_renewal', expire=3, auto_renewal=True)
    lock.acquire()

    assert isinstance(lock._lock_renewal_thread, InterruptableThread)
    assert not lock._lock_renewal_thread.should_exit
    assert lock._lock_renewal_interval == 2

    time.sleep(3)
    assert conn.get(lock._name) == lock.id, "Key expired but it should have been getting renewed"

    lock.release()
    assert lock._lock_renewal_thread is None

def test_auto_renewal(conn):
    lock = Lock(conn, 'lock_renewal', expire=3, auto_renewal=True)
    lock.acquire()

    assert isinstance(lock._lock_renewal_thread, threading.Thread)
    assert not lock._lock_renewal_stop.is_set()
    assert isinstance(lock._lock_renewal_interval, float)
    assert lock._lock_renewal_interval == 2

    time.sleep(3)
    assert maybe_decode(conn.get(lock._name)) == lock.id, "Key expired but it should have been getting renewed"

    lock.release()
    assert lock._lock_renewal_thread is None

def test_expire_without_timeout(conn):
    first_lock = Lock(conn, 'expire', expire=2)
    second_lock = Lock(conn, 'expire', expire=1)
    first_lock.acquire()
    assert second_lock.acquire(blocking=False) is False
    assert second_lock.acquire() is True
    second_lock.release()

def test_reset(conn):
    lock = Lock(conn, "foobar")
    lock.reset()
    new_lock = Lock(conn, "foobar")
    new_lock.acquire(blocking=False)
    new_lock.release()
    pytest.raises(NotAcquired, lock.release)

def redis_locking_context(lock_name, expire=120, auto_renewal=True):
    """Locked Context Manager to perform operations on Redis."""
    if not lock_name:
        raise RedisLockError('Lock name not specified.')

    redis_url = app.config.get('CACHE_REDIS_URL')

    redis = StrictRedis.from_url(redis_url)
    lock = Lock(redis, lock_name, expire=expire, auto_renewal=auto_renewal)

    if lock.acquire(blocking=False):
        try:
            yield redis
        finally:
            lock.release()
    else:
        raise RedisLockError('Can not acquire Redis lock for %s', lock_name)

 def test_expire(self):
     conn = StrictRedis(unix_socket_path=UDS_PATH)
     with Lock(conn, "foobar", expire=TIMEOUT/4):
         with TestProcess(sys.executable, __file__, 'daemon', 'test_expire') as proc:
             with self.dump_on_error(proc.read):
                 name = 'lock:foobar'
                 self.wait_for_strings(proc.read, TIMEOUT,
                     'Getting %r ...' % name,
                     'Got lock for %r.' % name,
                     'Releasing %r.' % name,
                     'UNLOCK_SCRIPT not cached.',
                     'DIED.',
                 )
     lock = Lock(conn, "foobar")
     try:
         self.assertEqual(lock.acquire(blocking=False), True)
     finally:
         lock.release()

def legacy_orcid_arrays():
    """Generator to fetch token data from redis.

    Yields:
        list: user data in the form of [orcid, token, email, name]
    """
    redis_url = current_app.config.get('CACHE_REDIS_URL')
    r = StrictRedis.from_url(redis_url)
    lock = Lock(r, 'import_legacy_orcid_tokens', expire=120, auto_renewal=True)
    if lock.acquire(blocking=False):
        try:
            while r.llen('legacy_orcid_tokens'):
                yield loads(r.lrange('legacy_orcid_tokens', 0, 1)[0])
                r.lpop('legacy_orcid_tokens')
        finally:
            lock.release()
    else:
        logger.info("Import_legacy_orcid_tokens already executed. Skipping.")

def test_expire(conn):
    with Lock(conn, "foobar", expire=TIMEOUT/4):
        with TestProcess(sys.executable, HELPER, 'test_expire') as proc:
            with dump_on_error(proc.read):
                name = 'lock:foobar'
                wait_for_strings(
                    proc.read, TIMEOUT,
                    'Getting %r ...' % name,
                    'Got lock for %r.' % name,
                    'Releasing %r.' % name,
                    'UNLOCK_SCRIPT not cached.',
                    'DIED.',
                )
    lock = Lock(conn, "foobar")
    try:
        assert lock.acquire(blocking=False) == True
    finally:
        lock.release()

def test_expire(conn):
    with Lock(conn, "foobar", expire=TIMEOUT / 4):
        with TestProcess(sys.executable, HELPER, "test_expire") as proc:
            with dump_on_error(proc.read):
                name = "lock:foobar"
                wait_for_strings(
                    proc.read,
                    TIMEOUT,
                    "Getting %r ..." % name,
                    "Got lock for %r." % name,
                    "Releasing %r." % name,
                    "UNLOCK_SCRIPT not cached.",
                    "DIED.",
                )
    lock = Lock(conn, "foobar")
    try:
        assert lock.acquire(blocking=False) == True
    finally:
        lock.release()