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_abort_called_on_validation_error(mock_abort):
app = Flask("testapp")
def validate(x):
return x == 42
argmap = {"value": fields.Field(validate=validate)}
with app.test_request_context(
"/foo",
method="post",
data=json.dumps({"value": 41}),
content_type="application/json",
):
parser.parse(argmap)
mock_abort.assert_called()
abort_args, abort_kwargs = mock_abort.call_args
assert abort_args[0] == 422
expected_msg = "Invalid value."
assert abort_kwargs["messages"]["value"] == [expected_msg]
assert type(abort_kwargs["exc"]) == ValidationError
def thread_fn(value):
with app.test_request_context(
"/foo",
method="post",
data=json.dumps({"value": value}),
content_type="application/json",
):
results[value] = parser.parse(argmap)["value"]
def load_json_viewargs_data(request: Request, schema: Schema) -> Dict[str, Any]:
"""Load data from a request accepting either json or view_args encoded data"""
view_args = parser.load_view_args(request, schema)
data = parser.load_json(request, schema)
if data is missing:
return data
data = _combine_data_and_view_args(data, view_args, schema)
return data
def __init__(self, urlmap={}):
self.release = None
self.endpoint = None
self.dapver = None
self.urlmap = urlmap
self.base_args = {'release': fields.String(required=True,
validate=validate.Regexp('^(MPL-|DR)([0-9]{1,2}$)'))}
self.use_params = None
self._required = None
self._setmissing = None
self._main_kwargs = {}
self.final_args = {}
self.final_args.update(self.base_args)
self._parser = parser
self.use_args = use_args
self.use_kwargs = use_kwargs
def user_list():
args = parser.parse({
'email': wf.Str(missing=None),
'permissions': wf.DelimitedList(wf.Str(), delimiter=',', missing=[]),
})
user_dbs, cursors = model.User.get_dbs(
email=args['email'], prev_cursor=True,
)
permissions = list(UserUpdateForm._permission_choices)
permissions += args['permissions']
return flask.render_template(
'user/user_list.html',
html_class='user-list',
title='User List',
user_dbs=user_dbs,
next_url=util.generate_next_url(cursors['next']),
prev_url=util.generate_next_url(cursors['prev']),
api_url=flask.url_for('api.admin.user.list'),
@parser.location_loader('json_and_view_args') # type: ignore
def load_json_viewargs_data(request: Request, schema: Schema) -> Dict[str, Any]:
"""Load data from a request accepting either json or view_args encoded data"""
view_args = parser.load_view_args(request, schema)
data = parser.load_json(request, schema)
if data is missing:
return data
data = _combine_data_and_view_args(data, view_args, schema)
return data
@parser.error_handler
def webargs_error_handler(err):
"""
webargs error handler that uses Flask-RESTful's abort function to return
a JSON error response to the client.
"""
code, msg = (
getattr(err, 'status_code', 400),
getattr(err, 'message', 'Invalid Request')
)
abort(code, status='error', code=code, message=msg)
@flaskparser.error_handler
def handle_parser_error(
err, req, schema, error_status_code, error_headers):
"""
configure webargs error handler
"""
raise FDSNHTTPError.create(400, service_version=service_version,
error_desc_long=str(err))
def wrapped(*args, **kwargs):
req_args = webargs_parser.parse({
'submission': webargs_fields.Int(),
'submission_id': webargs_fields.Int()
})
submission_id = req_args.get('submission',
req_args.get('submission_id'))
if submission_id is None:
raise ResponseException(
"submission_id is required", StatusCode.CLIENT_ERROR)
return fn(submission_id, *args, **kwargs)
return wrapped
def user_merge():
args = parser.parse({
'user_key': wf.Str(missing=None),
'user_keys': wf.DelimitedList(wf.Str(), delimiter=',', required=True),
})
user_db_keys = [ndb.Key(urlsafe=k) for k in args['user_keys']]
user_dbs = ndb.get_multi(user_db_keys)
if len(user_dbs) < 2:
flask.abort(400)
user_dbs.sort(key=lambda user_db: user_db.created)
merged_user_db = user_dbs[0]
auth_ids = []
permissions = []
is_admin = False
is_active = False
for user_db in user_dbs: