Adding unit tests for utilities.py

scripts/plugins/config/parser.py

@@ -17,9 +17,15 @@ def get_config_list(config_file, schema_file):
     \n\t PLUGIN SCHEMA FILE PATH:           [{schema_file}]    \
     \n\n"
     )
-
-    schema_yaml = load_yaml(schema_file)
-    config_yaml = load_yaml(config_file)
+    try:
+        schema_yaml = load_yaml(schema_file)
+        config_yaml = load_yaml(config_file)
+    except FileNotFoundError as e:
+        logger.error(
+            f"Required config file was not found. \
+                To fix this please add the following file: [{e.filename}]"
+        )
+        sys.exit(2)
     schema = convert_yaml_to_dict(schema_yaml)
     schema_properties_list = generate_schema_keys(schema)
     schema_list = generate_populated_schema_list(schema, schema_properties_list, config_yaml)

scripts/plugins/deploy_plugins.py

@@ -271,19 +271,25 @@ def deploy_plugins():  # pylint: disable=too-many-locals,too-many-branches,too-m
                         \n\t\t\tSTACK ACTION:   [{stack_action}]"
         )
 
-        result = run_cmd(
-            [
-                plugin_deploy_language,
-                plugin_deploy_script_path,
-                stack_action,
-            ]
-        )
-        if result.returncode != 0:
-            logger.warning(f"\n~#~#~#~DEPLOYING OPS REPO [{deployment}] FAILED~#~#~#~")
-            if BITOPS_FAST_FAIL_MODE:
-                sys.exit(result.returncode)
-        else:
-            logger.info(f"\n~#~#~#~DEPLOYING OPS REPO [{deployment}] SUCCESSFULLY COMPLETED~#~#~#~")
+        try:
+            result = run_cmd(
+                [
+                    plugin_deploy_language,
+                    plugin_deploy_script_path,
+                    stack_action,
+                ]
+            )
+            if result.returncode != 0:
+                logger.warning(f"\n~#~#~#~DEPLOYING OPS REPO [{deployment}] FAILED~#~#~#~")
+                if BITOPS_FAST_FAIL_MODE:
+                    sys.exit(result.returncode)
+            else:
+                logger.info(
+                    f"\n~#~#~#~DEPLOYING OPS REPO [{deployment}] SUCCESSFULLY COMPLETED~#~#~#~"
+                )
+        except Exception as e:
+            logger.error(f"Error running deployment script: {e}")
+            sys.exit(101)
 
         # ~#~#~#~#~#~# STAGE 5 - AFTER HOOKS #~#~#~#~#~#~#
         if plugin_deploy_after_hook_scripts_flag:

scripts/plugins/install_plugins.py

@@ -223,7 +223,13 @@ def install_plugins():  # pylint: disable=too-many-locals,too-many-statements,to
             logger.error(f"File does not exist: [{plugin_install_script_path}]")
             sys.exit(1)
 
-        result = run_cmd([plugin_install_language, plugin_install_script_path])
+        try:
+            result = run_cmd([plugin_install_language, plugin_install_script_path])
+        except Exception as e:
+            logger.error(
+                f"Failed to run plugin install script: [{plugin_install_script_path}]. Error: [{e}]"
+            )
+            sys.exit(101)
         if result.returncode == 0:
             logger.info(f"~#~#~#~INSTALLING PLUGIN [{plugin_config}] SUCCESSFULLY COMPLETED~#~#~#~")
             logger.debug(

scripts/plugins/logging.py

@@ -63,6 +63,13 @@ def formatter_message(message, use_color=BITOPS_LOGGING_COLOR):
     return message
 
 
+def turn_off_logger():
+    """
+    Disables the logger from printing.
+    """
+    logging.getLogger("bitops-logger").disabled = True
+
+
 class BitOpsFormatter(logging.Formatter):
     """
     Class that controls the formatting of logging text, adds colors if enabled.
@@ -93,7 +100,7 @@ def format(self, record):
 )
 
 
-logger = logging.getLogger()
+logger = logging.getLogger("bitops-logger")
 logger.setLevel(BITOPS_LOGGING_LEVEL)
 
 handler = logging.StreamHandler(sys.stdout)
@@ -104,7 +111,7 @@ def format(self, record):
 logger.addHandler(handler)
 
 
-if BITOPS_LOGGING_FILENAME is not None:
+if BITOPS_LOGGING_FILENAME not in [None, "", "false"]:
     # This assumes that the user wants to save output to a filename
 
     # Create the directory if it doesn't exist

scripts/plugins/settings.py

@@ -86,6 +86,7 @@ def parse_config(dictionary, dotted_key_list):
 BITOPS_ENV_fast_fail_mode = os.environ.get("BITOPS_FAST_FAIL")
 BITOPS_ENV_run_mode = os.environ.get("BITOPS_MODE")  # TODO: CLEAN
 BITOPS_ENV_logging_level = os.environ.get("BITOPS_LOGGING_LEVEL")
+BITOPS_ENV_logging_filename = os.environ.get("BITOPS_LOGGING_FILENAME")
 BITOPS_ENV_plugin_dir = os.environ.get("BITOPS_PLUGIN_DIR")
 
 BITOPS_ENV_default_folder = os.environ.get("BITOPS_DEFAULT_FOLDER")
@@ -135,6 +136,7 @@ def parse_config(dictionary, dotted_key_list):
 )
 
 BITOPS_LOGGING_FILENAME = get_first(
+    BITOPS_ENV_logging_filename,
     parse_config(bitops_user_configuration, "bitops.logging.filename"),
     parse_config(bitops_build_configuration, "bitops.logging.filename"),
     None,

scripts/plugins/utilities.py

@@ -6,7 +6,6 @@
 
 from .settings import BITOPS_FAST_FAIL_MODE
 from .logging import logger, mask_message
-from .doc import get_doc
 
 
 def add_value_to_env(export_env, value):
@@ -46,56 +45,46 @@ def add_value_to_env(export_env, value):
         )
 
 
-def load_yaml(inc_yaml):
+def load_yaml(filename: str) -> Union[dict, None]:
     """
     This function attempts to load a YAML file from a given location,
     and exits if the file is not found. It returns the loaded YAML file if successful.
     """
     out_yaml = None
-    try:
-        with open(inc_yaml, "r", encoding="utf8") as stream:
-            out_yaml = yaml.load(stream, Loader=yaml.FullLoader)
-    except FileNotFoundError as e:
-        msg, exit_code = get_doc("missing_required_file")
-        logger.error(f"{msg} [{e.filename}]")
-        logger.debug(e)
-        sys.exit(exit_code)
+    with open(filename, "r", encoding="utf8") as stream:
+        out_yaml = yaml.load(stream, Loader=yaml.FullLoader)
+
     return out_yaml
 
 
 def run_cmd(command: Union[list, str]) -> subprocess.Popen:
     """Run a linux command and return Popen instance as a result"""
-    try:
-        with subprocess.Popen(
-            command,
-            stdout=subprocess.PIPE,
-            stderr=subprocess.STDOUT,
-            universal_newlines=True,
-        ) as process:
-            for combined_output in process.stdout:
-                # TODO: parse output for secrets
-                # TODO: specify plugin and output tight output (no extra newlines)
-                # TODO: can we modify a specific handler to add handler.terminator = "" ?
-                sys.stdout.write(mask_message(combined_output))
-
-            # This polls the async function to get information
-            # about the status of the process execution.
-            # Namely the return code which is used elsewhere.
-            process.communicate()
-
-    except Exception as exc:
-        logger.error(exc)
-        sys.exit(101)
+    with subprocess.Popen(
+        command,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        universal_newlines=True,
+    ) as process:
+        for combined_output in process.stdout:
+            # TODO: specify plugin and output tight output (no extra newlines)
+            sys.stdout.write(mask_message(combined_output))
+        # This polls the async function to get information
+        # about the status of the process execution.
+        # Namely the return code which is used elsewhere.
+        process.communicate()
+
     return process
 
 
-def handle_hooks(mode, hooks_folder, source_folder):
+def handle_hooks(mode, hooks_folder, source_folder) -> bool:
     """
     Processes a bitops before/after hook by invoking bash script(s) within the hooks folder(s).
     """
     # Checks if the folder exists, if not, move on
     if not os.path.isdir(hooks_folder):
-        return
+        return False
+    if mode not in ["before", "after"]:
+        return False
 
     original_directory = os.getcwd()
     os.chdir(source_folder)
@@ -115,7 +104,11 @@ def handle_hooks(mode, hooks_folder, source_folder):
         plugin_before_hook_script_path = hooks_folder + "/" + hook_script
         os.chmod(plugin_before_hook_script_path, 775)
 
-        result = run_cmd(["bash", plugin_before_hook_script_path])
+        try:
+            result = run_cmd(["bash", plugin_before_hook_script_path])
+        except Exception as e:
+            logger.error(f"Failed to execute before_hook script command. Error: {e}")
+            sys.exit(101)
         if result.returncode == 0:
             logger.info(f"~#~#~#~{umode} HOOK [{hook_script}] SUCCESSFULLY COMPLETED~#~#~#~")
             logger.debug(result.stdout)
@@ -126,3 +119,4 @@ def handle_hooks(mode, hooks_folder, source_folder):
                 sys.exit(result.returncode)
 
     os.chdir(original_directory)
+    return True

scripts/tests/unit/__init__.py

@@ -0,0 +1,4 @@
+import os
+
+# Disable logging to file when running tests
+os.environ["BITOPS_LOGGING_FILENAME"] = "false"

scripts/tests/unit/assets/bitops.before-deploy.d/before_test.sh

@@ -0,0 +1,6 @@
+#!/bin/bash
+
+echo "In before_test.sh"
+
+ls ../../
+

scripts/tests/unit/test_utilities.py

@@ -0,0 +1,123 @@
+import os
+import subprocess
+from io import StringIO
+from unittest import mock, TestCase
+from plugins.utilities import add_value_to_env, load_yaml, run_cmd, handle_hooks
+
+
+class TestAddValueToEnv(TestCase):
+    """Testing add_value_to_env utilties function"""
+
+    def setUp(self):
+        self.export_env = ""
+        self.value = ""
+        os.environ.clear()
+
+    def test_add_value_to_env_with_value(self):
+        """Test the add_value_to_env() function with valid value"""
+        self.export_env = "ANSIBLE_VERBOSITY"
+        self.value = "1"
+
+        add_value_to_env(self.export_env, self.value)
+        self.assertEqual(os.environ[self.export_env], self.value)
+        self.assertEqual(os.environ["BITOPS_" + self.export_env], self.value)
+
+    def test_add_value_to_env_with_none(self):
+        """Test the add_value_to_env() function with None"""
+        self.export_env = "ANSIBLE_VERBOSITY"
+        self.value = None
+
+        add_value_to_env(self.export_env, self.value)
+        self.assertNotIn(self.export_env, os.environ)
+        self.assertNotIn("BITOPS_" + self.export_env, os.environ)
+
+    def test_add_value_to_env_with_list(self):
+        """Test the add_value_to_env() function with a list"""
+        self.export_env = "ANSIBLE_VERBOSITY"
+        self.value = ["1", "2", "3"]
+
+        add_value_to_env(self.export_env, self.value)
+        self.assertEqual(os.environ[self.export_env], " ".join(self.value))
+        self.assertEqual(os.environ["BITOPS_" + self.export_env], " ".join(self.value))
+
+
+class TestLoadYAML(TestCase):
+    """Testing load_yaml utilties function"""
+
+    def setUp(self):
+        root_dir = os.getcwd()
+        self.inc_yaml = f"{root_dir}/prebuilt-config/omnibus/bitops.config.yaml"
+
+    def test_load_yaml(self):
+        """
+        Test the load_yaml function.
+        """
+        out_yaml = load_yaml(self.inc_yaml)
+        self.assertIsNotNone(out_yaml)
+        self.assertIsInstance(out_yaml, dict)
+
+    def test_load_yaml_with_invalid_filename(self):
+        """
+        Test the load_yaml function with a non-existent file.
+        """
+        with self.assertRaises(FileNotFoundError):
+            load_yaml("invalid_file.yaml")
+
+
+class TestRunCmd(TestCase):
+    """Testing run_cmd utilties function"""
+
+    @mock.patch("sys.stdout", new_callable=StringIO)
+    def test_valid_run_cmd(self, stdout):
+        """
+        Test the run_cmd function with a valid command
+        """
+        process = run_cmd("ls")
+        self.assertIsInstance(process, subprocess.Popen)
+        self.assertEqual(process.returncode, 0)
+        self.assertEqual(process.args, "ls")
+        self.assertIn("README.md", stdout.getvalue())
+        self.assertIn("LICENSE.md", stdout.getvalue())
+
+    def test_invalid_run_cmd(self):
+        """
+        Test the run_cmd function with an invalid command should throw an exception
+        """
+        with self.assertRaises(Exception) as context:
+            run_cmd("not_a_real_command")
+        self.assertIsInstance(context.exception, FileNotFoundError)
+
+
+class TestHandleHooks(TestCase):
+    """Testing handle_hooks utilties function"""
+
+    def setUp(self):
+        self.original_cwd = os.getcwd()
+        self.hooks_folder = f"{self.original_cwd}/scripts/tests/unit/assets/bitops.before-deploy.d"
+        self.source_folder = f"{self.original_cwd}/scripts/tests/unit/assets"
+
+    def tearDown(self):
+        os.chdir(self.original_cwd)
+
+    def test_handle_hooks_called_with_invalid_folder(self):
+        """
+        Test handle_hooks with invalid folder path
+        """
+        result = handle_hooks("before", "./invalid_folder", self.source_folder)
+        self.assertFalse(result)
+
+    def test_handle_hooks_called_with_invalid_mode(self):
+        """
+        Test handle_hooks with invalid mode
+        """
+        result = handle_hooks("random_mode.exe", self.hooks_folder, self.source_folder)
+        self.assertFalse(result)
+
+    @mock.patch("sys.stdout", new_callable=StringIO)
+    def test_handle_hooks_called_with_valid_folder(self, stdout):
+        """
+        Test handle_hooks with valid folder path
+        """
+        result = handle_hooks("before", self.hooks_folder, self.source_folder)
+        self.assertTrue(result)
+        self.assertIn("In before_test.sh", stdout.getvalue())