def test_get_auth_from_url(self):
     """ Ensures that username and password in well-encoded URI as per RFC 3986 are correclty extracted """
     from requests.utils import get_auth_from_url
     from requests.compat import quote
     percent_encoding_test_chars = "%!*'();:@&=+$,/?#[] "
     url_address = "request.com/url.html#test"
     url = "http://" + quote(percent_encoding_test_chars, '') + ':' + quote(percent_encoding_test_chars, '') + '@' + url_address
     (username, password) = get_auth_from_url(url)
     assert username == percent_encoding_test_chars
     assert password == percent_encoding_test_chars

    def uri(self):
        """
        """
        uri = self.application_uri

        path_info = quote(self.path_info, safe='/=;,')
        if not self.env['SCRIPT_NAME']:
            uri += path_info[1:]
        else:
            uri += path_info

        if self.query_string:
            uri += '?' + quote(self.query_string, safe='&=;,')

        return uri

    def get_entities(self, group_name, expression=None, min_insert_date=None, max_insert_date=None, tags=None,
                     limit=None):
        """Retrieve a list of entities that are members of the specified entity group and match the specified expression filter.

        :param group_name: `str`
        :param expression: `str`
        :param min_insert_date: `str` | `int` | :class:`datetime`
        :param max_insert_date: `str` | `int` | :class:`datetime`
        :param tags: `dict`
        :param limit: `int`
        :return: `list` of :class:`.Entity` objects
        """
        _check_name(group_name)
        params = {}
        if expression is not None:
            params["expression"] = expression
        if min_insert_date is not None:
            params["minInsertDate"] = to_iso(min_insert_date)
        if max_insert_date is not None:
            params["maxInsertDate"] = to_iso(max_insert_date)
        if tags is not None:
            params["tags"] = tags
        if limit is not None:
            params["limit"] = limit
        resp = self.conn.get(eg_get_entities_url.format(group=quote(group_name, '')), params)
        return _jsonutil.deserialize(resp, Entity)

    def series(self, metric, entity=None, tags=None, min_insert_date=None, max_insert_date=None):
        """Retrieve series for the specified metric.

        :param metric: `str` | :class:`.Metric`
        :param entity: `str` | :class:`.Entity`
        :param tags: `dict`
        :param min_insert_date: `int` | `str` | None | :class:`datetime`
        :param max_insert_date: `int` | `str` | None | :class:`datetime`

        :return: :class:`.Series`
        """
        metric_name = metric.name if isinstance(metric, Metric) else metric
        _check_name(metric_name)

        params = {}
        if entity is not None:
            params['entity'] = entity.name if isinstance(entity, Entity) else entity
        if tags is not None and isinstance(tags, dict):
            for k, v in tags.items():
                params['tags.%s' % k] = v
        if min_insert_date is not None:
            params['minInsertDate'] = to_iso(min_insert_date)
        if max_insert_date is not None:
            params['maxInsertDate'] = to_iso(max_insert_date)

        try:
            response = self.conn.get(metric_series_url.format(metric=quote(metric_name, '')),
                                     params)
        except ServerException as e:
            if e.status_code == 404:
                return []
            else:
                raise e
        return _jsonutil.deserialize(response, Series)

    def metrics(self, entity, expression=None, min_insert_date=None, max_insert_date=None, use_entity_insert_time=False,
                limit=None, tags=None):
        """Retrieve a `list` of metrics matching the specified filters.

        :param entity: `str` | :class:`.Entity`
        :param expression: `str`
        :param min_insert_date: `int` | `str` | None | :class:`datetime`
        :param max_insert_date: `int` | `str` | None | :class:`datetime`
        :param use_entity_insert_time: `bool` If true, last_insert_date is calculated for the specified entity and metric
        :param limit: `int`
        :param tags: `str`
        :return: :class:`.Metric` objects
        """
        entity_name = entity.name if isinstance(entity, Entity) else entity
        _check_name(entity_name)
        params = {}
        if expression is not None:
            params['expression'] = expression
        if min_insert_date is not None:
            params['minInsertDate'] = to_iso(min_insert_date)
        if max_insert_date is not None:
            params['maxInsertDate'] = to_iso(max_insert_date)
        params['useEntityInsertTime'] = use_entity_insert_time
        if limit is not None:
            params['limit'] = limit
        if tags is not None:
            params['tags'] = tags
        response = self.conn.get(ent_metrics_url.format(entity=quote(entity_name, '')), params)
        return _jsonutil.deserialize(response, Metric)

    def request(self, method, endpoint, body='', **kwargs):
        """Make a request to the PyCon website, and return the result."""
        # The PyCon website runs using HTTPS, but localhost doesn't.
        # Determine the right thing.
        protocol = 'https'
        if 'localhost' in self.host:
            protocol = 'http'

        # Construct the base URI.
        uri = '/2014/pycon_api/%s/' % endpoint

        # If keyword arguments are provided, append them to
        # the URI.
        if kwargs:
            uri += '?' + '&'.join(
                ['%s=%s' % (k, quote(str(v))) for k, v in kwargs.items()],
            )

        # Construct the full URL.
        url = '{protocol}://{host}{uri}'.format(
            host=self.host,
            protocol=protocol,
            uri=uri,
        )

        # Generate the appropriate request signature to certify
        # that this is a valid request.
        signature = self._sign_request(uri, method, body)

        # Add the appropriate content-type header.
        if method == 'POST':
            signature['Content-Type'] = 'application/json'

        # Make the actual request to the PyCon website.
        r = requests.request(method, url, data=body, headers=signature,
                                          verify=False)

        # Sanity check: Did we get a bad request of some kind?
        if r.status_code >= 400:
            # If we got a 500 response, we can't really know what to do
            if r.status_code >= 500:
                raise InternalServerError(r)

            # What exception class shall I use?
            exc_class = APIError
            if r.status_code == 403:
                exc_class = AuthenticationError
            if r.status_code == 404:
                exc_class = NotFound

            # Create and raise the exception
            try:
                ex = exc_class(r.json()['error'])
            except ValueError:
                raise InternalServerError(r)
            ex.request = r
            raise ex

        # OK, all is well; return the response.
        return r.json()

 def config(self):
     url = 'job/%s/config.xml' % quote(self.name)
     res = self.server.get(url)
     if res.status_code != 200 or res.headers.get('content-type', '') != 'application/xml':
         msg = 'fetching configuration for job "%s" did not return a xml document'
         raise JenkinsError(msg % self.name)
     return res.text

    def create_or_replace(self, entity):
        """Create an entity or update an existing entity.

        :param entity: :class:`.Entity`
        :return: True if success
        """
        self.conn.put(ent_create_or_replace_url.format(entity=quote(entity.name, '')), entity)
        return True

    def update(self, entity):
        """Update the specified entity.

        :param entity: :class:`.Entity`
        :return: True if success
        """
        self.conn.patch(ent_update_url.format(entity=quote(entity.name, '')), entity)
        return True

    def delete(self, metric_name):
        """Delete the specified metric.

        :param metric_name: :class:`.Metric`
        :return: True if success
        """
        self.conn.delete(metric_delete_url.format(metric=quote(metric_name, '')))
        return True

    def type_query(self, entity):
        """Returns an array of property types for the entity.

        :param entity: :class:`.Entity`
        :return: returns `list` of property types for the entity.
        """
        entity_name = entity.name if isinstance(entity, Entity) else entity
        response = self.conn.get(properties_types_url.format(entity=quote(entity_name, '')))
        return response

    def delete(self):
        '''Permanently remove job.'''

        self._not_exist_raise()
        url = 'job/%s/doDelete' % quote(self.name)

        self.server.post(url)
        if self.exists():
            raise JenkinsError('delete of job "%s" failed' % self.name)

    def __init__(self, content_type="application/json", sock_path=None):
        if sock_path is None:
            sock_path = self.secrets_sock_path

        self.content_type = content_type
        self.session = requests.Session()
        self.session.mount("http+unix://", HTTPUnixAdapter())
        self.headers = dict({"Content-Type": content_type})
        self.url = "http+unix://" + quote(sock_path, safe="") + "/" + self.secrets_container
        self._last_response = None

    def _sendFreeMobileSMS(self, title, msg, cust_id=None, apiKey=None):
        """
        Sends a SMS notification

        msg: The message to send (unicode)
        title: The title of the message
        userKey: The pushover user id to send the message to (or to subscribe with)

        returns: True if the message succeeded, False otherwise
        """

        if cust_id is None:
            cust_id = sickbeard.FREEMOBILE_ID
        if apiKey is None:
            apiKey = sickbeard.FREEMOBILE_APIKEY

        logger.log(u"Free Mobile in use with API KEY: " + apiKey, logger.DEBUG)

        # build up the URL and parameters
        msg = msg.strip()
        msg_quoted = quote(title.encode('utf-8') + ": " + msg.encode('utf-8'))
        URL = "https://smsapi.free-mobile.fr/sendmsg?user=" + cust_id + "&pass=" + apiKey + "&msg=" + msg_quoted

        req = Request(URL)
        # send the request to Free Mobile
        try:
            urlopen(req)
        except IOError as e:
            if hasattr(e, 'code'):
                if e.code == 400:
                    message = "Missing parameter(s)."
                    logger.log(message, logger.ERROR)
                    return False, message
                if e.code == 402:
                    message = "Too much SMS sent in a short time."
                    logger.log(message, logger.ERROR)
                    return False, message
                if e.code == 403:
                    message = "API service isn't enabled in your account or ID / API key is incorrect."
                    logger.log(message, logger.ERROR)
                    return False, message
                if e.code == 500:
                    message = "Server error. Please retry in few moment."
                    logger.log(message, logger.ERROR)
                    return False, message
        except Exception as e:
            message = u"Error while sending SMS: {0}".format(e)
            logger.log(message, logger.ERROR)
            return False, message

        message = "Free Mobile SMS successful."
        logger.log(message, logger.INFO)
        return True, message

    def __init__(self, content_type='application/json', sock_path=None):
        if sock_path is None:
            sock_path = self.secrets_sock_path

        self.content_type = content_type
        self.session = requests.Session()
        self.session.mount('http+unix://', HTTPUnixAdapter())
        self.headers = dict({'Content-Type': content_type})
        self.url = 'http+unix://' + \
            quote(sock_path, safe='') + \
            '/' + \
            self.secrets_container
        self._last_response = None

    def get(self, name):
        """Retrieve metric.

        :param name: `str` metric name
        :return: :class:`.Metric`
        """
        _check_name(name)
        try:
            response = self.conn.get(metric_get_url.format(metric=quote(name, '')))
        except ServerException as e:
            if e.status_code == 404:
                return None
            else:
                raise e
        return _jsonutil.deserialize(response, Metric)

    def format(self, record):
        """Strip censored items from string.

        :param record: to censor
        :type record: logging.LogRecord
        :return:
        :rtype: str
        """
        privacy_level = sickbeard.common.privacy_levels[sickbeard.PRIVACY_LEVEL]
        if not privacy_level:
            return super(CensoredFormatter, self).format(record)
        elif privacy_level == sickbeard.common.privacy_levels['absurd']:
            return re.sub(r'[\d\w]', '*', super(CensoredFormatter, self).format(record))
        else:
            msg = super(CensoredFormatter, self).format(record)

        if not isinstance(msg, text_type):
            msg = msg.decode(self.encoding, 'replace')  # Convert to unicode

        # Change the SSL error to a warning with a link to information about how to fix it.
        # Check for u'error [SSL: SSLV3_ALERT_HANDSHAKE_FAILURE] sslv3 alert handshake failure (_ssl.c:590)'
        for ssl_error in SSL_ERRORS:
            if re.findall(ssl_error, msg):
                record.levelno = WARNING
                record.levelname = logging.getLevelName(record.levelno)
                msg = super(CensoredFormatter, self).format(record)
                msg = re.sub(ssl_error, SSL_ERROR_HELP_MSG, msg)

        # set of censored items
        censored = {value for value in itervalues(censored_items) if value}
        # set of censored items and urlencoded counterparts
        censored = censored | {quote(item) for item in censored}
        # convert set items to unicode and typecast to list
        censored = list({
                            item.decode(self.encoding, 'replace')
                            if not isinstance(item, text_type) else item
                            for item in censored
                            })
        # sort the list in order of descending length so that entire item is censored
        # e.g. password and password_1 both get censored instead of getting ********_1
        censored.sort(key=len, reverse=True)

        for item in censored:
            msg = msg.replace(item, len(item) * '*')

        # Needed because Newznab apikey isn't stored as key=value in a section.
        msg = re.sub(r'([&?]r|[&?]apikey|[&?]api_key)(?:=|%3D)[^&]*([&\w]?)', r'\1=**********\2', msg, re.I)
        return msg

    def build_params(self, params):
        """
        quotes values and converts keys of params to camelCase from underscore
        :param params: dict
        :return:
        """
        out = {}
        for key, value in params.iteritems():
            if not isinstance(value, types.StringTypes):
                value = str(value)

            elif isinstance(value, unicode):
                value = value.encode("utf-8")

            key = key.split('_')[0] + ''.join(x.capitalize() for x in key.split('_')[1:])
            out[key] = quote(value)
        return out

    def _sendFreeMobileSMS(self, title, msg, cust_id=None, apiKey=None):
        """
        Send a SMS notification

        msg: The message to send (unicode)
        title: The title of the message
        userKey: The pushover user id to send the message to (or to subscribe with)

        return: True if the message succeeded, False otherwise
        """
        if cust_id is None:
            cust_id = app.FREEMOBILE_ID
        if apiKey is None:
            apiKey = app.FREEMOBILE_APIKEY

        log.debug(u'Free Mobile in use with API KEY: {0}', apiKey)

        # build up the URL and parameters
        msg = msg.strip()
        msg_quoted = quote(title.encode('utf-8') + ': ' + msg.encode('utf-8'))
        URL = 'https://smsapi.free-mobile.fr/sendmsg?user=' + cust_id + '&pass=' + apiKey + '&msg=' + msg_quoted

        req = Request(URL)
        # send the request to Free Mobile
        try:
            urlopen(req)
        except IOError as e:
            if hasattr(e, 'code'):
                error_message = {
                    400: 'Missing parameter(s).',
                    402: 'Too much SMS sent in a short time.',
                    403: 'API service is not enabled in your account or ID / API key is incorrect.',
                    500: 'Server error. Please retry in few moment.',
                }
                message = error_message.get(e.code)
                if message:
                    log.error(message)
                    return False, message
        except Exception as e:
            message = u'Error while sending SMS: {0}'.format(e)
            log.error(message)
            return False, message

        message = 'Free Mobile SMS successful.'
        log.info(message)
        return True, message

    def delete_entities(self, group_name, entities):
        """Remove specified entities from members of the specified entity group.
        To delete all entities, submit an empty list [] using the set_entities method.
        Changing members of expression-based groups is not supported.

        :param group_name: `str`
        :param entities: `list` of :class:`.Entity` objects | `list` of `str` entity names
        :return: True if success
        """
        _check_name(group_name)
        data = []
        for e in entities:
            data.append(e.name if isinstance(e, Entity) else e)

        response = self.conn.post(eg_delete_entities_url.format(group=quote(group_name, '')),
                                  data=data)
        return True