HC-611: Fix closed-index params in PIT methods (OpenSearch + base class) (#68)

* HC-611: Fix closed-index params in PIT methods (OpenSearch + base class)

* bump version

* updates based on testing

* restore closed-index params for OSS search

---------

Co-authored-by: Mike Cayanan <michael.d.cayanan@jpl.nasa.gov>

Author: mcayanan

hysds_commons/__init__.py

@@ -1,3 +1,3 @@
-__version__ = "2.1.0"
+__version__ = "2.1.1"
 __description__ = "Common HySDS Functions, Utilities, Etc."
 __url__ = "https://github.jpl.nasa.gov/hysds-org/hysds_commons"

hysds_commons/opensearch_utils.py

@@ -44,9 +44,22 @@ def _pit(self, **kwargs):
         if index is None:
             raise RuntimeError("OpenSearchUtility._pit: the search_after API must specify a index/alias")
 
-        pit = self.es.create_point_in_time(index=index, keep_alive=keep_alive)
+        # Apply closed index params to PIT open call (HC-600).
+        # PIT APIs only accept ignore_unavailable and expand_wildcards (not allow_no_indices).
+        # Extract caller-specified params from kwargs first, then apply defaults.
+        pit_params = {}
+        for key, default_value in self.CLOSED_INDEX_PARAMS.items():
+            if key != "allow_no_indices":  # PIT APIs don't accept this param
+                pit_params[key] = kwargs.pop(key, default_value)
+        
+        pit = self.es.create_point_in_time(index=index, keep_alive=keep_alive, **pit_params)
         pit_id = pit["pit_id"]
 
+        # Once the PIT is open, strip any remaining indicesOptions from kwargs — OpenSearch/ES
+        # rejects them on _search calls when a PIT is in the body.
+        for key in self.CLOSED_INDEX_PARAMS:
+            kwargs.pop(key, None)
+
         size = kwargs.get("size", body.get("size"))
         if not size:
             kwargs["size"] = 1000

hysds_commons/search_utils.py

@@ -170,12 +170,13 @@ def _pit(self, **kwargs):
         if index is None:
             raise RuntimeError("ElasticsearchUtility._pit: the search_after API must specify a index/alias")
 
-        # Apply closed index params (HC-600) - always apply since aliases can
-        # resolve to multiple indices, some of which may be closed
+        # Apply closed index params to PIT open call (HC-600).
+        # PIT APIs only accept ignore_unavailable and expand_wildcards (not allow_no_indices).
+        # Extract caller-specified params from kwargs first, then apply defaults.
         pit_params = {}
-        for key, value in self.CLOSED_INDEX_PARAMS.items():
-            kwargs.setdefault(key, value)
-            pit_params[key] = kwargs[key]
+        for key, default_value in self.CLOSED_INDEX_PARAMS.items():
+            if key != "allow_no_indices":  # PIT APIs don't accept this param
+                pit_params[key] = kwargs.pop(key, default_value)
 
         size = kwargs.get("size", body.get("size"))
         if not size:
@@ -188,12 +189,21 @@ def _pit(self, **kwargs):
         pit = None
         if self.flavor != "oss":
             pit = self.es.open_point_in_time(index=index, keep_alive=keep_alive, **pit_params)
+
+            # Once the PIT is open, strip any remaining indicesOptions from kwargs — OpenSearch/ES
+            # rejects them on _search calls when a PIT is in the body.
+            for key in self.CLOSED_INDEX_PARAMS:
+                kwargs.pop(key, None)
+
             body = {
                 **body,
                 **{"pit": {**pit, **{"keep_alive": keep_alive}}},
             }
         else:
             warnings.warn("Elasticsearch OSS does not support _pit, will use search_after without _pit...")
+            # Restore closed-index params for OSS search (no PIT to conflict with)
+            for key, default_value in self.CLOSED_INDEX_PARAMS.items():
+                kwargs.setdefault(key, default_value)
         res = self.es.search(body=body, **kwargs)
 
         records = []

test/test_search_utils.py

@@ -10,6 +10,7 @@
 from unittest.mock import MagicMock, patch
 
 from hysds_commons.search_utils import SearchUtility
+from hysds_commons.opensearch_utils import OpenSearchUtility
 
 
 class ConcreteSearchUtility(SearchUtility):
@@ -23,6 +24,15 @@ def __init__(self):
         self.flavor = "default"
 
 
+class MockOpenSearchUtility(OpenSearchUtility):
+    """Mock OpenSearchUtility that skips real OpenSearch client initialization."""
+    def __init__(self):
+        self.es = MagicMock()
+        self.engine = "opensearch"
+        self.version = None
+        self.flavor = None
+
+
 class TestIsWildcardIndex:
     """Tests for _is_wildcard_index() static method."""
 
@@ -294,27 +304,125 @@ def setup_method(self):
         }
         self.utility.es.close_point_in_time.return_value = {"succeeded": True}
 
+    def test_pit_does_not_pass_allow_no_indices_to_open_pit(self):
+        """open_point_in_time() should NOT receive allow_no_indices."""
+        self.utility._pit(index="job_status-*", body={"query": {"match_all": {}}})
+        call_kwargs = self.utility.es.open_point_in_time.call_args[1]
+        assert "allow_no_indices" not in call_kwargs
+        assert call_kwargs["ignore_unavailable"] is True
+        assert call_kwargs["expand_wildcards"] == "open"
+
+    def test_pit_search_does_not_pass_indices_options(self):
+        """_search calls with PIT body must NOT include indicesOptions."""
+        self.utility._pit(index="job_status-*", body={"query": {"match_all": {}}})
+        call_kwargs = self.utility.es.search.call_args[1]
+        assert "ignore_unavailable" not in call_kwargs
+        assert "allow_no_indices" not in call_kwargs
+        assert "expand_wildcards" not in call_kwargs
+
     def test_pit_with_wildcard_applies_params_to_open_point_in_time(self):
         """_pit() should apply closed index params to open_point_in_time for wildcard patterns."""
         self.utility._pit(index="job_status-*", body={"query": {"match_all": {}}})
 
-        # Verify open_point_in_time was called with closed index params
+        # Verify open_point_in_time was called with closed index params (except allow_no_indices)
         call_kwargs = self.utility.es.open_point_in_time.call_args[1]
         assert call_kwargs["ignore_unavailable"] is True
-        assert call_kwargs["allow_no_indices"] is True
         assert call_kwargs["expand_wildcards"] == "open"
 
     def test_pit_with_single_index_applies_params_to_open_point_in_time(self):
         """_pit() should apply params to open_point_in_time for single index (could be alias)."""
         self.utility._pit(index="job_status-current", body={"query": {"match_all": {}}})
 
-        # Verify open_point_in_time was called with closed index params
+        # Verify open_point_in_time was called with closed index params (except allow_no_indices)
         call_kwargs = self.utility.es.open_point_in_time.call_args[1]
         assert call_kwargs["ignore_unavailable"] is True
-        assert call_kwargs["allow_no_indices"] is True
         assert call_kwargs["expand_wildcards"] == "open"
 
 
+class TestOpenSearchPitMethod:
+    """Tests for OpenSearchUtility._pit() with closed index handling (HC-600)."""
+
+    def setup_method(self):
+        """Set up test fixtures."""
+        self.utility = MockOpenSearchUtility()
+        self.utility.es.create_point_in_time.return_value = {
+            "pit_id": "opensearch_pit_id_123"
+        }
+        self.utility.es.search.return_value = {
+            "hits": {"total": {"value": 0}, "hits": []}
+        }
+        self.utility.es.delete_point_in_time.return_value = {"succeeded": True}
+
+    def test_create_pit_applies_closed_index_params(self):
+        """create_point_in_time() should receive ignore_unavailable and expand_wildcards."""
+        self.utility._pit(index="grq_v1.0_product-*", body={"query": {"match_all": {}}})
+        call_kwargs = self.utility.es.create_point_in_time.call_args[1]
+        assert call_kwargs["ignore_unavailable"] is True
+        assert call_kwargs["expand_wildcards"] == "open"
+
+    def test_create_pit_does_not_pass_allow_no_indices(self):
+        """PIT open APIs don't accept allow_no_indices -- must not be passed."""
+        self.utility._pit(index="grq", body={"query": {"match_all": {}}})
+        call_kwargs = self.utility.es.create_point_in_time.call_args[1]
+        assert "allow_no_indices" not in call_kwargs
+
+    def test_create_pit_applies_params_for_alias(self):
+        """create_point_in_time() should apply params for aliases (no wildcard in name)."""
+        self.utility._pit(index="grq", body={"query": {"match_all": {}}})
+        call_kwargs = self.utility.es.create_point_in_time.call_args[1]
+        assert call_kwargs["ignore_unavailable"] is True
+        assert call_kwargs["expand_wildcards"] == "open"
+
+    def test_create_pit_does_not_override_caller_params(self):
+        """Caller-specified closed-index params should not be overridden."""
+        self.utility._pit(
+            index="grq",
+            body={"query": {"match_all": {}}},
+            ignore_unavailable=False,
+            expand_wildcards="all",
+        )
+        call_kwargs = self.utility.es.create_point_in_time.call_args[1]
+        assert call_kwargs["ignore_unavailable"] is False
+        assert call_kwargs["expand_wildcards"] == "all"
+
+    def test_pit_search_does_not_pass_indices_options(self):
+        """_search calls with PIT must NOT include indicesOptions."""
+        self.utility._pit(index="grq", body={"query": {"match_all": {}}})
+        call_kwargs = self.utility.es.search.call_args[1]
+        assert "ignore_unavailable" not in call_kwargs
+        assert "allow_no_indices" not in call_kwargs
+        assert "expand_wildcards" not in call_kwargs
+
+    def test_pit_pagination_still_works(self):
+        """Verify pagination loop + PIT cleanup still functions correctly."""
+        page1 = {
+            "hits": {
+                "total": {"value": 2},
+                "hits": [
+                    {"_id": "1", "_source": {"id": "a"}, "sort": [1, "a"]},
+                    {"_id": "2", "_source": {"id": "b"}, "sort": [2, "b"]},
+                ],
+            }
+        }
+        page2 = {"hits": {"total": {"value": 2}, "hits": []}}
+        self.utility.es.search.side_effect = [page1, page2]
+
+        records = self.utility._pit(index="grq", body={"query": {"match_all": {}}})
+
+        assert len(records) == 2
+        assert records[0]["_id"] == "1"
+        assert records[1]["_id"] == "2"
+        # Verify PIT was cleaned up
+        self.utility.es.delete_point_in_time.assert_called_once_with(
+            body={"pit_id": ["opensearch_pit_id_123"]}
+        )
+
+    def test_missing_index_raises_runtime_error(self):
+        """_pit() should raise RuntimeError when no index is provided."""
+        with pytest.raises(RuntimeError, match="must specify a index/alias"):
+            self.utility._pit(body={"query": {"match_all": {}}})
+
+
 class TestClosedIndexParamsConstant:
     """Tests for CLOSED_INDEX_PARAMS class constant."""