Index State Management support (#398)

* feat(plugins): add index management client plugin

Signed-off-by: florian <[email protected]>

* chore(CHANGELOG): added entry for ISM api support in changelog

Signed-off-by: florian <[email protected]>

* test(plugins): use assertEqual to compare call parameters

Signed-off-by: florian <[email protected]>

* test(plugins): edit policy to support older versions of opensearch

Signed-off-by: florian <[email protected]>

* test(plugins): ignore plugin tests when opensearch is unreleased

Signed-off-by: florian <[email protected]>

* test(plugins): move plugin tests into separate files

Signed-off-by: florian <[email protected]>

* test(plugins): fix import of OpenSearchTestCase

Signed-off-by: florian <[email protected]>

* chore(USER_GUIDE): add a index management plugin part

Signed-off-by: florian <[email protected]>

---------

Signed-off-by: florian <[email protected]>

CHANGELOG.md

@@ -13,6 +13,7 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 - Added 'allow_redirects' parameter in perform_request function for RequestsHttpConnection ([#401](https://github.com/opensearch-project/opensearch-py/pull/401))
 - Enhanced YAML test runner to use OpenSearch rest-api-spec YAML tests ([#414](https://github.com/opensearch-project/opensearch-py/pull/414)
 - Added `Search#collapse` ([#409](https://github.com/opensearch-project/opensearch-py/issues/409))
+- Added support for the ISM API ([#398](https://github.com/opensearch-project/opensearch-py/pull/398))
 ### Changed
 - Upgrading pytest-asyncio to latest version - 0.21.0 ([#339](https://github.com/opensearch-project/opensearch-py/pull/339))
 - Fixed flaky CI tests by replacing httpbin with a simple http_server ([#395](https://github.com/opensearch-project/opensearch-py/pull/395))

USER_GUIDE.md

@@ -21,6 +21,10 @@
       - [**Creating a destination**](#creating-a-destination)
       - [**Getting alerts**](#getting-alerts)
       - [**Acknowledge alerts**](#acknowledge-alerts)
+    - [Index management plugin](#index-management-plugin)
+      - [Creating a policy](#creating-a-policy)
+      - [Getting a policy](#getting-a-policy)
+      - [Deleting a policy](#deleting-a-policy)
     - [Security plugin](#security-plugin)
       - [Creating a role](#creating-a-role)
       - [Getting a role](#getting-a-role)
@@ -426,6 +430,73 @@ response = client.plugins.alerting.acknowledge_alert(query)
 print(response)
 ```
 
+### Index management plugin
+
+#### Creating a policy
+[API definition](https://opensearch.org/docs/latest/im-plugin/ism/api/#create-policy)
+```python
+print('\Creating a policy:')
+
+policy_name = "test-policy"
+policy_content = {
+    "policy": {
+        "description": "hot warm delete workflow",
+        "default_state": "hot",
+        "schema_version": 1,
+        "states": [
+            {
+                "name": "hot",
+                "actions": [{"rollover": {"min_index_age": "1d"}}],
+                "transitions": [{"state_name": "warm"}],
+            },
+            {
+                "name": "warm",
+                "actions": [{"replica_count": {"number_of_replicas": 5}}],
+                "transitions": [{"state_name": "delete", "conditions": {"min_index_age": "30d"}}],
+            },
+            {
+                "name": "delete",
+                "actions": [
+                    {
+                        "notification": {
+                            "destination": {"chime": {"url": "<URL>"}},
+                            "message_template": {"source": "The index {{ctx.index}} is being deleted"},
+                        }
+                    },
+                    {"delete": {}},
+                ],
+            },
+        ],
+        "ism_template": {"index_patterns": ["log*"], "priority": 100},
+    }
+}
+
+response = client.index_managment.put_policy(policy_name, body=policy_content)
+print(response)
+```
+
+#### Getting a policy
+[API definition](https://opensearch.org/docs/latest/im-plugin/ism/api/#get-policy)
+```python
+print('\Getting a policy:')
+
+policy_name = "test-policy"
+
+response = client.index_managment.get_policy(policy_name)
+print(response)
+```
+
+#### Deleting a policy
+[API definition](https://opensearch.org/docs/latest/index_managment/access-control/api/#create-user)
+```python
+print('\Deleting a policy:')
+
+policy_name = "test-policy"
+
+response = client.index_managment.delete_policy(policy_name)
+print(response)
+```
+
 ### Security plugin
 
 #### Creating a role

docs/source/api-ref/plugins.md

@@ -8,5 +8,6 @@ maxdepth: 1
 ---
 
 plugins/alerting_plugin
+plugins/index_management_plugin
 plugins/security_plugin
 ```

docs/source/api-ref/plugins/index_management_plugin.md

@@ -0,0 +1,5 @@
+# Index Management Plugin
+
+```{eval-rst}
+.. autoclass:: opensearchpy.plugins.index_management.IndexManagementClient
+```

opensearchpy/_async/client/plugins.py

@@ -10,6 +10,7 @@
 import warnings
 
 from ..plugins.alerting import AlertingClient
+from ..plugins.index_management import IndexManagementClient
 from ..plugins.security import SecurityClient
 from .utils import NamespacedClient
 
@@ -23,7 +24,7 @@ class PluginsClient(NamespacedClient):
         self.alerting = AlertingClient(client)
         # self.anomaly_detection = AnomalyDetectionClient(client)
         # self.trace_analytics = TraceAnalyticsClient(client)
-        # self.index_management = IndexManagementClient(client)
+        self.index_management = IndexManagementClient(client)
         self.security = SecurityClient(client)
 
         self._dynamic_lookup(client)
@@ -38,7 +39,7 @@ class PluginsClient(NamespacedClient):
             "alerting",
             # "anomaly_detection",
             # "trace_analytics",
-            # "index_management",
+            "index_management",
             "security",
         ]
         for plugin in plugins:

opensearchpy/_async/client/plugins.pyi

@@ -14,5 +14,6 @@ from .utils import NamespacedClient as NamespacedClient
 
 class PluginsClient(NamespacedClient):
     alerting: Any
+    index_management: Any
     security: Any
     def __init__(self, client: AsyncOpenSearch) -> None: ...

opensearchpy/_async/plugins/index_management.py

@@ -0,0 +1,153 @@
+# SPDX-License-Identifier: Apache-2.0
+#
+# The OpenSearch Contributors require contributions made to
+# this file be licensed under the Apache-2.0 license or a
+# compatible open source license.
+#
+# Modifications Copyright OpenSearch Contributors. See
+# GitHub history for details.
+
+
+from ..client.utils import SKIP_IN_PATH, NamespacedClient, _make_path, query_params
+
+
+class IndexManagementClient(NamespacedClient):
+    @query_params()
+    async def put_policy(self, policy, body=None, params=None, headers=None):
+        """
+        Creates, or updates, a policy.
+
+        :arg policy: The name of the policy
+        """
+        if policy in SKIP_IN_PATH:
+            raise ValueError("Empty value passed for a required argument 'policy'.")
+
+        return await self.transport.perform_request(
+            "PUT",
+            _make_path("_plugins", "_ism", "policies", policy),
+            params=params,
+            headers=headers,
+            body=body,
+        )
+
+    @query_params()
+    async def add_policy(self, index, body=None, params=None, headers=None):
+        """
+        Adds a policy to an index. This operation does not change the policy if the index already has one.
+
+        :arg index: The name of the index to add policy on
+        """
+        if index in SKIP_IN_PATH:
+            raise ValueError("Empty value passed for a required argument 'index'.")
+
+        return await self.transport.perform_request(
+            "POST",
+            _make_path("_plugins", "_ism", "add", index),
+            params=params,
+            headers=headers,
+            body=body,
+        )
+
+    @query_params()
+    async def get_policy(self, policy, params=None, headers=None):
+        """
+        Gets the policy by `policy_id`.
+
+        :arg policy: The name of the policy
+        """
+        if policy in SKIP_IN_PATH:
+            raise ValueError("Empty value passed for a required argument 'policy'.")
+
+        return await self.transport.perform_request(
+            "GET",
+            _make_path("_plugins", "_ism", "policies", policy),
+            params=params,
+            headers=headers,
+        )
+
+    @query_params()
+    async def remove_policy_from_index(self, index, params=None, headers=None):
+        """
+        Removes any ISM policy from the index.
+
+        :arg index: The name of the index to remove policy on
+        """
+        if index in SKIP_IN_PATH:
+            raise ValueError("Empty value passed for a required argument 'index'.")
+
+        return await self.transport.perform_request(
+            "POST",
+            _make_path("_plugins", "_ism", "remove", index),
+            params=params,
+            headers=headers,
+        )
+
+    @query_params()
+    async def change_policy(self, index, body=None, params=None, headers=None):
+        """
+        Updates the managed index policy to a new policy (or to a new version of the policy).
+
+        :arg index: The name of the index to change policy on
+        """
+        if index in SKIP_IN_PATH:
+            raise ValueError("Empty value passed for a required argument 'index'.")
+
+        return await self.transport.perform_request(
+            "POST",
+            _make_path("_plugins", "_ism", "change_policy", index),
+            params=params,
+            headers=headers,
+            body=body,
+        )
+
+    @query_params()
+    async def retry(self, index, body=None, params=None, headers=None):
+        """
+        Retries the failed action for an index.
+
+        :arg index: The name of the index whose is in a failed state
+        """
+        if index in SKIP_IN_PATH:
+            raise ValueError("Empty value passed for a required argument 'index'.")
+
+        return await self.transport.perform_request(
+            "POST",
+            _make_path("_plugins", "_ism", "retry", index),
+            params=params,
+            headers=headers,
+            body=body,
+        )
+
+    @query_params("show_policy")
+    async def explain_index(self, index, params=None, headers=None):
+        """
+        Gets the current state of the index.
+
+        :arg index: The name of the index to explain
+        """
+        if index in SKIP_IN_PATH:
+            raise ValueError("Empty value passed for a required argument 'index'.")
+
+        return await self.transport.perform_request(
+            "GET",
+            _make_path("_plugins", "_ism", "explain", index),
+            params=params,
+            headers=headers,
+        )
+
+    @query_params()
+    async def delete_policy(self, policy, params=None, headers=None):
+        """
+        Deletes the policy by `policy_id`.
+
+        :arg policy: The name of the policy to delete
+        """
+        if policy in SKIP_IN_PATH:
+            raise ValueError("Empty value passed for a required argument 'policy'.")
+
+        return await self.transport.perform_request(
+            "DELETE",
+            _make_path("_plugins", "_ism", "policies", policy),
+            params=params,
+            headers=headers,
+        )

opensearchpy/_async/plugins/index_management.pyi

@@ -0,0 +1,71 @@
+# SPDX-License-Identifier: Apache-2.0
+#
+# The OpenSearch Contributors require contributions made to
+# this file be licensed under the Apache-2.0 license or a
+# compatible open source license.
+#
+# Modifications Copyright OpenSearch Contributors. See
+# GitHub history for details.
+
+from typing import Any, Union
+
+from ..client.utils import NamespacedClient as NamespacedClient
+from ..client.utils import query_params as query_params
+
+class IndexManagementClient(NamespacedClient):
+    async def put_policy(
+        self,
+        policy: Any,
+        body: Any | None = ...,
+        params: Any | None = ...,
+        headers: Any | None = ...,
+    ) -> Union[bool, Any]: ...
+    async def add_policy(
+        self,
+        index: Any,
+        body: Any | None = ...,
+        params: Any | None = ...,
+        headers: Any | None = ...,
+    ) -> Union[bool, Any]: ...
+    async def get_policy(
+        self,
+        policy: Any,
+        body: Any | None = ...,
+        params: Any | None = ...,
+        headers: Any | None = ...,
+    ) -> Union[bool, Any]: ...
+    async def remove_policy_from_index(
+        self,
+        index: Any,
+        body: Any | None = ...,
+        params: Any | None = ...,
+        headers: Any | None = ...,
+    ) -> Union[bool, Any]: ...
+    async def change_policy(
+        self,
+        index: Any,
+        body: Any | None = ...,
+        params: Any | None = ...,
+        headers: Any | None = ...,
+    ) -> Union[bool, Any]: ...
+    async def retry(
+        self,
+        index: Any,
+        body: Any | None = ...,
+        params: Any | None = ...,
+        headers: Any | None = ...,
+    ) -> Union[bool, Any]: ...
+    async def explain_index(
+        self,
+        index: Any,
+        body: Any | None = ...,
+        params: Any | None = ...,
+        headers: Any | None = ...,
+    ) -> Union[bool, Any]: ...
+    async def delete_policy(
+        self,
+        policy: Any,
+        body: Any | None = ...,
+        params: Any | None = ...,
+        headers: Any | None = ...,
+    ) -> Union[bool, Any]: ...

opensearchpy/client/plugins.py

@@ -11,6 +11,7 @@
 import warnings
 
 from ..plugins.alerting import AlertingClient
+from ..plugins.index_management import IndexManagementClient
 from ..plugins.security import SecurityClient
 from .utils import NamespacedClient
 
@@ -24,7 +25,7 @@ class PluginsClient(NamespacedClient):
         self.alerting = AlertingClient(client)
         # self.anomaly_detection = AnomalyDetectionClient(client)
         # self.trace_analytics = TraceAnalyticsClient(client)
-        # self.index_management = IndexManagementClient(client)
+        self.index_management = IndexManagementClient(client)
         self.security = SecurityClient(client)
 
         self._dynamic_lookup(client)
@@ -39,7 +40,7 @@ class PluginsClient(NamespacedClient):
             "alerting",
             # "anomaly_detection",
             # "trace_analytics",
-            # "index_management",
+            "index_management",
             "security",
         ]
         for plugin in plugins:

opensearchpy/client/plugins.pyi

@@ -14,5 +14,6 @@ from .utils import NamespacedClient as NamespacedClient
 
 class PluginsClient(NamespacedClient):
     alerting: Any
+    index_management: Any
     security: Any
     def __init__(self, client: OpenSearch) -> None: ...

opensearchpy/plugins/index_management.py

@@ -0,0 +1,153 @@
+# SPDX-License-Identifier: Apache-2.0
+#
+# The OpenSearch Contributors require contributions made to
+# this file be licensed under the Apache-2.0 license or a
+# compatible open source license.
+#
+# Modifications Copyright OpenSearch Contributors. See
+# GitHub history for details.
+
+
+from ..client.utils import SKIP_IN_PATH, NamespacedClient, _make_path, query_params
+
+
+class IndexManagementClient(NamespacedClient):
+    @query_params()
+    def put_policy(self, policy, body=None, params=None, headers=None):
+        """
+        Creates, or updates, a policy.
+
+        :arg policy: The name of the policy
+        """
+        if policy in SKIP_IN_PATH:
+            raise ValueError("Empty value passed for a required argument 'policy'.")
+
+        return self.transport.perform_request(
+            "PUT",
+            _make_path("_plugins", "_ism", "policies", policy),
+            params=params,
+            headers=headers,
+            body=body,
+        )
+
+    @query_params()
+    def add_policy(self, index, body=None, params=None, headers=None):
+        """
+        Adds a policy to an index. This operation does not change the policy if the index already has one.
+
+        :arg index: The name of the index to add policy on
+        """
+        if index in SKIP_IN_PATH:
+            raise ValueError("Empty value passed for a required argument 'index'.")
+
+        return self.transport.perform_request(
+            "POST",
+            _make_path("_plugins", "_ism", "add", index),
+            params=params,
+            headers=headers,
+            body=body,
+        )
+
+    @query_params()
+    def get_policy(self, policy, params=None, headers=None):
+        """
+        Gets the policy by `policy_id`.
+
+        :arg policy: The name of the policy
+        """
+        if policy in SKIP_IN_PATH:
+            raise ValueError("Empty value passed for a required argument 'policy'.")
+
+        return self.transport.perform_request(
+            "GET",
+            _make_path("_plugins", "_ism", "policies", policy),
+            params=params,
+            headers=headers,
+        )
+
+    @query_params()
+    def remove_policy_from_index(self, index, params=None, headers=None):
+        """
+        Removes any ISM policy from the index.
+
+        :arg index: The name of the index to remove policy on
+        """
+        if index in SKIP_IN_PATH:
+            raise ValueError("Empty value passed for a required argument 'index'.")
+
+        return self.transport.perform_request(
+            "POST",
+            _make_path("_plugins", "_ism", "remove", index),
+            params=params,
+            headers=headers,
+        )
+
+    @query_params()
+    def change_policy(self, index, body=None, params=None, headers=None):
+        """
+        Updates the managed index policy to a new policy (or to a new version of the policy).
+
+        :arg index: The name of the index to change policy on
+        """
+        if index in SKIP_IN_PATH:
+            raise ValueError("Empty value passed for a required argument 'index'.")
+
+        return self.transport.perform_request(
+            "POST",
+            _make_path("_plugins", "_ism", "change_policy", index),
+            params=params,
+            headers=headers,
+            body=body,
+        )
+
+    @query_params()
+    def retry(self, index, body=None, params=None, headers=None):
+        """
+        Retries the failed action for an index.
+
+        :arg index: The name of the index whose is in a failed state
+        """
+        if index in SKIP_IN_PATH:
+            raise ValueError("Empty value passed for a required argument 'index'.")
+
+        return self.transport.perform_request(
+            "POST",
+            _make_path("_plugins", "_ism", "retry", index),
+            params=params,
+            headers=headers,
+            body=body,
+        )
+
+    @query_params("show_policy")
+    def explain_index(self, index, params=None, headers=None):
+        """
+        Gets the current state of the index.
+
+        :arg index: The name of the index to explain
+        """
+        if index in SKIP_IN_PATH:
+            raise ValueError("Empty value passed for a required argument 'index'.")
+
+        return self.transport.perform_request(
+            "GET",
+            _make_path("_plugins", "_ism", "explain", index),
+            params=params,
+            headers=headers,
+        )
+
+    @query_params()
+    def delete_policy(self, policy, params=None, headers=None):
+        """
+        Deletes the policy by `policy_id`.
+
+        :arg policy: The name of the policy to delete
+        """
+        if policy in SKIP_IN_PATH:
+            raise ValueError("Empty value passed for a required argument 'policy'.")
+
+        return self.transport.perform_request(
+            "DELETE",
+            _make_path("_plugins", "_ism", "policies", policy),
+            params=params,
+            headers=headers,
+        )

opensearchpy/plugins/index_management.pyi

@@ -0,0 +1,71 @@
+# SPDX-License-Identifier: Apache-2.0
+#
+# The OpenSearch Contributors require contributions made to
+# this file be licensed under the Apache-2.0 license or a
+# compatible open source license.
+#
+# Modifications Copyright OpenSearch Contributors. See
+# GitHub history for details.
+
+from typing import Any, Union
+
+from ..client.utils import NamespacedClient as NamespacedClient
+from ..client.utils import query_params as query_params
+
+class IndexManagementClient(NamespacedClient):
+    def put_policy(
+        self,
+        policy: Any,
+        body: Any | None = ...,
+        params: Any | None = ...,
+        headers: Any | None = ...,
+    ) -> Union[bool, Any]: ...
+    def add_policy(
+        self,
+        index: Any,
+        body: Any | None = ...,
+        params: Any | None = ...,
+        headers: Any | None = ...,
+    ) -> Union[bool, Any]: ...
+    def get_policy(
+        self,
+        policy: Any,
+        body: Any | None = ...,
+        params: Any | None = ...,
+        headers: Any | None = ...,
+    ) -> Union[bool, Any]: ...
+    def remove_policy_from_index(
+        self,
+        index: Any,
+        body: Any | None = ...,
+        params: Any | None = ...,
+        headers: Any | None = ...,
+    ) -> Union[bool, Any]: ...
+    def change_policy(
+        self,
+        index: Any,
+        body: Any | None = ...,
+        params: Any | None = ...,
+        headers: Any | None = ...,
+    ) -> Union[bool, Any]: ...
+    def retry(
+        self,
+        index: Any,
+        body: Any | None = ...,
+        params: Any | None = ...,
+        headers: Any | None = ...,
+    ) -> Union[bool, Any]: ...
+    def explain_index(
+        self,
+        index: Any,
+        body: Any | None = ...,
+        params: Any | None = ...,
+        headers: Any | None = ...,
+    ) -> Union[bool, Any]: ...
+    def delete_policy(
+        self,
+        policy: Any,
+        body: Any | None = ...,
+        params: Any | None = ...,
+        headers: Any | None = ...,
+    ) -> Union[bool, Any]: ...

test_opensearchpy/run_tests.py

@@ -148,6 +148,15 @@ def run_all(argv=None):
                     ]
                 )
 
+        # There are no plugins for unreleased versions of opensearch
+        if environ.get("OPENSEARCH_VERSION") == "SNAPSHOT":
+            ignores.extend(
+                [
+                    "test_opensearchpy/test_server/test_plugins/",
+                    "test_opensearchpy/test_async/test_server/test_plugins/",
+                ]
+            )
+
         if ignores:
             argv.extend(["--ignore=%s" % ignore for ignore in ignores])
 

test_opensearchpy/test_client/test_plugins/__init__.py

@@ -0,0 +1,25 @@
+# SPDX-License-Identifier: Apache-2.0
+#
+# The OpenSearch Contributors require contributions made to
+# this file be licensed under the Apache-2.0 license or a
+# compatible open source license.
+#
+# Modifications Copyright OpenSearch Contributors. See
+# GitHub history for details.
+#
+#  Licensed to Elasticsearch B.V. under one or more contributor
+#  license agreements. See the NOTICE file distributed with
+#  this work for additional information regarding copyright
+#  ownership. Elasticsearch B.V. licenses this file to you under
+#  the Apache License, Version 2.0 (the "License"); you may
+#  not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+# 	http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing,
+#  software distributed under the License is distributed on an
+#  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+#  KIND, either express or implied.  See the License for the
+#  specific language governing permissions and limitations
+#  under the License.

test_opensearchpy/test_client/test_plugins/test_index_management.py

@@ -0,0 +1,56 @@
+# SPDX-License-Identifier: Apache-2.0
+#
+# The OpenSearch Contributors require contributions made to
+# this file be licensed under the Apache-2.0 license or a
+# compatible open source license.
+#
+# Modifications Copyright OpenSearch Contributors. See
+# GitHub history for details.
+
+from test_opensearchpy.test_cases import OpenSearchTestCase
+
+
+class TestIndexManagement(OpenSearchTestCase):
+    def test_create_policy(self):
+        self.client.index_management.put_policy("...")
+        self.assert_url_called("PUT", "/_plugins/_ism/policies/...")
+
+    def test_update_policy(self):
+        self.client.index_management.put_policy(
+            "...", params={"if_seq_no": 7, "if_primary_term": 1}
+        )
+        self.assertEqual(
+            [({"if_seq_no": 7, "if_primary_term": 1}, {}, None)],
+            self.assert_url_called("PUT", "/_plugins/_ism/policies/..."),
+        )
+
+    def test_add_policy(self):
+        self.client.index_management.add_policy("...")
+        self.assert_url_called("POST", "/_plugins/_ism/add/...")
+
+    def test_get_policy(self):
+        self.client.index_management.get_policy("...")
+        self.assert_url_called("GET", "/_plugins/_ism/policies/...")
+
+    def test_remove_policy_from_index(self):
+        self.client.index_management.remove_policy_from_index("...")
+        self.assert_url_called("POST", "/_plugins/_ism/remove/...")
+
+    def test_change_policy(self):
+        self.client.index_management.change_policy("...")
+        self.assert_url_called("POST", "/_plugins/_ism/change_policy/...")
+
+    def test_retry(self):
+        self.client.index_management.retry("...")
+        self.assert_url_called("POST", "/_plugins/_ism/retry/...")
+
+    def test_explain_index(self):
+        self.client.index_management.explain_index("...", show_policy=True)
+        self.assertEqual(
+            [({"show_policy": b"true"}, {}, None)],
+            self.assert_url_called("GET", "/_plugins/_ism/explain/..."),
+        )
+
+    def test_delete_policy(self):
+        self.client.index_management.delete_policy("...")
+        self.assert_url_called("DELETE", "/_plugins/_ism/policies/...")

test_opensearchpy/test_server/test_plugins/__init__.py

@@ -0,0 +1,25 @@
+# SPDX-License-Identifier: Apache-2.0
+#
+# The OpenSearch Contributors require contributions made to
+# this file be licensed under the Apache-2.0 license or a
+# compatible open source license.
+#
+# Modifications Copyright OpenSearch Contributors. See
+# GitHub history for details.
+#
+#  Licensed to Elasticsearch B.V. under one or more contributor
+#  license agreements. See the NOTICE file distributed with
+#  this work for additional information regarding copyright
+#  ownership. Elasticsearch B.V. licenses this file to you under
+#  the Apache License, Version 2.0 (the "License"); you may
+#  not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+# 	http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing,
+#  software distributed under the License is distributed on an
+#  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+#  KIND, either express or implied.  See the License for the
+#  specific language governing permissions and limitations
+#  under the License.

test_opensearchpy/test_server/test_plugins/test_alerting.py

@@ -15,7 +15,7 @@ import unittest
 
 from opensearchpy.helpers.test import OPENSEARCH_VERSION
 
-from . import OpenSearchTestCase
+from .. import OpenSearchTestCase
 
 
 class TestAlertingPlugin(OpenSearchTestCase):

test_opensearchpy/test_server/test_plugins/test_index_management.py

@@ -0,0 +1,120 @@
+# -*- coding: utf-8 -*-
+# SPDX-License-Identifier: Apache-2.0
+#
+# The OpenSearch Contributors require contributions made to
+# this file be licensed under the Apache-2.0 license or a
+# compatible open source license.
+#
+# Modifications Copyright OpenSearch Contributors. See
+# GitHub history for details.
+
+
+from __future__ import unicode_literals
+
+from opensearchpy.exceptions import NotFoundError
+
+from .. import OpenSearchTestCase
+
+
+class TestIndexManagementPlugin(OpenSearchTestCase):
+    POLICY_NAME = "example-policy"
+    POLICY_CONTENT = {
+        "policy": {
+            "description": "hot warm delete workflow",
+            "default_state": "hot",
+            "schema_version": 1,
+            "states": [
+                {
+                    "name": "hot",
+                    "actions": [
+                        {
+                            "rollover": {
+                                "min_index_age": "1d",
+                            }
+                        }
+                    ],
+                    "transitions": [{"state_name": "warm"}],
+                },
+                {
+                    "name": "warm",
+                    "actions": [{"replica_count": {"number_of_replicas": 5}}],
+                    "transitions": [
+                        {
+                            "state_name": "delete",
+                            "conditions": {"min_index_age": "30d"},
+                        }
+                    ],
+                },
+                {
+                    "name": "delete",
+                    "actions": [
+                        {
+                            "notification": {
+                                "destination": {"chime": {"url": "<URL>"}},
+                                "message_template": {
+                                    "source": "The index {{ctx.index}} is being deleted"
+                                },
+                            }
+                        },
+                        {"delete": {}},
+                    ],
+                },
+            ],
+            "ism_template": {"index_patterns": ["log*"], "priority": 100},
+        }
+    }
+
+    def test_create_policy(self):
+        # Test to create policy
+        response = self.client.index_management.put_policy(
+            policy=self.POLICY_NAME, body=self.POLICY_CONTENT
+        )
+
+        self.assertNotIn("errors", response)
+        self.assertIn("_id", response)
+
+    def test_get_policy(self):
+        # Create a policy
+        self.test_create_policy()
+
+        # Test to fetch the policy
+        response = self.client.index_management.get_policy(self.POLICY_NAME)
+
+        self.assertNotIn("errors", response)
+        self.assertIn("_id", response)
+        self.assertEqual(response["_id"], self.POLICY_NAME)
+
+    def test_update_policy(self):
+        # Create a policy
+        self.test_create_policy()
+
+        # Fetch the policy
+        response = self.client.index_management.get_policy(self.POLICY_NAME)
+        params = {
+            "if_seq_no": response["_seq_no"],
+            "if_primary_term": response["_primary_term"],
+        }
+
+        policy_content = self.POLICY_CONTENT.copy()
+        policy_content["policy"]["description"] = "example workflow"
+
+        # Test to update policy
+        response = self.client.index_management.put_policy(
+            policy=self.POLICY_NAME, body=policy_content, params=params
+        )
+
+        self.assertNotIn("errors", response)
+        self.assertIn("_id", response)
+
+    def test_delete_policy(self):
+        # Create a policy
+        self.test_create_policy()
+
+        # Test to delete the policy
+        response = self.client.index_management.delete_policy(self.POLICY_NAME)
+
+        self.assertNotIn("errors", response)
+
+        # Try fetching the policy
+        with self.assertRaises(NotFoundError):
+            response = self.client.index_management.get_policy(self.POLICY_NAME)