Add async tests for plugins (#425)

* test(plugins): add async tests for plugins

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

* chore(USER_GUIDE): remove **

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

* test(async): remove the event loop session scoped fixture

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

---------

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

USER_GUIDE.md

@@ -15,12 +15,12 @@
     - [Searching for documents with filters](#searching-for-documents-with-filters)
   - [Using plugins](#using-plugins)
     - [Alerting plugin](#alerting-plugin)
-      - [**Searching for monitors**](#searching-for-monitors)
+      - [Searching for monitors](#searching-for-monitors)
-      - [**Getting a monitor**](#getting-a-monitor)
+      - [Getting a monitor](#getting-a-monitor)
-      - [**Creating a monitor**](#creating-a-monitor)
+      - [Creating a monitor](#creating-a-monitor)
-      - [**Creating a destination**](#creating-a-destination)
+      - [Creating a destination](#creating-a-destination)
-      - [**Getting alerts**](#getting-alerts)
+      - [Getting alerts](#getting-alerts)
-      - [**Acknowledge alerts**](#acknowledge-alerts)
+      - [Acknowledge alerts](#acknowledge-alerts)
     - [Index management plugin](#index-management-plugin)
       - [Creating a policy](#creating-a-policy)
       - [Getting a policy](#getting-a-policy)
@@ -283,7 +283,7 @@ Plugin client definitions can be found here --
 
 ### Alerting plugin
 
-#### **Searching for monitors**
+#### Searching for monitors
 [API definition](https://opensearch.org/docs/latest/monitoring-plugins/alerting/api/#search-monitors)
 ```python
 print('\Searching for monitors:')
@@ -300,7 +300,7 @@ response = client.plugins.alerting.search_monitor(query)
 print(response)
 ```
 
-#### **Getting a monitor**
+#### Getting a monitor
 [API definition](https://opensearch.org/docs/latest/monitoring-plugins/alerting/api/#get-monitor)
 ```python
 print('\Getting a monitor:')
@@ -309,7 +309,7 @@ response = client.plugins.alerting.get_monitor("monitorID")
 print(response)
 ```
 
-#### **Creating a monitor**
+#### Creating a monitor
 [API definition](https://opensearch.org/docs/latest/monitoring-plugins/alerting/api/#create-a-bucket-level-monitor)
 ```python
 print('\Creating a bucket level monitor:')
@@ -382,7 +382,7 @@ response = client.plugins.alerting.create_monitor(query)
 print(response)
 ```
 
-#### **Creating a destination**
+#### Creating a destination
 [API definition](https://opensearch.org/docs/latest/monitoring-plugins/alerting/api/#create-destination)
 ```python
 print('\Creating an email destination:')
@@ -409,7 +409,7 @@ response = client.plugins.alerting.create_destination(query)
 print(response)
 ```
 
-#### **Getting alerts**
+#### Getting alerts
 [API definition](https://opensearch.org/docs/latest/monitoring-plugins/alerting/api/#get-alerts)
 ```python
 print('\Getting alerts:')
@@ -418,7 +418,7 @@ response = client.plugins.alerting.get_alerts()
 print(response)
 ```
 
-#### **Acknowledge alerts**
+#### Acknowledge alerts
 [API definition](https://opensearch.org/docs/latest/monitoring-plugins/alerting/api/#acknowledge-alert)
 ```python
 print('\Acknowledge alerts:')

test_opensearchpy/test_async/test_server/__init__.py

@@ -23,3 +23,24 @@
 #  KIND, either express or implied.  See the License for the
 #  specific language governing permissions and limitations
 #  under the License.
+
+
+from unittest import IsolatedAsyncioTestCase
+
+from opensearchpy._async.helpers.test import get_test_client
+from opensearchpy.connection.async_connections import add_connection
+
+from ...utils import wipe_cluster
+
+
+class AsyncOpenSearchTestCase(IsolatedAsyncioTestCase):
+    async def asyncSetUp(self):
+        self.client = await get_test_client(
+            verify_certs=False, http_auth=("admin", "admin")
+        )
+        await add_connection("default", self.client)
+
+    async def asyncTearDown(self):
+        wipe_cluster(self.client)
+        if self.client:
+            await self.client.close()

test_opensearchpy/test_async/test_server/test_helpers/conftest.py

@@ -7,7 +7,6 @@
 # Modifications Copyright OpenSearch Contributors. See
 # GitHub history for details.
 
-import asyncio
 import re
 from datetime import datetime
 
@@ -34,21 +33,14 @@ from test_opensearchpy.test_async.test_server.test_helpers.test_document import
 pytestmark = pytest.mark.asyncio
 
 
-@pytest.fixture(scope="session")
+@fixture(scope="function")
-def event_loop():
-    loop = asyncio.get_event_loop()
-    yield loop
-    loop.close()
-
-
-@fixture(scope="session")
 async def client():
     client = await get_test_client(verify_certs=False, http_auth=("admin", "admin"))
     await add_connection("default", client)
     return client
 
 
-@fixture(scope="session")
+@fixture(scope="function")
 async def opensearch_version(client):
     info = await client.info()
     print(info)

test_opensearchpy/test_async/test_server/test_helpers/test_faceted_search.py

@@ -52,7 +52,7 @@ class MetricSearch(AsyncFacetedSearch):
     }
 
 
-@pytest.fixture(scope="session")
+@pytest.fixture(scope="function")
 def commit_search_cls(opensearch_version):
     interval_kwargs = {"fixed_interval": "1d"}
 
@@ -77,7 +77,7 @@ def commit_search_cls(opensearch_version):
     return CommitSearch
 
 
-@pytest.fixture(scope="session")
+@pytest.fixture(scope="function")
 def repo_search_cls(opensearch_version):
     interval_type = "calendar_interval"
 
@@ -98,7 +98,7 @@ def repo_search_cls(opensearch_version):
     return RepoSearch
 
 
-@pytest.fixture(scope="session")
+@pytest.fixture(scope="function")
 def pr_search_cls(opensearch_version):
     interval_type = "calendar_interval"
 

test_opensearchpy/test_async/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_async/test_server/test_plugins/test_alerting.py

@@ -0,0 +1,187 @@
+# -*- 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
+
+import unittest
+
+import pytest
+
+from opensearchpy.helpers.test import OPENSEARCH_VERSION
+
+from .. import AsyncOpenSearchTestCase
+
+pytestmark = pytest.mark.asyncio
+
+
+class TestAlertingPlugin(AsyncOpenSearchTestCase):
+    @unittest.skipUnless(
+        (OPENSEARCH_VERSION) and (OPENSEARCH_VERSION < (2, 0, 0)),
+        "Plugin not supported for opensearch version",
+    )
+    async def test_create_destination(self):
+        # Test to create alert destination
+        dummy_destination = {
+            "name": "my-destination",
+            "type": "slack",
+            "slack": {"url": "http://www.example.com"},
+        }
+        response = await self.client.alerting.create_destination(dummy_destination)
+
+        self.assertNotIn("errors", response)
+        self.assertIn("_id", response)
+
+    @unittest.skipUnless(
+        (OPENSEARCH_VERSION) and (OPENSEARCH_VERSION < (2, 0, 0)),
+        "Plugin not supported for opensearch version",
+    )
+    async def test_get_destination(self):
+        # Create a dummy destination
+        await self.test_create_destination()
+
+        # Try fetching the destination
+        response = await self.client.alerting.get_destination()
+
+        self.assertNotIn("errors", response)
+        self.assertGreaterEqual(response["totalDestinations"], 1)
+        self.assertEqual(response["totalDestinations"], len(response["destinations"]))
+
+    @unittest.skipUnless(
+        (OPENSEARCH_VERSION) and (OPENSEARCH_VERSION < (2, 0, 0)),
+        "Plugin not supported for opensearch version",
+    )
+    async def test_create_monitor(self):
+        # Create a dummy destination
+        await self.test_create_destination()
+
+        # Try fetching the destination
+        destination = await self.client.alerting.get_destination()
+        self.assertGreaterEqual(
+            destination["totalDestinations"],
+            1,
+            "No destination entries found in the database.",
+        )
+
+        # Select the first destination available
+        destination = destination["destinations"][0]
+
+        # A dummy schedule for 1 minute interval
+        schedule = {"period": {"interval": 1, "unit": "MINUTES"}}
+
+        # A dummy query fetching everything
+        query = {"query": {"query_string": {"query": "*"}}}
+
+        # A dummy action with the dummy destination
+        action = {
+            "name": "test-action",
+            "destination_id": destination["id"],
+            "message_template": {"source": "This is my message body."},
+            "throttle_enabled": True,
+            "throttle": {"value": 27, "unit": "MINUTES"},
+            "subject_template": {"source": "TheSubject"},
+        }
+
+        # A dummy trigger with the dummy action
+        triggers = {
+            "name": "test-trigger",
+            "severity": "1",
+            "condition": {
+                "script": {
+                    "source": "ctx.results[0].hits.total.value > 0",
+                    "lang": "painless",
+                }
+            },
+            "actions": [action],
+        }
+
+        # A dummy monitor with the dummy schedule, dummy query, dummy trigger
+        monitor = {
+            "type": "monitor",
+            "name": "test-monitor",
+            "monitor_type": "query_level_monitor",
+            "enabled": True,
+            "schedule": schedule,
+            "inputs": [{"search": {"indices": ["*"], "query": query}}],
+            "triggers": [triggers],
+        }
+
+        response = await self.client.alerting.create_monitor(monitor)
+
+        self.assertNotIn("errors", response)
+        self.assertIn("_id", response)
+        self.assertIn("monitor", response)
+
+    @unittest.skipUnless(
+        (OPENSEARCH_VERSION) and (OPENSEARCH_VERSION < (2, 0, 0)),
+        "Plugin not supported for opensearch version",
+    )
+    async def test_search_monitor(self):
+        # Create a dummy monitor
+        await self.test_create_monitor()
+
+        # Create a monitor search query by it's name
+        query = {"query": {"match": {"monitor.name": "test-monitor"}}}
+
+        # Perform the search with the above query
+        response = await self.client.alerting.search_monitor(query)
+
+        self.assertNotIn("errors", response)
+        self.assertIn("hits", response)
+        self.assertEqual(response["hits"]["total"]["value"], 1, "No monitor found.")
+
+    @unittest.skipUnless(
+        (OPENSEARCH_VERSION) and (OPENSEARCH_VERSION < (2, 0, 0)),
+        "Plugin not supported for opensearch version",
+    )
+    async def test_get_monitor(self):
+        # Create a dummy monitor
+        await self.test_create_monitor()
+
+        # Create a monitor search query by it's name
+        query = {"query": {"match": {"monitor.name": "test-monitor"}}}
+
+        # Perform the search with the above query
+        response = await self.client.alerting.search_monitor(query)
+
+        # Select the first monitor
+        monitor = response["hits"]["hits"][0]
+
+        # Fetch the monitor by id
+        response = await self.client.alerting.get_monitor(monitor["_id"])
+
+        self.assertNotIn("errors", response)
+        self.assertIn("_id", response)
+        self.assertIn("monitor", response)
+
+    @unittest.skipUnless(
+        (OPENSEARCH_VERSION) and (OPENSEARCH_VERSION < (2, 0, 0)),
+        "Plugin not supported for opensearch version",
+    )
+    async def test_run_monitor(self):
+        # Create a dummy monitor
+        await self.test_create_monitor()
+
+        # Create a monitor search query by it's name
+        query = {"query": {"match": {"monitor.name": "test-monitor"}}}
+
+        # Perform the search with the above query
+        response = await self.client.alerting.search_monitor(query)
+
+        # Select the first monitor
+        monitor = response["hits"]["hits"][0]
+
+        # Run the monitor by id
+        response = await self.client.alerting.run_monitor(monitor["_id"])
+
+        self.assertEqual(response["error"], None)
+        self.assertIn("monitor_name", response)
+        self.assertIn("period_start", response)
+        self.assertIn("period_end", response)

test_opensearchpy/test_async/test_server/test_plugins/test_index_management.py

@@ -0,0 +1,124 @@
+# -*- 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
+
+import pytest
+
+from opensearchpy.exceptions import NotFoundError
+
+from .. import AsyncOpenSearchTestCase
+
+pytestmark = pytest.mark.asyncio
+
+
+class TestIndexManagementPlugin(AsyncOpenSearchTestCase):
+    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},
+        }
+    }
+
+    async def test_create_policy(self):
+        # Test to create policy
+        response = await self.client.index_management.put_policy(
+            policy=self.POLICY_NAME, body=self.POLICY_CONTENT
+        )
+
+        self.assertNotIn("errors", response)
+        self.assertIn("_id", response)
+
+    async def test_get_policy(self):
+        # Create a policy
+        await self.test_create_policy()
+
+        # Test to fetch the policy
+        response = await self.client.index_management.get_policy(self.POLICY_NAME)
+
+        self.assertNotIn("errors", response)
+        self.assertIn("_id", response)
+        self.assertEqual(response["_id"], self.POLICY_NAME)
+
+    async def test_update_policy(self):
+        # Create a policy
+        await self.test_create_policy()
+
+        # Fetch the policy
+        response = await 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 = await self.client.index_management.put_policy(
+            policy=self.POLICY_NAME, body=policy_content, params=params
+        )
+
+        self.assertNotIn("errors", response)
+        self.assertIn("_id", response)
+
+    async def test_delete_policy(self):
+        # Create a policy
+        await self.test_create_policy()
+
+        # Test to delete the policy
+        response = await self.client.index_management.delete_policy(self.POLICY_NAME)
+
+        self.assertNotIn("errors", response)
+
+        # Try fetching the policy
+        with self.assertRaises(NotFoundError):
+            response = await self.client.index_management.get_policy(self.POLICY_NAME)