XPath Vars in Case Search

corehq/apps/case_search/models.py

@@ -21,6 +21,7 @@
 FUZZY_PROPERTIES = "fuzzy_properties"
 CASE_SEARCH_BLACKLISTED_OWNER_ID_KEY = 'commcare_blacklisted_owner_ids'
 CASE_SEARCH_XPATH_QUERY_KEY = '_xpath_query'
+CASE_SEARCH_XPATH_VAR_PREFIX = '_xpath_var_'
 CASE_SEARCH_CASE_TYPE_KEY = "case_type"
 CASE_SEARCH_INDEX_KEY_PREFIX = "indices."
 CASE_SEARCH_SORT_KEY = "commcare_sort"
@@ -187,6 +188,7 @@ class CaseSearchRequestConfig:
     custom_related_case_property = attr.ib(kw_only=True, default=None, converter=_flatten_singleton_list)
     include_all_related_cases = attr.ib(kw_only=True, default=None, converter=_flatten_singleton_list)
     commcare_sort = attr.ib(kw_only=True, default=None, converter=_parse_commcare_sort_properties)
+    xpath_vars = attr.ib(kw_only=True, factory=dict)
 
     @case_types.validator
     def _require_case_type(self, attribute, value):
@@ -209,8 +211,12 @@ def extract_search_request_config(request_dict):
         config_name: params.pop(param_name, None)
         for param_name, config_name in CONFIG_KEYS_MAPPING.items()
     }
+    xpath_vars = {
+        k.removeprefix(CASE_SEARCH_XPATH_VAR_PREFIX): _flatten_singleton_list(params.pop(k))
+        for k in list(params.keys()) if k.startswith(CASE_SEARCH_XPATH_VAR_PREFIX)
+    }
     criteria = criteria_dict_to_criteria_list(params)
-    return CaseSearchRequestConfig(criteria=criteria, **kwargs_from_params)
+    return CaseSearchRequestConfig(criteria=criteria, xpath_vars=xpath_vars, **kwargs_from_params)
 
 
 class GetOrNoneManager(models.Manager):

corehq/apps/case_search/tests/test_case_search_endpoint.py

@@ -12,13 +12,14 @@
 )
 from corehq.apps.app_manager.tests.app_factory import AppFactory
 from corehq.apps.case_search.const import IS_RELATED_CASE
-from corehq.apps.case_search.models import CaseSearchConfig, SearchCriteria
+from corehq.apps.case_search.models import (
+    CaseSearchConfig,
+    CaseSearchRequestConfig,
+    SearchCriteria,
+)
 from corehq.apps.domain.shortcuts import create_user
 from corehq.apps.es.case_search import case_search_adapter
-from corehq.apps.es.tests.utils import (
-    case_search_es_setup,
-    es_test,
-)
+from corehq.apps.es.tests.utils import case_search_es_setup, es_test
 from corehq.form_processor.tests.utils import FormProcessorTestUtils
 
 from ..utils import get_case_search_results
@@ -71,23 +72,28 @@ def tearDownClass(cls):
         FormProcessorTestUtils.delete_all_cases()
         super().tearDownClass()
 
+    def get_case_search_results(self, case_types, criteria, app_id=None):
+        return get_case_search_results(self.domain, CaseSearchRequestConfig(
+            criteria=criteria, case_types=case_types,
+        ), app_id=app_id)
+
     def test_basic(self):
-        res = get_case_search_results(self.domain, ['person'], [])
+        res = self.get_case_search_results(['person'], [])
         self.assertItemsEqual(["Jane", "Xiomara", "Alba", "Rogelio", "Jane"], [
             case.name for case in res
         ])
 
     def test_case_id_criteia(self):
-        res = get_case_search_results(self.domain, ['household'], [SearchCriteria('case_id', self.household_1)])
+        res = self.get_case_search_results(['household'], [SearchCriteria('case_id', self.household_1)])
         self.assertItemsEqual(["Villanueva"], [case.name for case in res])
 
     def test_dynamic_property(self):
-        res = get_case_search_results(self.domain, ['person'], [SearchCriteria('family', 'Ramos')])
+        res = self.get_case_search_results(['person'], [SearchCriteria('family', 'Ramos')])
         self.assertItemsEqual(["Jane"], [case.name for case in res])
 
     def test_app_aware_related_cases(self):
         with mock.patch('corehq.apps.case_search.utils.get_app_cached', new=lambda _, __: self.factory.app):
-            res = get_case_search_results(self.domain, ['person'], [], app_id='fake_app_id')
+            res = self.get_case_search_results(['person'], [], app_id='fake_app_id')
         self.assertItemsEqual([
             (case.name, case.get_case_property(IS_RELATED_CASE)) for case in res
         ], [

corehq/apps/case_search/tests/test_case_search_registry.py

@@ -14,6 +14,7 @@
 from corehq.apps.case_search.const import COMMCARE_PROJECT
 from corehq.apps.case_search.models import (
     CaseSearchConfig,
+    CaseSearchRequestConfig,
     SearchCriteria,
     criteria_dict_to_criteria_list,
 )
@@ -153,8 +154,11 @@ def tearDownClass(cls):
         super().tearDownClass()
 
     def _run_query(self, domain, case_types, criteria_dict, registry_slug):
-        criteria = criteria_dict_to_criteria_list(criteria_dict)
-        results = get_case_search_results(domain, case_types, criteria, registry_slug=registry_slug)
+        results = get_case_search_results(domain, CaseSearchRequestConfig(
+            criteria=criteria_dict_to_criteria_list(criteria_dict),
+            case_types=case_types,
+            data_registry=registry_slug,
+        ))
         return [(case.name, case.domain) for case in results]
 
     def test_query_all_domains_in_registry(self):
@@ -273,12 +277,11 @@ def test_commcare_project_field_doesnt_expand_access(self):
         self.assertItemsEqual([], results)
 
     def test_includes_project_property(self):
-        results = get_case_search_results(
-            self.domain_1,
-            ["person"],
-            [SearchCriteria("name", "Jane")],
-            registry_slug=self.registry_slug
-        )
+        results = get_case_search_results(self.domain_1, CaseSearchRequestConfig(
+            criteria=[SearchCriteria("name", "Jane")],
+            case_types=["person"],
+            data_registry=self.registry_slug,
+        ))
         self.assertItemsEqual([
             ("Jane", self.domain_1, self.domain_1),
             ("Jane", self.domain_1, self.domain_1),
@@ -291,13 +294,11 @@ def test_includes_project_property(self):
 
     def test_related_cases_included(self):
         with patch_get_app_cached:
-            results = get_case_search_results(
-                self.domain_1,
-                ["creative_work"],
-                [SearchCriteria("name", "Jane Eyre")],  # from domain 2
-                app_id="mock_app_id",
-                registry_slug=self.registry_slug
-            )
+            results = get_case_search_results(self.domain_1, CaseSearchRequestConfig(
+                criteria=[SearchCriteria("name", "Jane Eyre")],  # from domain 2
+                case_types=["creative_work"],
+                data_registry=self.registry_slug,
+            ), app_id="mock_app_id")
         self.assertItemsEqual([
             ("Charlotte Brontë", "creator", self.domain_2),
             ("Jane Eyre", "creative_work", self.domain_2),
@@ -310,7 +311,7 @@ def test_primary_cases_not_included_with_related_cases(self):
         with patch_get_app_cached:
             registry_helper = _get_helper(None, self.domain_1, ["creative_work"], self.registry_slug)
             primary_cases = get_primary_case_search_results(
-                registry_helper, ["creative_work"], [SearchCriteria("name", "Jane Eyre")])
+                registry_helper, ["creative_work"], [SearchCriteria("name", "Jane Eyre")], None, None)
             related_cases = registry_helper.get_all_related_live_cases(primary_cases)
 
             self.assertItemsEqual([

corehq/apps/case_search/tests/test_get_related_cases.py

@@ -11,6 +11,7 @@
     Module,
 )
 from corehq.apps.case_search.const import IS_RELATED_CASE
+from corehq.apps.case_search.models import CaseSearchRequestConfig
 from corehq.apps.case_search.utils import (
     _get_all_related_cases,
     QueryHelper,
@@ -152,8 +153,16 @@ def test_get_related_cases_duplicates_and_tags(self):
         def get_related_cases_helper(include_all_related_cases):
             with patch("corehq.apps.case_search.utils.get_app_context",
                        return_value=({"parent", "parent/parent"}, {"c", "d"})):
-                return get_and_tag_related_cases(QueryHelper(self.domain), None, {"c"}, source_cases, None,
-                    include_all_related_cases)
+                return get_and_tag_related_cases(
+                    QueryHelper(self.domain),
+                    None,
+                    request_config=CaseSearchRequestConfig(
+                        criteria=None,
+                        case_types={"c"},
+                        include_all_related_cases=include_all_related_cases,
+                    ),
+                    cases=source_cases,
+                )
 
         partial_related_cases = get_related_cases_helper(include_all_related_cases=False)
         EXPECTED_PARTIAL_RELATED_CASES = (source_cases_related["PATH_RELATED_CASE_ID"]
@@ -202,9 +211,17 @@ def test_get_related_cases_expanded_results(self):
 
         with patch("corehq.apps.case_search.utils.get_app_context",
                    return_value=({"parent"}, {"c"})):
-            include_all_related_cases = False
-            partial_related_cases = get_and_tag_related_cases(QueryHelper(self.domain), None, {"a"}, source_cases,
-                'custom_related_case_id', include_all_related_cases)
+            partial_related_cases = get_and_tag_related_cases(
+                QueryHelper(self.domain),
+                app_id=None,
+                request_config=CaseSearchRequestConfig(
+                    criteria=None,
+                    case_types={"a"},
+                    custom_related_case_property='custom_related_case_id',
+                    include_all_related_cases=False,
+                ),
+                cases=source_cases,
+            )
 
         EXPECTED_PARTIAL_RELATED_CASES = (source_cases_related["EXPANDED_CASE_ID"]
             | source_cases_related["PATH_RELATED_CASE_ID"]

corehq/apps/case_search/tests/test_models.py

@@ -7,16 +7,20 @@
 from corehq.apps.case_search.exceptions import CaseSearchUserError
 from corehq.apps.case_search.filter_dsl import CaseFilterError
 from corehq.apps.case_search.models import (
+    CASE_SEARCH_BLACKLISTED_OWNER_ID_KEY,
+    CASE_SEARCH_CASE_TYPE_KEY,
     CASE_SEARCH_CUSTOM_RELATED_CASE_PROPERTY_KEY,
-    CASE_SEARCH_REGISTRY_ID_KEY,
     CASE_SEARCH_INCLUDE_ALL_RELATED_CASES_KEY,
-    CASE_SEARCH_SORT_KEY,
     CASE_SEARCH_MODULE_NAME_TAG_KEY,
+    CASE_SEARCH_REGISTRY_ID_KEY,
+    CASE_SEARCH_SORT_KEY,
+    CASE_SEARCH_XPATH_QUERY_KEY,
+    CASE_SEARCH_XPATH_VAR_PREFIX,
     CaseSearchRequestConfig,
+    SearchCriteria,
     disable_case_search,
     enable_case_search,
     extract_search_request_config,
-    CASE_SEARCH_CASE_TYPE_KEY, SearchCriteria, CASE_SEARCH_BLACKLISTED_OWNER_ID_KEY,
 )
 from corehq.util.test_utils import generate_cases
 
@@ -90,6 +94,22 @@ def _make_request_dict(params):
     }
 
 
+def test_extract_xpath_vars():
+    config = extract_search_request_config({
+        CASE_SEARCH_CASE_TYPE_KEY: 'jelly',
+        CASE_SEARCH_XPATH_QUERY_KEY: "@status='open' and {has_parent}",
+        f'{CASE_SEARCH_XPATH_VAR_PREFIX}has_parent': "ancestor-exists('parent', @case_type='sandwich')",
+        f'{CASE_SEARCH_XPATH_VAR_PREFIX}old': "last_modified < '2022-01-01'",
+    })
+    assert config.xpath_vars == {
+        'has_parent': "ancestor-exists('parent', @case_type='sandwich')",
+        'old': "last_modified < '2022-01-01'",
+    }
+    assert len(config.criteria) == 1
+    assert config.criteria[0] == SearchCriteria(
+        CASE_SEARCH_XPATH_QUERY_KEY, "@status='open' and {has_parent}")
+
+
 @generate_cases([
     (CASE_SEARCH_BLACKLISTED_OWNER_ID_KEY, ["a", "b"]),
     ("owner_id", ["a", "b"]),

corehq/apps/case_search/tests/test_utils.py

@@ -1,6 +1,14 @@
 from unittest.mock import patch
 
-from corehq.apps.case_search.utils import get_expanded_case_results
+import pytest
+
+from corehq.apps.case_search.exceptions import CaseSearchUserError
+from corehq.apps.case_search.models import CaseSearchConfig
+from corehq.apps.case_search.utils import (
+    CaseSearchQueryBuilder,
+    QueryHelper,
+    get_expanded_case_results,
+)
 from corehq.form_processor.models import CommCareCase
 
 
@@ -16,3 +24,31 @@ def test_get_expanded_case_results(get_cases_mock):
     helper = None
     get_expanded_case_results(helper, "potential_duplicate_id", cases)
     get_cases_mock.assert_called_with(helper, {"123", "456"})
+
+
+@pytest.mark.parametrize("xpath_vars, xpath_query, result", [
+    ({'name': 'Ethan'}, "name = '{name}'", "name = 'Ethan'"),
+    ({'ssn_matches': 'social_security_number = "123abc"'},
+     '{ssn_matches} or subcase-exists("parent", @case_type = "alias" and {ssn_matches})',
+     'social_security_number = "123abc" or subcase-exists('
+     '"parent", @case_type = "alias" and social_security_number = "123abc")'),
+    ({'ssn_matches': 'social_security_number = "123abc"'}, 'match-all()', 'match-all()'),
+])
+def test_xpath_vars(xpath_vars, xpath_query, result):
+    helper = QueryHelper('mydomain')
+    helper.config = CaseSearchConfig(domain='mydomain')
+    builder = CaseSearchQueryBuilder(helper, ['mycasetype'], xpath_vars)
+    with patch("corehq.apps.case_search.utils.build_filter_from_xpath") as build:
+        builder._build_filter_from_xpath(xpath_query)
+        assert build.call_args.args[0] == result
+
+
+def test_xpath_vars_misconfigured():
+    xpath_vars = {}  # No variables defined!
+    xpath_query = "name = '{name}'"  # 'name' is not specified
+    helper = QueryHelper('mydomain')
+    helper.config = CaseSearchConfig(domain='mydomain')
+    builder = CaseSearchQueryBuilder(helper, ['mycasetype'], xpath_vars)
+    with pytest.raises(CaseSearchUserError) as e_info:
+        builder._build_filter_from_xpath(xpath_query)
+    e_info.match("Variable 'name' not found")