def test_flavor_create_incorrect_param(self):
flavor = self.flavor
flavor.pop('uuid')
# Test invalid value
flavor['properties']['memory']['capacity_mib'] = 10
response = self.app.post('/v1/flavors',
content_type='application/json',
data=json.dumps(self.flavor))
response = json.loads(response.data.decode())
self.assertEqual(http_client.BAD_REQUEST, response['status'])
self.assertEqual('Validation Error', response['title'])
# Test invalid key
flavor['properties']['invalid_key'] = 'invalid'
response = self.app.post('/v1/flavors',
content_type='application/json',
data=json.dumps(self.flavor))
response = json.loads(response.data.decode())
self.assertEqual(http_client.BAD_REQUEST, response['status'])
self.assertEqual('Validation Error', response['title'])
python类BAD_REQUEST的实例源码
def test_bad_payload(self):
response = self.app.post(
'/runs',
data=json.dumps(dict(inventory_file='localhost,',
options={'connection': 'local'})),
content_type='application/json')
self.assertEqual(http_client.BAD_REQUEST, response.status_code)
pb = 'tests/fixtures/playbooks/hello_world_with_fail.yml'
response = self.app.post(
'/runs',
data=json.dumps(dict(playbook_path=pb,
inventory_file='localhost,',
options={'connection': 'local',
'bad_option': 'bad'})),
content_type='application/json')
self.assertEqual(http_client.BAD_REQUEST, response.status_code)
def _handle_error(url, response):
"""Handle response status codes."""
handlers = {
http_client.NOT_FOUND: NotFoundError(
'Resource not found: {}'.format(url)
),
http_client.FOUND: AlreadyExistsError(
'Resource already exists: {}'.format(url)
),
http_client.FAILED_DEPENDENCY: ValidationError(response),
http_client.UNAUTHORIZED: NotAuthorizedError(response),
http_client.BAD_REQUEST: BadRequestError(response),
}
if response.status_code in handlers:
raise handlers[response.status_code]
def _is_safe_to_update_metadata(self, patch, aggregate_uuid):
"""Check if it's safe to update aggregate metadata"""
keys = ['availability_zone', 'affinity_zone']
# Check if this tries to change *keys* to empty.
for patch_dict in patch:
for key in keys:
if patch_dict['path'] == '/metadata/' + key \
and patch_dict['op'] != 'remove':
if not patch_dict['value']:
msg = _("Aggregate %(uuid)s does not support empty "
"named %(key)s") % {"uuid": aggregate_uuid,
"key": key}
raise wsme.exc.ClientSideError(
msg, status_code=http_client.BAD_REQUEST)
else:
self._check_metadata_conflicts(
aggregate_uuid, key, patch_dict['value'])
def check_validation_error(self, method, body, expected_detail, req=None):
if not req:
req = FakeRequest()
try:
method(body=body, req=req,)
except exception.ValidationError as ex:
self.assertEqual(http.BAD_REQUEST, ex.kwargs['code'])
if isinstance(expected_detail, list):
self.assertIn(ex.kwargs['detail'], expected_detail,
'Exception details did not match expected')
elif not re.match(expected_detail, ex.kwargs['detail']):
self.assertEqual(expected_detail, ex.kwargs['detail'],
'Exception details did not match expected')
except Exception as ex:
self.fail('An unexpected exception happens: %s' % ex)
else:
self.fail('Any exception does not happen.')
def do_GET(self):
"""
Handle a ping request
"""
if not self.path.endswith("/"): self.path += "/"
if self.path == "/ping/":
msg = "pong".encode("UTF-8")
self.send_response(HTTPStatus.OK)
self.send_header("Content-Type", "text/application")
self.send_header("Content-Length", len(msg))
self.end_headers()
self.wfile.write(msg)
else:
self.send_response(HTTPStatus.BAD_REQUEST)
self.end_headers()
def abandon_0(self, *tokens):
"""Aborts an in-flight transaction for the Transaction ID
specified in the request path. Returns no output."""
try:
# cherrypy decoded it, but we actually need it encoded.
trans_id = quote(tokens[0], "")
except IndexError:
trans_id = None
try:
self.repo.abandon(trans_id)
except srepo.RepositoryError as e:
# Assume a bad request was made. A 404 can't be
# returned here as misc.versioned_urlopen will interpret
# that to mean that the server doesn't support this
# operation.
raise cherrypy.HTTPError(http_client.BAD_REQUEST, str(e))
def after(self, state):
# Omit empty body. Some errors may not have body at this level yet.
if not state.response.body:
return
# Do nothing if there is no error.
# Status codes in the range 200 (OK) to 399 (400 = BAD_REQUEST) are not
# an error.
if (http_client.OK <= state.response.status_int <
http_client.BAD_REQUEST):
return
json_body = state.response.json
# Do not remove traceback when traceback config is set
if cfg.CONF.debug_tracebacks_in_api:
return
faultstring = json_body.get('faultstring')
traceback_marker = 'Traceback (most recent call last):'
if faultstring and traceback_marker in faultstring:
# Cut-off traceback.
faultstring = faultstring.split(traceback_marker, 1)[0]
# Remove trailing newlines and spaces if any.
json_body['faultstring'] = faultstring.rstrip()
# Replace the whole json. Cannot change original one because it's
# generated on the fly.
state.response.json = json_body
def test_known_http_error(self):
self.request.return_value.status_code = http_client.BAD_REQUEST
with open('sushy/tests/unit/json_samples/error.json', 'r') as f:
self.request.return_value.json.return_value = json.load(f)
with self.assertRaisesRegex(exceptions.BadRequestError,
'A general error has occurred') as cm:
self.conn._op('GET', 'http://foo.bar')
exc = cm.exception
self.assertEqual(http_client.BAD_REQUEST, exc.status_code)
self.assertIsNotNone(exc.body)
self.assertIn('A general error has occurred', exc.detail)
def raise_for_response(method, url, response):
"""Raise a correct error class, if needed."""
if response.status_code < http_client.BAD_REQUEST:
return
elif response.status_code == http_client.NOT_FOUND:
raise ResourceNotFoundError(method, url, response)
elif response.status_code == http_client.BAD_REQUEST:
raise BadRequestError(method, url, response)
elif response.status_code in (http_client.UNAUTHORIZED,
http_client.FORBIDDEN):
raise AccessError(method, url, response)
elif response.status_code >= http_client.INTERNAL_SERVER_ERROR:
raise ServerSideError(method, url, response)
else:
raise HTTPError(method, url, response)
def test_token_refresh_failure(self):
for status_code in client.REFRESH_STATUS_CODES:
http = http_mock.HttpMockSequence([
({'status': status_code}, b''),
({'status': http_client.BAD_REQUEST},
b'{"error":"access_denied"}'),
])
http = self.credentials.authorize(http)
with self.assertRaises(
client.HttpAccessTokenRefreshError) as exc_manager:
transport.request(http, 'http://example.com')
self.assertEqual(http_client.BAD_REQUEST,
exc_manager.exception.status)
self.assertTrue(self.credentials.access_token_expired)
self.assertEqual(None, self.credentials.token_response)
def test_token_revoke_failure(self):
http = http_mock.HttpMock(headers={'status': http_client.BAD_REQUEST})
_token_revoke_test_helper(
self, revoke_raise=True, valid_bool_value=False,
token_attr='refresh_token', http_mock=http)
def test_non_401_error_response(self):
http = http_mock.HttpMock(headers={'status': http_client.BAD_REQUEST})
http = self.credentials.authorize(http)
resp, content = transport.request(http, 'http://example.com')
self.assertEqual(http_client.BAD_REQUEST, resp.status)
self.assertEqual(None, self.credentials.token_response)
def test__do_refresh_request_non_json_failure(self):
response = http_mock.ResponseMock({'status': http_client.BAD_REQUEST})
content = u'Bad request'
error_msg = 'Invalid response {0}.'.format(int(response.status))
self._do_refresh_request_test_helper(response, content, error_msg)
def test__do_revoke_non_json_failure(self):
response = http_mock.ResponseMock({'status': http_client.BAD_REQUEST})
content = u'Bad request'
error_msg = 'Invalid response {0}.'.format(response.status)
self._do_revoke_test_helper(response, content, error_msg)
def test_token_revoke_failure(self):
http = http_mock.HttpMock(headers={'status': http_client.BAD_REQUEST})
_token_revoke_test_helper(
self, revoke_raise=True, valid_bool_value=False,
token_attr='access_token', http_mock=http)
def test_non_401_error_response(self):
http = http_mock.HttpMock(headers={'status': http_client.BAD_REQUEST})
http = self.credentials.authorize(http)
resp, content = transport.request(http, 'http://example.com')
self.assertEqual(http_client.BAD_REQUEST, resp.status)
def test_token_revoke_failure(self):
http = http_mock.HttpMock(headers={'status': http_client.BAD_REQUEST})
_token_revoke_test_helper(
self, revoke_raise=True, valid_bool_value=False,
token_attr='access_token', http_mock=http)
def test_step1_get_device_and_user_codes_non_json_failure(self):
status = int(http_client.BAD_REQUEST)
content = 'Nope not JSON.'
error_msg = 'Invalid response {0}.'.format(status)
self._step1_get_device_and_user_codes_fail_helper(status, content,
error_msg)
def test_exchange_failure(self):
http = http_mock.HttpMock(
headers={'status': http_client.BAD_REQUEST},
data=b'{"error":"invalid_request"}',
)
with self.assertRaises(client.FlowExchangeError):
self.flow.step2_exchange(code='some random code', http=http)
def test_exchange_code_for_token_fail(self):
http = http_mock.HttpMock(
headers={'status': http_client.BAD_REQUEST},
data=b'{"error":"invalid_request"}',
)
with self.assertRaises(client.FlowExchangeError):
client.credentials_from_code(
self.client_id, self.client_secret, self.scope,
self.code, http=http, redirect_uri=self.redirect_uri)
def test_exchange_code_and_file_for_token_fail(self):
http = http_mock.HttpMock(
headers={'status': http_client.BAD_REQUEST},
data=b'{"error":"invalid_request"}',
)
with self.assertRaises(client.FlowExchangeError):
client.credentials_from_clientsecrets_and_code(
datafile('client_secrets.json'), self.scope,
self.code, http=http)
def test_token_refresh_failure(self):
for status_code in client.REFRESH_STATUS_CODES:
http = http_mock.HttpMockSequence([
({'status': status_code}, b''),
({'status': http_client.BAD_REQUEST},
b'{"error":"access_denied"}'),
])
http = self.credentials.authorize(http)
with self.assertRaises(
client.HttpAccessTokenRefreshError) as exc_manager:
transport.request(http, 'http://example.com')
self.assertEqual(http_client.BAD_REQUEST,
exc_manager.exception.status)
self.assertTrue(self.credentials.access_token_expired)
self.assertEqual(None, self.credentials.token_response)
def test_token_revoke_failure(self):
http = http_mock.HttpMock(headers={'status': http_client.BAD_REQUEST})
_token_revoke_test_helper(
self, revoke_raise=True, valid_bool_value=False,
token_attr='refresh_token', http_mock=http)
def test_non_401_error_response(self):
http = http_mock.HttpMock(headers={'status': http_client.BAD_REQUEST})
http = self.credentials.authorize(http)
resp, content = transport.request(http, 'http://example.com')
self.assertEqual(http_client.BAD_REQUEST, resp.status)
self.assertEqual(None, self.credentials.token_response)
def test__do_refresh_request_non_json_failure(self):
response = http_mock.ResponseMock({'status': http_client.BAD_REQUEST})
content = u'Bad request'
error_msg = 'Invalid response {0}.'.format(int(response.status))
self._do_refresh_request_test_helper(response, content, error_msg)
def test__do_revoke_non_json_failure(self):
response = http_mock.ResponseMock({'status': http_client.BAD_REQUEST})
content = u'Bad request'
error_msg = 'Invalid response {0}.'.format(response.status)
self._do_revoke_test_helper(response, content, error_msg)
def test_token_revoke_failure(self):
http = http_mock.HttpMock(headers={'status': http_client.BAD_REQUEST})
_token_revoke_test_helper(
self, revoke_raise=True, valid_bool_value=False,
token_attr='access_token', http_mock=http)
def test_non_401_error_response(self):
http = http_mock.HttpMock(headers={'status': http_client.BAD_REQUEST})
http = self.credentials.authorize(http)
resp, content = transport.request(http, 'http://example.com')
self.assertEqual(http_client.BAD_REQUEST, resp.status)
def test_token_revoke_failure(self):
http = http_mock.HttpMock(headers={'status': http_client.BAD_REQUEST})
_token_revoke_test_helper(
self, revoke_raise=True, valid_bool_value=False,
token_attr='access_token', http_mock=http)
