How to use the hiredis.ReplyError function in hiredis

def test_execute_ReplyError_to_many_retries(self):
        self.client._get_slot_info = Mock()
        self.client._get_slot_info.side_effect = [
            'host1_12345',
            'host2_12345',
            'host3_12345',
            'host4_12345'
        ]
        conn1 = Mock()
        conn1.read.side_effect = [ReplyError('MOVED 42 host2:12345')]
        conn2 = Mock()
        conn2.read.side_effect = [ReplyError('MOVED 42 host2:12345')]
        conn3 = Mock()
        conn3.read.side_effect = [ReplyError('MOVED 42 host2:12345')]
        conn4 = Mock()
        conn4.read.side_effect = [ReplyError('MOVED 42 host2:12345')]
        self.connection_mock.side_effect = [conn1, conn2, conn3, conn4]

        self.client._cleanup_conns = Mock()

        self.assertRaises(PyRedisError, self.client.execute, 'GET', 'test', shard_key='test')

        conn1.write.assert_called_with('GET', 'test')
        conn2.write.assert_called_with('GET', 'test')
        conn3.write.assert_called_with('GET', 'test')
        self.assertFalse(conn4.write.called)

def test_execute_ReplyError_ASK(self):
        self.client._get_slot_info = Mock()
        self.client._get_slot_info.return_value = 'host1_12345'
        conn1 = Mock()
        conn1.read.side_effect = [ReplyError('ASK 42 host2:12345')]
        conn2 = Mock()
        conn2.read.side_effect = ['success']
        self.connection_mock.side_effect = [conn1, conn2]

        result = self.client.execute('GET', 'test', shard_key='test')
        self.assertEqual(result, 'success')
        conn1.write.assert_called_with('GET', 'test')
        conn2.write.assert_called_with('ASKING', 'GET', 'test')

def test_execute_ReplyError_MOVED(self):
        self.client._get_slot_info = Mock()
        self.client._get_slot_info.side_effect = ['host1_12345', 'host2_12345']
        conn1 = Mock()
        conn1.read.side_effect = [ReplyError('MOVED 42 host2:12345')]
        conn2 = Mock()
        conn2.read.side_effect = ['success']
        self.connection_mock.side_effect = [conn1, conn2]

        self.client._cleanup_conns = Mock()

        id = self.client._map_id

        result = self.client.execute('GET', 'test', shard_key='test')
        self.assertEqual(result, 'success')
        conn1.write.assert_called_with('GET', 'test')
        conn2.write.assert_called_with('GET', 'test')
        self.assertTrue(self.client._cleanup_conns.called)
        self.clustermap_inst.update.assert_called_with(id)

def test_error_string(self):
    self.reader.feed(b"-error\r\n")
    error = self.reply()

    self.assertEquals(hiredis.ReplyError, type(error))
    self.assertEquals(("error",), error.args)

def execute(self, task_map):
        if self.start_time is None:
            self.start_time = datetime.now()
            db.session.add(self)
            db.session.commit()

        try:
            if task_map[self.command](self, **self.args):
                self.complete(None)
            return True
        except (ValueError, LookupError, IOError, SocketError, HiredisError,
                ProtocolError, ReplyError, RedisStatusError):
            self.complete(traceback.format_exc())
            return False

def _info_slots(t):
    try:
        for line in t.talk_raw(CMD_CLUSTER_NODES).split('\n'):
            if len(line) == 0 or 'fail' in line or 'myself' not in line:
                continue
            node = ClusterNode(*line.split(' '))
            return {
                'node_id': node.node_id,
                'slave': node.role_in_cluster != 'master',
                'master_id': node.master_id if node.master_id != '-' else None,
                'slots': node.assigned_slots,
                'slots_migrating': node.slots_migrating,
            }
    except (ValueError, LookupError, IOError, ReplyError):
        return {
            'node_id': None,
            'slave': False,
            'master_id': None,
            'slots': [],
        }

'Got %s' % m
            ]))

    @retry(stop_max_attempt_number=16, wait_fixed=100)
    def setslot_stable(conn, slot, node_id):
        m = conn.execute('cluster', 'setslot', slot, 'node', node_id)
        expect_exec_ok(m, conn, slot)

    source_conn = source_node.get_conn()
    target_conn = target_node.get_conn()

    try:
        expect_exec_ok(
            target_conn.execute('cluster', 'setslot', slot, 'importing',
                                source_node.node_id), target_conn, slot)
    except hiredis.ReplyError as e:
        if 'already the owner of' not in str(e):
            target_conn.raise_(str(e))

    try:
        expect_exec_ok(
            source_conn.execute('cluster', 'setslot', slot, 'migrating',
                                target_node.node_id), source_conn, slot)
    except hiredis.ReplyError as e:
        if 'not the owner of' not in str(e):
            source_conn.raise_(str(e))

    keys = _migr_keys(source_conn, target_node.host, target_node.port, slot)
    setslot_stable(source_conn, slot, target_node.node_id)
    for node in nodes:
        if node.master:
            setslot_stable(node.get_conn(), slot, target_node.node_id)

def send_raw(self, command, recv=None):
        recv = recv or self._recv
        for c in command:
            self.sock.send(c)
        r = recv()
        if r is None:
            raise ValueError('No reply')
        if isinstance(r, hiredis.ReplyError):
            raise r

        if isinstance(r, list):
            return [i.decode(ENCODING) for i in r]
        return r.decode(ENCODING)

def _read(self, command, future):
        """Invoked when a command is executed to read and parse its results.
        It will loop on the IOLoop until the response is complete and then
        set the value of the response in the execution future.

        :param command: The command that was being executed
        :type command: tredis.client.Command
        :param future: The execution future
        :type future: tornado.concurrent.Future

        """
        response = self._reader.gets()
        if response is not False:
            if isinstance(response, hiredis.ReplyError):
                if response.args[0].startswith('MOVED '):
                    self._on_cluster_data_moved(response.args[0], command,
                                                future)
                elif response.args[0].startswith('READONLY '):
                    self._on_read_only_error(command, future)
                else:
                    future.set_exception(exceptions.RedisError(response))
            elif command.callback is not None:
                future.set_result(command.callback(response))
            elif command.expectation is not None:
                self._eval_expectation(command, response, future)
            else:
                future.set_result(response)
        else:

            def on_data(data):

def collect_stats(self):
        try:
            self._collect_stats()
            self.app.stats_write(self.addr, self.stats_data())
        except (IOError, ValueError, LookupError, ReplyError) as e:
            logging.exception(e)
            self.set_unavailable()
            self.send_alarm(
                '%s failed: %s:%d - %s' % (
                    self.typename, self.host, self.port, e), e)