我们从Python开源项目中,提取了以下50个代码示例,用于说明如何使用six.moves.http_client.BAD_REQUEST。
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'])
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_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_assemble_node_failed(self, mock_request, mock_get_url, mock_delete_node): """Test allocate resource conflict when compose node""" CONF.set_override('url', 'http://localhost:8442/', group='podm') mock_get_url.return_value = '/redfish/v1/Nodes' # Fake response for getting nodes root fake_node_root_resp = fakes.mock_request_get(fakes.fake_nodes_root(), http_client.OK) # Fake response for allocating node fake_node_allocation_conflict = mock.MagicMock() fake_node_allocation_conflict.status_code = http_client.CREATED fake_node_allocation_conflict.headers['Location'] = \ os.path.normpath("/".join([CONF.podm.url, 'redfish/v1/Nodes/1'])) # Fake response for getting url of node assembling fake_node_detail = fakes.mock_request_get(fakes.fake_node_detail(), http_client.OK) # Fake response for assembling node fake_node_assemble_failed = fakes.mock_request_get( fakes.fake_assemble_node_failed(), http_client.BAD_REQUEST) mock_request.side_effect = [fake_node_root_resp, fake_node_allocation_conflict, fake_node_detail, fake_node_assemble_failed] with self.assertRaises(exception.RedfishException): redfish.compose_node({"name": "test_node"}) mock_delete_node.assert_called_once()
def test_check_creation_incomplete_parameters(self): incomplete_values = { 'name': 'name', 'url': 'url' } response = self.app.post('/v1/pod_managers', content_type='application/json', data=json.dumps(incomplete_values)) response = json.loads(response.data.decode()) self.assertEqual(http_client.BAD_REQUEST, response['status']) self.assertEqual('Validation Error', response['title'])
def test_check_creation_invalid_authentication(self): invalid_auth_values = { "name": "podm_name", "url": "https://10.0.0.2", 'authentication': { "username": "username", "password": "password" } } response = self.app.post('/v1/pod_managers', content_type='application/json', data=json.dumps(invalid_auth_values)) response = json.loads(response.data.decode()) self.assertEqual(http_client.BAD_REQUEST, response['status']) self.assertEqual('Validation Error', response['title'])
def test_compose_request_invalid_params(self, mock_conn): req = { "name": "test_request1", "properties": {"invalid_key": "invalid_value"} } resp = self.app.post('/v1/nodes', content_type='application/json', data=json.dumps(req)) response = json.loads(resp.data.decode()) self.assertEqual(http_client.BAD_REQUEST, response['status']) self.assertEqual('Validation Error', response['title'])
def test_node_action_request_invalid(self, mock_node, mock_connection): req = { "Boot": { "Type": "On" } } resp = self.app.post('/v1/nodes/fake-node/action', content_type='application/json', data=json.dumps(req)) response = json.loads(resp.data.decode()) self.assertEqual(http_client.BAD_REQUEST, response['status']) self.assertEqual('Validation Error', response['title'])
def test_server_exception_description_only(self): error_msg = "test error msg" error_body = _get_error_body(detail=error_msg) kwargs = {"valence_url": "http://localhost/"} client = http.HTTPClient(**kwargs) client.session = utils.mockSession( {'Content-Type': 'application/json'}, error_body, version=1, status_code=http_client.BAD_REQUEST) self.assertRaises(exc.BadRequest, client.json_request, 'GET', 'redfish/v1')
def test_from_response_know(self): method = "GET" url = "/fake" status_code = http_client.BAD_REQUEST json_data = {"error": {"message": "fake message", "details": "fake details"}} self.assert_exception(exceptions.BadRequest, method, url, status_code, json_data)
def test_from_response(self, mock_apiclient): fake_response = mock.Mock(status_code=http_client.BAD_REQUEST) exc.from_response(fake_response, error=self.error, method=self.method, url=self.url) self.assertEqual(http_client.BAD_REQUEST, fake_response.status_code) self.assertEqual(self.expected_json, fake_response.json()) mock_apiclient.assert_called_once_with( fake_response, method=self.method, url=self.url)
def test_from_response_status(self, mock_apiclient): fake_response = mock.Mock(status=http_client.BAD_REQUEST) fake_response.getheader.return_value = 'fake-header' delattr(fake_response, 'status_code') exc.from_response(fake_response, error=self.error, method=self.method, url=self.url) expected_header = {'Content-Type': 'fake-header'} self.assertEqual(expected_header, fake_response.headers) self.assertEqual(http_client.BAD_REQUEST, fake_response.status_code) self.assertEqual(self.expected_json, fake_response.json()) mock_apiclient.assert_called_once_with( fake_response, method=self.method, url=self.url)
def test_bad_payload(self): response = self.app.get('/credentials') self.assertEqual(http_client.BAD_REQUEST, response.status_code) response = self.app.put('/credentials', data=json.dumps(dict(bad_payload='test',)), content_type='application/json') self.assertEqual(http_client.BAD_REQUEST, response.status_code)
def test_validate_href(self, head_mock): href = 'http://1.2.3.4/abc.iso' response = head_mock.return_value response.status_code = http_client.OK utils.validate_href(href) head_mock.assert_called_once_with(href) response.status_code = http_client.NO_CONTENT self.assertRaises(exception.ImageRefValidationFailed, utils.validate_href, href) response.status_code = http_client.BAD_REQUEST self.assertRaises(exception.ImageRefValidationFailed, utils.validate_href, href)
def test_validate_href_error_code(self, head_mock): href = 'http://1.2.3.4/abc.iso' head_mock.return_value.status_code = http_client.BAD_REQUEST self.assertRaises(exception.ImageRefValidationFailed, utils.validate_href, href) head_mock.assert_called_once_with(href)
def set_key(self, req, dpid, key, **_kwargs): def _set_val(dpid, key): try: val = req.json if req.body else {} except ValueError: return Response(status=http_client.BAD_REQUEST, body='invalid syntax %s' % req.body) self.conf_switch.set_key(dpid, key, val) return None def _ret(_ret): return Response(status=http_client.CREATED) return self._do_key(dpid, key, _set_val, _ret)
def test_redir_not_allowed(self, *_args): """Test redirection not allowed error when protocol not http""" resp = self.app.get('/redir/srv/type2/cell1/foo/bar') self.assertEqual(resp.status_code, http_client.BAD_REQUEST) payload = json.loads(resp.data) self.assertEqual('Redirection not allowed for p2 protocol', payload['message'])
def test_aggregate_post_with_empty_az(self): body = {"name": "test_empty", "metadata": {"availability_zone": ""}} response = self.post_json( '/aggregates', body, headers=self.headers, expect_errors=True) self.assertEqual(http_client.BAD_REQUEST, response.status_code) self.assertEqual('application/json', response.content_type) self.assertTrue(response.json['error_message'])
def test_aggregate_post_with_empty_affinity_zone(self): body = {"name": "test_empty", "metadata": {"affinity_zone": ""}} response = self.post_json( '/aggregates', body, headers=self.headers, expect_errors=True) self.assertEqual(http_client.BAD_REQUEST, response.status_code) self.assertEqual('application/json', response.content_type) self.assertTrue(response.json['error_message'])