Add tests for opencas.py

.gitignore

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

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

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

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

@@ -0,0 +1,122 @@
+#
+# Copyright(c) 2012-2019 Intel Corporation
+# SPDX-License-Identifier: BSD-3-Clause-Clear
+#
+
+import mock
+import re
+import os
+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

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

@@ -0,0 +1,156 @@
+#
+# Copyright(c) 2012-2019 Intel Corporation
+# SPDX-License-Identifier: BSD-3-Clause-Clear
+#
+
+import pytest
+import mock
+import stat
+
+import helpers as h
+import opencas
+
+
+@pytest.mark.parametrize(
+    "line",
+    [
+        "",
+        " ",
+        "#",
+        "                      #                      ",
+        ("TG9yZW0gaXBzdW0gZG9sb3Igc2l0IGFtZXQsIGNvbnNlY3RldHVyIGFkaXBpc2Npbmcg"
+            "ZWxpdCwgc2VkIGRvIGVpdXNtb2QgdGVtcG9yIGluY2lkaWR1bnQgdXQgbGFib3JlI"
+            "GV0IGRvbG9yZSBtYWduYSBhbGlxdWEu"),
+        " # ? } { ! ",
+        "1 1 /dev/sda /dev/sdb",
+        "1 2    1 /dev/sda ",
+    ],
+)
+@mock.patch("opencas.cas_config.core_config.validate_config")
+def test_core_config_from_line_parsing_checks_01(mock_validate, line):
+    with pytest.raises(ValueError):
+        opencas.cas_config.core_config.from_line(line)
+
+
+@pytest.mark.parametrize("line", ["1 1 /dev/sda", "1     1 /dev/sda "])
+@mock.patch("opencas.cas_config.core_config.validate_config")
+def test_core_config_from_line_parsing_checks_02(mock_validate, line):
+    opencas.cas_config.core_config.from_line(line)
+
+
+@mock.patch("os.path.exists")
+@mock.patch("os.stat")
+def test_core_config_from_line_device_is_directory(
+    mock_stat, mock_path_exists
+):
+    mock_path_exists.side_effect = h.get_mock_os_exists(["/home/user/stuff"])
+    mock_stat.return_value = mock.Mock(st_mode=stat.S_IFDIR)
+
+    with pytest.raises(ValueError):
+        opencas.cas_config.core_config.from_line("1    1   /home/user/stuff")
+
+
+@mock.patch("os.path.exists")
+@mock.patch("os.stat")
+def test_core_config_from_line_device_not_present(mock_stat, mock_path_exists):
+    mock_path_exists.side_effect = h.get_mock_os_exists([])
+    mock_stat.side_effect = ValueError()
+
+    with pytest.raises(ValueError):
+        opencas.cas_config.core_config.from_line("1    1   /dev/sda")
+
+
+def test_core_config_from_line_recursive_multilevel():
+    with pytest.raises(ValueError):
+        opencas.cas_config.core_config.from_line("1    1   /dev/cas1-1")
+
+
+def test_core_config_from_line_multilevel():
+    opencas.cas_config.core_config.from_line("1    1   /dev/cas2-1")
+
+
+@mock.patch("opencas.cas_config.check_block_device")
+def test_core_config_from_line_allow_incomplete(mock_check_block,):
+    opencas.cas_config.core_config.from_line(
+        "1    1   /dev/sda", allow_incomplete=True
+    )
+
+    assert not mock_check_block.called
+
+
+@pytest.mark.parametrize(
+    "cache_id,core_id",
+    [
+        ("AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA", "bbbbbbb"),
+        ("lizard", "chicken"),
+        ("0", "0"),
+        ("0", "100"),
+        ("0", "-1"),
+        ("-1", "0"),
+        ("-1", "1"),
+        ("-1", "-1"),
+        ("16385", "4095"),
+        ("16384", "4096"),
+        ("0", "0"),
+        ("1", "-1"),
+    ],
+)
+@mock.patch("os.path.exists")
+@mock.patch("os.stat")
+def test_core_config_from_line_cache_id_validation_01(
+    mock_stat, mock_path_exists, cache_id, core_id
+):
+    mock_path_exists.side_effect = h.get_mock_os_exists(["/dev/sda"])
+    mock_stat.return_value = mock.Mock(st_mode=stat.S_IFBLK)
+
+    line = "{0}   {1}   /dev/sda".format(cache_id, core_id)
+
+    with pytest.raises(ValueError):
+        opencas.cas_config.core_config.from_line(line)
+
+
+@pytest.mark.parametrize(
+    "cache_id,core_id", [("16384", "4095"), ("1", "0"), ("1", "10")]
+)
+@mock.patch("os.path.exists")
+@mock.patch("os.stat")
+def test_core_config_from_line_cache_id_validation_02(
+    mock_stat, mock_path_exists, cache_id, core_id
+):
+    mock_path_exists.side_effect = h.get_mock_os_exists(["/dev/sda"])
+    mock_stat.return_value = mock.Mock(st_mode=stat.S_IFBLK)
+
+    line = "{0}   {1}   /dev/sda".format(cache_id, core_id)
+
+    opencas.cas_config.core_config.from_line(line)
+
+
+@pytest.mark.parametrize(
+    "cache_id,core_id,device",
+    [
+        ("1", "1", "/dev/sda"),
+        ("16384", "4095", "/dev/sda1"),
+        ("16384", "0", "/dev/nvme0n1p"),
+        ("100", "5", "/dev/dm-10"),
+    ],
+)
+@mock.patch("os.path.exists")
+@mock.patch("os.stat")
+def test_core_config_from_line_cache_id_validation(
+    mock_stat, mock_path_exists, cache_id, core_id, device
+):
+    mock_path_exists.side_effect = h.get_mock_os_exists([device])
+    mock_stat.return_value = mock.Mock(st_mode=stat.S_IFBLK)
+
+    core_reference = opencas.cas_config.core_config(
+        cache_id=cache_id, core_id=core_id, path=device
+    )
+
+    core_reference.validate_config()
+
+    core_after = opencas.cas_config.core_config.from_line(
+        core_reference.to_line()
+    )
+    assert core_after.cache_id == core_reference.cache_id
+    assert core_after.core_id == core_reference.core_id
+    assert core_after.device == core_reference.device

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

@@ -0,0 +1,53 @@
+#
+# Copyright(c) 2012-2019 Intel Corporation
+# SPDX-License-Identifier: BSD-3-Clause-Clear
+#
+
+import pytest
+import subprocess
+import mock
+
+from opencas import casadm
+from helpers import get_process_mock
+
+
+@mock.patch("subprocess.Popen")
+def test_run_cmd_01(mock_popen):
+    mock_popen.return_value = get_process_mock(0, "successes", "errors")
+    result = casadm.run_cmd(["casadm", "-L"])
+
+    assert result.exit_code == 0
+    assert result.stdout == "successes"
+    assert result.stderr == "errors"
+    mock_popen.assert_called_once_with(
+        ["casadm", "-L"], stdout=subprocess.PIPE, stderr=subprocess.PIPE
+    )
+
+
+@mock.patch("subprocess.Popen")
+def test_run_cmd_02(mock_popen):
+    mock_popen.return_value = get_process_mock(4, "successes", "errors")
+    with pytest.raises(casadm.CasadmError):
+        casadm.run_cmd(["casadm", "-L"])
+
+
+@mock.patch("subprocess.Popen")
+def test_get_version_01(mock_popen):
+    mock_popen.return_value = get_process_mock(0, "0.0.1", "errors")
+    result = casadm.get_version()
+
+    assert result.exit_code == 0
+    assert result.stdout == "0.0.1"
+    assert result.stderr == "errors"
+    mock_popen.assert_called_once_with(
+        [casadm.casadm_path, "--version", "--output-format", "csv"],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+    )
+
+
+@mock.patch("subprocess.Popen")
+def test_get_version_02(mock_popen):
+    mock_popen.return_value = get_process_mock(4, "successes", "errors")
+    with pytest.raises(casadm.CasadmError):
+        casadm.get_version()

utils/opencas.py

@@ -1,4 +1,3 @@
-#!/usr/bin/env python2
 #
 # Copyright(c) 2012-2019 Intel Corporation
 # SPDX-License-Identifier: BSD-3-Clause-Clear
@@ -223,7 +222,7 @@ def validate_config(self, force, allow_incomplete=False):
             type(self).check_cache_id_valid(self.cache_id)
             self.check_recursive()
             self.check_cache_mode_valid(self.cache_mode)
-            for param_name, param_value in self.params.iteritems():
+            for param_name, param_value in self.params.items():
                 self.validate_parameter(param_name, param_value)
 
             if not allow_incomplete:
@@ -289,7 +288,7 @@ def to_line(self):
             ret = '{0}\t{1}\t{2}'.format(self.cache_id, self.device, self.cache_mode)
             if len(self.params) > 0:
                 i = 0
-                for param, value in self.params.iteritems():
+                for param, value in self.params.items():
                     if i > 0:
                         ret += ','
                     else:
@@ -408,14 +407,14 @@ def insert_cache(self, new_cache_config):
                 raise cas_config.AlreadyConfiguredException(
                                 'Cache already configured')
 
-        for cache_id, cache in self.caches.iteritems():
+        for cache_id, cache in self.caches.items():
             if cache_id != new_cache_config.cache_id:
                 if (os.path.realpath(new_cache_config.device)
                         == os.path.realpath(cache.device)):
                     raise cas_config.ConflictingConfigException(
                             'This cache device is already configured as a cache')
 
-            for _, core in cache.cores.iteritems():
+            for _, core in cache.cores.items():
                 if (os.path.realpath(core.device)
                         == os.path.realpath(new_cache_config.device)):
                     raise cas_config.ConflictingConfigException(
@@ -433,13 +432,13 @@ def insert_core(self, new_core_config):
             raise KeyError('Cache id {0} doesn\'t exist'.format(new_core_config.cache_id))
 
         try:
-            for cache_id, cache in self.caches.iteritems():
+            for cache_id, cache in self.caches.items():
                 if (os.path.realpath(cache.device)
                         == os.path.realpath(new_core_config.device)):
                     raise cas_config.ConflictingConfigException(
                             'Core device already configured as a cache')
 
-                for core_id, core in cache.cores.iteritems():
+                for core_id, core in cache.cores.items():
                     if (cache_id == new_core_config.cache_id
                             and core_id == new_core_config.core_id):
                         if (os.path.realpath(core.device)
@@ -478,7 +477,7 @@ def write(self, config_file):
                 conf.write('# This config was automatically generated\n')
 
                 conf.write('[caches]\n')
-                for _, cache in self.caches.iteritems():
+                for _, cache in self.caches.items():
                     conf.write(cache.to_line())
 
                 conf.write('\n[cores]\n')