From 053ac450c0b23d7405ae8a90c7cfd96a5f439bdc Mon Sep 17 00:00:00 2001 From: Andrea Cecchi Date: Thu, 19 Oct 2023 10:15:11 +0200 Subject: [PATCH] Set search/querystring-search limit patch only for anonymous users --- CHANGES.rst | 4 ++ .../restapi/services/querystringsearch/get.py | 21 ++++--- .../volto/restapi/services/search/get.py | 36 ++++++----- .../volto/tests/test_catalog_limit_patches.py | 63 +++++++++++++++++-- 4 files changed, 94 insertions(+), 30 deletions(-) diff --git a/CHANGES.rst b/CHANGES.rst index b9a0d7d5..08f3019f 100644 --- a/CHANGES.rst +++ b/CHANGES.rst @@ -9,10 +9,14 @@ Changelog site, even when requesting replacement with a link from a different website. [lucabel]. + - plone.app.redirector.FourOhFourView.search_for_similar patch to enable conditionally the search for similar [folix-01] +- Set search/querystring-search limit patch only for anonymous users. + Auth users can need to perform an higher query (in contents view for example). + [cekk] 5.2.4 (2023-09-26) ------------------ diff --git a/src/redturtle/volto/restapi/services/querystringsearch/get.py b/src/redturtle/volto/restapi/services/querystringsearch/get.py index ac5166e7..6360963d 100644 --- a/src/redturtle/volto/restapi/services/querystringsearch/get.py +++ b/src/redturtle/volto/restapi/services/querystringsearch/get.py @@ -1,6 +1,7 @@ # -*- coding: utf-8 -*- from datetime import datetime from DateTime import DateTime +from plone import api from plone.app.event.base import get_events from plone.app.querystring import queryparser from plone.restapi.batching import HypermediaBatch @@ -96,21 +97,25 @@ def get_limit(self, data): """ If limit is <= 0 or higher than MAX_LIMIT, set it to MAX_LIMIT """ + is_anon = api.user.is_anonymous() + default_value = is_anon and MAX_LIMIT or max(1000, MAX_LIMIT) + if "limit" not in data: + return default_value try: - limit = int(data.get("limit", MAX_LIMIT)) - except ValueError: - raise BadRequest("Invalid limit") + limit = int(data.get("limit", default_value)) + except ValueError as exc: + raise BadRequest("Invalid limit") from exc - if "limit" in data and limit <= 0: + if limit <= 0: del data["limit"] - limit = MAX_LIMIT - if limit > MAX_LIMIT: + limit = default_value + if limit > default_value: logger.warning( '[wrong query] limit is too high: "{}". Set to default ({}).'.format( - data["query"], MAX_LIMIT + data["query"], default_value ) ) - limit = MAX_LIMIT + limit = default_value return limit def is_event_search(self, query): diff --git a/src/redturtle/volto/restapi/services/search/get.py b/src/redturtle/volto/restapi/services/search/get.py index 780d584e..783c210d 100644 --- a/src/redturtle/volto/restapi/services/search/get.py +++ b/src/redturtle/volto/restapi/services/search/get.py @@ -124,26 +124,30 @@ def search(self, query=None): return super(SearchHandler, self).search(query) def _parse_query(self, query): + """ + set a max limit for anonymous calls + """ query = super()._parse_query(query) - for idx in ["sort_limit", "b_size"]: - if idx not in query: - continue - value = query.get(idx, MAX_LIMIT) - if value <= 0: - logger.warning( - '[wrong query] {} is wrong: "{}". Set to default ({}).'.format( - idx, query, MAX_LIMIT + if api.user.is_anonymous(): + for idx in ["sort_limit", "b_size"]: + if idx not in query: + continue + value = query.get(idx, MAX_LIMIT) + if value <= 0: + logger.warning( + '[wrong query] {} is wrong: "{}". Set to default ({}).'.format( + idx, query, MAX_LIMIT + ) ) - ) - query[idx] = MAX_LIMIT + query[idx] = MAX_LIMIT - if value > MAX_LIMIT: - logger.warning( - '[wrong query] {} is too high: "{}". Set to default ({}).'.format( - idx, query, MAX_LIMIT + if value > MAX_LIMIT: + logger.warning( + '[wrong query] {} is too high: "{}". Set to default ({}).'.format( + idx, query, MAX_LIMIT + ) ) - ) - query[idx] = MAX_LIMIT + query[idx] = MAX_LIMIT return query diff --git a/src/redturtle/volto/tests/test_catalog_limit_patches.py b/src/redturtle/volto/tests/test_catalog_limit_patches.py index 872e3571..1ba26966 100644 --- a/src/redturtle/volto/tests/test_catalog_limit_patches.py +++ b/src/redturtle/volto/tests/test_catalog_limit_patches.py @@ -25,28 +25,33 @@ def setUp(self): setRoles(self.portal, TEST_USER_ID, ["Manager"]) for i in range(self.MAX_LIMIT + 1): - api.content.create( + doc = api.content.create( container=self.portal, type="Document", title=f"Document {i}", ) + api.content.transition(obj=doc, transition="publish") commit() self.api_session = RelativeSession(self.portal_url) self.api_session.headers.update({"Accept": "application/json"}) self.api_session.auth = (SITE_OWNER_NAME, SITE_OWNER_PASSWORD) + self.api_session_anon = RelativeSession(self.portal_url) + self.api_session_anon.headers.update({"Accept": "application/json"}) + def tearDown(self): self.api_session.close() + self.api_session_anon.close() - def test_search_b_size_default_to_500(self): + def test_search_b_size_not_limited_for_auth(self): response = self.api_session.get( "/@search", params={"portal_type": "Document", "b_size": 1000} ) result = response.json() - self.assertEqual(len(result["items"]), self.MAX_LIMIT) + self.assertEqual(len(result["items"]), self.MAX_LIMIT + 1) - def test_querystringsearch_post_default_limit_500(self): + def test_querystringsearch_post_not_limited_for_auth(self): response = self.api_session.post( "/@querystring-search", json={ @@ -62,10 +67,56 @@ def test_querystringsearch_post_default_limit_500(self): }, ) result = response.json() + self.assertEqual(len(result["items"]), self.MAX_LIMIT + 1) + self.assertEqual(result["items_total"], self.MAX_LIMIT + 1) + + def test_querystringsearch_get_not_limited_for_auth(self): + query = { + "query": [ + { + "i": "portal_type", + "o": "plone.app.querystring.operation.selection.any", + "v": ["Document"], + } + ], + "b_size": 2000, + "limit": 2000, + } + response = self.api_session.get( + f"/@querystring-search?query={quote(json.dumps(query))}", + ) + + result = response.json() + self.assertEqual(len(result["items"]), self.MAX_LIMIT + 1) + self.assertEqual(result["items_total"], self.MAX_LIMIT + 1) + + def test_search_b_size_default_to_500_for_anon(self): + response = self.api_session_anon.get( + "/@search", params={"portal_type": "Document", "b_size": 1000} + ) + result = response.json() + self.assertEqual(len(result["items"]), self.MAX_LIMIT) + + def test_querystringsearch_post_default_limit_500_for_anon(self): + response = self.api_session_anon.post( + "/@querystring-search", + json={ + "query": [ + { + "i": "portal_type", + "o": "plone.app.querystring.operation.selection.is", + "v": ["Document"], + } + ], + "limit": 2000, + "b_size": 2000, + }, + ) + result = response.json() self.assertEqual(len(result["items"]), self.MAX_LIMIT) self.assertEqual(result["items_total"], self.MAX_LIMIT) - def test_querystringsearch_get_default_limit_500(self): + def test_querystringsearch_get_default_limit_500_for_anon(self): query = { "query": [ { @@ -77,7 +128,7 @@ def test_querystringsearch_get_default_limit_500(self): "b_size": 2000, "limit": 2000, } - response = self.api_session.get( + response = self.api_session_anon.get( f"/@querystring-search?query={quote(json.dumps(query))}", )