Syndication - translate queries before passing them to external source - tests

2 files changed, 108 insertions, 46 deletions

diff --git a/archaeological_operations/tests.py b/archaeological_operations/tests.py
index b13bf99ff..cc0a8cb1b 100644
--- a/archaeological_operations/tests.py
+++ b/archaeological_operations/tests.py
@@ -24,6 +24,7 @@ from io import StringIO, BytesIO
 import tempfile
 import locale
 from unittest.mock import patch
+from urllib.parse import quote as url_quote
 import zipfile
 
 import requests
@@ -35,7 +36,7 @@ from django.core.files import File as DjangoFile
 from django.core.files.uploadedfile import SimpleUploadedFile
 from django.db.models import Q
 from django.test import tag
-from django.test.client import Client
+from django.test.client import Client, RequestFactory
 from django.urls import reverse
 from django.utils.text import slugify
 
@@ -50,6 +51,7 @@ from . import models
 from ishtar_common import models_rest
 from ishtar_common.admin import update_types_from_source
 from ishtar_common.views import document_deletion_steps
+from ishtar_common.views_item import get_item, adapt_distant_search
 from ishtar_common.serializers import document_serialization
 from archaeological_operations import views, serializers
 
@@ -4442,15 +4444,25 @@ class ApiTest(OperationInitTest, APITestCase):
         self.create_operation(self.user, self.orgas[0])
         self.create_operation(self.user, self.orgas[0])
         self.create_operation(self.user, self.orgas[0])
+
+        keys = ("old-neolithic", "middle-neolithic", "recent-neolithic",
+                "final-neolithic")
+        old_neo = models.Period.objects.get(txt_idx='old-neolithic')
+        middle_neo = models.Period.objects.get(txt_idx='middle-neolithic')
+        recent_neo = models.Period.objects.get(txt_idx='recent-neolithic')
+        final_neo = models.Period.objects.get(txt_idx='final-neolithic')
         self.operation_0 = models.Operation.objects.get(pk=self.operations[0].pk)
         self.operation_0.code_patriarche = "28123"
         self.operation_0.save()
+        self.operation_0.periods.add(old_neo)
         self.operation_1 = models.Operation.objects.get(pk=self.operations[1].pk)
         self.operation_1.code_patriarche = "28124"
         self.operation_1.save()
+        self.operation_1.periods.add(middle_neo)
         self.operation_2 = models.Operation.objects.get(pk=self.operations[2].pk)
         self.operation_2.code_patriarche = "29123"
         self.operation_2.save()
+        self.operation_2.periods.add(recent_neo)
         self.operation_3 = models.Operation.objects.get(pk=self.operations[3].pk)
         self.operation_3.code_patriarche = "29124"
         self.operation_3.save()
@@ -4755,36 +4767,83 @@ class ApiTest(OperationInitTest, APITestCase):
         self.assertIn(msg, response)
         # TODO: mark with no update
 
+    def _mock_search(self, mock_get, model, url):
+        _get_item = get_item(
+            model,
+            "get_" + model.SLUG,
+            model.SLUG,
+            no_permission_check=True
+        )
+        mock_get.__call__ = _get_item
+
+        def __json():
+            return json.loads(mock_get.return_value.text)
+
+        mock_get.return_value.json = __json
+        mock_get.return_value.status_code = 200
+
+        factory = RequestFactory()
+        request = factory.get(url)
+        request.user = self.user
+        mock_get.return_value.text = _get_item(request).content
+
+    def __construct_search(self, source):
+        params = {"submited": "1", "search_vector": ['periode="Néolithique final"']}
+        adapt_distant_search(params, source, "operation")
+        attr = "?"
+        for idx, k in enumerate(params.keys()):
+            if idx:
+                attr += "&"
+            value = params[k]
+            if isinstance(value, list):
+                value = value[0]
+            attr += f"{k}={url_quote(value)}"
+        return attr
+
     @patch("requests.get")
     def test_query_transformation(self, mock_get):
-        # POV: local
+        # POV: external
         # change query terms from a source Ishtar to match distant Ishtar
-        self._mock_request(mock_get)
-        self.client.login(username=self.username, password=self.password)
+        api_search_model = self.create_api_search_model()
         source = self._get_source()
 
-        url = "/admin/{}/{}/".format(
-            "ishtar_common", models_rest.ApiExternalSource.__name__.lower()
+        content_type = ContentType.objects.get_for_model(models.Period)
+        api_key, __ = models_rest.ApiKeyMatch.objects.get_or_create(
+            source=source,
+            associated_type=content_type,
+            search_model=api_search_model.content_type,
+            search_keys=["period", "periode"],
+            distant_slug="middle-neolithic",
+            distant_label="Néolithique moyen",
+            local_slug="middle-neolithic",
+            local_label="Néolithique moyen",
+        )
+
+        base_url = reverse("search-external", args=["operation", source.pk])
+        params = self.__construct_search(source)
+        url = base_url + params
+        self._mock_search(mock_get, models.Operation, "/get-operation/" + params)
+        response = self.client.get(
+            url, format="json", HTTP_AUTHORIZATION=self.auth_token,
         )
-        params = {
-            "action": "update_types_from_source",
-            "_selected_action": [source.pk],
-        }
-        self.client.post(url, params, follow=True)
+        self.assertEqual(response.status_code, 200)
+        j = json.loads(response.content.decode())
+        self.assertEqual(j["recordsTotal"], 0, "error on classic search")
+
         # change a key match
-        content_type = ContentType.objects.get_for_model(models.Period)
-        # no republic for local
-        models_rest.ApiKeyMatch.objects.get(
-            source=source, associated_type=content_type,
-            distant_key="republic"
-        ).delete()
-        # local have only neolithic period no detail
-        keys = ("old-neolithic", "middle-neolithic", "recent-neolithic",
-                "final-neolithic")
-        models_rest.ApiKeyMatch.objects.filter(
-            source=source, associated_type=content_type,
-            distant_key__in=keys).update(
-            local_slug="neolithic", local_label="Néolithique")
+        # final neolithic -> middle-neolitic
+        api_key.local_slug = "final-neolithic"
+        api_key.local_label = "Néolithique final"
+        api_key.save()
+        params = self.__construct_search(source)
+        url = base_url + params
+        self._mock_search(mock_get, models.Operation, "/get-operation/" + params)
+        response = self.client.get(
+            url, format="json", HTTP_AUTHORIZATION=self.auth_token,
+        )
+        self.assertEqual(response.status_code, 200)
+        j = json.loads(response.content.decode())
+        self.assertEqual(j["recordsTotal"], 1, "error on query transformation")
 
     def test_external_source_query(self):
         # POV: local
@@ -4808,4 +4867,3 @@ class ApiTest(OperationInitTest, APITestCase):
         )
         self.assertEqual(response.status_code, 200)
         j = json.loads(response.content.decode())
-        print(j)
diff --git a/ishtar_common/views_item.py b/ishtar_common/views_item.py
index 242e57b2a..88f20020b 100644
--- a/ishtar_common/views_item.py
+++ b/ishtar_common/views_item.py
@@ -1886,7 +1886,7 @@ def get_item(
                     my_base_request,
                     my_relative_session_names,
                 )
-        elif func_name and request:
+        elif func_name and request and hasattr(request, "session"):
             request.session[func_name] = dct
 
         for k in request_keys:
@@ -2394,26 +2394,7 @@ def get_item(
     return func
 
 
-def get_distant_item(
-        request,
-        model,
-        external_source_id,
-        data_type=None
-    ):
-    # TODO: check permissions
-    try:
-        src = models_rest.ApiExternalSource.objects.get(
-            pk=external_source_id)
-    except (models_rest.ApiExternalSource.DoesNotExist, ValueError):
-        return HttpResponse("{}", content_type="text/plain")
-    url = src.url
-    url += reverse(f"api-search-{model}")
-
-    params = {k: v for k, v in dict(request.GET).items() if not k.startswith("columns")}
-    params["submited"] = 1
-    params["data_type"] = "json"
-    if data_type:
-        params["data_type"] = data_type
+def adapt_distant_search(params, src, model):
     if "search_vector" in params and params["search_vector"]:
         search_vector = params["search_vector"][0]
         match = RE_FACET.search(search_vector)
@@ -2437,6 +2418,29 @@ def get_distant_item(
             match = RE_FACET.search(search_vector)
         final_search_vector += search_vector
         params["search_vector"] = [final_search_vector]
+
+
+def get_distant_item(
+        request,
+        model,
+        external_source_id,
+        data_type=None
+    ):
+    # TODO: check permissions
+    try:
+        src = models_rest.ApiExternalSource.objects.get(
+            pk=external_source_id)
+    except (models_rest.ApiExternalSource.DoesNotExist, ValueError):
+        return HttpResponse("{}", content_type="text/plain")
+    url = src.url
+    url += reverse(f"api-search-{model}")
+
+    params = {k: v for k, v in dict(request.GET).items() if not k.startswith("columns")}
+    params["submited"] = 1
+    params["data_type"] = "json"
+    if data_type:
+        params["data_type"] = data_type
+    adapt_distant_search(params, src, model)
     try:
         response = requests.get(
             url,