Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_send(self):
with app.app_context():
responses.add(responses.GET, '{}/foo1'.format(self.rmq_base_url), status=200)
self.assertEqual(send('get', 'foo1').status_code, 200)
#
# timeouts and other requests errors
#
def callback(request):
raise requests.RequestException('Requests exception')
responses.add_callback(
responses.GET, '{}/foo2'.format(self.rmq_base_url), callback=callback,
)
with self.assertRaises(InternalServerError):
send('get', 'foo2')
#
# Test non-200 resposes with raise_for_status=True
#
responses.add(
responses.GET, '{}/foo3'.format(self.rmq_base_url), status=400,
)
with self.assertRaises(InternalServerError):
send('get', 'foo3')
#
# Test that we return a resposes with raise_for_status=False
#
responses.add(
responses.GET, '{}/foo4'.format(self.rmq_base_url), status=400,
got_request_exception.send(self, exception=e)
handler = self._find_error_handler(InternalServerError())
if self.propagate_exceptions:
# if we want to repropagate the exception, we can attempt to
# raise it with the whole traceback in case we can do that
# (the function was actually called from the except part)
# otherwise, we just raise the error again
if exc_value is e:
reraise(exc_type, exc_value, tb)
else:
raise e
self.log_exception((exc_type, exc_value, tb))
if handler is None:
return InternalServerError()
return handler(e)
def internal_error(self, e):
tmp_file = self.template_loader.resolve_path('500'+settings.GLOBAL.TEMPLATE_SUFFIX)
if tmp_file:
response = self.render(tmp_file, {'url':local.request.path}, status=500)
else:
response = InternalServerError()
log.exception(e)
return response
def wrap(*args, **kwargs):
try:
return f(*args, **kwargs)
except Exception as e:
_logger.exception("An exception occured during an http request")
se = _serialize_exception(e)
error = {
'code': 200,
'message': "Flectra Server Error",
'data': se
}
return werkzeug.exceptions.InternalServerError(json.dumps(error))
return wrap
got_request_exception.send(self, exception=e)
handler = self._find_error_handler(InternalServerError())
if self.propagate_exceptions:
# if we want to repropagate the exception, we can attempt to
# raise it with the whole traceback in case we can do that
# (the function was actually called from the except part)
# otherwise, we just raise the error again
if exc_value is e:
reraise(exc_type, exc_value, tb)
else:
raise e
self.log_exception((exc_type, exc_value, tb))
if handler is None:
return InternalServerError()
return handler(e)
def handle_exception(self, e):
"""Default exception handling that kicks in when an exception
occurs that is not caught. In debug mode the exception will
be re-raised immediately, otherwise it is logged and the handler
for a 500 internal server error is used. If no such handler
exists, a default 500 internal server error message is displayed.
.. versionadded:: 0.3
"""
exc_type, exc_value, tb = sys.exc_info()
got_request_exception.send(self, exception=e)
handler = self._find_error_handler(InternalServerError())
if self.propagate_exceptions:
# if we want to repropagate the exception, we can attempt to
# raise it with the whole traceback in case we can do that
# (the function was actually called from the except part)
# otherwise, we just raise the error again
if exc_value is e:
reraise(exc_type, exc_value, tb)
else:
raise e
self.log_exception((exc_type, exc_value, tb))
if handler is None:
return InternalServerError()
return self.finalize_request(handler(e), from_error_handler=True)
if mediatype in self.representations:
resp = self.representations[mediatype](data, *args, **kwargs)
resp.headers['Access-Control-Allow-Origin'] = current_app.config['ACCESS_CONTROL_ALLOW_ORIGIN_HEADER']
resp.headers['Access-Control-Allow-Headers'] = current_app.config['ACCESS_CONTROL_ALLOW_HEADERS']
resp.headers['Access-Control-Allow-Methods'] = current_app.config['ACCESS_CONTROL_ALLOW_METHODS']
resp.headers['Content-Type'] = mediatype
return resp
elif mediatype == 'text/plain':
resp = original_flask_make_response(str(data), *args, **kwargs)
resp.headers['Access-Control-Allow-Origin'] = current_app.config['ACCESS_CONTROL_ALLOW_ORIGIN_HEADER']
resp.headers['Access-Control-Allow-Headers'] = current_app.config['ACCESS_CONTROL_ALLOW_HEADERS']
resp.headers['Access-Control-Allow-Methods'] = current_app.config['ACCESS_CONTROL_ALLOW_METHODS']
resp.headers['Content-Type'] = 'text/plain'
return resp
else:
raise InternalServerError()
got_request_exception.send(self, exception=e)
handler = self._find_error_handler(InternalServerError())
if self.propagate_exceptions:
# if we want to repropagate the exception, we can attempt to
# raise it with the whole traceback in case we can do that
# (the function was actually called from the except part)
# otherwise, we just raise the error again
if exc_value is e:
reraise(exc_type, exc_value, tb)
else:
raise e
self.log_exception((exc_type, exc_value, tb))
if handler is None:
return InternalServerError()
return handler(e)
def show_endorsers(arxiv_id: str) -> Response:
"""Show endorsers for an article."""
raise InternalServerError(f'Not yet implemented {arxiv_id}')
create_time=self.util.get_now(),
update_time=self.util.get_now(),
description=args[BaseTemplate.DESCRIPTION],
virtual_environment_count=len(args[BaseTemplate.VIRTUAL_ENVIRONMENTS]))
else:
# update record
self.db.update_object(template,
url=context.url,
local_path=context.get("physical_path"),
update_time=self.util.get_now(),
description=args[BaseTemplate.DESCRIPTION],
virtual_environment_count=len(args[BaseTemplate.VIRTUAL_ENVIRONMENTS]))
self.cache_manager.invalidate(self.__get_template_cache_key(template.id))
except Exception as ex:
self.log.error(ex)
raise InternalServerError(description="insert or update record in db failed")