我们从Python开源项目中，提取了以下36个代码示例，用于说明如何使用six.moves.http_client.INTERNAL_SERVER_ERROR。
def add(self): params = None try: params = utils.CNIParameters(flask.request.get_json()) LOG.debug('Received addNetwork request. CNI Params: %s', params) vif = self.plugin.add(params) data = jsonutils.dumps(vif.obj_to_primitive()) except exceptions.ResourceNotReady as e: LOG.error("Timed out waiting for requested pod to appear in " "registry: %s.", e) return '', httplib.GATEWAY_TIMEOUT, self.headers except Exception: LOG.exception('Error when processing addNetwork request. CNI ' 'Params: %s', params) return '', httplib.INTERNAL_SERVER_ERROR, self.headers return data, httplib.ACCEPTED, self.headers
def delete(self): params = None try: params = utils.CNIParameters(flask.request.get_json()) LOG.debug('Received delNetwork request. CNI Params: %s', params) self.plugin.delete(params) except exceptions.ResourceNotReady as e: # NOTE(dulek): It's better to ignore this error - most of the time # it will happen when pod is long gone and kubelet # overzealously tries to delete it from the network. # We cannot really do anything without VIF annotation, # so let's just tell kubelet to move along. LOG.warning("Timed out waiting for requested pod to appear in " "registry: %s. Ignoring.", e) return '', httplib.NO_CONTENT, self.headers except Exception: LOG.exception('Error when processing delNetwork request. CNI ' 'Params: %s.', params) return '', httplib.INTERNAL_SERVER_ERROR, self.headers return '', httplib.NO_CONTENT, self.headers
def test_get_new_token_should_throw_ecsclientexception_500(self): self.requests_mock.register_uri('GET', 'https://127.0.0.1:4443/login', status_code=http_client.INTERNAL_SERVER_ERROR) with super(testtools.TestCase, self).assertRaises(ECSClientException) as error: self.token_request.get_new_token() exception = error.exception self.assertEqual(exception.http_status, http_client.INTERNAL_SERVER_ERROR)
def test_token_validation_500(self, mock_get_existing_token): self.requests_mock.register_uri('GET', 'https://127.0.0.1:4443/user/whoami', status_code=http_client.INTERNAL_SERVER_ERROR) mock_get_existing_token.return_value = 'EXISTING-TOKEN-123' with super(testtools.TestCase, self).assertRaises(ECSClientException) as error: self.token_request.get_token() exception = error.exception self.assertEqual(exception.message, "Token validation error (Code: 500)") self.assertEqual(exception.http_status, http_client.INTERNAL_SERVER_ERROR)
def test_list_nodes_throw_exception(self, mock_get_token): self.requests_mock.register_uri('GET', 'https://127.0.0.1:4443/vdc/nodes', status_code=http_client.INTERNAL_SERVER_ERROR, text='Server Error') mock_get_token.return_value = 'FAKE-TOKEN-123' with super(testtools.TestCase, self).assertRaises(ECSClientException) as error: self.client.node.list() exception = error.exception self.assertEqual(self.requests_mock.last_request.method, 'GET') self.assertEqual(self.requests_mock.last_request.url, 'https://127.0.0.1:4443/vdc/nodes') self.assertEqual(self.requests_mock.last_request.headers['x-sds-auth-token'], 'FAKE-TOKEN-123') self.assertEqual(exception.http_response_content, 'Server Error') self.assertEqual(exception.http_status, http_client.INTERNAL_SERVER_ERROR)
def test_server_error(self): self.request.return_value.status_code = ( http_client.INTERNAL_SERVER_ERROR) self.request.return_value.json.side_effect = ValueError('no json') with self.assertRaisesRegex(exceptions.ServerSideError, 'unknown error') as cm: self.conn._op('GET', 'http://foo.bar') exc = cm.exception self.assertEqual(http_client.INTERNAL_SERVER_ERROR, exc.status_code)
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__do_refresh_request_basic_failure(self): response = http_mock.ResponseMock( {'status': http_client.INTERNAL_SERVER_ERROR}) content = u'{}' error_msg = 'Invalid response {0}.'.format(int(response.status)) self._do_refresh_request_test_helper(response, content, error_msg)
def test__do_revoke_basic_failure(self): response = http_mock.ResponseMock( {'status': http_client.INTERNAL_SERVER_ERROR}) content = u'{}' error_msg = 'Invalid response {0}.'.format(response.status) self._do_revoke_test_helper(response, content, error_msg)
def test__do_retrieve_scopes_basic_failure(self): response = http_mock.ResponseMock( {'status': http_client.INTERNAL_SERVER_ERROR}) content = u'{}' error_msg = 'Invalid response {0}.'.format(response.status) self._do_retrieve_scopes_test_helper(response, content, error_msg)
def test_step1_get_device_and_user_codes_basic_failure(self): status = int(http_client.INTERNAL_SERVER_ERROR) content = b'{}' error_msg = 'Invalid response {0}.'.format(status) self._step1_get_device_and_user_codes_fail_helper(status, content, error_msg)
def test_delete_composednode_fail(self, mock_request, mock_get_url, mock_make_response): mock_get_url.return_value = '/redfish/v1/Nodes' delete_result = fakes.fake_delete_composednode_fail() fake_resp = fakes.mock_request_get(delete_result, http_client.INTERNAL_SERVER_ERROR) mock_request.return_value = fake_resp self.assertRaises(exception.RedfishException, redfish.delete_composed_node, 101) self.assertFalse(mock_make_response.called)
def handle_error(self, error): if issubclass(error.__class__, exception.ValenceError): LOG.debug(traceback.format_exc()) return utils.make_response(error.status, error.as_dict()) elif hasattr(error, 'status'): LOG.debug(traceback.format_exc()) return utils.make_response(error.code, exception.httpexception(error)) else: # Valence will not throw general exception in normal case, so use # LOG.error() to record it. LOG.error(traceback.format_exc()) exc = exception.generalexception(error, http_client.INTERNAL_SERVER_ERROR) return utils.make_response(http_client.INTERNAL_SERVER_ERROR, exc)
def test_server_exception_empty_body(self): error_body = _get_error_body() 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.INTERNAL_SERVER_ERROR) self.assertRaises(exc.InternalServerError, client.json_request, 'GET', 'redfish/v1')
def test_server_exception_msg_only(self): error_body = "test error msg" error_body = _get_error_body(detail=error_body) 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.INTERNAL_SERVER_ERROR) self.assertRaises(exc.InternalServerError, client.json_request, 'GET', 'redfish/v1')
def test_get_bad_json(self, resp_mock): """Test treadmill.restclient.get bad JSON""" resp_mock.return_value.status_code = http_client.INTERNAL_SERVER_ERROR resp_mock.return_value.text = '{"bad json"' resp_mock.return_value.json.side_effect = sjs.JSONDecodeError( 'Foo', '{"bad json"', 1 ) self.assertRaises( restclient.MaxRequestRetriesError, restclient.get, 'http://foo.com', '/', retries=1)
def unexpected_error(e): """Handle exceptions by returning swagger-compliant json.""" logging.exception('An error occured while processing the request.') response = jsonify({ 'code': http_client.INTERNAL_SERVER_ERROR, 'message': 'Exception: {}'.format(e)}) response.status_code = http_client.INTERNAL_SERVER_ERROR return response
def server(self): """Provides a test HTTP server. The test server is automatically created before a test and destroyed at the end. The server is serving a test application that can be used to verify requests. """ app = flask.Flask(__name__) app.debug = True # pylint: disable=unused-variable # (pylint thinks the flask routes are unusued.) @app.route('/basic') def index(): header_value = flask.request.headers.get('x-test-header', 'value') headers = {'X-Test-Header': header_value} return 'Basic Content', http_client.OK, headers @app.route('/server_error') def server_error(): return 'Error', http_client.INTERNAL_SERVER_ERROR @app.route('/wait') def wait(): time.sleep(3) return 'Waited' # pylint: enable=unused-variable server = WSGIServer(application=app.wsgi_app) server.start() yield server server.stop()
def test_request_error(self, server): request = self.make_request() response = request(url=server.url + '/server_error', method='GET') assert response.status == http_client.INTERNAL_SERVER_ERROR assert response.data == b'Error'
def test_default_error_msg_with_kwargs(self): class FakeMasakariException(exception.MasakariException): msg_fmt = "default message: %(code)s" exc = FakeMasakariException(code=int(http.INTERNAL_SERVER_ERROR)) self.assertEqual('default message: 500', six.text_type(exc)) self.assertEqual('default message: 500', exc.message)
def test_error_msg_exception_with_kwargs(self): class FakeMasakariException(exception.MasakariException): msg_fmt = "default message: %(misspelled_code)s" exc = FakeMasakariException(code=int(http.INTERNAL_SERVER_ERROR), misspelled_code='blah') self.assertEqual('default message: blah', six.text_type(exc)) self.assertEqual('default message: blah', exc.message)
def test_error_code_from_kwarg(self): class FakeMasakariException(exception.MasakariException): code = http.INTERNAL_SERVER_ERROR exc = FakeMasakariException(code=http.NOT_FOUND) self.assertEqual(exc.kwargs['code'], http.NOT_FOUND)
def test_instantiate_without_title_known_code(self): exc = exception.ConvertedException(int(http.INTERNAL_SERVER_ERROR)) self.assertEqual(exc.title, status_reasons[http.INTERNAL_SERVER_ERROR])
def _raise_exc(exc): raise exc(int(http.INTERNAL_SERVER_ERROR))
def test_call_raises_exception(self): class FakeController(object): def index(self, shirt, pants=None): return (shirt, pants) resource = wsgi.Resource(FakeController(), None, None) with mock.patch('glare.common.wsgi.Resource.dispatch', side_effect=Exception("test exception")): request = wsgi.Request.blank('/') response = resource.__call__(request) self.assertIsInstance(response, webob.exc.HTTPInternalServerError) self.assertEqual(http.INTERNAL_SERVER_ERROR, response.status_code)
def __clear_cache(depot, pub): if not pub: shutil.rmtree(os.path.join(depot.tmp_root, "feed"), True) return pathname = __get_cache_pathname(depot, pub) try: if os.path.exists(pathname): os.remove(pathname) except IOError: raise cherrypy.HTTPError( http_client.INTERNAL_SERVER_ERROR, "Unable to clear feed cache.")
def default_error_page(status=http_client.NOT_FOUND, message="oops", traceback=None, version=None): """This function is registered as the default error page for CherryPy errors. This sets the response headers to be uncacheable, and then returns a HTTP response.""" response = cherrypy.response for key in ('Cache-Control', 'Pragma'): if key in response.headers: del response.headers[key] # Server errors are interesting, so let's log them. In the case # of an internal server error, we send a 404 to the client. but # log the full details in the server log. if (status == http_client.INTERNAL_SERVER_ERROR or status.startswith("500 ")): # Convert the error to a 404 to obscure implementation # from the client, but log the original error to the # server logs. error = cherrypy._cperror._HTTPErrorTemplate % \ {"status": http_client.NOT_FOUND, "message": http_client.responses[http_client.NOT_FOUND], "traceback": "", "version": cherrypy.__version__} print("Path that raised exception was {0}".format( cherrypy.request.path_info)) print(message) return error else: error = cherrypy._cperror._HTTPErrorTemplate % \ {"status": http_client.NOT_FOUND, "message": message, "traceback": "", "version": cherrypy.__version__} return error
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))
def from_response(response, method, url): """Returns an instance of :class:`HttpError` or subclass based on response. :param response: instance of `requests.Response` class :param method: HTTP method used for request :param url: URL used for request """ req_id = response.headers.get('X-openstack-request-id') # NOTE(hdd) true for older versions of nova and cinder if not req_id: req_id = response.headers.get('X-compute-request-id') kwargs = { 'http_status': response.status_code, 'response': response, 'method': method, 'url': url, 'request_id': req_id } if 'retry_after' in response.headers: kwargs['retry_after'] = response.headers['retry_after'] content_type = response.headers.get('Content-Type', "") if content_type.startswith('application/json'): try: body = response.json() except ValueError: pass else: if isinstance(body, dict): error = body.get(list(body)[0]) if isinstance(error, dict): kwargs['message'] = (error.get('message') or error.get('faultstring')) kwargs['details'] = (error.get('details') or six.text_type(body)) elif content_type.startswith("text/"): kwargs['details'] = getattr(response, 'text', '') try: cls = _code_map[response.status_code] except KeyError: # 5XX status codes are server errors if response.status_code >= http_client.INTERNAL_SERVER_ERROR: cls = HttpServerError # 4XX status codes are client request errors elif (http_client.BAD_REQUEST <= response.status_code < http_client.INTERNAL_SERVER_ERROR): cls = HTTPClientError else: cls = HttpError return cls(**kwargs)
def test_callback_view_errors(self): # Error supplied to callback with self.app.test_client() as client: with client.session_transaction() as session: session['google_oauth2_csrf_token'] = 'tokenz' response = client.get('/oauth2callback?state={}&error=something') self.assertEqual(response.status_code, httplib.BAD_REQUEST) self.assertIn('something', response.data.decode('utf-8')) # CSRF mismatch with self.app.test_client() as client: with client.session_transaction() as session: session['google_oauth2_csrf_token'] = 'goodstate' state = json.dumps({ 'csrf_token': 'badstate', 'return_url': '/return_url' }) response = client.get( '/oauth2callback?state={0}&code=codez'.format(state)) self.assertEqual(response.status_code, httplib.BAD_REQUEST) # KeyError, no CSRF state. with self.app.test_client() as client: response = client.get('/oauth2callback?state={}&code=codez') self.assertEqual(response.status_code, httplib.BAD_REQUEST) # Code exchange error with self.app.test_client() as client: state = self._setup_callback_state(client) with Http2Mock(status=httplib.INTERNAL_SERVER_ERROR): response = client.get( '/oauth2callback?state={0}&code=codez'.format(state)) self.assertEqual(response.status_code, httplib.BAD_REQUEST) # Invalid state json with self.app.test_client() as client: with client.session_transaction() as session: session['google_oauth2_csrf_token'] = 'tokenz' state = '[{' response = client.get( '/oauth2callback?state={0}&code=codez'.format(state)) self.assertEqual(response.status_code, httplib.BAD_REQUEST) # Missing flow. with self.app.test_client() as client: with client.session_transaction() as session: session['google_oauth2_csrf_token'] = 'tokenz' state = json.dumps({ 'csrf_token': 'tokenz', 'return_url': '/return_url' }) response = client.get( '/oauth2callback?state={0}&code=codez'.format(state)) self.assertEqual(response.status_code, httplib.BAD_REQUEST)
def from_response(response, method, url): """Returns an instance of :class:`HttpError` or subclass based on response. :param response: instance of `requests.Response` class :param method: HTTP method used for request :param url: URL used for request """ req_id = response.headers.get("x-openstack-request-id") # NOTE(hdd) true for older versions of nova and cinder if not req_id: req_id = response.headers.get("x-compute-request-id") kwargs = { "http_status": response.status_code, "response": response, "method": method, "url": url, "request_id": req_id, } if "retry-after" in response.headers: kwargs["retry_after"] = response.headers["retry-after"] content_type = response.headers.get("Content-Type", "") if content_type.startswith("application/json"): try: body = response.json() except ValueError: pass else: if isinstance(body, dict): error = body.get(list(body)[0]) if isinstance(error, dict): kwargs["message"] = (error.get("message") or error.get("faultstring")) kwargs["details"] = (error.get("details") or six.text_type(body)) elif content_type.startswith("text/"): kwargs["details"] = getattr(response, 'text', '') try: cls = _code_map[response.status_code] except KeyError: # 5XX status codes are server errors if response.status_code >= http_client.INTERNAL_SERVER_ERROR: cls = HttpServerError # 4XX status codes are client request errors elif (http_client.BAD_REQUEST <= response.status_code < http_client.INTERNAL_SERVER_ERROR): cls = HTTPClientError else: cls = HttpError return cls(**kwargs)
