Add support for an autocomplete livesearch widget

.github/workflows/test-plone-5.2.yml

@@ -29,10 +29,12 @@ jobs:
 
       # python setup
       - name: Set up Python ${{ matrix.python-version }} with Plone 5.2.5
-        uses: plone/[email protected]
-        with:
-          python-version: ${{ matrix.python-version }}
-          plone-version: "5.2.5"
+        run: |
+          make install-plone-5.2
+
+        ## Note: setup-plone action is retired, because we need to force
+        ## an older pip version, but the action seems to fetch the new version
+        ## regardless what I try.
 
       # python cache
       - uses: actions/cache@v1

Makefile

@@ -106,6 +106,7 @@ install-plone-5.2: bin/mxdev config ## pip install Plone packages
 	@echo "$(GREEN)==> Setup Build$(RESET)"
 	cp constraints-5.2.txt constraints.txt
 	cp requirements-5.2.txt requirements.txt
+	pip install pip\<24
 	bin/tox -e init
 	bin/mxdev -c mx.ini
 	bin/pip install -r requirements-mxdev.txt

news/28.feature

@@ -0,0 +1 @@
+Add support for an autocomplete livesearch widget @reebalazs

requirements-5.2.txt

@@ -1,4 +1,5 @@
 -c constraints.txt
+pip<24
 
 # Cannot install new testing framework with 5.2
 # -e ".[test]"

src/kitconcept/solr/services/configure.zcml

@@ -12,4 +12,12 @@
       name="@solr"
       />
 
+  <plone:service
+      method="GET"
+      factory=".suggest.SolrSuggest"
+      for="zope.interface.Interface"
+      permission="zope2.View"
+      name="@solr-suggest"
+      />
+
 </configure>

src/kitconcept/solr/services/suggest.py

@@ -0,0 +1,69 @@
+from collective.solr.interfaces import ISolrConnectionManager
+from collective.solr.utils import removeSpecialCharactersAndOperators
+from plone import api
+from plone.restapi.interfaces import ISerializeToJson
+from plone.restapi.interfaces import ISerializeToJsonSummary
+from plone.restapi.services import Service
+from zope.component import getMultiAdapter
+from zope.component import queryUtility
+
+import json
+import urllib
+
+
+class SolrSuggest(Service):
+    def query_suggest(self, query):
+        language = api.portal.get_current_language(self.context)
+        manager = queryUtility(ISolrConnectionManager)
+        if manager is None:
+            return {"error": "Solr is not installed or activated"}
+        connection = manager.getConnection()
+        if connection is None:
+            return {"error": "Solr is not installed or activated"}
+        data = {"error": "no response"}
+        parameters = {
+            "q": removeSpecialCharactersAndOperators(query),
+            "fq": "-showinsearch:False -portal_type:Image -portal_type:Glossary -portal_type:FAQ -portal_type:(FAQ Item) -portal_type:(FAQ Category) -portal_type:Link +Language:%s"
+            % language,
+        }
+        querystring = urllib.parse.urlencode(parameters)
+        url = "{}/{}".format(connection.solrBase, "suggest?%s" % querystring)
+        try:
+            res = connection.doGet(url, {"Accept": "application/json"})
+            data = json.loads(res.read())
+        finally:
+            if not connection.persistent:
+                connection.conn.close()
+        return data
+
+    def serialize_brain(self, brain):
+        if brain["portal_type"] in ["Member"]:
+            obj = brain.getObject()
+            data = getMultiAdapter((obj, self.request), ISerializeToJson)()
+            data["@id"] = obj.absolute_url()
+            return data
+
+        return getMultiAdapter(
+            (brain, self.request), ISerializeToJsonSummary
+        )()
+
+    def parse_response(self, data):
+        if "error" in data or "response" not in data:
+            error = {"suggestions": [], "error": "No response from solr"}
+            if "error" in data:
+                error["error"] = data["error"]
+            return error
+        uids = [doc["UID"] for doc in data["response"]["docs"]]
+
+        brains = {brain["UID"]: brain for brain in api.content.find(UID=uids)}
+        return [
+            self.serialize_brain(brains[uid]) for uid in uids if uid in brains
+        ]
+
+    def reply(self):
+        query = self.request.form.get("query", "")
+        data = self.query_suggest(query)
+        data = self.parse_response(data)
+        if isinstance(data, dict):
+            return data
+        return {"suggestions": data}

tests/services/suggest/test_suggest.py

@@ -0,0 +1,90 @@
+import pytest
+import urllib.parse
+
+
+class TestSuggestDefault:
+    @pytest.fixture(autouse=True)
+    def _init(self, portal_with_content, manager_request):
+        self.portal = portal_with_content
+        response = manager_request.get(self.url)
+        self.data = response.json()
+
+
+@pytest.fixture
+def get_suggest_result_props():
+    def func(
+        a: dict,
+    ) -> bool:
+        ac = dict(a)
+        del ac["@id"]
+        return ac
+
+    return func
+
+
+@pytest.fixture
+def get_suggest_result_path():
+    def func(a: dict) -> bool:
+        return urllib.parse.urlparse(a["@id"]).path
+
+    return func
+
+
+@pytest.fixture
+def get_suggest_item():
+    def func(data, index: int) -> dict:
+        return data.get("suggestions")[index]
+
+    return func
+
+
+class TestSuggestDefaultBaseSearch(TestSuggestDefault):
+    url = "/@solr-suggest?query=chomsky"
+    expected_result = [
+        {
+            "@id": "http://localhost:59793/plone/mydocument",
+            "@type": "Document",
+            "description": "",
+            "review_state": "private",
+            "title": "My Document about Noam Chomsky",
+            "type_title": "Page",
+        },
+        {
+            "@id": "http://localhost:59793/plone/mynews",
+            "@type": "News Item",
+            "description": "",
+            "review_state": "private",
+            "title": "My News Item with Noam Chomsky",
+            "type_title": "News Item",
+        },
+    ]
+
+    @pytest.mark.parametrize(
+        "index,expected_dict",
+        enumerate(expected_result),
+    )
+    def test_suggest_result_path(
+        self,
+        get_suggest_item,
+        get_suggest_result_path,
+        index: int,
+        expected_dict: dict,
+    ):
+        assert get_suggest_result_path(
+            get_suggest_item(self.data, index)
+        ) == get_suggest_result_path(expected_dict)
+
+    @pytest.mark.parametrize(
+        "index,expected_dict",
+        enumerate(expected_result),
+    )
+    def test_suggest_result_props(
+        self,
+        get_suggest_item,
+        get_suggest_result_props,
+        index: int,
+        expected_dict: dict,
+    ):
+        assert get_suggest_result_props(
+            get_suggest_item(self.data, index)
+        ) == get_suggest_result_props(expected_dict)