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)
python类METHOD_NOT_ALLOWED的实例源码
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.
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.
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()
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)
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()
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)
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()
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()
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)
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))
