Adding unittests to ocp-diag-autoval

src/autoval/unittest/mock/lib/mock_autoval.py

@@ -0,0 +1,99 @@
+# pyre-unsafe
+
+import csv
+import logging
+import unittest.mock as mock
+
+from autoval.lib.transport.ssh import SSHConn
+from autoval.lib.utils.autoval_exceptions import CmdError
+from autoval.lib.utils.autoval_log import AutovalLog
+from autoval.lib.utils.autoval_utils import AutovalUtils, CmdResult
+from autoval.lib.utils.file_actions import FileActions
+from autoval.unittest.mock.lib.mock_host import MockHost
+
+MOCK_INPUT_PATH = "autoval/unittest/mock/util_outputs/"
+MOCK_COMMAND_MAP_PATH = "autoval/unittest/mock/testbed/cmd_map"
+
+
+class MockAutovalUtils(MockHost):
+    def __init__(self, cmd_map=None):
+        self.autovalLog = AutovalLog._init_logs()
+        self.logger = logging.getLogger("cmdlog")
+        self.cmd_output_dict = None
+        self.get_result_obj_rc = 0
+        if cmd_map:
+            self.cmd_map = cmd_map
+        else:
+            self.cmd_map = self.generate_cmp_map()
+        super(MockAutovalUtils, self).__init__(self.cmd_map)
+
+    @staticmethod
+    def generate_cmp_map():
+        """This function will convert the cmd_map file into the list of
+        dict with in format [{"cmd"="cmd","file"="file_path"},..]"""
+        try:
+            file_path = FileActions.get_resource_file_path(MOCK_COMMAND_MAP_PATH[14:])
+            with open(file_path, "r") as file_context:
+                cmd_map_reader = csv.reader(
+                    file_context, delimiter=":", quoting=csv.QUOTE_ALL
+                )
+                """in case cmd has the delimiter part of it, csv reader
+                 will consider the last element as "file" and will join
+                 the rest of elements to command"""
+                cmd_map = [
+                    {
+                        "cmd": ":".join(each_cmd_map[0:-1]).strip(),
+                        "file": each_cmd_map[-1].strip(),
+                    }
+                    for each_cmd_map in cmd_map_reader
+                ]
+                return cmd_map
+        except Exception:
+            raise Exception(
+                f"Failed to generate the cmd_map from file {MOCK_COMMAND_MAP_PATH}"
+            )
+
+    def run(self, *params, **kparams):
+        """Function is side effect of mocking run method and
+        will be give a mock output based on the cmd_map
+        *params will contain values of cmd from run method
+        **kparams will contain the key argument values of get_result_obj,
+        ignore_status,custom_logfile cmd_output_dict is used
+        in case cmd_map is not to be referred  which would be and optimised way
+        in case we have single line output instead of creating file
+        get_result_obj_rc return code of command run by default set to 0
+        """
+        data = None
+        cmd = params[0]
+        get_result_obj = kparams.get("get_result_obj")
+        ignore_status = kparams.get("ignore_status")
+        if self.cmd_output_dict and cmd in self.cmd_output_dict:
+            data = self.cmd_output_dict[cmd]
+            if isinstance(data, Exception):
+                raise data
+        else:
+            if get_result_obj:
+                data = self.run_get_result(cmd, ignore_status)
+            else:
+                data = super(MockAutovalUtils, self).run(cmd, ignore_status)
+        if get_result_obj:
+            data = CmdResult(cmd, data, "", self.get_result_obj_rc)
+            if self.get_result_obj_rc and not ignore_status:
+                raise CmdError(cmd, data, "command failed")
+        return data
+
+    def get_mock_data(self, funct, *args, **kwargs):
+        """Function will mock the methods which should run on Dut
+        such as run"""
+        self.cmd_output_dict = kwargs.pop("cmd_output_dict", None)
+        self.get_result_obj_rc = kwargs.pop("get_result_obj_rc", 0)
+        with mock.patch.object(
+            SSHConn, "scp_file", return_value="pass"
+        ), mock.patch.object(SSHConn, "run", side_effect=self.run), mock.patch.object(
+            AutovalUtils, "run_get_output", side_effect=self.run
+        ), mock.patch.object(
+            AutovalLog,
+            "log_info",
+            side_effect=self.logger.info,
+        ):
+            return funct(*args, **kwargs)

src/autoval/unittest/mock/lib/mock_connection_dispatcher.py

@@ -0,0 +1,25 @@
+# pyre-unsafe
+
+from autoval.lib.transport.ssh import SSHConn
+
+MOCK_HOSTS = {
+    "hostname": "using.fake.host",
+    "ipv6": "abcd:db00:0012:700e:face:0000:0023:0000",
+    "oob_addr": "using-oob.fake.host",
+    "rack_sub_position_slot": 1,
+    "is_container": False,
+}
+
+
+class MockConnectionDispatcher:
+    def __init__(self):
+        self.oob_only = None
+        self.host_connection = SSHConn(None)
+        self.bmc_connections = [SSHConn(None)]
+        self._bmc_connections = [SSHConn(None)]
+        self.oob_addr = MOCK_HOSTS.get("oob_addr")
+        self.rack_sub_position = MOCK_HOSTS.get("rack_sub_position")
+        self.rack_sub_position_slot = MOCK_HOSTS.get("rack_sub_position_slot")
+        self.hostname = MOCK_HOSTS.get("hostname")
+        self.localhost = None
+        self.host_dict = MOCK_HOSTS

src/autoval/unittest/mock/lib/mock_host.py

@@ -0,0 +1,231 @@
+# pyre-unsafe
+
+import os
+import unittest.mock as mock  # noqa
+from typing import List, Optional
+
+from autoval.lib.connection.connection_abstract import ConnectionAbstract
+from autoval.lib.connection.connection_utils import CmdResult
+from autoval.lib.host.bmc import BMC
+from autoval.lib.host.system import System
+from autoval.lib.utils.autoval_exceptions import CmdError
+from autoval.lib.utils.file_actions import FileActions
+from autoval.unittest.mock.lib.mock_connection_dispatcher import (
+    MOCK_HOSTS,
+    MockConnectionDispatcher,
+)
+from autoval.unittest.mock.lib.mock_openbmc import MockOpenBMC
+
+
+MOCK_INPUT_PATH = "autoval/unittest/mock/util_outputs/"
+MOCK_INPUT_RELATIVE_PATH = "autoval/unittest/mock/util_outputs/"
+
+
+class MockHost(ConnectionAbstract):
+    def __init__(self, cmd_map, run_return_plain_stdout=False):
+        self.run_return_plain_stdout = run_return_plain_stdout
+        self.cmd_map = cmd_map
+        self.bmc_type = "Openbmc"
+        self.connection_obj = MockConnectionDispatcher()
+        self.hostname = self.connection_obj.hostname
+        self.connection = self.connection_obj.host_connection
+        self.localhost = self
+        self.oob_addr = self.connection_obj.oob_addr
+        self.oob_only = self.connection_obj.oob_only
+        self.rack_sub_position = self.connection_obj.rack_sub_position
+        self.rack_sub_position_slot = self.connection_obj.rack_sub_position_slot
+        self.product_obj = self._get_product_obj()
+        self.host.product_obj.product_name = "TestPlatform V2"
+        self._openbmc_obj = MockOpenBMC(self, self.cmd_map)._get_openbmc()
+        self.host_dict = MOCK_HOSTS
+        self.oobs = [self.oob]
+        # storage for iterator over run() results
+        self._iter_result = {}
+
+    @property
+    def openbmc_obj(self):
+        return self._openbmc_obj
+
+    @openbmc_obj.setter
+    def openbmc_obj(self, cls_obj):
+        if isinstance(cls_obj, BMC):
+            self._openbmc_obj = cls_obj
+        else:
+            raise Exception("cls_obj is not an openBMC obj")
+
+    # override run just like for sshpass
+    def run(
+        self,
+        cmd: str,
+        ignore_status: bool = False,
+        timeout: int = 600,
+        working_directory: Optional[str] = None,
+        custom_logfile: Optional[str] = None,
+        get_pty: bool = False,
+        sudo: bool = False,
+        sudo_options: Optional[List[str]] = None,
+        connection_timeout: int = 60,
+        background: bool = False,
+        keepalive: int = 0,
+        forward_ssh_agent=False,
+        path_env=None,
+    ) -> str:
+        if self.run_return_plain_stdout:
+            return self.run_get_result(
+                cmd,
+                ignore_status,
+                timeout,
+                working_directory,
+                custom_logfile,
+                get_pty,
+                sudo,
+                sudo_options,
+                connection_timeout,
+                background,
+                keepalive,
+                forward_ssh_agent,
+                path_env,
+            ).stdout
+
+        return super(MockHost, self).run(
+            cmd,
+            ignore_status,
+            timeout,
+            working_directory,
+            custom_logfile,
+            get_pty,
+            sudo,
+            sudo_options,
+            connection_timeout,
+            background,
+            keepalive,
+            forward_ssh_agent,
+            path_env,
+        )
+
+    def run_get_result(
+        self,
+        cmd: str,
+        ignore_status: bool = False,
+        timeout: int = 600,
+        working_directory: Optional[str] = None,
+        custom_logfile: Optional[str] = None,
+        get_pty: bool = False,
+        sudo: bool = False,
+        sudo_options: Optional[List[str]] = None,
+        connection_timeout: int = 60,
+        background: bool = False,
+        keepalive: int = 0,
+        forward_ssh_agent=False,
+        path_env=None,
+    ) -> CmdResult:
+        # whether to iterate over multiple result values
+        iterate = False
+        data = None
+        return_code = 0
+        for c in self.cmd_map:
+            if cmd == c["cmd"]:
+                if "return_code" in c:
+                    return_code = c["return_code"]
+                if "result" in c:
+                    data = c["result"]
+                else:
+                    data = self._read_file(c["file"])
+                if "iterate" in c and c["iterate"]:
+                    iterate = True
+                break
+        if iterate and type(data) in (list, tuple):
+            if cmd not in self._iter_result:
+                self._iter_result[cmd] = iter(data)
+        if isinstance(data, CmdError):
+            raise data
+        if not data:
+            data = ""
+        if cmd in self._iter_result:
+            try:
+                return next(self._iter_result[cmd])
+            except StopIteration:
+                return None  # pyre-fixme[7]
+        return CmdResult(cmd, str(data), "", return_code, 1)
+
+    @classmethod
+    def read_file(cls, file_path, **kwargs):
+        return cls._read_file(file_path, **kwargs)
+
+    def get_file(self, file_path, target, **kwargs):
+        pass
+
+    def put_file(self, file_path, target, **kwargs):
+        pass
+
+    def scp_file(
+        self, source_location: str, file_tocopy: str, destination_location: str
+    ) -> str:
+        pass
+
+    def get_host_arch(self) -> str:
+        pass
+
+    def _connect(self):
+        pass
+
+    def _get_product_obj(self):
+        system = System(self)
+        return system
+
+    @staticmethod
+    def _read_file(_file, json_file=False):
+        _file = os.path.join(MOCK_INPUT_RELATIVE_PATH, _file)
+        # file_path = FileActions.get_resource_file_path(_file[14:])
+        return FileActions.read_data(path=_file, json_file=json_file)
+
+    def update_cmd_map(
+        self,
+        cmd: str,
+        mock_output: str,
+        is_file: bool = False,
+        return_code: str = "0",
+    ) -> None:
+        """This method will update the command map with the cmd values
+        in case if the command is already present it would update the file
+        or result value of the command."""
+        for each_cmd_map in self.cmd_map:
+            if each_cmd_map["cmd"] == cmd:
+                each_cmd_map["return_code"] = return_code
+                if is_file:
+                    each_cmd_map["file"] = mock_output
+                else:
+                    each_cmd_map["result"] = mock_output
+                return
+        cmd_map_dict = {"cmd": cmd}
+        cmd_map_dict["return_code"] = return_code
+        if is_file:
+            cmd_map_dict["file"] = mock_output
+        else:
+            cmd_map_dict["result"] = mock_output
+        self.cmd_map.append(cmd_map_dict)
+
+    def add_test_start_msg(self, test_name):
+
+        pass
+
+    def add_test_end_msg(self, test_name):
+        pass
+
+    def deploy_tool(self, tool) -> str:
+        return ""
+
+    def __getattr__(self, attr):
+        if attr == "oob":
+            _oob = MockOpenBMC(self, self.openbmc_obj)._get_openbmc()
+            setattr(self, "oob", _oob)  # noqa
+            return _oob
+        return getattr(self.product_obj, attr)
+
+    def revert_cmd_map(self, cmd_map: List):
+        for each_cmd_map in cmd_map:
+            cmd = each_cmd_map["cmd"]
+            if "result" in each_cmd_map:
+                self.update_cmd_map(cmd, each_cmd_map["result"], is_file=False)
+            else:
+                self.update_cmd_map(cmd, each_cmd_map["file"], is_file=True)

src/autoval/unittest/mock/lib/mock_openbmc.py

@@ -0,0 +1,39 @@
+# pyre-unsafe
+
+import os
+import unittest.mock as mock
+
+from autoval.lib.host.bmc import BMC
+from autoval.lib.utils.file_actions import FileActions
+
+
+MOCK_INPUT_PATH = "autoval/unittest/mock/util_outputs/"
+
+
+class MockOpenBMC:
+    def __init__(self, host, cmd_map):
+        self.host = host
+        self.cmd_map = cmd_map
+        self.bmc_host = host
+
+    def run(self, cmd, ignore_status=True):
+        data = None
+        for c in self.cmd_map:
+            if cmd == c["cmd"]:
+                data = self._read_file(c["file"])
+                break
+        return data
+
+    def _get_openbmc(self):
+        with mock.patch.object(BMC, "__init__", lambda a, b, c, d: None):
+            openbmc = BMC(None, None, None)
+            openbmc.slot_info = "slot4"
+            openbmc.host = self.host
+            openbmc.bmc_host = self.host
+            openbmc.config_filter = {}
+            return openbmc
+
+    def _read_file(self, _file):
+        _file = os.path.join(MOCK_INPUT_PATH, _file)
+        file_path = FileActions.get_resource_file_path(_file)
+        return FileActions.read_data(file_path)