Commit 95d08def authored by David Sikter's avatar David Sikter
Browse files

Added tests and restricted urls to get requests

parent cec19bd1
Pipeline #97497 passed with stages
in 22 minutes and 52 seconds
......@@ -575,15 +575,15 @@ class PaginationResponse(Pagination):
'''))
page_url: Optional[str] = Field(
None, description=strip('''
The url of the current page.
The url of the current page. Only applicable for GET requests.
'''))
next_page_url: Optional[str] = Field(
None, description=strip('''
The url to get the next page.
The url to get the next page. Only applicable for GET requests.
'''))
first_page_url: Optional[str] = Field(
None, description=strip('''
The url to get the first page.
The url to get the first page. Only applicable for GET requests.
'''))
@validator('order_by')
......@@ -609,8 +609,9 @@ class PaginationResponse(Pagination):
def populate_urls(self, request: Request):
'''
Populates the urls (`page_url`, `next_page_url`, `first_page_url` from the
request and `next_page_after_value`.
request and `next_page_after_value`. Only applicable for GET requests.
'''
assert request.method.upper() == 'GET'
original_url = str(request.url)
self.page_url = original_url
self.first_page_url = update_url_query_arguments(
......@@ -630,8 +631,9 @@ class IndexBasedPaginationResponse(PaginationResponse):
'''
Provided that `page` and `total` are populated, populates all other references:
`page_after_value`, `next_page_after_value`, `page_url`, `next_page_url`,
`prev_page_url`, and `first_page_url`.
`prev_page_url`, and `first_page_url`. Only applicable for GET requests.
'''
assert request.method.upper() == 'GET'
has_more_pages = self.total > self.page * self.page_size
self.page_after_value = str((self.page - 1) * self.page_size - 1) if self.page > 1 else None
self.next_page_after_value = str(self.page * self.page_size - 1) if has_more_pages else None
......
......@@ -122,7 +122,7 @@ async def post_entries_metadata_query(
and aggregated data over all search results.
'''
res = perform_search(
return perform_search(
owner=data.owner,
query=data.query,
pagination=data.pagination,
......@@ -130,8 +130,6 @@ async def post_entries_metadata_query(
statistics=data.statistics,
aggregations=data.aggregations,
user_id=user.user_id if user is not None else None)
res.pagination.populate_urls(request)
return res
@router.get(
......@@ -358,10 +356,8 @@ _entries_raw_query_docstring = strip('''
async def post_entries_raw_query(
request: Request, data: EntriesRaw, user: User = Depends(get_optional_user)):
res = _answer_entries_raw_request(
return _answer_entries_raw_request(
owner=data.owner, query=data.query, pagination=data.pagination, user=user)
res.pagination.populate_urls(request)
return res
@router.get(
......@@ -541,11 +537,9 @@ _entries_archive_docstring = strip('''
async def post_entries_archive_query(
request: Request, data: EntriesArchive, user: User = Depends(get_optional_user)):
res = _answer_entries_archive_request(
return _answer_entries_archive_request(
owner=data.owner, query=data.query, pagination=data.pagination,
required=data.required, user=user)
res.pagination.populate_urls(request)
return res
@router.get(
......
......@@ -96,6 +96,21 @@ def data(elastic, raw_files, mongo, test_user, other_test_user):
return data
def assert_pagination(pagination):
page = pagination['page']
total = pagination['total']
page_size = pagination['page_size']
assert page >= 1
assert ('page_after_value' in pagination) == (page != 1)
assert ('next_page_after_value' in pagination) == (total > page * page_size)
assert 'page_url' in pagination
assert 'first_page_url' in pagination
if 'next_page_after_value' in pagination:
assert 'next_page_url' in pagination
if page > 1:
assert 'prev_page_url' in pagination
def assert_dataset(dataset, query: Query = None, entries: List[str] = None, n_entries: int = -1, **kwargs):
for key, value in kwargs.items():
assert dataset[key] == value
......@@ -158,6 +173,7 @@ def test_datasets(client, data, query, size, status_code):
assert len(json_response['data']) == size
for dataset in json_response['data']:
assert_dataset(dataset, **query)
assert_pagination(json_response['pagination'])
@pytest.mark.parametrize('dataset_id, result, status_code', [
......
......@@ -25,7 +25,7 @@ import json
from nomad.metainfo.search_extension import search_quantities
from nomad.app.v1.models import AggregateableQuantity, Metric
from tests.utils import assert_at_least
from tests.utils import assert_at_least, assert_url_query_args
from .common import assert_response
from tests.app.conftest import example_data as data # pylint: disable=unused-import
......@@ -323,9 +323,9 @@ def assert_aggregations(response_json, name, agg, total: int, size: int):
agg_data = [{agg['quantity']: value} for value in agg_response['data']]
if 'pagination' in agg:
assert_pagination(agg['pagination'], agg_response['pagination'], agg_data)
assert_pagination(agg['pagination'], agg_response['pagination'], agg_data, is_get=False)
else:
assert_pagination({}, agg_response['pagination'], agg_data, order_by=agg['quantity'])
assert_pagination({}, agg_response['pagination'], agg_data, order_by=agg['quantity'], is_get=False)
if 'entries' in agg:
for item in agg_response['data'].values():
......@@ -336,7 +336,7 @@ def assert_aggregations(response_json, name, agg, total: int, size: int):
assert_required(entry, agg['entries']['required'])
def assert_pagination(pagination, pagination_response, data, order_by=None, order=None):
def assert_pagination(pagination, pagination_response, data, order_by=None, order=None, is_get=True):
assert_at_least(pagination, pagination_response)
assert len(data) <= pagination_response['page_size']
assert len(data) <= pagination_response['total']
......@@ -354,6 +354,19 @@ def assert_pagination(pagination, pagination_response, data, order_by=None, orde
else:
assert item[order_by] <= data[index + 1][order_by]
if is_get:
page_url = pagination_response.get('page_url')
first_page_url = pagination_response.get('first_page_url')
next_page_url = pagination_response.get('next_page_url')
next_page_after_value = pagination_response.get('next_page_after_value')
assert page_url
assert first_page_url
assert_url_query_args(first_page_url, page_after_value=None)
if next_page_after_value:
assert next_page_url
assert_url_query_args(next_page_url, page_after_value=next_page_after_value)
def assert_raw_zip_file(
response, files: int = -1, manifest_entries: int = -1, compressed: bool = False):
......@@ -888,4 +901,4 @@ def test_entries_pagination(client, data, pagination, response_pagination, statu
if response_json is None:
return
assert_pagination(pagination, response_json['pagination'], response_json['data'])
assert_pagination(pagination, response_json['pagination'], response_json['data'], is_get=(http_method == 'get'))
......@@ -18,6 +18,7 @@
''' Methods to help with testing of nomad@FAIRDI.'''
import urllib.parse
import json
from logging import LogRecord
......@@ -68,3 +69,16 @@ def assert_at_least(source, target):
else:
assert value == target[key], '%s with value %s in %s is not equal the target value %s in %s' % (
key, source[key], source, target[key], target)
def assert_url_query_args(url: str, **kwargs):
'''
Parses the url, and checks that the query arguments match the values specified by kwargs.
'''
__, __, __, __, query, __ = urllib.parse.urlparse(url)
query_dict = urllib.parse.parse_qs(query)
for k, v in kwargs.items():
if v is None:
assert k not in query_dict
else:
assert query_dict[k][0] == v
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment