def parse_dcs(dcs):
    """
    Break up the provided dcs string
    >>> parse_dcs('localhost') == {'scheme': 'etcd', 'hostname': 'localhost', 'port': 4001}
    True
    >>> parse_dcs('localhost:8500') == {'scheme': 'consul', 'hostname': 'localhost', 'port': 8500}
    True
    >>> parse_dcs('zookeeper://localhost') == {'scheme': 'zookeeper', 'hostname': 'localhost', 'port': 2181}
    True
    """

    if not dcs:
        return {}

    parsed = urlparse(dcs)
    scheme = parsed.scheme
    if scheme == '' and parsed.netloc == '':
        parsed = urlparse('//' + dcs)

    if scheme == '':
        default_schemes = {'2181': 'zookeeper', '8500': 'consul'}
        scheme = default_schemes.get(str(parsed.port), 'etcd')

    port = parsed.port
    if port is None:
        default_ports = {'consul': 8500, 'zookeeper': 2181}
        port = default_ports.get(str(scheme), 4001)

    return {'scheme': str(scheme), 'hostname': str(parsed.hostname), 'port': int(port)}

def from_docker_envvars(config):
    # linked postgres database (link name 'pg' or 'postgres')
    if 'PG_PORT' in os.environ:
        pg_url = urlparse(os.environ['PG_PORT'])

        if not pg_url.scheme == 'tcp':
            raise ValueError('Only tcp scheme supported for postgres')

        host, port = pg_url.netloc.split(':')

        uri = 'postgres://{user}:{password}@{host}:{port}/{database}'.format(
            user=os.environ.get('PG_ENV_POSTGRES_USER', 'postgres'),
            password=os.environ.get('PG_ENV_POSTGRES_PASSWORD', ''),
            host=host,
            port=port,
            database=os.environ.get('PG_ENV_POSTGRES_DB', 'postgres'))

        config['SQLALCHEMY_DATABASE_URI'] = uri

    if 'REDIS_PORT' in os.environ:
        redis_url = urlparse(os.environ['REDIS_PORT'])

        if not redis_url.scheme == 'tcp':
            raise ValueError('Only tcp scheme supported for redis')

        host, port = redis_url.netloc.split(':')

        uri = 'redis://{host}:{port}/0'.format(host=host, port=port, )

        config['REDIS_URL'] = uri
        config['REDIS_HOST'] = host
        config['REDIS_PORT'] = int(port)

 def _connect(self, url=None, cdn=False):
     if not url:
         if cdn:
             if not self.cdn_url:
                 self.auth()
             url = self.cdn_url
         else:
             if not self.storage_url:
                 self.auth()
             url = self.storage_url
     parsed = urlparse.urlparse(url) if url else None
     http_proxy_parsed = \
         urlparse.urlparse(self.http_proxy) if self.http_proxy else None
     if not parsed and not http_proxy_parsed:
         return None, None
     netloc = (http_proxy_parsed if self.http_proxy else parsed).netloc
     if parsed.scheme == 'http':
         self.verbose('Establishing HTTP connection to %s', netloc)
         conn = self.HTTPConnection(netloc)
     elif parsed.scheme == 'https':
         self.verbose('Establishing HTTPS connection to %s', netloc)
         conn = self.HTTPSConnection(netloc)
     else:
         raise self.HTTPException(
             'Cannot handle protocol scheme %s for url %s' %
             (parsed.scheme, repr(url)))
     if self.http_proxy:
         self.verbose(
             'Setting tunnelling to %s:%s', parsed.hostname, parsed.port)
         conn._set_tunnel(parsed.hostname, parsed.port)
     return parsed, conn

def url_distance(preprocessor, url1, url2):
    url1 = urlparse(url1)
    url2 = urlparse(url2)

    process_fn = lambda s: preprocessor(unquote(s))
    path1 = map(process_fn, url1.path.strip('/').split('/'))
    path2 = map(process_fn, url2.path.strip('/').split('/'))
    path_distance = levenshtein_array(path1, path2)

    query_distance = dict_distance(preprocessor,
        parse_qs(url1.query, True),
        parse_qs(url2.query, True)
    )

    domain_distance = 4 * levenshtein_array(
        (url1.hostname or '').split('.'),
        (url2.hostname or '').split('.')
    )

    return (
        domain_distance +
        path_distance +
        query_distance +
        (url1.fragment != url2.fragment)
    )

    def get_canonicalized_resource(self, req, service):
        # /bucket/keyname
        parsed_req_path = urlparse.urlparse(req.url).path
        assert service.endpoint is not None
        parsed_svc_path = urlparse.urlparse(service.endpoint).path
        # IMPORTANT:  this only supports path-style requests
        assert parsed_req_path.startswith(parsed_svc_path)
        resource = parsed_req_path[len(parsed_svc_path):]
        if parsed_svc_path.endswith('/'):
            # The leading / got stripped off
            resource = '/' + resource
        if not resource:
            # This resource does not address a bucket
            resource = '/'

        # Now append sub-resources, a.k.a. query string parameters
        if getattr(req, 'params', None):
            # A regular Request
            params = req.params
        else:
            # A PreparedRequest
            params = _get_params_from_url(req.url)
        if params:
            subresources = []
            for key, val in sorted(params.iteritems()):
                if key in self.HASHED_PARAMS:
                    if val is None:
                        subresources.append(key)
                    else:
                        print '{0}={1}'.format(key, val), key + '=' + val
                        subresources.append(key + '=' + val)
                if subresources:
                    resource += '?' + '&'.join(subresources)
        self.log.debug('canonicalized resource: %s', repr(resource))
        return resource

def assert_urls_match(url_a, url_b):
    url_a = urlparse(url_a)
    url_b = urlparse(url_b)

    assert url_a.scheme == url_b.scheme
    assert url_a.netloc == url_b.netloc
    assert url_a.path == url_b.path
    assert cgi.parse_qs(url_a.query) == cgi.parse_qs(url_b.query)

def assert_url_equal(url, expected, compare_host=False):
    """Compare url paths and query strings."""
    parsed = urlparse(six.text_type(url))
    parsed_expected = urlparse(six.text_type(expected))
    compare_url_part(parsed.path, parsed_expected.path)
    compare_url_part(parse_qs(parsed.query), parse_qs(parsed_expected.query))
    if compare_host:
        compare_url_part(parsed.netloc, parsed_expected.netloc)

def is_safe_url(target):
    """ Checks that the target url is safe and sending to the current
    website not some other malicious one.
    """
    ref_url = urlparse.urlparse(flask.request.host_url)
    test_url = urlparse.urlparse(
        urlparse.urljoin(flask.request.host_url, target))
    return test_url.scheme in ('http', 'https') and \
        ref_url.netloc == test_url.netloc

def _rebase_url(url, base):
    base = list(urlparse.urlparse(base))
    url = list(urlparse.urlparse(url))
    if not url[0]:  # fix up schema
        url[0] = base[0] or "http"
    if not url[1]:  # fix up hostname
        url[1] = base[1]
        if not url[2].startswith('/'):
            url[2] = re.sub(r'/[^/]+$', '/', base[2]) + url[2]
    return urlparse.urlunparse(url)

    def __init__(self, raw=None, _address=None):

        if raw:
            raw = _to_parser_input(raw)
            url = url_parser.parse(raw, lexer.clone())
            self._address = urlparse(url.address)
        elif _address:
            self._address = urlparse(_address)
        else:
            raise SyntaxError('failed to create UrlAddress: bad parameters')

 def do_start(self):
     start_url = self.backend.start().url
     target_url = self.auth_handlers(start_url)
     response = requests.get(start_url)
     self.assertEqual(response.url, target_url)
     self.assertEqual(response.text, 'foobar')
     self.strategy.set_request_data(parse_qs(urlparse(start_url).query),
                                    self.backend)
     self.strategy.set_request_data(parse_qs(urlparse(target_url).query),
                                    self.backend)
     return self.backend.complete()

    def add_uri(self, uri, **kwargs):
        """

        .. WARNING::
            Deprecated, please use add_torrent.
        """
        if uri is None:
            raise ValueError('add_uri requires a URI.')
        # there has been some problem with T's built in torrent fetcher,
        # use a python one instead
        parsed_uri = urlparse(uri)
        torrent_data = None
        if parsed_uri.scheme in ['ftp', 'ftps', 'http', 'https']:
            torrent_file = urlopen(uri)
            torrent_data = torrent_file.read()
            torrent_data = base64.b64encode(torrent_data).decode('utf-8')
        if parsed_uri.scheme in ['file']:
            filepath = uri
            # uri decoded different on linux / windows ?
            if len(parsed_uri.path) > 0:
                filepath = parsed_uri.path
            elif len(parsed_uri.netloc) > 0:
                filepath = parsed_uri.netloc
            torrent_file = open(filepath, 'rb')
            torrent_data = torrent_file.read()
            torrent_data = base64.b64encode(torrent_data).decode('utf-8')
        warnings.warn('add_uri has been deprecated, please use add_torrent instead.', DeprecationWarning)
        if torrent_data:
            return self.add(torrent_data, **kwargs)
        else:
            return self.add(None, filename=uri, **kwargs)

 def post(self, queue_name, messages, client_uuid, project=None):
     """Send messages to the subscribers."""
     if self.subscription_controller:
         if not isinstance(self.subscription_controller,
                           pooling.SubscriptionController):
             marker = None
             while True:
                 subscribers = self.subscription_controller.list(
                     queue_name, project, marker=marker)
                 for sub in next(subscribers):
                     LOG.debug("Notifying subscriber %r" % (sub,))
                     s_type = urllib_parse.urlparse(
                         sub['subscriber']).scheme
                     # If the subscriber doesn't contain 'confirmed', it
                     # means that this kind of subscriber was created before
                     # the confirm feature be introduced into Zaqar. We
                     # should allow them be subscribed.
                     if (self.require_confirmation and
                             not sub.get('confirmed', True)):
                         LOG.info(_LI('The subscriber %s is not '
                                      'confirmed.'), sub['subscriber'])
                         continue
                     for msg in messages:
                         msg['Message_Type'] = MessageType.Notification.name
                     self._execute(s_type, sub, messages)
                 marker = next(subscribers)
                 if not marker:
                     break
     else:
         LOG.error(_LE('Failed to get subscription controller.'))

 def __new__(cls, urlstring):
     if isinstance(urlstring, binary_type):
         # here we make a safe-ish assumption it is  a utf-8 string
         urlstring = urlstring.decode('utf-8')
     url = super(URL, cls).__new__(cls, urlstring)
     url.parsed = urlparse(url)
     return url

    def __call__(self, value):
        parsed = urlparse(value)
        if parsed.scheme not in ("http", "https"):
            raise ValidationError(message=self.message)

        if parsed.hostname in ("127.0.0.1", "localhost"):
            raise ValidationError(message=self.message)

def get_agent(reactor, connect_timeout=None):
    """Return appropriate agent based on whether an http_proxy is used or not.

    :param connect_timeout: connection timeout in seconds
    :type connect_timeout: float
    :returns: :class:`twisted.web.client.ProxyAgent` when an http_proxy
        environment variable is present, :class:`twisted.web.client.Agent`
        otherwise.
    """

    # TODO: Would be nice to have https_proxy support too.
    http_proxy = os.environ.get('http_proxy')
    if http_proxy is None:
        return _twisted_web_client().Agent(
            reactor,
            connectTimeout=connect_timeout)

    parse_result = urlparse(http_proxy)
    http_proxy_endpoint = TCP4ClientEndpoint(
        reactor,
        parse_result.hostname,
        parse_result.port or 80,
        timeout=connect_timeout)

    return _twisted_web_client().ProxyAgent(http_proxy_endpoint)

 def parse_url(cls, url):
     parsed = urlparse(url)
     return cls(proxy_type=parsed.scheme,
                proxy_address=parsed.hostname,
                proxy_port=parsed.port,
                proxy_login=parsed.username,
                proxy_password=parsed.password)

def validate_image_proxies(node):
    """Check that the provided proxy parameters are valid.

    :param node: an Ironic node.
    :raises: InvalidParameterValue if any of the provided proxy parameters are
        incorrect.
    """
    invalid_proxies = {}
    for scheme in ('http', 'https'):
        proxy_param = 'image_%s_proxy' % scheme
        proxy = node.driver_info.get(proxy_param)
        if proxy:
            chunks = urlparse.urlparse(proxy)
            # NOTE(vdrok) If no scheme specified, this is still a valid
            # proxy address. It is also possible for a proxy to have a
            # scheme different from the one specified in the image URL,
            # e.g. it is possible to use https:// proxy for downloading
            # http:// image.
            if chunks.scheme not in ('', 'http', 'https'):
                invalid_proxies[proxy_param] = proxy
    msg = ''
    if invalid_proxies:
        msg += _("Proxy URL should either have HTTP(S) scheme "
                 "or no scheme at all, the following URLs are "
                 "invalid: %s.") % invalid_proxies
    no_proxy = node.driver_info.get('image_no_proxy')
    if no_proxy is not None and not utils.is_valid_no_proxy(no_proxy):
        msg += _(
            "image_no_proxy should be a list of host names, IP addresses "
            "or domain names to exclude from proxying, the specified list "
            "%s is incorrect. To denote a domain name, prefix it with a dot "
            "(instead of e.g. '.*').") % no_proxy
    if msg:
        raise exception.InvalidParameterValue(msg)

def _discover_auth_versions(session, auth_url):
    # discover the API versions the server is supporting base on the
    # given URL
    v2_auth_url = None
    v3_auth_url = None
    try:
        ks_discover = discover.Discover(session=session, auth_url=auth_url)
        v2_auth_url = ks_discover.url_for('2.0')
        v3_auth_url = ks_discover.url_for('3.0')
    except ks_exc.ClientException:
        # Identity service may not support discover API version.
        # Lets trying to figure out the API version from the original URL.
        path = urlparse.urlparse(auth_url).path
        path = path.lower()
        if path.startswith('/v3'):
            v3_auth_url = auth_url
        elif path.startswith('/v2'):
            v2_auth_url = auth_url
        else:
            # not enough information to determine the auth version
            msg = ('Unable to determine the Keystone version '
                   'to authenticate with using the given '
                   'auth_url. Identity service may not support API '
                   'version discovery. Please provide a versioned '
                   'auth_url instead.')
            raise exc.CommandError(msg)

    return v2_auth_url, v3_auth_url

def depaginate(api, result):
    """Depaginate the first (or only) page of a paginated result"""
    meta = result['meta']
    if meta['next'] is None:
        # No pages means we're done
        return result

    # make a copy of meta that we'll mess with and eventually return
    # since we'll be chewing on the 'meta' object with every new GET
    # same thing for objects, since we'll just be appending to it
    # while we pull more records
    ret_meta = meta.copy()
    ret_objects = result['objects']
    while meta['next']:
        # parse out url bits for constructing the new api req
        next_url = urlparse(meta['next'])
        # ugh...need to find the word after 'api/' in the next URL to
        # get the resource endpoint name; not sure how to make this better
        next_endpoint = next_url.path.strip('/').split('/')[-1]
        next_params = {k: v[0] for k, v in parse_qs(next_url.query).items()}
        result = getattr(api, next_endpoint).get(**next_params)
        ret_objects.extend(result['objects'])
        meta = result['meta']

    # fix meta up to not tell lies
    ret_meta['total_count'] = len(ret_objects)
    ret_meta['next'] = None
    ret_meta['limit'] = ret_meta['total_count']
    return {
        'meta': ret_meta,
        'objects': ret_objects
    }