Add requested changes

Signed-off-by: Divya Madala <[email protected]>

src/test_workflow/integ_test/utils.py

@@ -5,12 +5,16 @@
 # this file be licensed under the Apache-2.0 license or a
 # compatible open source license.
 
+import base64
+
 import semver
 
 
-def get_password(version: str) -> str:
+def str_to_base64(value: str) -> str:
+    return base64.b64encode(value.encode("utf-8")).decode("utf-8")
+
+
+def get_password(version: str, convert_to_base64: bool = False) -> str:
     # Starting in 2.12.0, demo configuration setup script requires a strong password
-    if semver.compare(version, '2.12.0') != -1:
-        return "myStrongPassword123!"
-    else:
-        return "admin"
+    password = "myStrongPassword123!" if semver.compare(version, '2.12.0') != -1 else "admin"
+    return str_to_base64(password) if convert_to_base64 else password

src/validation_workflow/api_request.py

@@ -6,12 +6,11 @@
 # this file be licensed under the Apache-2.0 license or a
 # compatible open source license.
 
-import base64
 from typing import Any
 
 import requests
 
-from test_workflow.integ_test.utils import get_password
+from test_workflow.integ_test.utils import get_password, str_to_base64
 
 """
 This class is to run API test againt on local OpenSearch API URL with default port 9200.
@@ -23,8 +22,7 @@ class ApiTest:
 
     def __init__(self, request_url: str, version: str) -> None:
         self.request_url = request_url
-        self.password = base64.b64encode(f"admin:{get_password(version)}".encode("utf-8")).decode("utf-8")
-        self.apiHeaders_auth = {"Authorization": f'Basic {self.password}'}  # user/pass "admin/pass" in Base64 format fetched from get_password() method
+        self.apiHeaders_auth = {"Authorization": f'Basic {str_to_base64("admin:" + get_password(version))}'}  # user/pass "admin/pass" in Base64 format fetched from get_password() method
         self.apiHeaders_accept = {"Accept": "*/*"}
         self.apiHeaders_content_type = {"Content-Type": "application/json"}
         self.apiHeaders = {}

src/validation_workflow/api_test_cases.py

@@ -23,17 +23,14 @@ def __init__(self) -> None:
     @staticmethod
     def test_apis(version: str, projects: list, security_plugin_exists: bool = True) -> Any:
         pass_counter, fail_counter = 0, 0
+        protocol_prefix = "https" if security_plugin_exists else "http"
 
         # the test case parameters are formated as ['<request_url>',<success_status_code>,'<validate_string(optional)>']
         test_apis = [
-            ['https://localhost:9200/', 200, ''],
-            ['https://localhost:9200/_cat/plugins?v', 200, ''],
-            ['https://localhost:9200/_cat/health?v', 200, 'green'],
+            [f'{protocol_prefix}://localhost:9200/', 200, ''],
+            [f'{protocol_prefix}://localhost:9200/_cat/plugins?v', 200, ''],
+            [f'{protocol_prefix}://localhost:9200/_cat/health?v', 200, 'green'],
         ]
-        # Use http in the request url if the security plugin is absent
-        if not security_plugin_exists:
-            for api in test_apis:
-                api[0] = "http" + api[0][5:]  # type: ignore
 
         if ("opensearch-dashboards" in projects):
             test_apis.append(['http://localhost:5601/api/status', 200, ''])

src/validation_workflow/rpm/validation_rpm.py

@@ -50,9 +50,6 @@ def installation(self) -> bool:
                 self.filename = os.path.basename(self.args.file_path.get(project))
                 execute(f'sudo yum remove {project} -y', ".")
                 execute(f'sudo env OPENSEARCH_INITIAL_ADMIN_PASSWORD={get_password(str(self.args.version))} rpm -ivh {os.path.join(self.tmp_dir.path, self.filename)}', str(self.tmp_dir.path), True, False)  # noqa: 501
-
-            if self.args.allow_without_security:
-                self.args.allow_without_security = self.test_security_plugin("/usr/")
         except:
             raise Exception('Failed to install Opensearch')
         return True
@@ -73,7 +70,7 @@ def start_cluster(self) -> bool:
         return True
 
     def validation(self) -> bool:
-        test_result, counter = ApiTestCases().test_apis(self.args.version, self.args.projects, self.args.allow_without_security)
+        test_result, counter = ApiTestCases().test_apis(self.args.version, self.args.projects, self.check_for_security_plugin(os.path.join(os.sep, "usr", "share", "opensearch"), "rpm") if not self.args.force_https_check else True)  # noqa: E501
         if (test_result):
             logging.info(f'All tests Pass : {counter}')
             return True

src/validation_workflow/tar/validation_tar.py

@@ -51,8 +51,6 @@ def installation(self) -> bool:
             for project in self.args.projects:
                 self.filename = os.path.basename(self.args.file_path.get(project))
                 execute('mkdir ' + os.path.join(self.tmp_dir.path, project) + ' | tar -xzf ' + os.path.join(str(self.tmp_dir.path), self.filename) + ' -C ' + os.path.join(self.tmp_dir.path, project) + ' --strip-components=1', ".", True, False)  # noqa: E501
-            if self.args.allow_without_security:
-                self.args.allow_without_security = self.test_security_plugin(str(self.tmp_dir.path))
         except:
             raise Exception('Failed to install Opensearch')
         return True
@@ -70,8 +68,7 @@ def start_cluster(self) -> bool:
         return True
 
     def validation(self) -> bool:
-
-        test_result, counter = ApiTestCases().test_apis(self.args.version, self.args.projects, self.args.allow_without_security)
+        test_result, counter = ApiTestCases().test_apis(self.args.version, self.args.projects, self.check_for_security_plugin(os.path.join(self.tmp_dir.path, "opensearch"), "tar") if not self.args.force_https_check else True)  # noqa: E501
         if (test_result):
             logging.info(f'All tests Pass : {counter}')
         else:

src/validation_workflow/validation.py

@@ -41,13 +41,10 @@ def copy_artifact(self, filepath: str, tempdir_path: str) -> bool:
         else:
             raise Exception("Provided path for local artifacts does not exist")
 
-    def test_security_plugin(self, work_dir: str) -> bool:
-        (_, path, _) = execute(f'find {work_dir} -type f -iname \'opensearch-plugin\'', ".", True, False)
-        if (path):
-            (_, list_plugins, _) = execute("./opensearch-plugin list", path.replace("opensearch-plugin", "").rstrip("\n"), True, False)
-            return "opensearch-security" in list_plugins
-        else:
-            raise Exception("Couldn't fetch the path to plugin folder")
+    def check_for_security_plugin(self, work_dir: str, distribution: str) -> bool:
+        list_cmd = "dir" if distribution == "zip" else "ls"
+        (_, plugins_list, _) = execute(f"{list_cmd} plugins", work_dir, True, False)
+        return "opensearch-security" in plugins_list
 
     def get_version(self, project: str) -> str:
         return re.search(r'(\d+\.\d+\.\d+)', os.path.basename(project)).group(1)

src/validation_workflow/validation_args.py

@@ -112,11 +112,11 @@ def __init__(self) -> None:
             dest="artifact_type"
         )
         parser.add_argument(
-            "-s",
-            "--allow-without-security",
+            "-f",
+            "--force-https-check",
             action="store_true",
-            default=False,
-            help="Allow Validation without security"
+            default=True,
+            help="If False then check for the existence of security plugin and use http/https as required, else force to check to https"
         )
         group = parser.add_mutually_exclusive_group()
         group.add_argument(
@@ -143,7 +143,7 @@ def __init__(self) -> None:
         self.version = args.version
         self.file_path = args.file_path
         self.artifact_type = args.artifact_type
-        self.allow_without_security = args.allow_without_security
+        self.force_https_check = args.force_https_check
         self.logging_level = args.logging_level
         self.distribution = args.distribution
         self.platform = args.platform

src/validation_workflow/yum/validation_yum.py

@@ -54,8 +54,7 @@ def installation(self) -> bool:
                 urllink = f"{self.args.file_path.get(project)} -o /etc/yum.repos.d/{os.path.basename(self.args.file_path.get(project))}"
                 execute(f'sudo curl -SL {urllink}', ".")
                 execute(f"sudo env OPENSEARCH_INITIAL_ADMIN_PASSWORD={get_password(str(self.args.version))} yum install '{project}-{self.args.version}' -y", ".")
-            if self.args.allow_without_security:
-                self.args.allow_without_security = self.test_security_plugin("/usr/")
+
         except:
             raise Exception('Failed to install Opensearch')
         return True
@@ -71,7 +70,7 @@ def start_cluster(self) -> bool:
         return True
 
     def validation(self) -> bool:
-        test_result, counter = ApiTestCases().test_apis(self.args.version, self.args.projects, self.args.allow_without_security)
+        test_result, counter = ApiTestCases().test_apis(self.args.version, self.args.projects, self.check_for_security_plugin(os.path.join(os.sep, "usr", "share", "opensearch"), "yum") if not self.args.force_https_check else True)  # noqa: E501
         if (test_result):
             logging.info(f'All tests Pass : {counter}')
             return True

tests/tests_test_workflow/test_integ_workflow/integ_test/test_utils.py

@@ -5,12 +5,20 @@
 # this file be licensed under the Apache-2.0 license or a
 # compatible open source license.
 
+import base64
 import unittest
 
-from test_workflow.integ_test.utils import get_password
+from test_workflow.integ_test.utils import get_password, str_to_base64
 
 
 class TestUtils(unittest.TestCase):
     def test_strong_password(self) -> None:
         self.assertEqual("admin", get_password("2.11.1"))
         self.assertEqual("myStrongPassword123!", get_password("3.0.0"))
+        self.assertEqual("YWRtaW4=", get_password("2.11.1", True))
+        self.assertEqual("bXlTdHJvbmdQYXNzd29yZDEyMyE=", get_password("2.12.0", True))
+
+    def test_str_to_base64(self) -> None:
+        value = "admin"
+        result = base64.b64encode(value.encode("utf-8")).decode("utf-8")
+        self.assertEqual(str_to_base64(value), result)

tests/tests_validation_workflow/test_api_test_cases.py

@@ -16,7 +16,7 @@ class TestTestCases(unittest.TestCase):
     def test_opensearch(self, mock_api_get: Mock) -> None:
         mock_api_get.return_value = (200, 'green')
         testcases = ApiTestCases()
-        result = testcases.test_apis("1.3.0", ['opensearch'])
+        result = testcases.test_apis("1.3.0", ['opensearch'], True)
 
         self.assertEqual(result[1], 'There are 3/3 test cases Pass')
         self.assertEqual(mock_api_get.call_count, 3)
@@ -25,7 +25,7 @@ def test_opensearch(self, mock_api_get: Mock) -> None:
     def test_both(self, mock_api_get: Mock) -> None:
         mock_api_get.return_value = (200, 'green')
         testcases = ApiTestCases()
-        result = testcases.test_apis("2.1.1", ['opensearch', 'opensearch-dashboards'])
+        result = testcases.test_apis("2.1.1", ['opensearch', 'opensearch-dashboards'], True)
 
         self.assertEqual(result[1], 'There are 4/4 test cases Pass')
         self.assertEqual(mock_api_get.call_count, 4)

tests/tests_validation_workflow/test_validation.py

@@ -64,37 +64,24 @@ def test_copy_artifact(self, mock_validation_args: Mock, mock_copy: Mock) -> Non
 
     @patch('validation_workflow.validation.execute')
     @patch('validation_workflow.tar.validation_tar.ValidationArgs')
-    def test_is_allow_with_security_true(self, mock_validation_args: Mock, mock_execute: Mock) -> None:
+    def test_check_for_security_plugin_true(self, mock_validation_args: Mock, mock_execute: Mock) -> None:
         mock_execute.return_value = (0, "opensearch-security", "")
 
         mock_validation_args.projects.return_value = ["opensearch"]
         mock_validation = ValidateTar(mock_validation_args.return_value)
 
-        result = mock_validation.test_security_plugin("/bin/opensearch")
+        result = mock_validation.check_for_security_plugin("/tmp/tmkuiuo/opensearch", "tar")
 
         self.assertTrue(result)
 
     @patch('validation_workflow.validation.execute')
     @patch('validation_workflow.tar.validation_tar.ValidationArgs')
-    def test_is_allow_with_security_false(self, mock_validation_args: Mock, mock_execute: Mock) -> None:
+    def test_check_for_security_plugin_false(self, mock_validation_args: Mock, mock_execute: Mock) -> None:
         mock_execute.return_value = (0, "opensearch", "")
 
         mock_validation_args.projects.return_value = ["opensearch"]
         mock_validation = ValidateTar(mock_validation_args.return_value)
 
-        result = mock_validation.test_security_plugin("/bin/opensearch")
+        result = mock_validation.check_for_security_plugin("/bin/opensearch", "tar")
 
         self.assertFalse(result)
-
-    @patch('validation_workflow.validation.execute')
-    @patch('validation_workflow.tar.validation_tar.ValidationArgs')
-    def test_is_allow_with_security_exception(self, mock_validation_args: Mock, mock_execute: Mock) -> None:
-        mock_execute.return_value = (0, "", "error")
-
-        mock_validation_args.projects.return_value = ["opensearch"]
-        mock_validation = ValidateTar(mock_validation_args.return_value)
-
-        with self.assertRaises(Exception) as context:
-            mock_validation.test_security_plugin("/bin/opensearch")
-
-        self.assertEqual(str(context.exception), "Couldn't fetch the path to plugin folder")

tests/tests_validation_workflow/test_validation_args.py

@@ -67,9 +67,9 @@ def test_file_path(self) -> None:
     def test_artifact_type(self) -> None:
         self.assertNotEqual(ValidationArgs().artifact_type, "production")
 
-    @patch("argparse._sys.argv", [VALIDATION_PY, "--version", "1.3.6", "--distribution", "rpm", "--artifact-type", "staging", "--os-build-number", "1234", "--osd-build-number", "2312", "--allow-without-security"])  # noqa: E501
-    def test_allow_without_security(self) -> None:
-        self.assertEqual(ValidationArgs().allow_without_security, True)
+    @patch("argparse._sys.argv", [VALIDATION_PY, "--version", "1.3.6", "--distribution", "rpm", "--artifact-type", "staging", "--os-build-number", "1234", "--osd-build-number", "2312", "--force-https-check"])  # noqa: E501
+    def test_force_https_check(self) -> None:
+        self.assertEqual(ValidationArgs().force_https_check, True)
 
     @patch("argparse._sys.argv", [VALIDATION_PY, "--version", "1.3.0", "--projects", "opensearch"])
     def test_set_projects(self) -> None:
@@ -82,7 +82,7 @@ def test_projects_exception(self) -> None:
             self.assertEqual(ValidationArgs().projects, ["opensearch-dashboards"])
         self.assertEqual(str(ctx.exception), "Missing OpenSearch OpenSearch artifact details! Please provide the same along with OpenSearch-Dashboards to validate")
 
-    @patch("argparse._sys.argv", [VALIDATION_PY, "--file-path", "opensearch=https://opensearch.org/releases/opensearch/2.8.0/opensearch-2.8.0-linux-x64.zip"])
+    @patch("argparse._sys.argv", [VALIDATION_PY, "--file-path", "opensearch=https://opensearch.org/releases/opensearch/2.8.0/opensearch-2.8.0-linux-x64.xyz"])
     def test_file_path_distribution_type(self) -> None:
         with self.assertRaises(Exception) as ctx:
             self.assertEqual(ValidationArgs().projects, ["opensearch"])

tests/tests_validation_workflow/test_validation_rpm.py

@@ -90,30 +90,14 @@ def test_installation(self, mock_system: Mock, mock_validation_args: Mock) -> No
         mock_validation_args.return_value.version = '2.3.0'
         mock_validation_args.return_value.arch = 'x64'
         mock_validation_args.return_value.platform = 'linux'
-        mock_validation_args.return_value.allow_without_security = False
+        mock_validation_args.return_value.force_https_check = True
         mock_validation_args.return_value.projects = ["opensearch"]
 
         validate_rpm = ValidateRpm(mock_validation_args.return_value)
         mock_system.side_effect = lambda *args, **kwargs: (0, "stdout_output", "stderr_output")
         result = validate_rpm.installation()
         self.assertTrue(result)
 
-    @patch('validation_workflow.rpm.validation_rpm.ValidationArgs')
-    @patch('os.path.basename')
-    @patch('validation_workflow.rpm.validation_rpm.execute')
-    @patch('validation_workflow.validation.Validation.test_security_plugin')
-    def test_installation_with_security_parameter(self, mock_security: Mock, mock_system: Mock, mock_basename: Mock, mock_validation_args: Mock) -> None:
-        mock_validation_args.return_value.version = '2.3.0'
-        mock_validation_args.return_value.allow_without_security = True
-        validate_rpm = ValidateRpm(mock_validation_args.return_value)
-        mock_basename.side_effect = lambda path: "mocked_filename"
-        mock_system.side_effect = lambda *args, **kwargs: (0, "stdout_output", "stderr_output")
-        mock_security.return_value = True
-
-        result = validate_rpm.installation()
-        self.assertTrue(result)
-        mock_security.assert_called_once()
-
     @patch("validation_workflow.rpm.validation_rpm.execute", return_value=True)
     @patch('validation_workflow.rpm.validation_rpm.ValidationArgs')
     @patch('time.sleep')
@@ -138,6 +122,25 @@ def test_validation(self, mock_test_apis: Mock, mock_validation_args: Mock) -> N
 
         self.assertEqual(mock_test_apis.call_count, 1)
 
+    @patch('validation_workflow.rpm.validation_rpm.ValidationArgs')
+    @patch('validation_workflow.rpm.validation_rpm.ApiTestCases')
+    @patch('os.path.basename')
+    @patch('validation_workflow.rpm.validation_rpm.execute')
+    @patch('validation_workflow.validation.Validation.check_for_security_plugin')
+    def test_validation_with_security_parameter(self, mock_security: Mock, mock_system: Mock, mock_basename: Mock, mock_test_apis: Mock, mock_validation_args: Mock) -> None:
+        mock_validation_args.return_value.version = '2.3.0'
+        mock_validation_args.return_value.force_https_check = False
+        validate_rpm = ValidateRpm(mock_validation_args.return_value)
+        mock_basename.side_effect = lambda path: "mocked_filename"
+        mock_system.side_effect = lambda *args, **kwargs: (0, "stdout_output", "stderr_output")
+        mock_security.return_value = True
+        mock_test_apis_instance = mock_test_apis.return_value
+        mock_test_apis_instance.test_apis.return_value = (True, 4)
+
+        result = validate_rpm.validation()
+        self.assertTrue(result)
+        mock_security.assert_called_once()
+
     @patch('validation_workflow.rpm.validation_rpm.ValidationArgs')
     @patch('validation_workflow.rpm.validation_rpm.ApiTestCases')
     def test_failed_testcases(self, mock_test_apis: Mock, mock_validation_args: Mock) -> None: