Fix tests after moving to python3
Signed-off-by: Jan Musial <jan.musial@intel.com>
This commit is contained in:
parent
2a0c302aba
commit
86ccc817a6
@ -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)
|
||||
|
||||
|
@ -54,7 +54,7 @@ def test_cache_config_from_line_device_is_directory(
|
||||
)
|
||||
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"
|
||||
)
|
||||
@ -68,69 +68,68 @@ def test_cache_config_from_line_device_not_present(
|
||||
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")
|
||||
|
||||
@ -215,7 +214,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)
|
||||
|
||||
|
||||
@ -279,7 +278,7 @@ 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 +376,12 @@ 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",
|
||||
},
|
||||
],
|
||||
)
|
||||
@mock.patch("os.path.exists")
|
||||
|
@ -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()
|
||||
|
@ -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):
|
||||
|
Loading…
Reference in New Issue
Block a user