python类string_types()的实例源码

def bool_from_string(value):
    """Interpret string value as boolean.

    Returns True if value translates to True otherwise False.
    """
    if isinstance(value, six.string_types):
        value = six.text_type(value)
    else:
        msg = "Unable to interpret non-string value '%s' as boolean" % (value)
        raise ValueError(msg)

    value = value.strip().lower()

    if value in ['y', 'yes', 'true', 't', 'on']:
        return True
    elif value in ['n', 'no', 'false', 'f', 'off']:
        return False

    msg = "Unable to interpret string value '%s' as boolean" % (value)
    raise ValueError(msg)

def log(message, level=None):
    """Write a message to the juju log"""
    command = ['juju-log']
    if level:
        command += ['-l', level]
    if not isinstance(message, six.string_types):
        message = repr(message)
    command += [message]
    # Missing juju-log should not cause failures in unit tests
    # Send log output to stderr
    try:
        subprocess.call(command)
    except OSError as e:
        if e.errno == errno.ENOENT:
            if level:
                message = "{}: {}".format(level, message)
            message = "juju-log: {}".format(message)
            print(message, file=sys.stderr)
        else:
            raise

def remove_cache_tier(self, cache_pool):
        """
        Removes a cache tier from Ceph.  Flushes all dirty objects from writeback pools and waits for that to complete.
        :param cache_pool: six.string_types.  The cache tier pool name to remove.
        :return: None
        """
        # read-only is easy, writeback is much harder
        mode = get_cache_mode(self.service, cache_pool)
        version = ceph_version()
        if mode == 'readonly':
            check_call(['ceph', '--id', self.service, 'osd', 'tier', 'cache-mode', cache_pool, 'none'])
            check_call(['ceph', '--id', self.service, 'osd', 'tier', 'remove', self.name, cache_pool])

        elif mode == 'writeback':
            pool_forward_cmd = ['ceph', '--id', self.service, 'osd', 'tier',
                                'cache-mode', cache_pool, 'forward']
            if version >= '10.1':
                # Jewel added a mandatory flag
                pool_forward_cmd.append('--yes-i-really-mean-it')

            check_call(pool_forward_cmd)
            # Flush the cache and wait for it to return
            check_call(['rados', '--id', self.service, '-p', cache_pool, 'cache-flush-evict-all'])
            check_call(['ceph', '--id', self.service, 'osd', 'tier', 'remove-overlay', self.name])
            check_call(['ceph', '--id', self.service, 'osd', 'tier', 'remove', self.name, cache_pool])

def get_mon_map(service):
    """
    Returns the current monitor map.
    :param service: six.string_types. The Ceph user name to run the command under
    :return: json string. :raise: ValueError if the monmap fails to parse.
      Also raises CalledProcessError if our ceph command fails
    """
    try:
        mon_status = check_output(
            ['ceph', '--id', service,
             'mon_status', '--format=json'])
        try:
            return json.loads(mon_status)
        except ValueError as v:
            log("Unable to parse mon_status json: {}. Error: {}".format(
                mon_status, v.message))
            raise
    except CalledProcessError as e:
        log("mon_status command failed with message: {}".format(
            e.message))
        raise

def hash_monitor_names(service):
    """
    Uses the get_mon_map() function to get information about the monitor
    cluster.
    Hash the name of each monitor.  Return a sorted list of monitor hashes
    in an ascending order.
    :param service: six.string_types. The Ceph user name to run the command under
    :rtype : dict.   json dict of monitor name, ip address and rank
    example: {
        'name': 'ip-172-31-13-165',
        'rank': 0,
        'addr': '172.31.13.165:6789/0'}
    """
    try:
        hash_list = []
        monitor_list = get_mon_map(service=service)
        if monitor_list['monmap']['mons']:
            for mon in monitor_list['monmap']['mons']:
                hash_list.append(
                    hashlib.sha224(mon['name'].encode('utf-8')).hexdigest())
            return sorted(hash_list)
        else:
            return None
    except (ValueError, CalledProcessError):
        raise

def monitor_key_set(service, key, value):
    """
    Sets a key value pair on the monitor cluster.
    :param service: six.string_types. The Ceph user name to run the command under
    :param key: six.string_types.  The key to set.
    :param value: The value to set.  This will be converted to a string
        before setting
    """
    try:
        check_output(
            ['ceph', '--id', service,
             'config-key', 'put', str(key), str(value)])
    except CalledProcessError as e:
        log("Monitor config-key put failed with message: {}".format(
            e.output))
        raise

def monitor_key_get(service, key):
    """
    Gets the value of an existing key in the monitor cluster.
    :param service: six.string_types. The Ceph user name to run the command under
    :param key: six.string_types.  The key to search for.
    :return: Returns the value of that key or None if not found.
    """
    try:
        output = check_output(
            ['ceph', '--id', service,
             'config-key', 'get', str(key)])
        return output
    except CalledProcessError as e:
        log("Monitor config-key get failed with message: {}".format(
            e.output))
        return None

def monitor_key_exists(service, key):
    """
    Searches for the existence of a key in the monitor cluster.
    :param service: six.string_types. The Ceph user name to run the command under
    :param key: six.string_types.  The key to search for
    :return: Returns True if the key exists, False if not and raises an
     exception if an unknown error occurs. :raise: CalledProcessError if
     an unknown error occurs
    """
    try:
        check_call(
            ['ceph', '--id', service,
             'config-key', 'exists', str(key)])
        # I can return true here regardless because Ceph returns
        # ENOENT if the key wasn't found
        return True
    except CalledProcessError as e:
        if e.returncode == errno.ENOENT:
            return False
        else:
            log("Unknown error from ceph config-get exists: {} {}".format(
                e.returncode, e.output))
            raise

def remove_pool_snapshot(service, pool_name, snapshot_name):
    """
    Remove a snapshot from a RADOS pool in ceph.
    :param service: six.string_types. The Ceph user name to run the command under
    :param pool_name: six.string_types
    :param snapshot_name: six.string_types
    :return: None.  Can raise CalledProcessError
    """
    cmd = ['ceph', '--id', service, 'osd', 'pool', 'rmsnap', pool_name, snapshot_name]
    try:
        check_call(cmd)
    except CalledProcessError:
        raise


# max_bytes should be an int or long

def get_cache_mode(service, pool_name):
    """
    Find the current caching mode of the pool_name given.
    :param service: six.string_types. The Ceph user name to run the command under
    :param pool_name: six.string_types
    :return: int or None
    """
    validator(value=service, valid_type=six.string_types)
    validator(value=pool_name, valid_type=six.string_types)
    out = check_output(['ceph', '--id', service, 'osd', 'dump', '--format=json'])
    try:
        osd_json = json.loads(out)
        for pool in osd_json['pools']:
            if pool['pool_name'] == pool_name:
                return pool['cache_mode']
        return None
    except ValueError:
        raise

def _validate_dict_data(self, expected, actual):
        """Validate dictionary data.

           Compare expected dictionary data vs actual dictionary data.
           The values in the 'expected' dictionary can be strings, bools, ints,
           longs, or can be a function that evaluates a variable and returns a
           bool.
           """
        self.log.debug('actual: {}'.format(repr(actual)))
        self.log.debug('expected: {}'.format(repr(expected)))

        for k, v in six.iteritems(expected):
            if k in actual:
                if (isinstance(v, six.string_types) or
                        isinstance(v, bool) or
                        isinstance(v, six.integer_types)):
                    # handle explicit values
                    if v != actual[k]:
                        return "{}:{}".format(k, actual[k])
                # handle function pointers, such as not_null or valid_ip
                elif not v(actual[k]):
                    return "{}:{}".format(k, actual[k])
            else:
                return "key '{}' does not exist".format(k)
        return None

def __getitem__(self, index):
        """
        Indexes an expiration state on this collection.

        Args:
            index (str): Name of the physical channel of which the
                expiration state to retrieve.
        Returns:
            nidaqmx.system._watchdog_modules.expiration_state.ExpirationState:

            The object representing the indexed expiration state.
        """
        if isinstance(index, six.string_types):
            return ExpirationState(self._handle, index)
        else:
            raise DaqError(
                'Invalid index type "{0}" used to access expiration states.'
                .format(type(index)), -1)

def _ConvertValueMessage(self, value, message):
    """Convert a JSON representation into Value message."""
    if isinstance(value, dict):
      self._ConvertStructMessage(value, message.struct_value)
    elif isinstance(value, list):
      self. _ConvertListValueMessage(value, message.list_value)
    elif value is None:
      message.null_value = 0
    elif isinstance(value, bool):
      message.bool_value = value
    elif isinstance(value, six.string_types):
      message.string_value = value
    elif isinstance(value, _INT_OR_FLOAT):
      message.number_value = value
    else:
      raise ParseError('Unexpected type for Value message.')

def generic_type_name(v):
    """Return a descriptive type name that isn't Python specific. For example,
    an int value will return 'integer' rather than 'int'."""
    if isinstance(v, numbers.Integral):
        # Must come before real numbers check since integrals are reals too
        return 'integer'
    elif isinstance(v, numbers.Real):
        return 'float'
    elif isinstance(v, (tuple, list)):
        return 'list'
    elif isinstance(v, six.string_types):
        return 'string'
    elif v is None:
        return 'null'
    else:
        return type(v).__name__

def _get_url(self):
        """
        Returns: string_type or parse.ParseResult or parse SplitResult
        We have a vague return type for _get_url method
        because we could not generalise regex support for all
        backends. So we are directly passing result to backends.
        That way users can use regex that their backend provides.
        """
        if self.url is NotImplemented:
            raise NotImplementedError('To use placebo, you need to either '
                                      'provide url attribute or '
                                      'overwrite get_url method in subclass.')
        else:
            url = invoke_or_get(self.url)
            # if url is a string convert it to ParsedUrl
            # if isinstance(url, six.string_types):
            #     url = parse.urlparse(url)
            # TODO: check return type
            return url

def _get_url(self):
        """
        Returns: string_type or parse.ParseResult or parse SplitResult
        We have a vague return type for _get_url method
        because we could not generalise regex support for all
        backends. So we are directly passing result to backends.
        That way users can use regex that their backend provides.
        """
        if self.url is NotImplemented:
            raise NotImplementedError('To use placebo, you need to either '
                                      'provide url attribute or '
                                      'overwrite get_url method in subclass.')
        else:
            url = invoke_or_get(self.url)
            # if url is a string convert it to ParsedUrl
            # if isinstance(url, six.string_types):
            #     url = parse.urlparse(url)
            # TODO: check return type
            return url

def bool_from_string(value):
    """Interpret string value as boolean.

    Returns True if value translates to True otherwise False.
    """
    if isinstance(value, six.string_types):
        value = six.text_type(value)
    else:
        msg = "Unable to interpret non-string value '%s' as boolean" % (value)
        raise ValueError(msg)

    value = value.strip().lower()

    if value in ['y', 'yes', 'true', 't', 'on']:
        return True
    elif value in ['n', 'no', 'false', 'f', 'off']:
        return False

    msg = "Unable to interpret string value '%s' as boolean" % (value)
    raise ValueError(msg)

def log(message, level=None):
    """Write a message to the juju log"""
    command = ['juju-log']
    if level:
        command += ['-l', level]
    if not isinstance(message, six.string_types):
        message = repr(message)
    command += [message]
    # Missing juju-log should not cause failures in unit tests
    # Send log output to stderr
    try:
        subprocess.call(command)
    except OSError as e:
        if e.errno == errno.ENOENT:
            if level:
                message = "{}: {}".format(level, message)
            message = "juju-log: {}".format(message)
            print(message, file=sys.stderr)
        else:
            raise

def remove_cache_tier(self, cache_pool):
        """
        Removes a cache tier from Ceph.  Flushes all dirty objects from writeback pools and waits for that to complete.
        :param cache_pool: six.string_types.  The cache tier pool name to remove.
        :return: None
        """
        # read-only is easy, writeback is much harder
        mode = get_cache_mode(self.service, cache_pool)
        version = ceph_version()
        if mode == 'readonly':
            check_call(['ceph', '--id', self.service, 'osd', 'tier', 'cache-mode', cache_pool, 'none'])
            check_call(['ceph', '--id', self.service, 'osd', 'tier', 'remove', self.name, cache_pool])

        elif mode == 'writeback':
            pool_forward_cmd = ['ceph', '--id', self.service, 'osd', 'tier',
                                'cache-mode', cache_pool, 'forward']
            if version >= '10.1':
                # Jewel added a mandatory flag
                pool_forward_cmd.append('--yes-i-really-mean-it')

            check_call(pool_forward_cmd)
            # Flush the cache and wait for it to return
            check_call(['rados', '--id', self.service, '-p', cache_pool, 'cache-flush-evict-all'])
            check_call(['ceph', '--id', self.service, 'osd', 'tier', 'remove-overlay', self.name])
            check_call(['ceph', '--id', self.service, 'osd', 'tier', 'remove', self.name, cache_pool])

def get_mon_map(service):
    """
    Returns the current monitor map.
    :param service: six.string_types. The Ceph user name to run the command under
    :return: json string. :raise: ValueError if the monmap fails to parse.
      Also raises CalledProcessError if our ceph command fails
    """
    try:
        mon_status = check_output(['ceph', '--id', service,
                                   'mon_status', '--format=json'])
        if six.PY3:
            mon_status = mon_status.decode('UTF-8')
        try:
            return json.loads(mon_status)
        except ValueError as v:
            log("Unable to parse mon_status json: {}. Error: {}"
                .format(mon_status, str(v)))
            raise
    except CalledProcessError as e:
        log("mon_status command failed with message: {}"
            .format(str(e)))
        raise

def hash_monitor_names(service):
    """
    Uses the get_mon_map() function to get information about the monitor
    cluster.
    Hash the name of each monitor.  Return a sorted list of monitor hashes
    in an ascending order.
    :param service: six.string_types. The Ceph user name to run the command under
    :rtype : dict.   json dict of monitor name, ip address and rank
    example: {
        'name': 'ip-172-31-13-165',
        'rank': 0,
        'addr': '172.31.13.165:6789/0'}
    """
    try:
        hash_list = []
        monitor_list = get_mon_map(service=service)
        if monitor_list['monmap']['mons']:
            for mon in monitor_list['monmap']['mons']:
                hash_list.append(
                    hashlib.sha224(mon['name'].encode('utf-8')).hexdigest())
            return sorted(hash_list)
        else:
            return None
    except (ValueError, CalledProcessError):
        raise

def monitor_key_set(service, key, value):
    """
    Sets a key value pair on the monitor cluster.
    :param service: six.string_types. The Ceph user name to run the command under
    :param key: six.string_types.  The key to set.
    :param value: The value to set.  This will be converted to a string
        before setting
    """
    try:
        check_output(
            ['ceph', '--id', service,
             'config-key', 'put', str(key), str(value)])
    except CalledProcessError as e:
        log("Monitor config-key put failed with message: {}".format(
            e.output))
        raise

def monitor_key_get(service, key):
    """
    Gets the value of an existing key in the monitor cluster.
    :param service: six.string_types. The Ceph user name to run the command under
    :param key: six.string_types.  The key to search for.
    :return: Returns the value of that key or None if not found.
    """
    try:
        output = check_output(
            ['ceph', '--id', service,
             'config-key', 'get', str(key)]).decode('UTF-8')
        return output
    except CalledProcessError as e:
        log("Monitor config-key get failed with message: {}".format(
            e.output))
        return None

def monitor_key_exists(service, key):
    """
    Searches for the existence of a key in the monitor cluster.
    :param service: six.string_types. The Ceph user name to run the command under
    :param key: six.string_types.  The key to search for
    :return: Returns True if the key exists, False if not and raises an
     exception if an unknown error occurs. :raise: CalledProcessError if
     an unknown error occurs
    """
    try:
        check_call(
            ['ceph', '--id', service,
             'config-key', 'exists', str(key)])
        # I can return true here regardless because Ceph returns
        # ENOENT if the key wasn't found
        return True
    except CalledProcessError as e:
        if e.returncode == errno.ENOENT:
            return False
        else:
            log("Unknown error from ceph config-get exists: {} {}".format(
                e.returncode, e.output))
            raise

def remove_pool_snapshot(service, pool_name, snapshot_name):
    """
    Remove a snapshot from a RADOS pool in ceph.
    :param service: six.string_types. The Ceph user name to run the command under
    :param pool_name: six.string_types
    :param snapshot_name: six.string_types
    :return: None.  Can raise CalledProcessError
    """
    cmd = ['ceph', '--id', service, 'osd', 'pool', 'rmsnap', pool_name, snapshot_name]
    try:
        check_call(cmd)
    except CalledProcessError:
        raise


# max_bytes should be an int or long

def get_cache_mode(service, pool_name):
    """
    Find the current caching mode of the pool_name given.
    :param service: six.string_types. The Ceph user name to run the command under
    :param pool_name: six.string_types
    :return: int or None
    """
    validator(value=service, valid_type=six.string_types)
    validator(value=pool_name, valid_type=six.string_types)
    out = check_output(['ceph', '--id', service,
                        'osd', 'dump', '--format=json'])
    if six.PY3:
        out = out.decode('UTF-8')
    try:
        osd_json = json.loads(out)
        for pool in osd_json['pools']:
            if pool['pool_name'] == pool_name:
                return pool['cache_mode']
        return None
    except ValueError:
        raise

def _validate_dict_data(self, expected, actual):
        """Validate dictionary data.

           Compare expected dictionary data vs actual dictionary data.
           The values in the 'expected' dictionary can be strings, bools, ints,
           longs, or can be a function that evaluates a variable and returns a
           bool.
           """
        self.log.debug('actual: {}'.format(repr(actual)))
        self.log.debug('expected: {}'.format(repr(expected)))

        for k, v in six.iteritems(expected):
            if k in actual:
                if (isinstance(v, six.string_types) or
                        isinstance(v, bool) or
                        isinstance(v, six.integer_types)):
                    # handle explicit values
                    if v != actual[k]:
                        return "{}:{}".format(k, actual[k])
                # handle function pointers, such as not_null or valid_ip
                elif not v(actual[k]):
                    return "{}:{}".format(k, actual[k])
            else:
                return "key '{}' does not exist".format(k)
        return None

def bool_from_string(value):
    """Interpret string value as boolean.

    Returns True if value translates to True otherwise False.
    """
    if isinstance(value, six.string_types):
        value = six.text_type(value)
    else:
        msg = "Unable to interpret non-string value '%s' as boolean" % (value)
        raise ValueError(msg)

    value = value.strip().lower()

    if value in ['y', 'yes', 'true', 't', 'on']:
        return True
    elif value in ['n', 'no', 'false', 'f', 'off']:
        return False

    msg = "Unable to interpret string value '%s' as boolean" % (value)
    raise ValueError(msg)

def log(message, level=None):
    """Write a message to the juju log"""
    command = ['juju-log']
    if level:
        command += ['-l', level]
    if not isinstance(message, six.string_types):
        message = repr(message)
    command += [message]
    # Missing juju-log should not cause failures in unit tests
    # Send log output to stderr
    try:
        subprocess.call(command)
    except OSError as e:
        if e.errno == errno.ENOENT:
            if level:
                message = "{}: {}".format(level, message)
            message = "juju-log: {}".format(message)
            print(message, file=sys.stderr)
        else:
            raise

def ns_query(address):
    try:
        import dns.resolver
    except ImportError:
        if six.PY2:
            apt_install('python-dnspython', fatal=True)
        else:
            apt_install('python3-dnspython', fatal=True)
        import dns.resolver

    if isinstance(address, dns.name.Name):
        rtype = 'PTR'
    elif isinstance(address, six.string_types):
        rtype = 'A'
    else:
        return None

    try:
        answers = dns.resolver.query(address, rtype)
    except dns.resolver.NXDOMAIN:
        return None

    if answers:
        return str(answers[0])
    return None