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_clusters_listing_with_no_etcd_result(self):
"""
Verify listing Clusters handles no etcd result properly.
"""
with mock.patch('cherrypy.engine.publish') as _publish:
_publish.return_value = [[[], etcd.EtcdKeyNotFound()]]
body = self.simulate_request('/api/v0/clusters')
self.assertEqual(self.srmock.status, falcon.HTTP_404)
self.assertEqual('{}', body[0])
def on_head(self, req, resp, project_id, queue_name):
if self._queue_controller.exists(queue_name, project=project_id):
resp.status = falcon.HTTP_204
else:
resp.status = falcon.HTTP_404
resp.content_location = req.path
def on_get(self, req, resp, session_id):
# GET /v1/sessions/{session_id} retrieves the specified session
# search in body
user_id = req.get_header('X-User-ID') or ''
obj = self.db.get_session(user_id=user_id, session_id=session_id)
if obj:
resp.body = obj
else:
resp.status = falcon.HTTP_404
}
ERR_DATABASE_ROLLBACK = {
'status': falcon.HTTP_500,
'code': 77,
'title': 'Database Rollback Error'
}
ERR_DATABASE_CONNECTION = {
'status': falcon.HTTP_500,
'code': 76,
'title': 'Database Connection Error'
}
ERR_NOT_SUPPORTED = {
'status': falcon.HTTP_404,
'code': 10,
'title': 'Not Supported'
}
ERR_USER_NOT_EXISTS = {
'status': falcon.HTTP_404,
'code': 21,
'title': 'User Not Exists'
}
ERR_USER_ALREADY_EXISTS = {
'status': falcon.HTTP_400,
'code': 25,
'message': 'Users Already Exists'
}
def on_get(self, request, response, user_id=None):
"""Responds to GET request for users."""
if user_id == self.config['relay_uid']:
response.status = falcon.HTTP_200
response.body = json.dumps({})
self.api.as_register(user_id, self.config['as_token'])
else:
response.status = falcon.HTTP_404
response.body = json.dumps({})
def process_response(self, req, resp, resource=''):
"""Intercept main 404 response by Falcon
If the API hits a non existing endpoint, it will trigger a customized
404 response that will redirect people to the documentation.
Raises:
HTTP 404: A falcon.HTTP_404 error
Returns:
JSON: A customized JSON response
"""
if resp.status == falcon.HTTP_404:
resp.body = json.dumps({"message": "Resource not found",
"documentation": settings.__docs__})
def _producer_not_found():
"""
sends an http 404 response to the caller
"""
api.abort(falcon.HTTP_404, 'Unable to locate event producer.')
def on_get(self, req, resp, site_name=None):
if site_name is None:
site_name = 'bizocean'
elif not self.site_exists(site_name, "swallow"):
resp.status = falcon.HTTP_404
message = 'site name not found:{0}'.format(site_name)
resp.body = message
self.logger.error(message)
return
resp.set_header('Access-Control-Allow-Origin', '*')
"""
item_dict = { key: (
user_data,
validate_schema
),
}
key is BigQuery's column name
"""
elif filetype == 'html':
resp.content_type = 'text/html'
elif filetype == 'css':
resp.content_type = 'text/css'
else:
resp.content_type = 'text/plain'
resp.set_header('Access-Control-Allow-Origin', '*')
name = os.path.normpath(name)
filetype = os.path.normpath(filetype)
file_path = os.path.join('static', filetype, name)
try:
with open(file_path, 'r', encoding='utf8') as f:
raw_body = f.read()
except Exception as e:
resp.body = "{0} {1}".format(e, name)
resp.status = falcon.HTTP_404
return
else:
resp.body = raw_body.replace('{%oceanus_host%}',
OCEANUS_SWALLOW_HOST)
def on_post(self, req, resp, site_name):
if not self.site_exists(site_name, "pirate"):
resp.status = falcon.HTTP_404
message = 'site name not found:{0}'.format(site_name)
resp.body = message
self.logger.error(message)
return
self.logger.debug("{}".format(req.query_string))
resp.set_header('Access-Control-Allow-Origin', '*')
"""
item_dict = { key: (
user_data,
validate_schema
),
}
key is BigQuery's column name
"""
item_dict = {
'dt': (