def on_put(self, request, response, txn_id=None):
"""Responds to PUT request containing events."""
response.body = "{}"
# Check whether repeat txn_id
if not self._is_new(txn_id):
response.status = falcon.HTTP_200
return
request.context["body"] = request.stream.read()
try:
events = json.loads(request.context["body"].decode("utf-8"))["events"]
except(KeyError, ValueError, UnicodeDecodeError):
response.status = falcon.HTTP_400
response.body = "Malformed request body"
return
if self.handler(EventStream(events, self.Api)):
response.status = falcon.HTTP_200
else:
response.status = falcon.HTTP_400
python类HTTP_400的实例源码
def on_get(self, req, resp, uni, term):
# The term must be a string since the threads represent them as such
if uni in uniThreads and term in uniThreads[uni].getTerms():
if uniThreads[uni].isScraping:
# we don't want to return data while scraping, send error (configure nginx to send stale data if it can)
resp.status = falcon.HTTP_500
resp.body = json.dumps(
{"error": "We're currently scraping this university, please check back in a couple minutes!"}
).encode('utf-8')
else:
# Get course/subject list
subject_list = json.dumps(uniThreads[uni].getSubjectListAll(term), sort_keys=True).encode('utf-8')
# set the etag and body
resp.etag = "W/" + hashlib.sha1(subject_list).hexdigest()
resp.body = subject_list
else:
# Couldn't find the uni or term, send error
resp.status = falcon.HTTP_400
resp.body = json.dumps(
{"error": "The specified university or term was not found"}
).encode('utf-8')
def login(username, password, response=None):
"""
Logs user in. Returns status 400 response if user doesn't exist or password is wrong.
:param username: a username
:param password: a password
:return: a token to authenticate with
"""
if not username or not username.strip():
logger.debug("Tried to log in with empty username")
response.status = falcon.HTTP_400
return "empty username"
username = username.strip().lower()
client = _get_client(username)
if not client:
logger.debug("Tried to log in with unknown username: %s", username)
response.status = falcon.HTTP_400
return "unknown"
success = bcrypt_sha256.verify(password, client.pw_hash)
if not success:
logger.debug("Tried to log in with wrong password as user %s", username)
response.status = falcon.HTTP_400
return "wrong password"
logger.debug("New user login: %s", username)
return _create_user_token(username, client.permissions)
def suggestions(api_name, max_fetch: hug.types.number = 10, response=None):
max_fetch = min(10, max(1, max_fetch))
try:
api = music_api_names[api_name]
except KeyError:
logger.debug("Requested suggestions for unknown API %s", api_name)
response.status = falcon.HTTP_400
return "Unknown API"
if isinstance(api, AbstractSongProvider):
return list(map(Song.to_json, api.get_suggestions(max_fetch)))
else:
logger.debug("Tried to get suggestions for API %s which isn't a SongProvider", api_name)
response.status = falcon.HTTP_400
return []
def get_active_playlist(user: hug.directives.user, response=None):
"""
Get the active playlist of the offline API.
:return: the active playlist of the form {'playlist_id': <id>, 'playlist_name': <name>}
"""
if not has_permission(user, ["admin", "mod", "select_playlist"]):
response.status = falcon.HTTP_FORBIDDEN
return "Forbidden"
try:
api = music_api_names['offline_api']
playlist_tuple = api.get_active_playlist()
if not playlist_tuple:
return None
return {'playlist_id': playlist_tuple[0], 'playlist_name': playlist_tuple[1]}
except KeyError:
response.status = falcon.HTTP_400
return "Not in offline mode"
def mark_active(playlist_id: str, user: hug.directives.user, response=None):
"""
Mark a playlist as active in offline mode.
Needs admin, mod or select_playlist permission.
:param playlist_id: the playlist ID as returned by get_available_playlists
"""
if not has_permission(user, ["admin", "mod", "select_playlist"]):
response.status = falcon.HTTP_FORBIDDEN
return "Forbidden"
try:
try:
api = music_api_names['offline_api']
except KeyError:
response.status = falcon.HTTP_400
return "Not in offline mode"
api.set_active_playlist(playlist_id)
return "OK"
except ValueError:
response.status = falcon.HTTP_UNPROCESSABLE_ENTITY
return "Unknown ID"
def on_post(self, req, resp):
try:
manifest = self.req_yaml(req)
documents = list(manifest)
message = {
'valid': validate_armada_documents(documents)
}
resp.status = falcon.HTTP_200
resp.body = json.dumps(message)
resp.content_type = 'application/json'
except Exception:
err_message = 'Failed to validate Armada Manifest'
self.error(req.context, err_message)
self.return_error(
resp, falcon.HTTP_400, message=err_message)
def task_validate_design(self, req, resp, json_data):
"""Create async task for validate design."""
action = json_data.get('action', None)
if action != 'validate_design':
self.error(
req.context,
"Task body ended up in wrong handler: action %s in task_validate_design"
% action)
self.return_error(
resp, falcon.HTTP_500, message="Error", retry=False)
try:
task = self.create_task(json_data, req.context)
resp.body = json.dumps(task.to_dict())
resp.append_header('Location',
"/api/v1.0/tasks/%s" % str(task.task_id))
resp.status = falcon.HTTP_201
except errors.InvalidFormat as ex:
self.error(req.context, ex.msg)
self.return_error(
resp, falcon.HTTP_400, message=ex.msg, retry=False)
def task_prepare_site(self, req, resp, json_data):
"""Create async task for prepare site."""
action = json_data.get('action', None)
if action != 'prepare_site':
self.error(
req.context,
"Task body ended up in wrong handler: action %s in task_prepare_site"
% action)
self.return_error(
resp, falcon.HTTP_500, message="Error", retry=False)
try:
task = self.create_task(json_data, req.context)
resp.body = json.dumps(task.to_dict())
resp.append_header('Location',
"/api/v1.0/tasks/%s" % str(task.task_id))
resp.status = falcon.HTTP_201
except errors.InvalidFormat as ex:
self.error(req.context, ex.msg)
self.return_error(
resp, falcon.HTTP_400, message=ex.msg, retry=False)
def task_verify_nodes(self, req, resp, json_data):
"""Create async task for verify node."""
action = json_data.get('action', None)
if action != 'verify_nodes':
self.error(
req.context,
"Task body ended up in wrong handler: action %s in task_verify_nodes"
% action)
self.return_error(
resp, falcon.HTTP_500, message="Error", retry=False)
try:
task = self.create_task(json_data, req.context)
resp.body = json.dumps(task.to_dict())
resp.append_header('Location',
"/api/v1.0/tasks/%s" % str(task.task_id))
resp.status = falcon.HTTP_201
except errors.InvalidFormat as ex:
self.error(req.context, ex.msg)
self.return_error(
resp, falcon.HTTP_400, message=ex.msg, retry=False)
def task_prepare_nodes(self, req, resp, json_data):
"""Create async task for prepare node."""
action = json_data.get('action', None)
if action != 'prepare_nodes':
self.error(
req.context,
"Task body ended up in wrong handler: action %s in task_prepare_nodes"
% action)
self.return_error(
resp, falcon.HTTP_500, message="Error", retry=False)
try:
task = self.create_task(json_data, req.context)
resp.body = json.dumps(task.to_dict())
resp.append_header('Location',
"/api/v1.0/tasks/%s" % str(task.task_id))
resp.status = falcon.HTTP_201
except errors.InvalidFormat as ex:
self.error(req.context, ex.msg)
self.return_error(
resp, falcon.HTTP_400, message=ex.msg, retry=False)
def task_deploy_nodes(self, req, resp, json_data):
"""Create async task for deploy node."""
action = json_data.get('action', None)
if action != 'deploy_nodes':
self.error(
req.context,
"Task body ended up in wrong handler: action %s in task_deploy_nodes"
% action)
self.return_error(
resp, falcon.HTTP_500, message="Error", retry=False)
try:
task = self.create_task(json_data, req.context)
resp.body = json.dumps(task.to_dict())
resp.append_header('Location',
"/api/v1.0/tasks/%s" % str(task.task_id))
resp.status = falcon.HTTP_201
except errors.InvalidFormat as ex:
self.error(req.context, ex.msg)
self.return_error(
resp, falcon.HTTP_400, message=ex.msg, retry=False)
def task_destroy_nodes(self, req, resp, json_data):
"""Create async task for destroy node."""
action = json_data.get('action', None)
if action != 'destroy_nodes':
self.error(
req.context,
"Task body ended up in wrong handler: action %s in task_destroy_nodes"
% action)
self.return_error(
resp, falcon.HTTP_500, message="Error", retry=False)
try:
task = self.create_task(json_data, req.context)
resp.body = json.dumps(task.to_dict())
resp.append_header('Location',
"/api/v1.0/tasks/%s" % str(task.task_id))
resp.status = falcon.HTTP_201
except errors.InvalidFormat as ex:
self.error(req.context, ex.msg)
self.return_error(
resp, falcon.HTTP_400, message=ex.msg, retry=False)
def on_post(self, req, resp):
payload = json.loads(req.stream.read())
pi = PredictionInput()
if ('url' in payload.keys()) and ('modelId' in payload.keys()):
pi.url = payload['url']
pi.model_id = payload['modelId']
else:
resp.status = falcon.HTTP_400
raise falcon.HTTPBadRequest("Bad Request", "Url and(or) modelId missing in the payload")
po = self.caffe_worker_client.predict(pi)
if po.bo.status == 'Success':
resp.status = falcon.HTTP_200
resp.body = (str(po.values))
elif po.bo.status == 'Failure':
resp.body = json.dumps({'status': 'Failure', 'message' : 'Error occurred'})
resp.status = falcon.HTTP_500
raise falcon.HTTPInternalServerError('Internal Server Error', 'Predict failed! ')
def on_get(self, req, resp):
q = req.get_param('q')
resp.content_type = 'text/plain; charset=utf-8'
if not q:
resp.body = 'Bad Request'
resp.status = falcon.HTTP_400
return
try:
zenline = zenlines[q]
except KeyError:
resp.body = 'Not Found'
resp.status = falcon.HTTP_404
return
resp.body = zenline
resp.status = falcon.HTTP_200
def test_process_request_failed_request_no_body(self):
bad_headers = {
'Content-Length': '0',
'Content-Type': 'application/json',
'URL-METHODS': 'POST, GET, PUT',
}
result = self.simulate_post(headers=bad_headers)
self.assertEqual(falcon.HTTP_400, result.status)
def test_process_request_failed_request_MalformedJSON(self):
headers = {
'Content-Type': 'application/json',
'URL-METHODS': 'POST, GET, PUT',
}
bad_body = '{"cloud":"cloud1"'
result = self.simulate_post(headers=headers, body=bad_body)
self.assertEqual(falcon.HTTP_400, result.status)
def queue(body, remove: hug.types.boolean = False, user: hug.directives.user = None, response=None):
try:
song = Song.from_json(body, music_api_names)
song.user = user['name']
except ValueError as e:
logger.debug("Received bad json %s", e)
response.status = falcon.HTTP_400
return str(e)
def _remove():
try:
logger.debug("Song %s removed by %s", song, user['name'])
queue.remove(song)
except ValueError:
logger.debug("%s tried to remove Song not in queue: %s", user['name'], song)
response.status = falcon.HTTP_400
return "song {} is not in queue".format(song)
if remove:
if not has_permission(user, ["admin", "mod", "queue_remove"]):
try:
queue_song = queue[queue.index(song)]
if queue_song.user == user['name']:
logger.debug("User %s removes own song %s", user['name'], song)
_remove()
return "OK"
except (ValueError, IndexError):
pass
logger.debug("Unauthorized attempt to remove song from queue by %s", user['name'])
response.status = falcon.HTTP_FORBIDDEN
return "Not permitted"
_remove()
else:
logger.debug("Song %s added by %s", song, user['name'])
queue.append(song)
return "OK"
def search(api_name, query, max_fetch: hug.types.number = 50, response=None):
max_fetch = min(50, max(1, max_fetch))
try:
api = music_api_names[api_name]
except KeyError:
logger.debug("Tried to search on unknown API %s", api_name)
response.status = falcon.HTTP_400
return "Unknown API"
if not query:
response.status = falcon.HTTP_400
return "Invalid query"
return list(map(Song.to_json, api.search_song(query, max_fetch=max_fetch)))
def move(moving_song_json, other_song_json, after_other: hug.types.boolean = False, user: hug.directives.user = None,
response=None):
"""
Move a song before or after another one.
Needs mod permission.
:param moving_song_json: the song to move
:param other_song_json: the song to move before/after
:param after_other: whether to move after the other song. Passing anything will be evaluated to true.
"""
if not has_permission(user, ['admin', 'mod']):
logger.debug("%s tried to move a song, but isn't allowed to.", user['name'])
response.status = falcon.HTTP_FORBIDDEN
return "Not a mod"
try:
moving_song = Song.from_json(moving_song_json, music_api_names)
other_song = Song.from_json(other_song_json, music_api_names)
except ValueError as e:
logger.debug("Received invalid json (%s): %s , %s", str(e), moving_song_json, other_song_json)
response.status = falcon.HTTP_400
return str(e)
try:
queue.remove(moving_song)
except ValueError:
logger.debug("Tried to move song %s that wasn't in the queue", moving_song)
response.status = falcon.HTTP_400
return "song {} is not in queue".format(moving_song)
try:
index = queue.index(other_song)
if after_other:
index += 1
queue.insert(index, moving_song)
logger.debug("Moved song %s after/before(%s) %s", moving_song, after_other, other_song)
return player_state()
except ValueError:
logger.debug("Couldn't move song %s, because other song %s was removed from queue", moving_song, other_song)
response.status = falcon.HTTP_400
return "song {} is not in queue".format(other_song)
def grant_permission(target_username, body, user: hug.directives.user = None, response=None):
"""
Grant a permission to a user
Note that the new permissions only apply after the user logged out and in again.
Needs admin permission.
:param target_username: the username of the user the permission should be granted
:param body: the permission as returned by get_permissions
:return: 'OK', error message or None
"""
try:
permission = _Permission.from_json(body)
except ValueError:
response.status = falcon.HTTP_UNPROCESSABLE_ENTITY
return "invalid JSON"
if not is_admin(user):
logger.debug("%s tried to grant a permission %s to %s but is not admin.", user['name'], permission,
target_username)
response.status = falcon.HTTP_FORBIDDEN
return None
target_username = target_username.strip().lower()
client = _get_client(target_username)
if not client:
logger.debug("%s tried to grant permission to unknown user %s", user['name'], target_username)
response.status = falcon.HTTP_400
return "unknown target user"
permissions = client.permissions
permissions.append(permission.name)
permissions = list(set(permissions))
db = _get_db_conn()
try:
with db:
db.execute("UPDATE clients SET permissions=? WHERE username=?", [",".join(permissions), target_username])
finally:
db.close()
return "OK"
def revoke_permission(target_username, body, user: hug.directives.user = None, response=None):
"""
Revokes a granted permission.
Note that the new permissions only apply after the user logged out and in again.
Needs admin permission.
:param target_username: the username of the user whos permission should be revoked
:param body: the permission as returned by get_permissions
:return: 'OK' or error_message or None
"""
try:
permission = _Permission.from_json(body)
except ValueError:
response.status = falcon.HTTP_UNPROCESSABLE_ENTITY
return "invalid JSON"
if not is_admin(user):
logger.debug("%s tried to revoke a permission %s from %s but is not admin.", user['name'], permission,
target_username)
response.status = falcon.HTTP_FORBIDDEN
return None
target_username = target_username.strip().lower()
client = _get_client(target_username)
if not client:
logger.debug("%s tried to revoke permission from unknown user %s", user['name'], target_username)
response.status = falcon.HTTP_400
return "unknown target user"
permissions = client.permissions
try:
permissions.remove(permission.name)
except ValueError:
pass
db = _get_db_conn()
try:
with db:
db.execute("UPDATE clients SET permissions=? WHERE username=?", [",".join(permissions), target_username])
finally:
db.close()
return "OK"
def get_available_offline_playlists(user: hug.directives.user, response=None):
"""
Get all available playlists of the offline API.
:return: a list of playlists of the form {'playlist_id': <id>, 'playlist_name': <name>}
"""
if not has_permission(user, ["admin", "mod", "select_playlist"]):
response.status = falcon.HTTP_FORBIDDEN
return "Forbidden"
try:
api = music_api_names['offline_api']
playlists = api.get_available_playlists()
return list(map(lambda p: {'playlist_id': p[0], 'playlist_name': p[1]}, playlists))
except KeyError:
response.status = falcon.HTTP_400
return "Not in offline mode"
def on_patch(self, req, resp):
raise falcon.HTTPError(falcon.HTTP_400)
def test_no_description_json(self, client):
response = client.simulate_patch('/fail')
assert response.status == falcon.HTTP_400
assert response.json == {'title': '400 Bad Request'}
def test_no_description_xml(self, client):
response = client.simulate_patch(
path='/fail', headers={'Accept': 'application/xml'}
)
assert response.status == falcon.HTTP_400
expected_xml = (b'<?xml version="1.0" encoding="UTF-8"?><error>'
b'<title>400 Bad Request</title></error>')
assert response.content == expected_xml
def test_invalid_header(self, client):
client.app.add_route('/400', InvalidHeaderResource())
response = client.simulate_request(path='/400')
expected_desc = (u'The value provided for the X-Auth-Token '
u'header is invalid. Please provide a valid token.')
expected_body = {
u'title': u'Invalid header value',
u'description': expected_desc,
u'code': u'A1001',
}
assert response.status == falcon.HTTP_400
assert response.json == expected_body
def test_missing_header(self, client):
client.app.add_route('/400', MissingHeaderResource())
response = client.simulate_request(path='/400')
expected_body = {
u'title': u'Missing header value',
u'description': u'The X-Auth-Token header is required.',
}
assert response.status == falcon.HTTP_400
assert response.json == expected_body
def test_missing_param(self, client):
client.app.add_route('/400', MissingParamResource())
response = client.simulate_request(path='/400')
expected_body = {
u'title': u'Missing parameter',
u'description': u'The "id" parameter is required.',
u'code': u'P1003',
}
assert response.status == falcon.HTTP_400
assert response.json == expected_body
def test_misc(self, client):
self._misc_test(client, falcon.HTTPBadRequest, falcon.HTTP_400)
self._misc_test(client, falcon.HTTPNotAcceptable, falcon.HTTP_406,
needs_title=False)
self._misc_test(client, falcon.HTTPConflict, falcon.HTTP_409)
self._misc_test(client, falcon.HTTPPreconditionFailed, falcon.HTTP_412)
self._misc_test(client, falcon.HTTPUnsupportedMediaType, falcon.HTTP_415,
needs_title=False)
self._misc_test(client, falcon.HTTPUnprocessableEntity, falcon.HTTP_422)
self._misc_test(client, falcon.HTTPUnavailableForLegalReasons, falcon.HTTP_451,
needs_title=False)
self._misc_test(client, falcon.HTTPInternalServerError, falcon.HTTP_500)
self._misc_test(client, falcon.HTTPBadGateway, falcon.HTTP_502)
