Add smoke test framework for opensearch bundle (opensearch-project#5185)

Signed-off-by: Zelin Hao <zelinhao@amazon.com>
Signed-off-by: Peter Zhu <zhujiaxi@amazon.com>
Co-authored-by: Peter Zhu <zhujiaxi@amazon.com>

Author: zelinh

Pipfile

@@ -9,7 +9,7 @@ types-PyYAML = "~=6.0.1"
 # TODO: The 'requests' package stays on 2.28 until we deprecate CentOS7.
 #       As newer version requires openssl1.1.1 where CentOS7 only provides openssl1.1.0.
 #       https://github.com/opensearch-project/opensearch-build/issues/3554
-requests = "<=2.28.1"
+requests = "==2.31.0"
 types-requests = "~=2.25"
 pre-commit = "~=2.15.0"
 isort = "~=5.9"
@@ -44,6 +44,7 @@ types-urllib3 = "~=1.26.25.14"
 charset-normalizer = "~=2.1.1"
 beautifulsoup4 = "~=4.12.3"
 lxml = "~=5.3.0"
+openapi-core = "~=0.19.4"
 
 [dev-packages]
 

Pipfile.lock

@@ -6,6 +6,9 @@ ci:
     name: opensearchstaging/ci-runner:ci-runner-al2-opensearch-build-v1
     args: -e JAVA_HOME=/opt/java/openjdk-21
 components:
+  - name: opensearch
+    smoke-test:
+      test-spec: opensearch.yml
   - name: alerting
     integ-test:
       test-configs:

manifests/2.19.0/opensearch-2.19.0-test.yml

@@ -0,0 +1,36 @@
+#!/usr/bin/env python
+# Copyright OpenSearch Contributors
+# 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.
+
+import sys
+
+from manifests.test_manifest import TestManifest
+from system import console
+from test_workflow.smoke_test.smoke_test_runners import SmokeTestRunners
+from test_workflow.test_args import TestArgs
+
+
+def main() -> int:
+    args = TestArgs()
+
+    # Any logging.info call preceding to next line in the execution chain will make the console output not displaying logs in console.
+    console.configure(level=args.logging_level)
+
+    test_manifest = TestManifest.from_path(args.test_manifest_path)
+
+    all_results = SmokeTestRunners.from_test_manifest(args, test_manifest).run()
+
+    all_results.log()
+
+    if all_results.failed():
+        return 1
+    else:
+        return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

src/run_smoke_test.py

@@ -0,0 +1,8 @@
+# Copyright OpenSearch Contributors
+# 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.
+#
+# This page intentionally left blank.

src/test_workflow/smoke_test/__init__.py

@@ -0,0 +1,102 @@
+# Copyright OpenSearch Contributors
+# 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.
+
+import logging
+import os
+import shutil
+import time
+
+import requests
+
+from git.git_repository import GitRepository
+from manifests.build_manifest import BuildManifest
+from manifests.bundle_manifest import BundleManifest
+from manifests.test_manifest import TestManifest
+from system.process import Process
+from test_workflow.integ_test.distributions import Distributions
+from test_workflow.test_args import TestArgs
+from test_workflow.test_recorder.test_recorder import TestRecorder
+
+
+class SmokeTestClusterOpenSearch():
+    # dependency_installer: DependencyInstallerOpenSearch
+    repo: GitRepository
+
+    def __init__(
+        self,
+        args: TestArgs,
+        work_dir: str,
+        test_recorder: TestRecorder
+    ) -> None:
+        self.args = args
+        self.work_dir = work_dir
+        self.test_recorder = test_recorder
+        self.process_handler = Process()
+        self.test_manifest = TestManifest.from_path(args.test_manifest_path)
+        self.product = self.test_manifest.name.lower().replace(" ", "-")
+        self.path = args.paths.get(self.product)
+        self.build_manifest = BuildManifest.from_urlpath(os.path.join(self.path, "builds", f"{self.product}", "manifest.yml"))
+        self.bundle_manifest = BundleManifest.from_urlpath(os.path.join(self.path, "dist", f"{self.product}", "manifest.yml"))
+        self.version = self.bundle_manifest.build.version
+        self.platform = self.bundle_manifest.build.platform
+        self.arch = self.bundle_manifest.build.architecture
+        self.dist = self.bundle_manifest.build.distribution
+        self.distribution = Distributions.get_distribution(self.product, self.dist, self.version, work_dir)
+
+    def cluster_version(self) -> str:
+        return self.version
+
+    def download_or_copy_bundle(self, work_dir: str) -> str:
+        extension = "tar.gz" if self.dist == "tar" else self.dist
+        artifact_name = f"{self.product}-{self.version}-{self.platform}-{self.arch}.{extension}"
+        src_path = '/'.join([self.path.rstrip("/"), "dist", f"{self.product}", f"{artifact_name}"]) \
+            if self.path.startswith("https://") else os.path.join(self.path, "dist",
+                                                                  f"{self.product}", f"{artifact_name}")
+        dest_path = os.path.join(work_dir, artifact_name)
+
+        if src_path.startswith("https://"):
+            logging.info(f"Downloading artifacts to {dest_path}")
+            response = requests.get(src_path)
+            with open(dest_path, "wb") as file:
+                file.write(response.content)
+        else:
+            logging.info(f"Trying to copy {src_path} to {dest_path}")
+            # Only copy if it's a file
+            if os.path.isfile(src_path):
+                shutil.copy2(src_path, dest_path)
+                logging.info(f"Copied {src_path} to {dest_path}")
+        return artifact_name
+
+    # Reason we don't re-use test-suite from integ-test is that it's too specific and not generic and lightweight.
+    def __installation__(self, work_dir: str) -> None:
+        self.distribution.install(self.download_or_copy_bundle(work_dir))
+        logging.info("Cluster is installed and ready to be start.")
+
+    # Start the cluster after installed and provide endpoint.
+    def __start_cluster__(self, work_dir: str) -> None:
+        self.__installation__(work_dir)
+        self.process_handler.start(self.distribution.start_cmd, self.distribution.install_dir, self.distribution.require_sudo)
+        logging.info(f"Started OpenSearch with parent PID {self.process_handler.pid}")
+        time.sleep(30)
+        logging.info("Cluster is started.")
+
+    # Check if the cluster is ready
+    def __check_cluster_ready__(self) -> bool:
+        url = "https://localhost:9200/"
+        logging.info(f"Pinging {url}")
+        try:
+            request = requests.get(url, verify=False, auth=("admin", "myStrongPassword123!"))
+            logging.info(f"Cluster response is {request.text}")
+            return 200 <= request.status_code < 300
+        except requests.RequestException as e:
+            logging.info(f"Request is {request.text}")
+            logging.info(f"Cluster check fails: {e}")
+            return False
+
+    def __uninstall__(self) -> None:
+        self.process_handler.terminate()
+        logging.info("Cluster is terminated.")

src/test_workflow/smoke_test/smoke_test_cluster_opensearch.py

@@ -0,0 +1,89 @@
+# Copyright OpenSearch Contributors
+# 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.
+
+import abc
+import json
+import logging
+import os
+import sys
+import time
+from pathlib import Path
+from typing import Any
+
+import yaml
+
+from manifests.component_manifest import Components
+from manifests.test_manifest import TestManifest
+from system.temporary_directory import TemporaryDirectory
+from test_workflow.smoke_test.smoke_test_cluster_opensearch import SmokeTestClusterOpenSearch
+from test_workflow.test_args import TestArgs
+from test_workflow.test_recorder.test_recorder import TestRecorder
+
+
+class SmokeTestRunner(abc.ABC):
+    args: TestArgs
+    test_manifest: TestManifest
+    tests_dir: str
+    test_recorder: TestRecorder
+    components: Components
+
+    def __init__(self, args: TestArgs, test_manifest: TestManifest) -> None:
+        self.args = args
+        self.test_manifest = test_manifest
+        self.tests_dir = os.path.join(os.getcwd(), "test-results")
+        os.makedirs(self.tests_dir, exist_ok=True)
+        self.test_recorder = TestRecorder(self.args.test_run_id, "smoke-test", self.tests_dir, args.base_path)
+        self.save_log = self.test_recorder.test_results_logs
+        self.version = ""
+
+    def start_test(self, work_dir: Path) -> Any:
+        pass
+
+    def extract_paths_from_yaml(self, component: str, version: str) -> Any:
+        base_path = os.path.dirname(os.path.abspath(__file__))
+        paths = [
+            os.path.join(base_path, "smoke_tests_spec", f"{version.split('.')[0]}.x", f"{component}.yml"),
+            os.path.join(base_path, "smoke_tests_spec", "default", f"{component}.yml")
+        ]
+        for file_path in paths:
+            if os.path.exists(file_path):
+                logging.info(f"Component spec for {component} with path {file_path} is found.")
+                with open(file_path, 'r') as file:
+                    data = yaml.safe_load(file)  # Load the YAML content
+                # Extract paths
+                paths = data.get('paths', {})
+                return paths
+        logging.error("No spec found.")
+        sys.exit(1)
+
+    def convert_parameter_json(self, data: list) -> str:
+        return "\n".join(json.dumps(item) for item in data) + "\n" if data else ""
+
+    # Essential of initiate the testing phase. This function is called by the run_smoke_test.py
+    def run(self) -> Any:
+        with TemporaryDirectory(keep=self.args.keep, chdir=True) as work_dir:
+
+            logging.info("Initiating smoke tests.")
+            test_cluster = SmokeTestClusterOpenSearch(self.args, os.path.join(work_dir.path), self.test_recorder)
+            self.version = test_cluster.cluster_version()
+            test_cluster.__start_cluster__(os.path.join(work_dir.path))
+            is_cluster_ready = False
+            for i in range(10):
+                logging.info(f"Attempt {i} of 10 to check cluster.")
+                if test_cluster.__check_cluster_ready__():
+                    is_cluster_ready = True
+                    break
+                time.sleep(10)
+            try:
+                if is_cluster_ready:
+                    results_data = self.start_test(work_dir.path)
+                else:
+                    logging.info("Cluster is not ready after 10 attempts.")
+            finally:
+                logging.info("Terminating and uninstalling the cluster.")
+                test_cluster.__uninstall__()
+        return results_data

src/test_workflow/smoke_test/smoke_test_runner.py

@@ -0,0 +1,81 @@
+# Copyright OpenSearch Contributors
+# 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.
+
+import logging
+import os
+from pathlib import Path
+from typing import Any
+
+import requests
+from openapi_core import Spec, validate_request, validate_response
+from openapi_core.contrib.requests import RequestsOpenAPIRequest, RequestsOpenAPIResponse
+
+from manifests.test_manifest import TestManifest
+from test_workflow.smoke_test.smoke_test_runner import SmokeTestRunner
+from test_workflow.test_args import TestArgs
+from test_workflow.test_result.test_component_results import TestComponentResults
+from test_workflow.test_result.test_result import TestResult
+from test_workflow.test_result.test_suite_results import TestSuiteResults
+
+
+class SmokeTestRunnerOpenSearch(SmokeTestRunner):
+
+    def __init__(self, args: TestArgs, test_manifest: TestManifest) -> None:
+        super().__init__(args, test_manifest)
+        logging.info("Entering Smoke test for OpenSearch Bundle.")
+
+        # TODO: Download the spec from https://github.com/opensearch-project/opensearch-api-specification/releases/download/main-latest/opensearch-openapi.yaml
+        spec_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), "smoke_tests_spec", "opensearch-openapi.yaml")
+        self.spec_ = Spec.from_file_path(spec_file)
+        self.mimetype = {
+            "Content-Type": "application/json"
+        }
+        # self.openapi = openapi_core.OpenAPI.from_file_path(spec_file)
+
+    def validate_request_swagger(self, request: Any) -> None:
+        request = RequestsOpenAPIRequest(request)
+        validate_request(request=request, spec=self.spec_)
+        logging.info("Request is validated.")
+
+    def validate_response_swagger(self, response: Any) -> None:
+        request = RequestsOpenAPIRequest(response.request)
+        response = RequestsOpenAPIResponse(response)
+        validate_response(response=response, spec=self.spec_, request=request)
+        logging.info("Response is validated.")
+
+    def start_test(self, work_dir: Path) -> TestSuiteResults:
+        url = "https://localhost:9200"
+
+        all_results = TestSuiteResults()
+        for component in self.test_manifest.components.select(self.args.components):
+            if component.smoke_test:
+                logging.info(f"Running smoke test on {component.name} component.")
+                component_spec = self.extract_paths_from_yaml(component.name, self.version)
+                logging.info(f"component spec is {component_spec}")
+                test_results = TestComponentResults()
+                for api_requests, api_details in component_spec.items():
+                    request_url = ''.join([url, api_requests])
+                    logging.info(f"Validating api request {api_requests}")
+                    logging.info(f"API request URL is {request_url}")
+                    for method in api_details.keys():  # Iterates over each method, e.g., "GET", "POST"
+                        requests_method = getattr(requests, method.lower())
+                        parameters_data = self.convert_parameter_json(api_details.get(method).get("parameters"))
+                        header = api_details.get(method).get("header", self.mimetype)
+                        logging.info(f"Parameter is {parameters_data} and type is {type(parameters_data)}")
+                        logging.info(f"header is {header}")
+                        status = 0
+                        try:
+                            response = requests_method(request_url, verify=False, auth=("admin", "myStrongPassword123!"), headers=header, data=parameters_data)
+                            logging.info(f"Response is {response.text}")
+                            self.validate_response_swagger(response)
+                        except:
+                            status = 1
+                        finally:
+                            test_result = TestResult(component.name, ' '.join([api_requests, method]), status)  # type: ignore
+                            test_results.append(test_result)
+                all_results.append(component.name, test_results)
+        return all_results

src/test_workflow/smoke_test/smoke_test_runner_opensearch.py

@@ -0,0 +1,22 @@
+# Copyright OpenSearch Contributors
+# 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.
+
+
+from manifests.test_manifest import TestManifest
+from test_workflow.smoke_test.smoke_test_runner import SmokeTestRunner
+from test_workflow.smoke_test.smoke_test_runner_opensearch import SmokeTestRunnerOpenSearch
+from test_workflow.test_args import TestArgs
+
+
+class SmokeTestRunners:
+    RUNNERS = {
+        "OpenSearch": SmokeTestRunnerOpenSearch
+    }
+
+    @classmethod
+    def from_test_manifest(cls, args: TestArgs, test_manifest: TestManifest) -> SmokeTestRunner:
+        return cls.RUNNERS[test_manifest.name](args, test_manifest)