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_elastic_sort_by_score_if_there_is_query(self):
with self.app.app_context():
self.app.data.insert('items', [
{'uri': 'foo', 'name': 'foo bar'},
{'uri': 'bar', 'name': 'foo bar'}
])
with self.app.test_request_context('/items/'):
req = parse_request('items')
req.args = {'q': 'foo'}
cursor = self.app.data.find('items', req, None)
self.assertEqual(2, cursor.count())
self.assertEqual('foo', cursor[0]['uri'])
def test_basic_search_query(self):
with self.app.app_context():
self.app.data.insert('items', [
{'uri': 'foo'},
{'uri': 'bar'}
])
with self.app.test_request_context('/items/?q=foo'):
req = parse_request('items')
cursor = self.app.data.find('items', req, None)
self.assertEquals(1, cursor.count())
def test_elastic_find_default_sort_no_mapping(self):
with self.app.test_request_context('/items/'):
req = parse_request('items')
req.args = {}
cursor = self.app.data.find('items', req, None)
self.assertEqual(0, cursor.count())
def test_phrase_search_query(self):
with self.app.app_context():
self.app.data.insert('items', [
{'uri': 'foo bar'},
{'uri': 'some text'}
])
with self.app.test_request_context('/items/?q="foo bar"'):
req = parse_request('items')
cursor = self.app.data.find('items', req, None)
self.assertEquals(1, cursor.count())
with self.app.test_request_context('/items/?q="bar foo"'):
req = parse_request('items')
cursor = self.app.data.find('items', req, None)
self.assertEquals(0, cursor.count())
def test_elastic_filter_callback(self):
with self.app.app_context():
self.app.data.insert('items_with_callback_filter', [
{'uri': 'foo'},
{'uri': 'bar'},
])
with self.app.test_request_context():
req = parse_request('items_with_callback_filter')
cursor = self.app.data.find('items_with_callback_filter', req, None)
self.assertEqual(1, cursor.count())
def on_pre_get_files(_, lookup):
# Override the HTTP header, we always want to fetch the document from
# MongoDB.
parsed_req = eve.utils.parse_request('files')
parsed_req.if_modified_since = None
# Only fetch it if the date got expired.
now = datetime.datetime.now(tz=bson.tz_util.utc)
lookup_expired = lookup.copy()
lookup_expired['link_expires'] = {'$lte': now}
cursor = current_app.data.find('files', parsed_req, lookup_expired)
for file_doc in cursor:
# log.debug('Updating expired links for file %r.', file_doc['_id'])
generate_all_links(file_doc, now)
.. versionchanged:: 0.6
Return soft deleted documents.
.. versionchanged:: 0.5
Concurrency control optional for internal functions.
ETAG are now stored with the document (#369).
.. versionchanged:: 0.0.9
More informative error messages.
.. versionchanged:: 0.0.5
Pass current resource to ``parse_request``, allowing for proper
processing of new configuration settings: `filters`, `sorting`, `paging`.
"""
req = parse_request(resource)
if config.DOMAIN[resource]["soft_delete"]:
# get_document should always fetch soft deleted documents from the db
# callers must handle soft deleted documents
req.show_deleted = True
if original:
document = original
else:
document = app.data.find_one(
resource, req, check_auth_value, force_auth_field_projection, **lookup
)
if document:
e_if_m = config.ENFORCE_IF_MATCH
if_m = config.IF_MATCH
if not req.if_match and e_if_m and if_m and concurrency_check:
def on_pre_get_files(_, lookup):
# Override the HTTP header, we always want to fetch the document from MongoDB.
parsed_req = eve.utils.parse_request('files')
parsed_req.if_modified_since = None
# Only fetch it if the date got expired.
now = datetime.datetime.now(tz=bson.tz_util.utc)
lookup_expired = lookup.copy()
lookup_expired['link_expires'] = {'$lte': now}
cursor = current_app.data.find('files', parsed_req, lookup_expired)
for file_doc in cursor:
log.debug('Updating expired links for file %r.', file_doc['_id'])
_generate_all_links(file_doc, now)
normalize_document = resource_def.get("normalize_document_for_patch")
validator = app.validator(
schema, resource=resource, allow_unknown=resource_def["allow_unknown"]
)
object_id = original[resource_def["id_field"]]
last_modified = None
etag = None
issues = {}
response = {}
if config.BANDWIDTH_SAVER is True:
embedded_fields = []
else:
req = parse_request(resource)
embedded_fields = resolve_embedded_fields(resource, req)
try:
updates = parse(payload, resource)
if skip_validation:
validation = True
else:
validation = validator.validate_update(
updates, object_id, original, normalize_document
)
updates = validator.document
if validation:
# Apply coerced values
# sneak in a shadow copy if it wasn't already there
.. versionchanged:: 0.0.6
ETag added to payload.
.. versionchanged:: 0.0.5
Support for user-restricted access to resources.
Support for LAST_UPDATED field missing from documents, because they were
created outside the API context.
.. versionchanged:: 0.0.4
Added the ``requires_auth`` decorator.
.. versionchanged:: 0.0.3
Superflous ``response`` container removed. Links wrapped with
``_links``. Links are now properly JSON formatted.
"""
req = parse_request(resource)
resource_def = config.DOMAIN[resource]
embedded_fields = resolve_embedded_fields(resource, req)
soft_delete_enabled = config.DOMAIN[resource]["soft_delete"]
if soft_delete_enabled:
# GET requests should always fetch soft deleted documents from the db
# They are handled and included in 404 responses below.
req.show_deleted = True
document = app.data.find_one(resource, req, **lookup)
if not document:
abort(404)
response = {}
etag = None
version = request.args.get(config.VERSION_PARAM)