interpolate xpath expressions from variables

corehq/apps/case_search/tests/test_case_search_registry.py

@@ -311,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_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")

corehq/apps/case_search/utils.py

@@ -130,6 +130,7 @@ def get_case_search_results(domain, request_config, app_id=None, couch_user=None
         request_config.case_types,
         request_config.criteria,
         request_config.commcare_sort,
+        request_config.xpath_vars,
     )
     helper.profiler.primary_count = len(cases)
     if app_id:
@@ -140,8 +141,8 @@ def get_case_search_results(domain, request_config, app_id=None, couch_user=None
 
 
 @time_function()
-def get_primary_case_search_results(helper, case_types, criteria, commcare_sort=None):
-    builder = CaseSearchQueryBuilder(helper, case_types)
+def get_primary_case_search_results(helper, case_types, criteria, commcare_sort, xpath_vars):
+    builder = CaseSearchQueryBuilder(helper, case_types, xpath_vars)
     try:
         with helper.profiler.timing_context('build_query'):
             search_es = builder.build_query(criteria, commcare_sort)
@@ -224,11 +225,12 @@ def get_all_related_live_cases(self, initial_cases):
 class CaseSearchQueryBuilder:
     """Compiles the case search object for the view"""
 
-    def __init__(self, helper, case_types):
+    def __init__(self, helper, case_types, xpath_vars=None):
         self.request_domain = helper.domain
         self.case_types = case_types
         self.helper = helper
         self.config = helper.config
+        self.xpath_vars = xpath_vars or {}
 
     def build_query(self, search_criteria, commcare_sort=None):
         search_es = self._get_initial_search_es()
@@ -264,8 +266,8 @@ def _apply_filter(self, search_es, criteria):
         if criteria.key == CASE_SEARCH_XPATH_QUERY_KEY:
             if not criteria.is_empty:
                 xpaths = criteria.value if criteria.has_multiple_terms else [criteria.value]
-                for xpath in xpaths:
-                    search_es = search_es.filter(self._build_filter_from_xpath(xpath))
+                for xpath_template in xpaths:
+                    search_es = search_es.filter(self._build_filter_from_xpath(xpath_template))
                 return search_es
         elif criteria.key == 'case_id':
             return search_es.filter(case_es.case_ids(criteria.value))
@@ -282,7 +284,11 @@ def _apply_filter(self, search_es, criteria):
             return search_es.add_query(self._get_case_property_query(criteria), queries.MUST)
         return search_es
 
-    def _build_filter_from_xpath(self, xpath, fuzzy=False):
+    def _build_filter_from_xpath(self, xpath_template, fuzzy=False):
+        try:
+            xpath = xpath_template.format(**self.xpath_vars)
+        except KeyError as e:
+            raise CaseSearchUserError(_("Variable '{}' not found").format(e.args[0]))
         context = SearchFilterContext(self.request_domain, fuzzy, self.helper)
         with self.helper.profiler.timing_context('_build_filter_from_xpath'):
             return build_filter_from_xpath(xpath, context=context)