Set search/querystring-search limit patch only for anonymous users

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)
 ------------------

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):

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
 
 

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))}",
         )