def test_vbucket_id_option(self):
        bucket = RestConnection(self.server_origin).get_bucket(self.buckets[0])
        self.num_items = self.num_items - (self.num_items % len(bucket.vbuckets))
        num_items_per_vb = self.num_items / len(bucket.vbuckets)
        template = '{{ "mutated" : 0, "age": {0}, "first_name": "{1}" }}'
        gen_load = DocumentGenerator('cbtransfer', template, range(5), ['james', 'john'], start=0, end=self.num_items)
        client = MemcachedClient(self.server_origin.ip,
                                 int(bucket.vbuckets[0].master.split(':')[1]))
        kv_value_dict = {}
        vb_id_to_check = bucket.vbuckets[-1].id
        for vb_id in xrange(len(bucket.vbuckets)):
            cur_items_per_vb = 0
            while cur_items_per_vb < num_items_per_vb:
                key, value = gen_load.next()

                client.set(key, 0, 0, value, vb_id)
                if vb_id_to_check == vb_id:
                    kv_value_dict[key] = value
                cur_items_per_vb += 1

        transfer_source = 'http://%s:%s' % (self.server_origin.ip, self.server_origin.port)
        transfer_destination = 'http://%s:%s' % (self.server_recovery.ip, self.server_recovery.port)
        output = self.shell.execute_cbtransfer(transfer_source, transfer_destination,
                                      "-b %s -B %s -i %s" % (bucket.name, bucket.name, vb_id_to_check))
        client = MemcachedClient(self.server_recovery.ip,
                                 int(bucket.vbuckets[0].master.split(':')[1]))
        for key, value in kv_value_dict.iteritems():
            _, _, d = client.get(key, vbucket=vb_id_to_check)
            self.assertEquals(d, value, 'Key: %s expected. Value expected %s. Value actual %s' % (
                                        key, value, d))

    def get_vBuckets_info(master):
        """
        return state and count items for all vbuckets for each node
        format: dict: {u'1node_ip1': {'vb_79': ['replica', '0'], 'vb_78': ['active', '0']..}, u'1node_ip1':....}
        """
        rest = RestConnection(master)
        port = rest.get_nodes_self().memcached
        nodes = rest.node_statuses()
        _nodes_stats= {}
        for node in nodes:
            stat={}
            buckets = []
            _server = {"ip": node.ip, "port": node.port, "username": master.rest_username,
                           "password": master.rest_password}
            try:
                buckets = rest.get_buckets()
                mc = MemcachedClient(node.ip, port)
                stat_hash = mc.stats("hash")
            except Exception:
                if not buckets:
                    log.error("There are not any buckets in {0}:{1} node".format(node.ip, node.port))
                else:
                    log.error("Impossible to get vBucket's information for {0}:{1} node".format(node.ip, node.port))
                    _nodes_stats[node.ip+":"+str(node.port)]
                continue
            mc.close()
            vb_names=[key[:key.index(":")] for key in stat_hash.keys()]

            for name in vb_names:
                stat[name]=[stat_hash[name + ":state"], stat_hash[name+":counted"]]
            _nodes_stats[node.ip+":"+str(port)] = stat
        log.info(_nodes_stats)
        return _nodes_stats

    def direct_client(server, bucket, timeout=30):
        log = logger.Logger.get_logger()
        rest = RestConnection(server)
        node = None
        try:
            node = rest.get_nodes_self()
        except ValueError as e:
            log.info("could not connect to server {0}, will try scanning all nodes".format(server))
        if not node:
            nodes = rest.get_nodes()
            for n in nodes:
                if n.ip == server.ip and n.port == server.port:
                    node = n

        if isinstance(server, dict):
            log.info("dict:{0}".format(server))
            log.info("creating direct client {0}:{1} {2}".format(server["ip"], node.memcached, bucket))
        else:
            log.info("creating direct client {0}:{1} {2}".format(server.ip, node.memcached, bucket))
        RestHelper(rest).vbucket_map_ready(bucket, 60)
        vBuckets = RestConnection(server).get_vbuckets(bucket)
        if isinstance(server, dict):
            client = MemcachedClient(server["ip"], node.memcached, timeout=timeout)
        else:
            client = MemcachedClient(server.ip, node.memcached, timeout=timeout)
        client.vbucket_count = len(vBuckets)
        bucket_info = rest.get_bucket(bucket)
        # todo raise exception for not bucket_info
        client.sasl_auth_plain(bucket_info.name.encode("ascii"), bucket_info.saslPassword.encode("ascii"))
        return client

 def flushctl_set(servers, key, val):
     log = logger.Logger.get_logger()
     for server in servers:
         c = MemcachedClient(server.ip, 11210)
         log.info("Setting flush param on server {0}, {1} to {2}".format(server, key, val))
         rv = c.set_flush_param(key, val)
         log.info("Setting flush param on server {0}, {1} to {2}, result: {3}".format(server, key, val, rv))
         c.close()

 def get_mb_stats(servers, key):
     log = logger.Logger.get_logger()
     for server in servers:
         c = MemcachedClient(server.ip, 11210)
         log.info("Get flush param on server {0}, {1}".format(server, key))
         value = c.stats().get(key, None)
         log.info("Get flush param on server {0}, {1}".format(server, value))
         c.close()

 def test_list_mechs(self):
     nodes = RestConnection(self.master).get_nodes()
     for n in nodes:
         if n.ip == self.master.ip and n.port == self.master.port:
             node = n
     client = MemcachedClient(self.master.ip, node.memcached)
     mechs = list(client.sasl_mechanisms())
     assert "CRAM-MD5" in mechs
     assert "PLAIN" in mechs
     assert len(list(mechs)) == 2

 def load_one_mutation_into_source_vb0(self, vb0_active_src_node):
     key = self.vb0_keys[self.key_counter]
     memc_client = MemcachedClient(vb0_active_src_node.ip, 11210)
     try:
         memc_client.set(key, exp=0, flags=0, val="dummy val")
         self.key_counter += 1
         self.keys_loaded.append(key)
         self.log.info("Loaded key {} onto vb0 in {}".format(key, vb0_active_src_node.ip))
         self.log.info ("deleted, flags, exp, rev_id, cas for key {} = {}".format(key, memc_client.getMeta(key)))
     except MemcachedError as e:
         self.log.error(e)

 def connection(self, client_ip, bucket_name, user,password, port=11210):
     log.info("Bucket name for connection is ---- {0}, username -- {1}, ----- password -- {2}".format(bucket_name,user, \
                                                                                                      password))
     try:
         mc = MemcachedClient(host=client_ip, port=port)
         mc.sasl_auth_plain(user,password)
         mc.bucket_select(bucket_name)
         return mc, True
     except Exception as e:
         log.info( "Exception is from connection function {0}".format(e))
         return False, False

 def test_list_mechs(self):
     nodes = RestConnection(self.master).get_nodes()
     for n in nodes:
         if n.ip == self.master.ip and n.port == self.master.port:
             node = n
     client = MemcachedClient(self.master.ip, node.memcached)
     mechs = list(client.sasl_mechanisms())
     self.log.info("Start check mech types")
     assert "SCRAM-SHA1" in mechs
     assert "SCRAM-SHA256" in mechs
     assert "SCRAM-SHA512" in mechs
     assert "PLAIN" in mechs
     assert len(list(mechs)) == 4

 def do_auth(self, bucket, password):
     """ default self.auth_mech is 'PLAIN' """
     self.log.info("Authenticate with {0} to {1}:{2}".format(self.auth_mech,
                                                             bucket,
                                                             password))
     ret = None
     nodes = RestConnection(self.master).get_nodes()
     for n in nodes:
         if n.ip == self.master.ip and n.port == self.master.port:
             node = n
     client = MemcachedClient(self.master.ip, node.memcached)
     try:
         if self.auth_mech == "PLAIN":
             ret = client.sasl_auth_plain(bucket, password)[2]
         else:
             self.fail("Invalid auth mechanism {0}".format(self.auth_mech))
     except MemcachedError, e:
         ret = e[0].split(' for vbucket')[0]

 def direct_client(server, bucket, timeout=30):
     rest = RestConnection(server)
     node = rest.get_nodes_self()
     log = logger.Logger.get_logger()
     if isinstance(server, dict):
         log.info("dict:{0}".format(server))
         log.info("creating direct client {0}:{1} {2}".format(server["ip"], node.memcached, bucket))
     else:
         log.info("creating direct client {0}:{1} {2}".format(server.ip, node.memcached, bucket))
     RestHelper(rest).vbucket_map_ready(bucket, 60)
     vBuckets = RestConnection(server).get_vbuckets(bucket)
     if isinstance(server, dict):
         client = MemcachedClient(server["ip"], node.memcached, timeout=timeout)
     else:
         client = MemcachedClient(server.ip, node.memcached, timeout=timeout)
     client.vbucket_count = len(vBuckets)
     bucket_info = rest.get_bucket(bucket)
     #todo raise exception for not bucket_info
     client.sasl_auth_plain(bucket_info.name.encode('ascii'),
                            bucket_info.saslPassword.encode('ascii'))
     return client

 def wait_for_mc_stats(master, bucket, stat_key, stat_value, timeout_in_seconds=120, verbose=True):
     log.info("waiting for bucket {0} stat : {1} to match {2} on {3}".format(bucket, stat_key, \
                                                                             stat_value, master.ip))
     start = time.time()
     verified = False
     while (time.time() - start) <= timeout_in_seconds:
         c = MemcachedClient(master.ip, 11210)
         stats = c.stats()
         c.close()
         if stats and stat_key in stats and str(stats[stat_key]) == str(stat_value):
             log.info("{0} : {1}".format(stat_key, stats[stat_key]))
             verified = True
             break
         else:
             if stats and stat_key in stats:
                 if verbose:
                     log.info("{0} : {1}".format(stat_key, stats[stat_key]))
             if not verbose:
                 time.sleep(0.1)
             else:
                 time.sleep(2)
     return verified

    def wait_for_mc_stats_no_timeout(master, bucket, stat_key, stat_value, timeout_in_seconds=-1, verbose=True):
        log.info("waiting for bucket {0} stat : {1} to match {2} on {3}".format(bucket, stat_key, \
                                                                                stat_value, master.ip))
        c = MemcachedClient(master.ip, 11210)
        stats = c.stats()
        c.close()

        while str(stats[stat_key]) != str(stat_value):
            c = MemcachedClient(master.ip, 11210)
            stats = c.stats()
            c.close()
            if verbose:
                log.info("{0} : {1}".format(stat_key, stats[stat_key]))
            time.sleep(5)
        return True

 def verify_revid(self):
     missing_keys = False
     src_node = self.get_active_vb0_node(self.src_master)
     dest_node = self.get_active_vb0_node(self.dest_master)
     src_client = MemcachedClient(src_node.ip, 11210)
     dest_client = MemcachedClient(dest_node.ip, 11210)
     for key in self.keys_loaded:
         try:
             src_meta = src_client.getMeta(key)
             dest_meta = dest_client.getMeta(key)
             self.log.info("deleted, flags, exp, rev_id, cas for key from Source({0}) {1} = {2}"
                            .format(src_node.ip, key, src_meta))
             self.log.info("deleted, flags, exp, rev_id, cas for key from Destination({0}) {1} = {2}"
                            .format(dest_node.ip, key, dest_meta))
             if src_meta == dest_meta:
                 self.log.info("RevID verification successful for key {}".format(key))
             else:
                 self.fail("RevID verification failed for key {}".format(key))
         except MemcachedError:
             self.log.error("Key {} is missing at destination".format(key))
             missing_keys = True
     if missing_keys:
         self.fail("Some keys are missing at destination")

def getDirectMC(key, ip, port = 8091, bucket = "default", password = ""):

    real_mc_client = None

    # get initial mc client
    client = MemcachedClient(ip, int(port))
    vbId = (((zlib.crc32(key)) >> 16) & 0x7fff) & (client.vbucket_count - 1)

    # get vbucket map
    rest = create_rest(ip, port)
    vbuckets = rest.get_vbuckets(bucket)

    # find vbucket responsible to this key and mapping host
    if vbuckets is not None:

        vbucket = [vbucket for vbucket in vbuckets if vbucket.id == vbId]

        if len(vbucket) == 1:
            mc_ip, mc_port = vbucket[0].master.split(":")
            real_mc_client = MemcachedClient(mc_ip, int(mc_port))
            real_mc_client.sasl_auth_plain(bucket, password)

    return real_mc_client

    def __init__(self,
                 host = cfg.COUCHBASE_IP,
                 port = 11210):
        self.host = host
        self.port = port
        self.client = None

        try:
            self.client = MemcachedClient(self.host, self.port)
        except MemcachedError as ex:
            msg = "error connecting to host %s for gathering postconditions"\
                   %  self.host
            logger.error(msg)
            pass

 def load_one_mutation_into_source_vb0(self, vb0_active_src_node):
     key = self.vb0_keys[self.key_counter]
     memc_client = MemcachedClient(vb0_active_src_node.ip, 11210)
     memc_client.sasl_auth_plain("cbadminbucket","password")
     memc_client.bucket_select("default")
     try:
         memc_client.set(key, exp=0, flags=0, val="dummy val")
         self.key_counter += 1
         self.keys_loaded.append(key)
         self.log.info("Loaded key {0} onto vb0 in {1}".format(key, vb0_active_src_node.ip))
         self.log.info ("deleted, flags, exp, rev_id, cas for key {0} = {1}".format(key, memc_client.getMeta(key)))
     except MemcachedError as e:
         self.log.error(e)

 def wait_for_vbuckets_ready_state(node, bucket, timeout_in_seconds=300, log_msg=''):
     log = logger.Logger.get_logger()
     start_time = time.time()
     end_time = start_time + timeout_in_seconds
     ready_vbuckets = {}
     rest = RestConnection(node)
     servers = rest.get_nodes()
     RestHelper(rest).vbucket_map_ready(bucket, 60)
     vbucket_count = len(rest.get_vbuckets(bucket))
     vbuckets = rest.get_vbuckets(bucket)
     obj = VBucketAwareMemcached(rest, bucket)
     memcacheds, vbucket_map, vbucket_map_replica = obj.request_map(rest, bucket)
     #Create dictionary with key:"ip:port" and value: a list of vbuckets
     server_dict = defaultdict(list)
     for everyID in range(0, vbucket_count):
         memcached_ip_port = str(vbucket_map[everyID])
         server_dict[memcached_ip_port].append(everyID)
     while time.time() < end_time and len(ready_vbuckets) < vbucket_count:
         for every_ip_port in server_dict:
             #Retrieve memcached ip and port
             ip, port = every_ip_port.split(":")
             client = MemcachedClient(ip, int(port), timeout=30)
             client.vbucket_count = len(vbuckets)
             bucket_info = rest.get_bucket(bucket)
             client.sasl_auth_plain(bucket_info.name.encode('ascii'),
                                 bucket_info.saslPassword.encode('ascii'))
             for i in server_dict[every_ip_port]:
                 try:
                     (a, b, c) = client.get_vbucket_state(i)
                 except mc_bin_client.MemcachedError as e:
                     ex_msg = str(e)
                     if "Not my vbucket" in log_msg:
                         log_msg = log_msg[:log_msg.find("vBucketMap") + 12] + "..."
                     if "Not my vbucket" in ex_msg:
                         #reduce output
                         ex_msg = str(e)[:str(e).find('Not my vbucket') + 14] + "..."
                     log.error("%s: %s" % (log_msg, ex_msg))
                     continue
                 if c.find("\x01") > 0 or c.find("\x02") > 0:
                     ready_vbuckets[i] = True
                 elif i in ready_vbuckets:
                     log.warning("vbucket state changed from active to {0}".format(c))
                     del ready_vbuckets[i]
             client.close()
     return len(ready_vbuckets) == vbucket_count

class EPStatChecker(StatChecker):

    def __init__(self,
                 host = cfg.COUCHBASE_IP,
                 port = 11210):
        self.host = host
        self.port = port
        self.client = None

        try:
            self.client = MemcachedClient(self.host, self.port)
        except MemcachedError as ex:
            msg = "error connecting to host %s for gathering postconditions"\
                   %  self.host
            logger.error(msg)
            pass

    @property
    def stat_keys(self):
        keys = []

        stats = self.get_stats()
        if stats:
            keys = stats.keys()
        return keys

    def get_stats(self):
        stats = None

        if self.client:
            try:
                stats = self.client.stats()
            except Exception:
                logger.info("unable to get stats from host: %s:%s" %\
                    (self.host, self.port))

        return stats

    def wait_for_mc_stats_no_timeout(master, bucket, stat_key, stat_value, timeout_in_seconds=-1, verbose=True):
        log.info(
            "waiting for bucket {0} stat : {1} to match {2} on {3}".format(bucket, stat_key, stat_value, master.ip)
        )
        # keep retrying until reaches the server
        stats = {}
        while not stats:
            try:
                c = MemcachedClient(master.ip, 11210)
                c.sasl_auth_plain(bucket, "")
                stats = c.stats()
            except Exception as e:
                log.info("Exception: {0}, retry in 2 seconds ...".format(str(e)))
                stats = {}
                time.sleep(2)
            finally:
                c.close()

        while str(stats[stat_key]) != str(stat_value):
            c = MemcachedClient(master.ip, 11210)
            c.sasl_auth_plain(bucket, "")
            stats = c.stats()
            c.close()
            if verbose:
                log.info("{0} : {1}".format(stat_key, stats[stat_key]))
            time.sleep(5)
        return True