github/open-cas-linux
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #110 from imjfckm/update-utils-tests

Browse Source 
Update utils tests
This commit is contained in:
Adam Rutkowski 2019-09-20 17:59:55 +02:00 committed by GitHub
commit 99cbafaac6
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 101 additions and 54 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
test/utils_tests/opencas-py-tests/helpers.py
View File

@ -27,8 +27,9 @@ def find_repo_root():
def get_process_mock(return_value, stdout, stderr):
process_mock = mock.Mock()
attrs = {
"wait.return_value": return_value,
"communicate.return_value": (stdout, stderr),
"returncode": return_value,
"stdout": stdout,
"stderr": stderr
}
process_mock.configure_mock(**attrs)

112
test/utils_tests/opencas-py-tests/test_cas_config_cache_01.py
View File

@ -18,9 +18,11 @@ import opencas
" ",
"#",
" # ",
("TG9yZW0gaXBzdW0gZG9sb3Igc2l0IGFtZXQsIGNvbnNlY3RldHVyIGFkaXBpc2Npbmcg"
(
"TG9yZW0gaXBzdW0gZG9sb3Igc2l0IGFtZXQsIGNvbnNlY3RldHVyIGFkaXBpc2Npbmcg"
"ZWxpdCwgc2VkIGRvIGVpdXNtb2QgdGVtcG9yIGluY2lkaWR1bnQgdXQgbGFib3JlI"
"GV0IGRvbG9yZSBtYWduYSBhbGlxdWEu"),
"GV0IGRvbG9yZSBtYWduYSBhbGlxdWEu"
),
" # ? } { ! ",
"1 /dev/nvme0n1 WT 1 2 3",
"1 /dev/nvme0n1 WT ioclass_file=ioclass.csv ,cache_line_size=4",
@ -46,15 +48,13 @@ def test_cache_config_from_line_parsing_checks_02(mock_validate, line):
@mock.patch("os.path.exists")
@mock.patch("os.stat")
def test_cache_config_from_line_device_is_directory(
mock_stat, mock_path_exists
):
def test_cache_config_from_line_device_is_directory(mock_stat, mock_path_exists):
mock_path_exists.side_effect = h.get_mock_os_exists(
["/home/user/catpictures"]
)
mock_stat.return_value = mock.Mock(st_mode=stat.S_IFDIR)
with pytest.raises(ValueError):
with pytest.raises(ValueError, match="is not block device"):
opencas.cas_config.cache_config.from_line(
"1 /home/user/catpictures WT"
)
@ -62,75 +62,72 @@ def test_cache_config_from_line_device_is_directory(
@mock.patch("os.path.exists")
@mock.patch("os.stat")
def test_cache_config_from_line_device_not_present(
mock_stat, mock_path_exists
):
def test_cache_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 = OSError()
with pytest.raises(ValueError):
with pytest.raises(ValueError, match="not found"):
opencas.cas_config.cache_config.from_line("1 /dev/nvme0n1 WT")
@mock.patch("os.path.exists")
@mock.patch("os.stat")
@mock.patch("subprocess.Popen")
@mock.patch("subprocess.run")
def test_cache_config_from_line_device_with_partitions(
mock_popen, mock_stat, mock_path_exists
mock_run, mock_stat, mock_path_exists
):
mock_path_exists.side_effect = h.get_mock_os_exists(["/dev/sda"])
mock_stat.return_value = mock.Mock(st_mode=stat.S_IFBLK)
mock_popen.return_value = h.get_process_mock(0, "sda\nsda1\nsda2", "")
mock_run.return_value = h.get_process_mock(0, "sda\nsda1\nsda2", "")
with pytest.raises(ValueError):
with pytest.raises(ValueError, match="Partitions"):
opencas.cas_config.cache_config.from_line("1 /dev/sda WT")
@mock.patch("os.path.exists")
@mock.patch("os.stat")
@mock.patch("subprocess.Popen")
@mock.patch("subprocess.run")
def test_cache_config_validate_device_with_partitions(
mock_popen, mock_stat, mock_path_exists
mock_run, mock_stat, mock_path_exists
):
mock_path_exists.side_effect = h.get_mock_os_exists(["/dev/sda"])
mock_stat.return_value = mock.Mock(st_mode=stat.S_IFBLK)
mock_popen.return_value = h.get_process_mock(0, "sda\nsda1\nsda2", "")
mock_run.return_value = h.get_process_mock(0, "sda\nsda1\nsda2", "")
cache = opencas.cas_config.cache_config(
cache_id="1", device="/dev/sda", cache_mode="WT", params=dict()
cache_id="1", device="/dev/sda", cache_mode="WT"
)
with pytest.raises(ValueError):
with pytest.raises(ValueError, match="Partitions"):
cache.validate_config(False)
@mock.patch("os.path.exists")
@mock.patch("os.stat")
@mock.patch("subprocess.Popen")
@mock.patch("subprocess.run")
def test_cache_config_validate_force_device_with_partitions(
mock_popen, mock_stat, mock_path_exists
mock_run, mock_stat, mock_path_exists
):
mock_path_exists.side_effect = h.get_mock_os_exists(["/dev/sda"])
mock_stat.return_value = mock.Mock(st_mode=stat.S_IFBLK)
mock_popen.return_value = h.get_process_mock(0, "sda\nsda1\nsda2", "")
mock_run.return_value = h.get_process_mock(0, "sda\nsda1\nsda2", "")
cache = opencas.cas_config.cache_config(
cache_id="1", device="/dev/sda", cache_mode="WT", params=dict()
cache_id="1", device="/dev/sda", cache_mode="WT"
)
with pytest.raises(ValueError):
cache.validate_config(True)
cache.validate_config(True)
@mock.patch("os.path.exists")
@mock.patch("os.stat")
@mock.patch("subprocess.Popen")
@mock.patch("subprocess.run")
def test_cache_config_from_line_device_without_partitions(
mock_popen, mock_stat, mock_path_exists
mock_run, mock_stat, mock_path_exists
):
mock_path_exists.side_effect = h.get_mock_os_exists(["/dev/sda"])
mock_stat.return_value = mock.Mock(st_mode=stat.S_IFBLK)
mock_popen.return_value = h.get_process_mock(0, "sda\n", "")
mock_run.return_value = h.get_process_mock(0, "sda\n", "")
opencas.cas_config.cache_config.from_line("1 /dev/sda WT")
@ -145,9 +142,7 @@ def test_cache_config_from_line_recursive_multilevel(
mock_stat.raises = OSError()
with pytest.raises(ValueError):
opencas.cas_config.cache_config.from_line(
"1 {0} WT".format(device)
)
opencas.cas_config.cache_config.from_line("1 {0} WT".format(device))
@mock.patch("os.path.exists")
@ -177,8 +172,10 @@ def test_cache_config_from_line_missing_ioclass_file(
with pytest.raises(ValueError):
opencas.cas_config.cache_config.from_line(
("11 /dev/nvme0n1 WT ioclass_file=ioclass.csv,"
"cleaning_policy=nop,cache_line_size=4")
(
"11 /dev/nvme0n1 WT ioclass_file=ioclass.csv,"
"cleaning_policy=nop,cache_line_size=4"
)
)
@ -201,6 +198,12 @@ def test_cache_config_from_line_missing_ioclass_file(
"cache_line_size=-1",
"cache_line_size=four",
"cache_line_size=128",
"promotion_policy=111111",
"promotion_policy=",
"promotion_policy=dinosaurs",
"promotion_policy=Robert'); DROP TABLE Students;--",
"promotion_policy=awlays",
"promotion_policy=nnhit",
],
)
@mock.patch("os.path.exists")
@ -215,7 +218,7 @@ def test_cache_config_from_line_parameter_validation_01(
line = "1 /dev/sda WT {0}".format(params)
with pytest.raises(ValueError):
with pytest.raises(ValueError, match="[Ii]nvalid"):
opencas.cas_config.cache_config.from_line(line)
@ -235,6 +238,9 @@ def test_cache_config_from_line_parameter_validation_01(
"cache_line_size=64",
"cache_line_size=4,cleaning_policy=nop",
"ioclass_file=ioclass.csv,cache_line_size=4,cleaning_policy=nop",
"promotion_policy=nhit",
"promotion_policy=always",
"ioclass_file=ioclass.csv,cache_line_size=4,cleaning_policy=nop,promotion_policy=always",
],
)
@mock.patch("os.path.exists")
@ -279,7 +285,23 @@ def test_cache_config_from_line_cache_mode_validation_01(
@pytest.mark.parametrize(
"mode", ["wt", "WT", "pt", "PT", "wb", "WB", "wa", "WA", "wA", "Wa"]
"mode",
[
"wt",
"WT",
"pt",
"PT",
"wb",
"WB",
"wa",
"WA",
"wA",
"Wa",
"wo",
"WO",
"wO",
"Wo",
],
)
@mock.patch("os.path.exists")
@mock.patch("opencas.cas_config.cache_config.check_cache_device_empty")
@ -377,6 +399,26 @@ def test_cache_config_from_line_cache_id_validation_02(
"cache_mode": "WT",
"cache_line_size": "4",
},
{
"cache_id": "1",
"device": "/dev/nvme0n1",
"cache_mode": "wo",
"cache_line_size": "16",
},
{
"cache_id": "1",
"device": "/dev/nvme0n1",
"cache_mode": "wo",
"promotion_policy": "always",
"cache_line_size": "16",
},
{
"cache_id": "1",
"device": "/dev/nvme0n1",
"cache_mode": "wo",
"promotion_policy": "nhit",
"cache_line_size": "16",
},
],
)
@mock.patch("os.path.exists")

34
test/utils_tests/opencas-py-tests/test_casadm_01.py
View File

@ -11,43 +11,47 @@ 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")
@mock.patch("subprocess.run")
def test_run_cmd_01(mock_run):
mock_run.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_run.assert_called_once_with(
["casadm", "-L"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
universal_newlines=mock.ANY,
)
@mock.patch("subprocess.Popen")
def test_run_cmd_02(mock_popen):
mock_popen.return_value = get_process_mock(4, "successes", "errors")
@mock.patch("subprocess.run")
def test_run_cmd_02(mock_run):
mock_run.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")
@mock.patch("subprocess.run")
def test_get_version_01(mock_run):
mock_run.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(
mock_run.assert_called_once_with(
[casadm.casadm_path, "--version", "--output-format", "csv"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
universal_newlines=mock.ANY,
)
@mock.patch("subprocess.Popen")
def test_get_version_02(mock_popen):
mock_popen.return_value = get_process_mock(4, "successes", "errors")
@mock.patch("subprocess.run")
def test_get_version_02(mock_run):
mock_run.return_value = get_process_mock(4, "successes", "errors")
with pytest.raises(casadm.CasadmError):
casadm.get_version()

4
utils/opencas.py
View File

@ -233,7 +233,7 @@ class cas_config(object):
def validate_parameter(self, param_name, param_value):
if param_name == 'ioclass_file':
if not os.path.exists(param_value):
raise ValueError('Incorrect path to io_class file')
raise ValueError('Invalid path to io_class file')
elif param_name == 'cleaning_policy':
self.check_cleaning_policy_valid(param_value)
elif param_name == 'promotion_policy':
@ -241,7 +241,7 @@ class cas_config(object):
elif param_name == 'cache_line_size':
self.check_cache_line_size_valid(param_value)
else:
raise ValueError('{0} is unknown parameter name'.format(param_name))
raise ValueError('{0} is invalid parameter name'.format(param_name))
@staticmethod
def check_cache_id_valid(cache_id):