Python six.moves.http_client 模块,METHOD_NOT_ALLOWED 实例源码

我们从Python开源项目中,提取了以下11个代码示例,用于说明如何使用six.moves.http_client.METHOD_NOT_ALLOWED

项目:glare    作者:openstack    | 项目源码 | 文件源码
def test_http_error_response_codes(self):
        """Makes sure v1 unallowed methods return 405"""
        unallowed_methods = [
            ('/schemas', ['PUT', 'DELETE', 'HEAD', 'PATCH', 'POST']),
            ('/schemas/type_name', ['PUT', 'DELETE', 'HEAD', 'PATCH', 'POST']),
            ('/artifacts/type_name', ['PUT', 'DELETE', 'HEAD', 'PATCH']),
            ('/artifacts/type_name/artifact_id', ['PUT', 'HEAD', 'POST']),
            ('/artifacts/type_name/artifact_id/blob)name',
             ['HEAD', 'PATCH', 'POST']),
        ]
        api = router.API(routes.Mapper())
        for uri, methods in unallowed_methods:
            for method in methods:
                req = webob.Request.blank(uri)
                req.method = method
                res = req.get_response(api)
                self.assertEqual(http.METHOD_NOT_ALLOWED, res.status_int)

        # Makes sure not implemented methods return 405
        req = webob.Request.blank('/schemas/image')
        req.method = 'NonexistentMethod'
        res = req.get_response(api)
        self.assertEqual(http.METHOD_NOT_ALLOWED, res.status_int)
项目:solaris-ips    作者:oracle    | 项目源码 | 文件源码
def manifest_1(self, *tokens):
                """Outputs the contents of the manifest or uploads the
                manifest."""

                method = cherrypy.request.method
                if method == "GET":
                        return self.manifest_0(*tokens)
                elif method in ("POST", "PUT"):
                        return self.__upload_manifest(*tokens)
                raise cherrypy.HTTPError(http_client.METHOD_NOT_ALLOWED,
                    "{0} is not allowed".format(method))

        # We need to prevent cherrypy from processing the request body so that
        # manifest can parse the request body itself.  In addition, we also need
        # to set the timeout higher since the default is five minutes; not
        # really enough for a slow connection to upload content.
项目:solaris-ips    作者:oracle    | 项目源码 | 文件源码
def file_1(self, *tokens):
                """Outputs the contents of the file, named by the SHA hash
                name in the request path, directly to the client."""

                method = cherrypy.request.method
                if method == "GET":
                        return self.file_0(*tokens)
                elif method in ("POST", "PUT"):
                        return self.__upload_file(*tokens)
                raise cherrypy.HTTPError(http_client.METHOD_NOT_ALLOWED,
                    "{0} is not allowed".format(method))

        # We need to prevent cherrypy from processing the request body so that
        # file can parse the request body itself.  In addition, we also need to
        # set the timeout higher since the default is five minutes; not really
        # enough for a slow connection to upload content.
项目:GAMADV-XTD    作者:taers232c    | 项目源码 | 文件源码
def _do_revoke(self, http, token):
        """Revokes this credential and deletes the stored copy (if it exists).

        Args:
            http: an object to be used to make HTTP requests.
            token: A string used as the token to be revoked. Can be either an
                   access_token or refresh_token.

        Raises:
            TokenRevokeError: If the revoke request does not return with a
                              200 OK.
        """
        logger.info('Revoking token')
        query_params = {'token': token}
        token_revoke_uri = _helpers.update_query_params(
            self.revoke_uri, query_params)
        resp, content = transport.request(http, token_revoke_uri)
        if resp.status == http_client.METHOD_NOT_ALLOWED:
            body = urllib.parse.urlencode(query_params)
            resp, content = transport.request(http, token_revoke_uri,
                                              method='POST', body=body)
        if resp.status == http_client.OK:
            self.invalid = True
        else:
            error_msg = 'Invalid response {0}.'.format(resp.status)
            try:
                d = json.loads(_helpers._from_bytes(content))
                if 'error' in d:
                    error_msg = d['error']
            except (TypeError, ValueError):
                pass
            raise TokenRevokeError(error_msg)

        if self.store:
            self.store.delete()
项目:deb-python-oauth2client    作者:openstack    | 项目源码 | 文件源码
def test_token_revoke_405(self):
        original_credentials = self.credentials.to_json()
        self.credentials.refresh_token = None

        http = http_mock.HttpMockSequence([
            ({'status': http_client.METHOD_NOT_ALLOWED}, b''),
            ({'status': http_client.OK}, b''),
        ])
        _token_revoke_test_helper(
            self, revoke_raise=False, valid_bool_value=True,
            token_attr='access_token', http_mock=http)
        self.credentials = self.credentials.from_json(original_credentials)
项目:deb-python-oauth2client    作者:openstack    | 项目源码 | 文件源码
def _do_revoke(self, http, token):
        """Revokes this credential and deletes the stored copy (if it exists).

        Args:
            http: an object to be used to make HTTP requests.
            token: A string used as the token to be revoked. Can be either an
                   access_token or refresh_token.

        Raises:
            TokenRevokeError: If the revoke request does not return with a
                              200 OK.
        """
        logger.info('Revoking token')
        query_params = {'token': token}
        token_revoke_uri = _helpers.update_query_params(
            self.revoke_uri, query_params)
        resp, content = transport.request(http, token_revoke_uri)
        if resp.status == http_client.METHOD_NOT_ALLOWED:
            body = urllib.parse.urlencode(query_params)
            resp, content = transport.request(http, token_revoke_uri,
                                              method='POST', body=body)
        if resp.status == http_client.OK:
            self.invalid = True
        else:
            error_msg = 'Invalid response {0}.'.format(resp.status)
            try:
                d = json.loads(_helpers._from_bytes(content))
                if 'error' in d:
                    error_msg = d['error']
            except (TypeError, ValueError):
                pass
            raise TokenRevokeError(error_msg)

        if self.store:
            self.store.delete()
项目:REMAP    作者:REMAPApp    | 项目源码 | 文件源码
def test_token_revoke_405(self):
        original_credentials = self.credentials.to_json()
        self.credentials.refresh_token = None

        http = http_mock.HttpMockSequence([
            ({'status': http_client.METHOD_NOT_ALLOWED}, b''),
            ({'status': http_client.OK}, b''),
        ])
        _token_revoke_test_helper(
            self, revoke_raise=False, valid_bool_value=True,
            token_attr='access_token', http_mock=http)
        self.credentials = self.credentials.from_json(original_credentials)
项目:Webradio_v2    作者:Acer54    | 项目源码 | 文件源码
def _do_revoke(self, http, token):
        """Revokes this credential and deletes the stored copy (if it exists).

        Args:
            http: an object to be used to make HTTP requests.
            token: A string used as the token to be revoked. Can be either an
                   access_token or refresh_token.

        Raises:
            TokenRevokeError: If the revoke request does not return with a
                              200 OK.
        """
        logger.info('Revoking token')
        query_params = {'token': token}
        token_revoke_uri = _helpers.update_query_params(
            self.revoke_uri, query_params)
        resp, content = transport.request(http, token_revoke_uri)
        if resp.status == http_client.METHOD_NOT_ALLOWED:
            body = urllib.parse.urlencode(query_params)
            resp, content = transport.request(http, token_revoke_uri,
                                              method='POST', body=body)
        if resp.status == http_client.OK:
            self.invalid = True
        else:
            error_msg = 'Invalid response {0}.'.format(resp.status)
            try:
                d = json.loads(_helpers._from_bytes(content))
                if 'error' in d:
                    error_msg = d['error']
            except (TypeError, ValueError):
                pass
            raise TokenRevokeError(error_msg)

        if self.store:
            self.store.delete()
项目:GAMADV-X    作者:taers232c    | 项目源码 | 文件源码
def _do_revoke(self, http, token):
        """Revokes this credential and deletes the stored copy (if it exists).

        Args:
            http: an object to be used to make HTTP requests.
            token: A string used as the token to be revoked. Can be either an
                   access_token or refresh_token.

        Raises:
            TokenRevokeError: If the revoke request does not return with a
                              200 OK.
        """
        logger.info('Revoking token')
        query_params = {'token': token}
        token_revoke_uri = _helpers.update_query_params(
            self.revoke_uri, query_params)
        resp, content = transport.request(http, token_revoke_uri)
        if resp.status == http_client.METHOD_NOT_ALLOWED:
            body = urllib.parse.urlencode(query_params)
            resp, content = transport.request(http, token_revoke_uri,
                                              method='POST', body=body)
        if resp.status == http_client.OK:
            self.invalid = True
        else:
            error_msg = 'Invalid response {0}.'.format(resp.status)
            try:
                d = json.loads(_helpers._from_bytes(content))
                if 'error' in d:
                    error_msg = d['error']
            except (TypeError, ValueError):
                pass
            raise TokenRevokeError(error_msg)

        if self.store:
            self.store.delete()
项目:masakari    作者:openstack    | 项目源码 | 文件源码
def test_delete_and_update_notification(self, method, mock_client):
        url = '/v1/notifications/%s' % uuidsentinel.fake_notification
        fake_req = fakes.HTTPRequest.blank(url, use_admin_context=True)
        fake_req.headers['Content-Type'] = 'application/json'
        fake_req.method = method
        resp = fake_req.get_response(self.app)
        self.assertEqual(http.METHOD_NOT_ALLOWED, resp.status_code)
项目:os-xenapi    作者:openstack    | 项目源码 | 文件源码
def check_resp_status_and_retry(resp, image_id, url):
    # Note(Jesse): This branch sorts errors into those that are permanent,
    # those that are ephemeral, and those that are unexpected.
    if resp.status in (httplib.BAD_REQUEST,                      # 400
                       httplib.UNAUTHORIZED,                     # 401
                       httplib.PAYMENT_REQUIRED,                 # 402
                       httplib.FORBIDDEN,                        # 403
                       httplib.METHOD_NOT_ALLOWED,               # 405
                       httplib.NOT_ACCEPTABLE,                   # 406
                       httplib.PROXY_AUTHENTICATION_REQUIRED,    # 407
                       httplib.CONFLICT,                         # 409
                       httplib.GONE,                             # 410
                       httplib.LENGTH_REQUIRED,                  # 411
                       httplib.PRECONDITION_FAILED,              # 412
                       httplib.REQUEST_ENTITY_TOO_LARGE,         # 413
                       httplib.REQUEST_URI_TOO_LONG,             # 414
                       httplib.UNSUPPORTED_MEDIA_TYPE,           # 415
                       httplib.REQUESTED_RANGE_NOT_SATISFIABLE,  # 416
                       httplib.EXPECTATION_FAILED,               # 417
                       httplib.UNPROCESSABLE_ENTITY,             # 422
                       httplib.LOCKED,                           # 423
                       httplib.FAILED_DEPENDENCY,                # 424
                       httplib.UPGRADE_REQUIRED,                 # 426
                       httplib.NOT_IMPLEMENTED,                  # 501
                       httplib.HTTP_VERSION_NOT_SUPPORTED,       # 505
                       httplib.NOT_EXTENDED,                     # 510
                       ):
        raise PluginError("Got Permanent Error response [%i] while "
                          "uploading image [%s] to glance [%s]"
                          % (resp.status, image_id, url))
    # Nova service would process the exception
    elif resp.status == httplib.NOT_FOUND:                        # 404
        exc = XenAPI.Failure('ImageNotFound')
        raise exc
    # NOTE(nikhil): Only a sub-set of the 500 errors are retryable. We
    # optimistically retry on 500 errors below.
    elif resp.status in (httplib.REQUEST_TIMEOUT,                # 408
                         httplib.INTERNAL_SERVER_ERROR,          # 500
                         httplib.BAD_GATEWAY,                    # 502
                         httplib.SERVICE_UNAVAILABLE,            # 503
                         httplib.GATEWAY_TIMEOUT,                # 504
                         httplib.INSUFFICIENT_STORAGE,           # 507
                         ):
        raise RetryableError("Got Ephemeral Error response [%i] while "
                             "uploading image [%s] to glance [%s]"
                             % (resp.status, image_id, url))
    else:
        # Note(Jesse): Assume unexpected errors are retryable. If you are
        # seeing this error message, the error should probably be added
        # to either the ephemeral or permanent error list.
        raise RetryableError("Got Unexpected Error response [%i] while "
                             "uploading image [%s] to glance [%s]"
                             % (resp.status, image_id, url))