test: add OpenSearch test suite

Add pytest fixture that runs OpenSearch client.
Add a simple test in the new suite to verify that OpenSearch in running correctly.

test.py

@@ -45,6 +45,7 @@
 from test.pylib.util import LogPrefixAdapter
 from test.pylib.scylla_cluster import ScyllaServer, ScyllaCluster, get_cluster_manager, merge_cmdline_options
 from test.pylib.minio_server import MinioServer
+from test.pylib.opensearch_cluster import OpenSearchCluster
 from typing import Dict, List, Callable, Any, Iterable, Optional, Awaitable, Union
 import logging
 from test.pylib import coverage_utils
@@ -719,6 +720,28 @@ def junit_tests(self):
         count in the CI report"""
         return []
 
+class OpensearchTestSuite(PythonTestSuite):
+    """A collection of Python pytests against an OpenSearch cluster."""
+
+    def build_test_list(self):
+        """Build list of OpenSearch python tests"""
+        return TestSuite.build_test_list(self)
+
+    async def add_test(self, shortname: str, casename: str) -> None:
+        """Add test to suite"""
+        test = OpensearchTest(self.next_id((shortname, 'opensearch', self.mode)), shortname, casename, self)
+        self.tests.append(test)
+
+    @property
+    def pattern(self) -> str:
+        """Python pattern"""
+        return "test_*.py"
+
+    # TODO: maybe remove?
+    def junit_tests(self):
+        """Return an empty list, since topology tests are excluded from an aggregated Junit report to prevent double
+        count in the CI report"""
+        return []
 
 class RunTestSuite(TestSuite):
     """TestSuite for test directory with a 'run' script """
@@ -1410,6 +1433,41 @@ async def run(self, options: argparse.Namespace) -> Test:
         return self
 
 
+class OpensearchTest(PythonTest):
+    """Run a pytest collection of cases against a standalone OpenSearch"""
+    status: bool
+
+    def __init__(self, test_no: int, shortname: str, casename: str, suite) -> None:
+        super().__init__(test_no, shortname, casename, suite)
+
+
+    # TODO: check, fix, adjust
+    async def run(self, options: argparse.Namespace) -> Test:
+
+        self._prepare_pytest_params(options)
+
+        test_path = os.path.join(self.suite.options.tmpdir, self.mode)
+        async with get_cluster_manager(self.uname, self.suite.clusters, test_path) as manager:
+            self.args.insert(0, "--tmpdir={}".format(options.tmpdir))
+            # self.args.insert(0, "--manager-api={}".format(manager.sock_path))
+            if options.artifacts_dir_url:
+                self.args.insert(0, "--artifacts_dir_url={}".format(options.artifacts_dir_url))
+
+            try:
+                # Note: start manager here so cluster (and its logs) is available in case of failure
+                await manager.start()
+                self.success = await run_test(self, options)
+            except Exception as e:
+                self.server_log = manager.cluster.read_server_log()
+                self.server_log_filename = manager.cluster.server_log_filename()
+                if not manager.is_before_test_ok:
+                    print("Test {} pre-check failed: {}".format(self.name, str(e)))
+                    print("Server log of the first server:\n{}".format(self.server_log))
+                    # Don't try to continue if the cluster is broken
+                    raise
+            manager.logger.info("Test %s %s", self.uname, "succeeded" if self.success else "failed ")
+        return self
+
 class ToolTest(Test):
     """Run a collection of pytest test cases
 
@@ -1900,6 +1958,10 @@ async def reap(done, pending, signaled):
     await proxy_s3_server.start()
     TestSuite.artifacts.add_exit_artifact(None, proxy_s3_server.stop)
 
+    os_cluster = OpenSearchCluster(options.tmpdir, '127.0.0.1', LogPrefixAdapter(logging.getLogger('opensearch'), {'prefix': 'opensearch'}))
+    await os_cluster.start()
+    TestSuite.artifacts.add_exit_artifact(None, os_cluster.stop)
+
     console.print_start_blurb()
     max_failures = options.max_failures
     failed = 0

test/opensearch/conftest.py

@@ -0,0 +1,28 @@
+#
+# Copyright (C) 2024-present ScyllaDB
+#
+# SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
+#
+# This file configures pytest for all tests in this directory, and also
+# defines common test fixtures for all of them to use.
+from opensearchpy import OpenSearch
+import pytest
+import os
+
+@pytest.fixture(scope="function")
+async def opensearch():
+    host = os.environ.get('OPENSEARCH_ADDRESS')
+    port = os.environ.get('OPENSEARCH_PORT')
+
+    client = OpenSearch(
+        hosts = [{'host': host, 'port': port}],
+        http_compress = True,
+        use_ssl = False,
+        verify_certs = False,
+        ssl_assert_hostname = False,
+        ssl_show_warn = False
+    )
+
+    yield client
+
+    client.close()

test/opensearch/suite.yaml

@@ -0,0 +1 @@
+type: OpenSearch

test/opensearch/test_opensearch.py

@@ -0,0 +1,25 @@
+#
+# Copyright (C) 2025-present ScyllaDB
+#
+# SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
+#
+import pytest
+
+@pytest.mark.asyncio
+async def test_opensearch_basic(opensearch):
+
+    index_name = 'python-test-index'
+    index_body = {
+    'settings': {
+        'index': {
+        'number_of_shards': 4
+        }
+    }
+    }
+
+    response = opensearch.indices.create(index_name, body=index_body)
+    print(f"Index creation response: {response}")
+
+    response = opensearch.cat.indices(format='json')
+    for index in response:
+        print(index['index'])