Add tests for opencas.py

Signed-off-by: Jan Musial <[email protected]>

Author: Jan Musial

.gitignore

@@ -13,3 +13,6 @@ Module.symvers
 Module.markers
 *.mod.c
 modules.order
+__pycache__/
+*.py[cod]
+*$py.class

ocf

@@ -0,0 +1,15 @@
+#
+# Copyright(c) 2012-2019 Intel Corporation
+# SPDX-License-Identifier: BSD-3-Clause-Clear
+#
+
+import sys
+import os
+
+def pytest_configure(config):
+    try:
+        import helpers
+    except:
+        raise Exception("Couldn't import helpers")
+
+    sys.path.append(helpers.find_repo_root() + "/utils")

test/utils_tests/opencas-py-tests/__init__.py

@@ -0,0 +1,114 @@
+#
+# Copyright(c) 2012-2019 Intel Corporation
+# SPDX-License-Identifier: BSD-3-Clause-Clear
+#
+
+import mock
+import re
+import os
+import sys
+import logging
+from io import StringIO
+from textwrap import dedent
+
+def find_repo_root():
+    path = os.getcwd()
+
+    while os.path.realpath(path) != "/":
+        if ".git" in os.listdir(path):
+            return path
+
+        path = os.path.dirname(path)
+
+    raise Exception("Couldn't find repository root - unable to locate opencas.py")
+
+def get_process_mock(return_value, stdout, stderr):
+    process_mock = mock.Mock()
+    attrs = {
+        "wait.return_value" : return_value,
+        "communicate.return_value" : (stdout, stderr),
+    }
+    process_mock.configure_mock(**attrs)
+
+    return process_mock
+
+def get_mock_os_exists(existing_files):
+    return lambda x : x in existing_files
+
+def get_hashed_config_list(conf):
+    """ Convert list of config lines to list of config lines hashes, drop empty lines """
+    hashed_conf = [get_conf_line_hash(x) for x in conf]
+
+    return [x for x in hashed_conf if x]
+
+def get_conf_line_hash(line):
+    """
+    Removes whitespace, lowercases, comments and sorts cache params if present.
+    Returns empty line for comment-only lines
+
+    We don't care about order of params and kinds of whitespace in config lines
+    so normalize it to compare. We do care about case in paths, but to simplify
+    testing we pretend we don't.
+    """
+
+    def sort_cache_params(params):
+        return ",".join(sorted(params.split(",")))
+
+    line = line.split("#")[0]
+
+    cache_params_pattern = re.compile(r"(.*?\s)(\S+=\S+)")
+    match = cache_params_pattern.search(line)
+    if match:
+        sorted_params = sort_cache_params(match.group(2))
+        line = match.group(1) + sorted_params
+
+    return "".join(line.lower().split())
+
+
+class MockConfigFile(object):
+    def __init__(self, buffer=""):
+        self.set_contents(buffer)
+
+    def __enter__(self):
+        return self.buffer
+
+    def __exit__(self, *args, **kwargs):
+        self.set_contents(self.buffer.getvalue())
+
+    def __call__(self, path, mode):
+        if (mode == "w"):
+            self.buffer = StringIO()
+
+        return self
+
+    def read(self):
+        return self.buffer.read()
+
+    def write(self, str):
+        return self.buffer.write(str)
+
+    def close(self):
+        self.set_contents(self.buffer.getvalue())
+
+    def readline(self):
+        return self.buffer.readline()
+
+    def __next__(self):
+        return self.buffer.__next__()
+
+    def __iter__(self):
+        return self
+
+    def set_contents(self, buffer):
+        self.buffer = StringIO(dedent(buffer).strip())
+
+class CopyableMock(mock.Mock):
+    def __init__(self, *args, **kwargs):
+        super(CopyableMock, self).__init__(*args, **kwargs)
+        self.copies = []
+
+    def __deepcopy__(self, memo):
+        copy = mock.Mock(spec=self)
+        self.copies += [copy]
+        return copy
+