saltstack/salt
1
0
Fork 0
mirror of https://github.com/saltstack/salt.git synced 2025-04-17 10:10:20 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

Stop using the @patch() decorator.

Browse source 
https://engblog.nextdoor.com/what-your-mocks-do-when-you-aren-t-looking-b278e0d9e201

Signed-off-by: Pedro Algarvio <palgarvio@vmware.com>
This commit is contained in:
Pedro Algarvio 2022-12-29 19:36:35 +00:00 • committed by Pedro Algarvio
parent 53a78e71fb
commit ac34089d5d
14 changed files with 689 additions and 690 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

46
tests/pytests/unit/modules/file/test_file_chattr.py
View file

@ -15,7 +15,15 @@ pytestmark = pytest.mark.skip_on_windows(
@pytest.fixture
def configure_loader_modules():
def _common_patches():
with patch("salt.utils.platform.is_aix", Mock(return_value=False)), patch(
"os.path.exists", Mock(return_value=True)
), patch("salt.utils.path.which", Mock(return_value="some/tune2fs")):
yield
@pytest.fixture
def configure_loader_modules(_common_patches):
return {
filemod: {
"__salt__": {"cmd.run": cmdmod.run},
@ -24,22 +32,12 @@ def configure_loader_modules():
}
@patch("salt.utils.platform.is_aix", Mock(return_value=False))
@patch("os.path.exists", Mock(return_value=True))
@patch("salt.utils.path.which", Mock(return_value="some/tune2fs"))
def test_chattr_version_returns_None_if_no_tune2fs_exists():
patch_which = patch(
"salt.utils.path.which",
Mock(return_value=""),
)
with patch_which:
with patch("salt.utils.path.which", Mock(return_value="")):
actual = filemod._chattr_version()
assert actual is None
@patch("salt.utils.platform.is_aix", Mock(return_value=False))
@patch("os.path.exists", Mock(return_value=True))
@patch("salt.utils.path.which", Mock(return_value="some/tune2fs"))
def test_on_aix_chattr_version_should_be_None_even_if_tune2fs_exists():
patch_which = patch(
"salt.utils.path.which",
@ -57,9 +55,6 @@ def test_on_aix_chattr_version_should_be_None_even_if_tune2fs_exists():
mock_run.assert_not_called()
@patch("salt.utils.platform.is_aix", Mock(return_value=False))
@patch("os.path.exists", Mock(return_value=True))
@patch("salt.utils.path.which", Mock(return_value="some/tune2fs"))
def test_chattr_version_should_return_version_from_tune2fs():
expected = "1.43.4"
sample_output = textwrap.dedent(
@ -88,9 +83,6 @@ def test_chattr_version_should_return_version_from_tune2fs():
assert actual == expected
@patch("salt.utils.platform.is_aix", Mock(return_value=False))
@patch("os.path.exists", Mock(return_value=True))
@patch("salt.utils.path.which", Mock(return_value="some/tune2fs"))
def test_if_tune2fs_has_no_version_version_should_be_None():
patch_which = patch(
"salt.utils.path.which",
@ -105,9 +97,6 @@ def test_if_tune2fs_has_no_version_version_should_be_None():
assert actual is None
@patch("salt.utils.platform.is_aix", Mock(return_value=False))
@patch("os.path.exists", Mock(return_value=True))
@patch("salt.utils.path.which", Mock(return_value="some/tune2fs"))
def test_chattr_has_extended_attrs_should_return_False_if_chattr_version_is_None():
patch_chattr = patch(
"salt.modules.file._chattr_version",
@ -118,9 +107,6 @@ def test_chattr_has_extended_attrs_should_return_False_if_chattr_version_is_None
assert not actual, actual
@patch("salt.utils.platform.is_aix", Mock(return_value=False))
@patch("os.path.exists", Mock(return_value=True))
@patch("salt.utils.path.which", Mock(return_value="some/tune2fs"))
def test_chattr_has_extended_attrs_should_return_False_if_version_is_too_low():
below_expected = "0.1.1"
patch_chattr = patch(
@ -132,9 +118,6 @@ def test_chattr_has_extended_attrs_should_return_False_if_version_is_too_low():
assert not actual, actual
@patch("salt.utils.platform.is_aix", Mock(return_value=False))
@patch("os.path.exists", Mock(return_value=True))
@patch("salt.utils.path.which", Mock(return_value="some/tune2fs"))
def test_chattr_has_extended_attrs_should_return_False_if_version_is_equal_threshold():
threshold = "1.41.12"
patch_chattr = patch(
@ -146,9 +129,6 @@ def test_chattr_has_extended_attrs_should_return_False_if_version_is_equal_thres
assert not actual, actual
@patch("salt.utils.platform.is_aix", Mock(return_value=False))
@patch("os.path.exists", Mock(return_value=True))
@patch("salt.utils.path.which", Mock(return_value="some/tune2fs"))
def test_chattr_has_extended_attrs_should_return_True_if_version_is_above_threshold():
higher_than = "1.41.13"
patch_chattr = patch(
@ -160,9 +140,6 @@ def test_chattr_has_extended_attrs_should_return_True_if_version_is_above_thresh
assert actual, actual
@patch("salt.utils.platform.is_aix", Mock(return_value=False))
@patch("os.path.exists", Mock(return_value=True))
@patch("salt.utils.path.which", Mock(return_value="some/tune2fs"))
def test_check_perms_should_report_no_attr_changes_if_there_are_none():
filename = "/path/to/fnord"
attrs = "aAcCdDeijPsStTu"
@ -197,9 +174,6 @@ def test_check_perms_should_report_no_attr_changes_if_there_are_none():
assert actual_ret.get("changes", {}).get("attrs") is None, actual_ret
@patch("salt.utils.platform.is_aix", Mock(return_value=False))
@patch("os.path.exists", Mock(return_value=True))
@patch("salt.utils.path.which", Mock(return_value="some/tune2fs"))
def test_check_perms_should_report_attrs_new_and_old_if_they_changed():
filename = "/path/to/fnord"
attrs = "aAcCdDeijPsStTu"

50
tests/pytests/unit/modules/file/test_file_check.py
View file

@ -152,28 +152,34 @@ def test_check_managed_changes_follow_symlinks(a_link, tfile):
@pytest.mark.skip_on_windows(reason="os.symlink is not available on Windows")
@patch("os.path.exists", Mock(return_value=True))
def test_check_perms_user_group_name_and_id():
filename = "/path/to/fnord"
tests = [
@pytest.mark.parametrize(
"input,expected",
[
# user/group changes needed by name
{
"input": {"user": "cuser", "group": "cgroup"},
"expected": {"user": "cuser", "group": "cgroup"},
},
(
{"user": "cuser", "group": "cgroup"},
{"user": "cuser", "group": "cgroup"},
),
# no changes needed by name
{"input": {"user": "luser", "group": "lgroup"}, "expected": {}},
(
{"user": "luser", "group": "lgroup"},
{},
),
# user/group changes needed by id
{
"input": {"user": 1001, "group": 2001},
"expected": {"user": 1001, "group": 2001},
},
(
{"user": 1001, "group": 2001},
{"user": 1001, "group": 2001},
),
# no user/group changes needed by id
{"input": {"user": 3001, "group": 4001}, "expected": {}},
]
for test in tests:
(
{"user": 3001, "group": 4001},
{},
),
],
)
def test_check_perms_user_group_name_and_id(input, expected):
filename = "/path/to/fnord"
with patch("os.path.exists", Mock(return_value=True)):
# Consistent initial file stats
stat_out = {
"user": "luser",
@ -191,7 +197,7 @@ def test_check_perms_user_group_name_and_id():
# "chown" the file to the permissions we want in test["input"]
# pylint: disable=W0640
def fake_chown(cmd, *args, **kwargs):
for k, v in test["input"].items():
for k, v in input.items():
stat_out.update({k: v})
patch_chown = patch(
@ -203,9 +209,9 @@ def test_check_perms_user_group_name_and_id():
ret, pre_post = filemod.check_perms(
name=filename,
ret={},
user=test["input"]["user"],
group=test["input"]["group"],
user=input["user"],
group=input["group"],
mode="123",
follow_symlinks=False,
)
assert ret["changes"] == test["expected"]
assert ret["changes"] == expected

780
tests/pytests/unit/modules/file/test_file_line.py
View file

@ -112,20 +112,23 @@ def test_set_line_should_raise_command_execution_error_with_unknown_mode():
assert str(err.value) == "Unknown mode: fnord"
def test_if_content_is_none_and_mode_is_valid_but_not_delete_it_should_raise_command_execution_error():
valid_modes = ("insert", "ensure", "replace")
for mode in valid_modes:
with pytest.raises(CommandExecutionError) as err:
filemod._set_line(lines=[], mode=mode)
assert str(err.value) == "Content can only be empty if mode is delete"
@pytest.mark.parametrize("mode", ("insert", "ensure", "replace"))
def test_if_content_is_none_and_mode_is_valid_but_not_delete_it_should_raise_command_execution_error(
mode,
):
with pytest.raises(CommandExecutionError) as err:
filemod._set_line(lines=[], mode=mode)
assert str(err.value) == "Content can only be empty if mode is delete"
def test_if_delete_or_replace_is_called_with_empty_lines_it_should_warn_and_return_empty_body():
for mode in ("delete", "replace"):
with patch("salt.modules.file.log.warning", MagicMock()) as fake_warn:
actual_lines = filemod._set_line(mode=mode, lines=[], content="roscivs")
assert actual_lines == []
fake_warn.assert_called_with("Cannot find text to %s. File is empty.", mode)
@pytest.mark.parametrize("mode", ("delete", "replace"))
def test_if_delete_or_replace_is_called_with_empty_lines_it_should_warn_and_return_empty_body(
mode,
):
with patch("salt.modules.file.log.warning", MagicMock()) as fake_warn:
actual_lines = filemod._set_line(mode=mode, lines=[], content="roscivs")
assert actual_lines == []
fake_warn.assert_called_with("Cannot find text to %s. File is empty.", mode)
def test_if_mode_is_delete_and_not_before_after_or_match_then_content_should_be_used_to_delete_line():
@ -297,64 +300,68 @@ def test_if_location_is_end_of_file_and_indent_is_True_then_line_should_match_pr
assert actual_lines == expected_lines
def test_if_location_is_not_set_but_before_and_after_are_then_line_should_appear_as_the_line_before_before():
for indent in ("", " \t \t\t\t "):
content = "roscivs"
after = "after"
before = "before"
original_lines = ["foo", "bar", indent + after, "belowme", indent + before]
expected_lines = [
"foo",
"bar",
indent + after,
"belowme",
indent + content,
indent + before,
]
@pytest.mark.parametrize("indent", ("", " \t \t\t\t "))
def test_if_location_is_not_set_but_before_and_after_are_then_line_should_appear_as_the_line_before_before(
indent,
):
content = "roscivs"
after = "after"
before = "before"
original_lines = ["foo", "bar", indent + after, "belowme", indent + before]
expected_lines = [
"foo",
"bar",
indent + after,
"belowme",
indent + content,
indent + before,
]
actual_lines = filemod._set_line(
lines=original_lines,
content=content,
mode="insert",
location=None,
before=before,
after=after,
)
actual_lines = filemod._set_line(
lines=original_lines,
content=content,
mode="insert",
location=None,
before=before,
after=after,
)
assert actual_lines == expected_lines
assert actual_lines == expected_lines
def test_insert_with_after_and_before_with_no_location_should_indent_to_match_before_indent():
for indent in ("", " \t \t\t\t "):
content = "roscivs"
after = "after"
before = "before"
original_lines = [
"foo",
"bar",
indent + after,
"belowme",
(indent * 2) + before,
]
expected_lines = [
"foo",
"bar",
indent + after,
"belowme",
(indent * 2) + content,
(indent * 2) + before,
]
@pytest.mark.parametrize("indent", ("", " \t \t\t\t "))
def test_insert_with_after_and_before_with_no_location_should_indent_to_match_before_indent(
indent,
):
content = "roscivs"
after = "after"
before = "before"
original_lines = [
"foo",
"bar",
indent + after,
"belowme",
(indent * 2) + before,
]
expected_lines = [
"foo",
"bar",
indent + after,
"belowme",
(indent * 2) + content,
(indent * 2) + before,
]
actual_lines = filemod._set_line(
lines=original_lines,
content=content,
mode="insert",
location=None,
before=before,
after=after,
)
actual_lines = filemod._set_line(
lines=original_lines,
content=content,
mode="insert",
location=None,
before=before,
after=after,
)
assert actual_lines == expected_lines
assert actual_lines == expected_lines
def test_if_not_location_but_before_and_after_and_more_than_one_after_it_should_CommandExecutionError():
@ -779,83 +786,83 @@ def test_ensure_with_too_many_lines_between_before_and_after_should_CommandExecu
)
def test_ensure_with_no_lines_between_before_and_after_should_insert_a_line():
for indent in ("", " \t \t\t\t "):
before = "before"
after = "after"
content = "roscivs"
original_lines = [indent + after, before]
expected_lines = [indent + after, indent + content, before]
@pytest.mark.parametrize("indent", ("", " \t \t\t\t "))
def test_ensure_with_no_lines_between_before_and_after_should_insert_a_line(indent):
before = "before"
after = "after"
content = "roscivs"
original_lines = [indent + after, before]
expected_lines = [indent + after, indent + content, before]
actual_lines = filemod._set_line(
lines=original_lines,
content=content,
before=before,
after=after,
mode="ensure",
indent=True,
)
actual_lines = filemod._set_line(
lines=original_lines,
content=content,
before=before,
after=after,
mode="ensure",
indent=True,
)
assert actual_lines == expected_lines
assert actual_lines == expected_lines
def test_ensure_with_existing_but_different_line_should_set_the_line():
for indent in ("", " \t \t\t\t "):
before = "before"
after = "after"
content = "roscivs"
original_lines = [indent + after, "fnord", before]
expected_lines = [indent + after, indent + content, before]
@pytest.mark.parametrize("indent", ("", " \t \t\t\t "))
def test_ensure_with_existing_but_different_line_should_set_the_line(indent):
before = "before"
after = "after"
content = "roscivs"
original_lines = [indent + after, "fnord", before]
expected_lines = [indent + after, indent + content, before]
actual_lines = filemod._set_line(
lines=original_lines,
content=content,
before=before,
after=after,
mode="ensure",
indent=True,
)
actual_lines = filemod._set_line(
lines=original_lines,
content=content,
before=before,
after=after,
mode="ensure",
indent=True,
)
assert actual_lines == expected_lines
assert actual_lines == expected_lines
def test_ensure_with_after_and_existing_content_should_return_same_lines():
for indent in ("", " \t \t\t\t "):
before = None
after = "after"
content = "roscivs"
original_lines = [indent + after, indent + content, "fnord"]
@pytest.mark.parametrize("indent", ("", " \t \t\t\t "))
def test_ensure_with_after_and_existing_content_should_return_same_lines(indent):
before = None
after = "after"
content = "roscivs"
original_lines = [indent + after, indent + content, "fnord"]
actual_lines = filemod._set_line(
lines=original_lines,
content=content,
before=before,
after=after,
mode="ensure",
indent=True,
)
actual_lines = filemod._set_line(
lines=original_lines,
content=content,
before=before,
after=after,
mode="ensure",
indent=True,
)
assert actual_lines == original_lines
assert actual_lines == original_lines
def test_ensure_with_after_and_missing_content_should_add_it():
for indent in ("", " \t \t\t\t "):
before = None
after = "after"
content = "roscivs"
original_lines = [indent + after, "more fnord", "fnord"]
expected_lines = [indent + after, indent + content, "more fnord", "fnord"]
@pytest.mark.parametrize("indent", ("", " \t \t\t\t "))
def test_ensure_with_after_and_missing_content_should_add_it(indent):
before = None
after = "after"
content = "roscivs"
original_lines = [indent + after, "more fnord", "fnord"]
expected_lines = [indent + after, indent + content, "more fnord", "fnord"]
actual_lines = filemod._set_line(
lines=original_lines,
content=content,
before=before,
after=after,
mode="ensure",
indent=True,
)
actual_lines = filemod._set_line(
lines=original_lines,
content=content,
before=before,
after=after,
mode="ensure",
indent=True,
)
assert actual_lines == expected_lines
assert actual_lines == expected_lines
def test_ensure_with_after_and_content_at_the_end_should_not_add_duplicate():
@ -873,48 +880,48 @@ def test_ensure_with_after_and_content_at_the_end_should_not_add_duplicate():
assert actual_lines == original_lines
def test_ensure_with_before_and_missing_content_should_add_it():
for indent in ("", " \t \t\t\t "):
before = "before"
after = None
content = "roscivs"
original_lines = [indent + "fnord", indent + "fnord", before]
expected_lines = [
indent + "fnord",
indent + "fnord",
indent + content,
before,
]
@pytest.mark.parametrize("indent", ("", " \t \t\t\t "))
def test_ensure_with_before_and_missing_content_should_add_it(indent):
before = "before"
after = None
content = "roscivs"
original_lines = [indent + "fnord", indent + "fnord", before]
expected_lines = [
indent + "fnord",
indent + "fnord",
indent + content,
before,
]
actual_lines = filemod._set_line(
lines=original_lines,
content=content,
before=before,
after=after,
mode="ensure",
indent=True,
)
actual_lines = filemod._set_line(
lines=original_lines,
content=content,
before=before,
after=after,
mode="ensure",
indent=True,
)
assert actual_lines == expected_lines
assert actual_lines == expected_lines
def test_ensure_with_before_and_existing_content_should_return_same_lines():
for indent in ("", " \t \t\t\t "):
before = "before"
after = None
content = "roscivs"
original_lines = [indent + "fnord", indent + content, before]
@pytest.mark.parametrize("indent", ("", " \t \t\t\t "))
def test_ensure_with_before_and_existing_content_should_return_same_lines(indent):
before = "before"
after = None
content = "roscivs"
original_lines = [indent + "fnord", indent + content, before]
actual_lines = filemod._set_line(
lines=original_lines,
content=content,
before=before,
after=after,
mode="ensure",
indent=True,
)
actual_lines = filemod._set_line(
lines=original_lines,
content=content,
before=before,
after=after,
mode="ensure",
indent=True,
)
assert actual_lines == original_lines
assert actual_lines == original_lines
def test_ensure_without_before_and_after_should_CommandExecutionError():
@ -937,9 +944,8 @@ def test_ensure_without_before_and_after_should_CommandExecutionError():
)
@patch("os.path.realpath", MagicMock(wraps=lambda x: x))
@patch("os.path.isfile", MagicMock(return_value=True))
def test_delete_line_in_empty_file(anyattr):
@pytest.mark.parametrize("mode", ["delete", "replace"])
def test_delete_line_in_empty_file(anyattr, mode):
"""
Tests that when calling file.line with ``mode=delete``,
the function doesn't stack trace if the file is empty.
@ -947,7 +953,9 @@ def test_delete_line_in_empty_file(anyattr):
See Issue #38438.
"""
for mode in ["delete", "replace"]:
with patch("os.path.realpath", MagicMock(wraps=lambda x: x)), patch(
"os.path.isfile", MagicMock(return_value=True)
):
_log = MagicMock()
with patch("salt.utils.files.fopen", mock_open(read_data="")), patch(
"os.stat", anyattr
@ -960,10 +968,8 @@ def test_delete_line_in_empty_file(anyattr):
assert "Cannot find text to {}".format(mode) in warning_log_msg
@patch("os.path.realpath", MagicMock())
@patch("os.path.isfile", MagicMock(return_value=True))
@patch("os.stat", MagicMock())
def test_line_delete_no_match():
@pytest.mark.parametrize("mode", ["delete", "replace"])
def test_line_delete_no_match(mode):
"""
Tests that when calling file.line with ``mode=delete``,
with not matching pattern to delete returns False
@ -973,7 +979,9 @@ def test_line_delete_no_match():
["file_roots:", " base:", " - /srv/salt", " - /srv/custom"]
)
match = "not matching"
for mode in ["delete", "replace"]:
with patch("os.path.realpath", MagicMock()), patch(
"os.path.isfile", MagicMock(return_value=True)
), patch("os.stat", MagicMock()):
files_fopen = mock_open(read_data=file_content)
with patch("salt.utils.files.fopen", files_fopen):
atomic_opener = mock_open()
@ -981,46 +989,51 @@ def test_line_delete_no_match():
assert not filemod.line("foo", content="foo", match=match, mode=mode)
@patch("os.path.realpath", MagicMock(wraps=lambda x: x))
@patch("os.path.isfile", MagicMock(return_value=True))
def test_line_modecheck_failure():
@pytest.mark.parametrize(
"mode,err_msg",
[
(None, "How to process the file"),
("nonsense", "Unknown mode"),
],
)
def test_line_modecheck_failure(mode, err_msg):
"""
Test for file.line for empty or wrong mode.
Calls unknown or empty mode and expects failure.
:return:
"""
for mode, err_msg in [
(None, "How to process the file"),
("nonsense", "Unknown mode"),
]:
with patch("os.path.realpath", MagicMock(wraps=lambda x: x)), patch(
"os.path.isfile", MagicMock(return_value=True)
):
with pytest.raises(CommandExecutionError) as exc_info:
filemod.line("foo", mode=mode)
assert err_msg in str(exc_info.value)
@patch("os.path.realpath", MagicMock(wraps=lambda x: x))
@patch("os.path.isfile", MagicMock(return_value=True))
def test_line_no_content():
@pytest.mark.parametrize("mode", ["insert", "ensure", "replace"])
def test_line_no_content(mode):
"""
Test for file.line for an empty content when not deleting anything.
:return:
"""
for mode in ["insert", "ensure", "replace"]:
with patch("os.path.realpath", MagicMock(wraps=lambda x: x)), patch(
"os.path.isfile", MagicMock(return_value=True)
):
with pytest.raises(CommandExecutionError) as exc_info:
filemod.line("foo", mode=mode)
assert 'Content can only be empty if mode is "delete"' in str(exc_info.value)
@patch("os.path.realpath", MagicMock(wraps=lambda x: x))
@patch("os.path.isfile", MagicMock(return_value=True))
@patch("os.stat", MagicMock())
def test_line_insert_no_location_no_before_no_after():
"""
Test for file.line for insertion but define no location/before/after.
:return:
"""
files_fopen = mock_open(read_data="test data")
with patch("salt.utils.files.fopen", files_fopen):
with patch("os.path.realpath", MagicMock(wraps=lambda x: x)), patch(
"os.path.isfile", MagicMock(return_value=True)
), patch("os.stat", MagicMock()), patch(
"salt.utils.files.fopen", mock_open(read_data="test data")
):
with pytest.raises(CommandExecutionError) as exc_info:
filemod.line("foo", content="test content", mode="insert")
assert '"location" or "before/after"' in str(exc_info.value)
@ -1063,7 +1076,8 @@ def test_line_insert_after_no_pattern(tempfile_name, get_body):
assert writelines_content[0] == expected, (writelines_content[0], expected)
def test_line_insert_after_pattern(tempfile_name, get_body):
@pytest.mark.parametrize("after_line", ["file_r.*", ".*roots"])
def test_line_insert_after_pattern(tempfile_name, get_body, after_line):
"""
Test for file.line for insertion after specific line, using pattern.
@ -1096,37 +1110,36 @@ def test_line_insert_after_pattern(tempfile_name, get_body):
isfile_mock = MagicMock(
side_effect=lambda x: True if x == tempfile_name else DEFAULT
)
for after_line in ["file_r.*", ".*roots"]:
with patch("os.path.isfile", isfile_mock), patch(
"os.stat", MagicMock(return_value=DummyStat())
), patch("salt.utils.files.fopen", mock_open(read_data=file_content)), patch(
"salt.utils.atomicfile.atomic_open", mock_open()
) as atomic_open_mock:
filemod.line(
tempfile_name,
content=cfg_content,
after=after_line,
mode="insert",
indent=False,
)
handles = atomic_open_mock.filehandles[tempfile_name]
# We should only have opened the file once
open_count = len(handles)
assert open_count == 1, open_count
# We should only have invoked .writelines() once...
writelines_content = handles[0].writelines_calls
writelines_count = len(writelines_content)
assert writelines_count == 1, writelines_count
# ... with the updated content
expected = get_body(file_modified)
# We passed cfg_content with a newline in the middle, so it
# will be written as two lines in the same element of the list
# passed to .writelines()
expected[3] = expected[3] + expected.pop(4)
assert writelines_content[0] == expected, (
writelines_content[0],
expected,
)
with patch("os.path.isfile", isfile_mock), patch(
"os.stat", MagicMock(return_value=DummyStat())
), patch("salt.utils.files.fopen", mock_open(read_data=file_content)), patch(
"salt.utils.atomicfile.atomic_open", mock_open()
) as atomic_open_mock:
filemod.line(
tempfile_name,
content=cfg_content,
after=after_line,
mode="insert",
indent=False,
)
handles = atomic_open_mock.filehandles[tempfile_name]
# We should only have opened the file once
open_count = len(handles)
assert open_count == 1, open_count
# We should only have invoked .writelines() once...
writelines_content = handles[0].writelines_calls
writelines_count = len(writelines_content)
assert writelines_count == 1, writelines_count
# ... with the updated content
expected = get_body(file_modified)
# We passed cfg_content with a newline in the middle, so it
# will be written as two lines in the same element of the list
# passed to .writelines()
expected[3] = expected[3] + expected.pop(4)
assert writelines_content[0] == expected, (
writelines_content[0],
expected,
)
def test_line_insert_multi_line_content_after_unicode(tempfile_name, get_body):
@ -1146,36 +1159,37 @@ def test_line_insert_multi_line_content_after_unicode(tempfile_name, get_body):
isfile_mock = MagicMock(
side_effect=lambda x: True if x == tempfile_name else DEFAULT
)
for after_line in ["This is another line"]:
with patch("os.path.isfile", isfile_mock), patch(
"os.stat", MagicMock(return_value=DummyStat())
), patch("salt.utils.files.fopen", mock_open(read_data=file_content)), patch(
"salt.utils.atomicfile.atomic_open", mock_open()
) as atomic_open_mock:
filemod.line(
tempfile_name,
content=cfg_content,
after=after_line,
mode="insert",
indent=False,
)
handles = atomic_open_mock.filehandles[tempfile_name]
# We should only have opened the file once
open_count = len(handles)
assert open_count == 1, open_count
# We should only have invoked .writelines() once...
writelines_content = handles[0].writelines_calls
writelines_count = len(writelines_content)
assert writelines_count == 1, writelines_count
# ... with the updated content
expected = get_body(file_modified)
assert writelines_content[0] == expected, (
writelines_content[0],
expected,
)
after_line = "This is another line"
with patch("os.path.isfile", isfile_mock), patch(
"os.stat", MagicMock(return_value=DummyStat())
), patch("salt.utils.files.fopen", mock_open(read_data=file_content)), patch(
"salt.utils.atomicfile.atomic_open", mock_open()
) as atomic_open_mock:
filemod.line(
tempfile_name,
content=cfg_content,
after=after_line,
mode="insert",
indent=False,
)
handles = atomic_open_mock.filehandles[tempfile_name]
# We should only have opened the file once
open_count = len(handles)
assert open_count == 1, open_count
# We should only have invoked .writelines() once...
writelines_content = handles[0].writelines_calls
writelines_count = len(writelines_content)
assert writelines_count == 1, writelines_count
# ... with the updated content
expected = get_body(file_modified)
assert writelines_content[0] == expected, (
writelines_content[0],
expected,
)
def test_line_insert_before(tempfile_name, get_body):
@pytest.mark.parametrize("before_line", ["/srv/salt", "/srv/sa.*t"])
def test_line_insert_before(tempfile_name, get_body, before_line):
"""
Test for file.line for insertion before specific line, using pattern and no patterns.
@ -1199,32 +1213,28 @@ def test_line_insert_before(tempfile_name, get_body):
isfile_mock = MagicMock(
side_effect=lambda x: True if x == tempfile_name else DEFAULT
)
for before_line in ["/srv/salt", "/srv/sa.*t"]:
with patch("os.path.isfile", isfile_mock), patch(
"os.stat", MagicMock(return_value=DummyStat())
), patch("salt.utils.files.fopen", mock_open(read_data=file_content)), patch(
"salt.utils.atomicfile.atomic_open", mock_open()
) as atomic_open_mock:
filemod.line(
tempfile_name, content=cfg_content, before=before_line, mode="insert"
)
handles = atomic_open_mock.filehandles[tempfile_name]
# We should only have opened the file once
open_count = len(handles)
assert open_count == 1, open_count
# We should only have invoked .writelines() once...
writelines_content = handles[0].writelines_calls
writelines_count = len(writelines_content)
assert writelines_count == 1, writelines_count
# ... with the updated content
expected = get_body(file_modified)
# assert writelines_content[0] == expected, (writelines_content[0], expected)
assert writelines_content[0] == expected
with patch("os.path.isfile", isfile_mock), patch(
"os.stat", MagicMock(return_value=DummyStat())
), patch("salt.utils.files.fopen", mock_open(read_data=file_content)), patch(
"salt.utils.atomicfile.atomic_open", mock_open()
) as atomic_open_mock:
filemod.line(
tempfile_name, content=cfg_content, before=before_line, mode="insert"
)
handles = atomic_open_mock.filehandles[tempfile_name]
# We should only have opened the file once
open_count = len(handles)
assert open_count == 1, open_count
# We should only have invoked .writelines() once...
writelines_content = handles[0].writelines_calls
writelines_count = len(writelines_content)
assert writelines_count == 1, writelines_count
# ... with the updated content
expected = get_body(file_modified)
# assert writelines_content[0] == expected, (writelines_content[0], expected)
assert writelines_content[0] == expected
@patch("os.path.realpath", MagicMock(wraps=lambda x: x))
@patch("os.path.isfile", MagicMock(return_value=True))
@patch("os.stat", MagicMock())
def test_line_assert_exception_pattern():
"""
Test for file.line for exception on insert with too general pattern.
@ -1235,22 +1245,25 @@ def test_line_assert_exception_pattern():
["file_roots:", " base:", " - /srv/salt", " - /srv/sugar"]
)
cfg_content = "- /srv/custom"
for before_line in ["/sr.*"]:
files_fopen = mock_open(read_data=file_content)
with patch("salt.utils.files.fopen", files_fopen):
atomic_opener = mock_open()
with patch("salt.utils.atomicfile.atomic_open", atomic_opener):
with pytest.raises(CommandExecutionError) as cm:
filemod.line(
"foo",
content=cfg_content,
before=before_line,
mode="insert",
)
assert (
str(cm.value)
== 'Found more than expected occurrences in "before" expression'
)
before_line = "/sr.*"
with patch("os.path.realpath", MagicMock(wraps=lambda x: x)), patch(
"os.path.isfile", MagicMock(return_value=True)
), patch("os.stat", MagicMock()), patch(
"salt.utils.files.fopen", mock_open(read_data=file_content)
), patch(
"salt.utils.atomicfile.atomic_open", mock_open()
):
with pytest.raises(CommandExecutionError) as cm:
filemod.line(
"foo",
content=cfg_content,
before=before_line,
mode="insert",
)
assert (
str(cm.value)
== 'Found more than expected occurrences in "before" expression'
)
def test_line_insert_before_after(tempfile_name, get_body):
@ -1284,30 +1297,31 @@ def test_line_insert_before_after(tempfile_name, get_body):
isfile_mock = MagicMock(
side_effect=lambda x: True if x == tempfile_name else DEFAULT
)
for b_line, a_line in [("/srv/sugar", "/srv/salt")]:
with patch("os.path.isfile", isfile_mock), patch(
"os.stat", MagicMock(return_value=DummyStat())
), patch("salt.utils.files.fopen", mock_open(read_data=file_content)), patch(
"salt.utils.atomicfile.atomic_open", mock_open()
) as atomic_open_mock:
filemod.line(
tempfile_name,
content=cfg_content,
before=b_line,
after=a_line,
mode="insert",
)
handles = atomic_open_mock.filehandles[tempfile_name]
# We should only have opened the file once
open_count = len(handles)
assert open_count == 1, open_count
# We should only have invoked .writelines() once...
writelines_content = handles[0].writelines_calls
writelines_count = len(writelines_content)
assert writelines_count == 1, writelines_count
# ... with the updated content
expected = get_body(file_modified)
assert writelines_content[0] == expected
b_line = "/srv/sugar"
a_line = "/srv/salt"
with patch("os.path.isfile", isfile_mock), patch(
"os.stat", MagicMock(return_value=DummyStat())
), patch("salt.utils.files.fopen", mock_open(read_data=file_content)), patch(
"salt.utils.atomicfile.atomic_open", mock_open()
) as atomic_open_mock:
filemod.line(
tempfile_name,
content=cfg_content,
before=b_line,
after=a_line,
mode="insert",
)
handles = atomic_open_mock.filehandles[tempfile_name]
# We should only have opened the file once
open_count = len(handles)
assert open_count == 1, open_count
# We should only have invoked .writelines() once...
writelines_content = handles[0].writelines_calls
writelines_count = len(writelines_content)
assert writelines_count == 1, writelines_count
# ... with the updated content
expected = get_body(file_modified)
assert writelines_content[0] == expected
def test_line_insert_start(tempfile_name, get_body):
@ -1645,9 +1659,6 @@ def test_line_insert_ensure_beforeafter_twolines_exists(tempfile_name):
assert result is False
@patch("os.path.realpath", MagicMock(wraps=lambda x: x))
@patch("os.path.isfile", MagicMock(return_value=True))
@patch("os.stat", MagicMock())
def test_line_insert_ensure_beforeafter_rangelines():
"""
Test for file.line for insertion ensuring the line is between two lines
@ -1667,25 +1678,31 @@ def test_line_insert_ensure_beforeafter_rangelines():
file_content.split(os.linesep)[-1],
)
for (_after, _before) in [(after, before), ("NAME_.*", "SKEL_.*")]:
files_fopen = mock_open(read_data=file_content)
with patch("salt.utils.files.fopen", files_fopen):
atomic_opener = mock_open()
with patch("salt.utils.atomicfile.atomic_open", atomic_opener):
with pytest.raises(CommandExecutionError) as exc_info:
filemod.line(
"foo",
content=cfg_content,
after=_after,
before=_before,
mode="ensure",
)
assert (
'Found more than one line between boundaries "before" and "after"'
in str(exc_info.value)
with patch("os.path.realpath", MagicMock(wraps=lambda x: x)), patch(
"os.path.isfile", MagicMock(return_value=True)
), patch("os.stat", MagicMock()), patch(
"salt.utils.files.fopen", mock_open(read_data=file_content)
), patch(
"salt.utils.atomicfile.atomic_open", mock_open()
):
with pytest.raises(CommandExecutionError) as exc_info:
filemod.line(
"foo",
content=cfg_content,
after=_after,
before=_before,
mode="ensure",
)
assert (
'Found more than one line between boundaries "before" and "after"'
in str(exc_info.value)
)
def test_line_delete(tempfile_name, get_body):
@pytest.mark.parametrize(
"content", ["/srv/pepper", "/srv/pepp*", "/srv/p.*", "/sr.*pe.*"]
)
def test_line_delete(tempfile_name, get_body, content):
"""
Test for file.line for deletion of specific line
:return:
@ -1706,31 +1723,33 @@ def test_line_delete(tempfile_name, get_body):
isfile_mock = MagicMock(
side_effect=lambda x: True if x == tempfile_name else DEFAULT
)
for content in ["/srv/pepper", "/srv/pepp*", "/srv/p.*", "/sr.*pe.*"]:
files_fopen = mock_open(read_data=file_content)
with patch("os.path.isfile", isfile_mock), patch(
"os.stat", MagicMock(return_value=DummyStat())
), patch("salt.utils.files.fopen", files_fopen), patch(
"salt.utils.atomicfile.atomic_open", mock_open()
) as atomic_open_mock:
filemod.line(tempfile_name, content=content, mode="delete")
handles = atomic_open_mock.filehandles[tempfile_name]
# We should only have opened the file once
open_count = len(handles)
assert open_count == 1, open_count
# We should only have invoked .writelines() once...
writelines_content = handles[0].writelines_calls
writelines_count = len(writelines_content)
assert writelines_count == 1, writelines_count
# ... with the updated content
expected = get_body(file_modified)
assert writelines_content[0] == expected, (
writelines_content[0],
expected,
)
files_fopen = mock_open(read_data=file_content)
with patch("os.path.isfile", isfile_mock), patch(
"os.stat", MagicMock(return_value=DummyStat())
), patch("salt.utils.files.fopen", files_fopen), patch(
"salt.utils.atomicfile.atomic_open", mock_open()
) as atomic_open_mock:
filemod.line(tempfile_name, content=content, mode="delete")
handles = atomic_open_mock.filehandles[tempfile_name]
# We should only have opened the file once
open_count = len(handles)
assert open_count == 1, open_count
# We should only have invoked .writelines() once...
writelines_content = handles[0].writelines_calls
writelines_count = len(writelines_content)
assert writelines_count == 1, writelines_count
# ... with the updated content
expected = get_body(file_modified)
assert writelines_content[0] == expected, (
writelines_content[0],
expected,
)
def test_line_replace(tempfile_name, get_body):
@pytest.mark.parametrize(
"match", ["/srv/pepper", "/srv/pepp*", "/srv/p.*", "/sr.*pe.*"]
)
def test_line_replace(tempfile_name, get_body, match):
"""
Test for file.line for replacement of specific line
:return:
@ -1757,30 +1776,29 @@ def test_line_replace(tempfile_name, get_body):
isfile_mock = MagicMock(
side_effect=lambda x: True if x == tempfile_name else DEFAULT
)
for match in ["/srv/pepper", "/srv/pepp*", "/srv/p.*", "/sr.*pe.*"]:
files_fopen = mock_open(read_data=file_content)
with patch("os.path.isfile", isfile_mock), patch(
"os.stat", MagicMock(return_value=DummyStat())
), patch("salt.utils.files.fopen", files_fopen), patch(
"salt.utils.atomicfile.atomic_open", mock_open()
) as atomic_open_mock:
filemod.line(
tempfile_name,
content="- /srv/natrium-chloride",
match=match,
mode="replace",
)
handles = atomic_open_mock.filehandles[tempfile_name]
# We should only have opened the file once
open_count = len(handles)
assert open_count == 1, open_count
# We should only have invoked .writelines() once...
writelines_content = handles[0].writelines_calls
writelines_count = len(writelines_content)
assert writelines_count == 1, writelines_count
# ... with the updated content
expected = get_body(file_modified)
assert writelines_content[0] == expected, (
writelines_content[0],
expected,
)
files_fopen = mock_open(read_data=file_content)
with patch("os.path.isfile", isfile_mock), patch(
"os.stat", MagicMock(return_value=DummyStat())
), patch("salt.utils.files.fopen", files_fopen), patch(
"salt.utils.atomicfile.atomic_open", mock_open()
) as atomic_open_mock:
filemod.line(
tempfile_name,
content="- /srv/natrium-chloride",
match=match,
mode="replace",
)
handles = atomic_open_mock.filehandles[tempfile_name]
# We should only have opened the file once
open_count = len(handles)
assert open_count == 1, open_count
# We should only have invoked .writelines() once...
writelines_content = handles[0].writelines_calls
writelines_count = len(writelines_content)
assert writelines_count == 1, writelines_count
# ... with the updated content
expected = get_body(file_modified)
assert writelines_content[0] == expected, (
writelines_content[0],
expected,
)

28
tests/pytests/unit/modules/file/test_file_lsattr.py
View file

@ -12,15 +12,20 @@ log = logging.getLogger(__name__)
@pytest.fixture
def configure_loader_modules():
def _common_patches():
with patch("salt.utils.platform.is_aix", Mock(return_value=False)), patch(
"os.path.exists", Mock(return_value=True)
), patch("salt.utils.path.which", Mock(return_value="fnord")):
yield
@pytest.fixture
def configure_loader_modules(_common_patches):
return {
filemod: {"__salt__": {"cmd.run": cmdmod.run}},
}
@patch("salt.utils.platform.is_aix", Mock(return_value=False))
@patch("os.path.exists", Mock(return_value=True))
@patch("salt.utils.path.which", Mock(return_value="fnord"))
def test_if_lsattr_is_missing_it_should_return_None():
patch_which = patch(
"salt.utils.path.which",
@ -31,9 +36,6 @@ def test_if_lsattr_is_missing_it_should_return_None():
assert actual is None, actual
@patch("salt.utils.platform.is_aix", Mock(return_value=False))
@patch("os.path.exists", Mock(return_value=True))
@patch("salt.utils.path.which", Mock(return_value="fnord"))
def test_on_aix_lsattr_should_be_None():
patch_aix = patch(
"salt.utils.platform.is_aix",
@ -46,9 +48,6 @@ def test_on_aix_lsattr_should_be_None():
assert actual is None
@patch("salt.utils.platform.is_aix", Mock(return_value=False))
@patch("os.path.exists", Mock(return_value=True))
@patch("salt.utils.path.which", Mock(return_value="fnord"))
def test_SaltInvocationError_should_be_raised_when_file_is_missing():
patch_exists = patch(
"os.path.exists",
@ -58,9 +57,6 @@ def test_SaltInvocationError_should_be_raised_when_file_is_missing():
filemod.lsattr("foo")
@patch("salt.utils.platform.is_aix", Mock(return_value=False))
@patch("os.path.exists", Mock(return_value=True))
@patch("salt.utils.path.which", Mock(return_value="fnord"))
def test_if_chattr_version_is_less_than_required_flags_should_ignore_extended():
fname = "/path/to/fnord"
with_extended = (
@ -89,9 +85,6 @@ def test_if_chattr_version_is_less_than_required_flags_should_ignore_extended():
assert actual == expected, msg
@patch("salt.utils.platform.is_aix", Mock(return_value=False))
@patch("os.path.exists", Mock(return_value=True))
@patch("salt.utils.path.which", Mock(return_value="fnord"))
def test_if_chattr_version_is_high_enough_then_extended_flags_should_be_returned():
fname = "/path/to/fnord"
with_extended = (
@ -120,9 +113,6 @@ def test_if_chattr_version_is_high_enough_then_extended_flags_should_be_returned
assert actual == expected, msg
@patch("salt.utils.platform.is_aix", Mock(return_value=False))
@patch("os.path.exists", Mock(return_value=True))
@patch("salt.utils.path.which", Mock(return_value="fnord"))
def test_if_supports_extended_but_there_are_no_flags_then_none_should_be_returned():
fname = "/path/to/fnord"
with_extended = (

28
tests/pytests/unit/modules/test_aptpkg.py
View file

@ -781,13 +781,6 @@ def test_mod_repo_match():
)
@patch("salt.utils.path.os_walk", MagicMock(return_value=[("test", "test", "test")]))
@patch("os.path.getsize", MagicMock(return_value=123456))
@patch("os.path.getctime", MagicMock(return_value=1234567890.123456))
@patch(
"fnmatch.filter",
MagicMock(return_value=["/var/cache/apt/archive/test_package.rpm"]),
)
def test_list_downloaded():
"""
Test downloaded packages listing.
@ -803,8 +796,14 @@ def test_list_downloaded():
}
}
}
with patch.dict(
with patch(
"salt.utils.path.os_walk", MagicMock(return_value=[("test", "test", "test")])
), patch("os.path.getsize", MagicMock(return_value=123456)), patch(
"os.path.getctime", MagicMock(return_value=1234567890.123456)
), patch(
"fnmatch.filter",
MagicMock(return_value=["/var/cache/apt/archive/test_package.rpm"]),
), patch.dict(
aptpkg.__salt__,
{
"lowpkg.bin_pkg_info": MagicMock(
@ -1184,13 +1183,14 @@ def test_call_apt_default():
)
@patch("salt.utils.systemd.has_scope", MagicMock(return_value=True))
def test_call_apt_in_scope():
"""
Call apt within the scope.
:return:
"""
with patch.dict(
with patch(
"salt.utils.systemd.has_scope", MagicMock(return_value=True)
), patch.dict(
aptpkg.__salt__,
{"cmd.run_all": MagicMock(), "config.get": MagicMock(return_value=True)},
):
@ -1292,7 +1292,6 @@ def test_services_need_restart_checkrestart_missing():
aptpkg.services_need_restart()
@patch("salt.utils.path.which_bin", Mock(return_value="/usr/sbin/checkrestart"))
def test_services_need_restart():
"""
Test that checkrestart output is parsed correctly
@ -1304,8 +1303,9 @@ PACKAGES: 8
SERVICE:rsyslog,385,/usr/sbin/rsyslogd
SERVICE:cups-daemon,390,/usr/sbin/cupsd
"""
with patch.dict(aptpkg.__salt__, {"cmd.run_stdout": Mock(return_value=cr_output)}):
with patch(
"salt.utils.path.which_bin", Mock(return_value="/usr/sbin/checkrestart")
), patch.dict(aptpkg.__salt__, {"cmd.run_stdout": Mock(return_value=cr_output)}):
assert sorted(aptpkg.services_need_restart()) == [
"cups-daemon",
"rsyslog",

44
tests/pytests/unit/modules/test_btrfs.py
View file

@ -384,26 +384,25 @@ def test_subvolume_create_fails_parameters():
btrfs.subvolume_create("var", qgroupids="1")
@patch("salt.modules.btrfs.subvolume_exists")
def test_subvolume_create_already_exists(subvolume_exists):
def test_subvolume_create_already_exists():
"""
Test btrfs subvolume create
"""
subvolume_exists.return_value = True
assert not btrfs.subvolume_create("var", dest="/mnt")
with patch("salt.modules.btrfs.subvolume_exists", return_value=True):
assert not btrfs.subvolume_create("var", dest="/mnt")
@patch("salt.modules.btrfs.subvolume_exists")
def test_subvolume_create(subvolume_exists):
def test_subvolume_create():
"""
Test btrfs subvolume create
"""
subvolume_exists.return_value = False
salt_mock = {
"cmd.run_all": MagicMock(return_value={"recode": 0}),
}
expected_path = os.path.join("/mnt", "var")
with patch.dict(btrfs.__salt__, salt_mock):
with patch(
"salt.modules.btrfs.subvolume_exists", return_value=False
) as subvolume_exists, patch.dict(btrfs.__salt__, salt_mock):
assert btrfs.subvolume_create("var", dest="/mnt")
subvolume_exists.assert_called_once()
salt_mock["cmd.run_all"].assert_called_once()
@ -433,43 +432,40 @@ def test_subvolume_delete_fails_parameter_commit():
btrfs.subvolume_delete(name="var", commit="maybe")
@patch("salt.modules.btrfs.subvolume_exists")
def test_subvolume_delete_already_missing(subvolume_exists):
def test_subvolume_delete_already_missing():
"""
Test btrfs subvolume delete
"""
subvolume_exists.return_value = False
assert not btrfs.subvolume_delete(name="var", names=["tmp"])
with patch("salt.modules.btrfs.subvolume_exists", return_value=False):
assert not btrfs.subvolume_delete(name="var", names=["tmp"])
@patch("salt.modules.btrfs.subvolume_exists")
def test_subvolume_delete_already_missing_name(subvolume_exists):
def test_subvolume_delete_already_missing_name():
"""
Test btrfs subvolume delete
"""
subvolume_exists.return_value = False
assert not btrfs.subvolume_delete(name="var")
with patch("salt.modules.btrfs.subvolume_exists", return_value=False):
assert not btrfs.subvolume_delete(name="var")
@patch("salt.modules.btrfs.subvolume_exists")
def test_subvolume_delete_already_missing_names(subvolume_exists):
def test_subvolume_delete_already_missing_names():
"""
Test btrfs subvolume delete
"""
subvolume_exists.return_value = False
assert not btrfs.subvolume_delete(names=["tmp"])
with patch("salt.modules.btrfs.subvolume_exists", return_value=False):
assert not btrfs.subvolume_delete(names=["tmp"])
@patch("salt.modules.btrfs.subvolume_exists")
def test_subvolume_delete(subvolume_exists):
def test_subvolume_delete():
"""
Test btrfs subvolume delete
"""
subvolume_exists.return_value = True
salt_mock = {
"cmd.run_all": MagicMock(return_value={"recode": 0}),
}
with patch.dict(btrfs.__salt__, salt_mock):
with patch("salt.modules.btrfs.subvolume_exists", return_value=True), patch.dict(
btrfs.__salt__, salt_mock
):
assert btrfs.subvolume_delete("var", names=["tmp"])
salt_mock["cmd.run_all"].assert_called_once()
salt_mock["cmd.run_all"].assert_called_with(

4
tests/pytests/unit/modules/test_ps.py
View file

@ -9,10 +9,10 @@ import salt.utils.data
from salt.exceptions import SaltInvocationError
from tests.support.mock import MagicMock, Mock, call, patch
# TestCase Exceptions are tested in tests/unit/modules/test_ps.py
psutil = pytest.importorskip("salt.utils.psutil_compat")
# TestCase Exceptions are tested in tests/unit/modules/test_ps.py
@pytest.fixture
def sample_process():

27
tests/pytests/unit/modules/test_yumpkg.py
View file

@ -1969,24 +1969,25 @@ def test_call_yum_default():
)
@patch("salt.utils.systemd.has_scope", MagicMock(return_value=True))
def test_call_yum_in_scope():
"""
Call Yum/Dnf within the scope.
:return:
"""
with patch.dict(yumpkg.__context__, {"yum_bin": "fake-yum"}):
with patch.dict(
yumpkg.__salt__,
{"cmd.run_all": MagicMock(), "config.get": MagicMock(return_value=True)},
):
yumpkg._call_yum(["-y", "--do-something"]) # pylint: disable=W0106
yumpkg.__salt__["cmd.run_all"].assert_called_once_with(
["systemd-run", "--scope", "fake-yum", "-y", "--do-something"],
env={},
output_loglevel="trace",
python_shell=False,
)
with patch(
"salt.utils.systemd.has_scope", MagicMock(return_value=True)
), patch.dict(yumpkg.__context__, {"yum_bin": "fake-yum"}), patch.dict(
yumpkg.__salt__,
{"cmd.run_all": MagicMock(), "config.get": MagicMock(return_value=True)},
):
yumpkg._call_yum(["-y", "--do-something"]) # pylint: disable=W0106
yumpkg.__salt__["cmd.run_all"].assert_called_once_with(
["systemd-run", "--scope", "fake-yum", "-y", "--do-something"],
env={},
output_loglevel="trace",
python_shell=False,
)
def test_call_yum_with_kwargs():

14
tests/pytests/unit/modules/test_zypperpkg.py
View file

@ -232,7 +232,6 @@ def test_pkg_list_holds():
("vim", "1.1", "1.2", [], "vim", [], None),
],
)
@patch.object(zypper, "refresh_db", MagicMock(return_value=True))
def test_upgrade(
package,
pre_version,
@ -242,7 +241,7 @@ def test_upgrade(
pkgs_param,
diff_attr_param,
):
with patch(
with patch.object(zypper, "refresh_db", MagicMock(return_value=True)), patch(
"salt.modules.zypperpkg.__zypper__.noraise.call"
) as zypper_mock, patch.object(
zypper,
@ -276,9 +275,8 @@ def test_upgrade(
("emacs", "1.1", "1.2", ["Dummy", "Dummy2"]),
],
)
@patch.object(zypper, "refresh_db", MagicMock(return_value=True))
def test_dist_upgrade(package, pre_version, post_version, fromrepo_param):
with patch(
with patch.object(zypper, "refresh_db", MagicMock(return_value=True)), patch(
"salt.modules.zypperpkg.__zypper__.noraise.call"
) as zypper_mock, patch.object(
zypper,
@ -302,9 +300,8 @@ def test_dist_upgrade(package, pre_version, post_version, fromrepo_param):
("emacs", "1.1", "1.1", ["Dummy", "Dummy2"]),
],
)
@patch.object(zypper, "refresh_db", MagicMock(return_value=True))
def test_dist_upgrade_dry_run(package, pre_version, post_version, fromrepo_param):
with patch(
with patch.object(zypper, "refresh_db", MagicMock(return_value=True)), patch(
"salt.modules.zypperpkg.__zypper__.noraise.call"
) as zypper_mock, patch.object(
zypper,
@ -323,7 +320,6 @@ def test_dist_upgrade_dry_run(package, pre_version, post_version, fromrepo_param
zypper_mock.assert_any_call(*expected_call)
@patch.object(zypper, "refresh_db", MagicMock(return_value=True))
def test_dist_upgrade_failure():
zypper_output = textwrap.dedent(
"""\
@ -340,7 +336,9 @@ def test_dist_upgrade_failure():
zypper_mock.stderr = ""
zypper_mock.exit_code = 3
zypper_mock.noraise.call = call_spy
with patch("salt.modules.zypperpkg.__zypper__", zypper_mock), patch.object(
with patch.object(zypper, "refresh_db", MagicMock(return_value=True)), patch(
"salt.modules.zypperpkg.__zypper__", zypper_mock
), patch.object(
zypper, "list_pkgs", MagicMock(side_effect=[{"vim": 1.1}, {"vim": 1.1}])
):
expected_call = [

9
tests/pytests/unit/runners/vault/test_app_role_auth.py
View file

@ -50,10 +50,6 @@ def configure_loader_modules():
}
@patch(
"salt.runners.vault._get_policies_cached",
Mock(return_value=["saltstack/minion/test-minion", "saltstack/minions"]),
)
def test_generate_token():
"""
Basic test for test_generate_token with approle (two vault calls)
@ -61,7 +57,10 @@ def test_generate_token():
mock = _mock_json_response(
{"auth": {"client_token": "test", "renewable": False, "lease_duration": 0}}
)
with patch("requests.post", mock):
with patch(
"salt.runners.vault._get_policies_cached",
Mock(return_value=["saltstack/minion/test-minion", "saltstack/minions"]),
), patch("requests.post", mock):
result = vault.generate_token("test-minion", "signature")
log.debug("generate_token result: %s", result)
assert isinstance(result, dict)

14
tests/pytests/unit/runners/vault/test_token_auth.py
View file

@ -33,7 +33,11 @@ def configure_loader_modules():
"salt.runners.vault._get_token_create_url",
MagicMock(return_value="http://fake_url"),
)
with sig_valid_mock, token_url_mock:
cached_policies = patch(
"salt.runners.vault._get_policies_cached",
Mock(return_value=["saltstack/minion/test-minion", "saltstack/minions"]),
)
with sig_valid_mock, token_url_mock, cached_policies:
yield {
vault: {
"__opts__": {
@ -50,10 +54,6 @@ def configure_loader_modules():
}
@patch(
"salt.runners.vault._get_policies_cached",
Mock(return_value=["saltstack/minion/test-minion", "saltstack/minions"]),
)
def test_generate_token():
"""
Basic tests for test_generate_token: all exits
@ -128,10 +128,6 @@ def test_generate_token():
assert result["error"] == "Test Exception Reason"
@patch(
"salt.runners.vault._get_policies_cached",
Mock(return_value=["saltstack/minion/test-minion", "saltstack/minions"]),
)
def test_generate_token_with_namespace():
"""
Basic tests for test_generate_token: all exits

310
tests/pytests/unit/states/test_btrfs.py
View file

@ -20,17 +20,16 @@ def configure_loader_modules():
return {btrfs: {"__salt__": {}, "__states__": {}, "__utils__": {}}}
@patch("salt.states.btrfs._umount")
@patch("tempfile.mkdtemp")
def test__mount_fails(mkdtemp, umount):
def test__mount_fails():
"""
Test mounting a device in a temporary place.
"""
mkdtemp.return_value = "/tmp/xxx"
states_mock = {
"mount.mounted": MagicMock(return_value={"result": False}),
}
with patch.dict(btrfs.__states__, states_mock):
with patch("salt.states.btrfs._umount") as umount, patch(
"tempfile.mkdtemp", return_value="/tmp/xxx"
) as mkdtemp, patch.dict(btrfs.__states__, states_mock):
assert btrfs._mount("/dev/sda1", use_default=False) is None
mkdtemp.assert_called_once()
states_mock["mount.mounted"].assert_called_with(
@ -43,17 +42,16 @@ def test__mount_fails(mkdtemp, umount):
umount.assert_called_with("/tmp/xxx")
@patch("salt.states.btrfs._umount")
@patch("tempfile.mkdtemp")
def test__mount(mkdtemp, umount):
def test__mount():
"""
Test mounting a device in a temporary place.
"""
mkdtemp.return_value = "/tmp/xxx"
states_mock = {
"mount.mounted": MagicMock(return_value={"result": True}),
}
with patch.dict(btrfs.__states__, states_mock):
with patch("salt.states.btrfs._umount") as umount, patch(
"tempfile.mkdtemp", return_value="/tmp/xxx"
) as mkdtemp, patch.dict(btrfs.__states__, states_mock):
assert btrfs._mount("/dev/sda1", use_default=False) == "/tmp/xxx"
mkdtemp.assert_called_once()
states_mock["mount.mounted"].assert_called_with(
@ -66,17 +64,16 @@ def test__mount(mkdtemp, umount):
umount.assert_not_called()
@patch("salt.states.btrfs._umount")
@patch("tempfile.mkdtemp")
def test__mount_use_default(mkdtemp, umount):
def test__mount_use_default():
"""
Test mounting a device in a temporary place.
"""
mkdtemp.return_value = "/tmp/xxx"
states_mock = {
"mount.mounted": MagicMock(return_value={"result": True}),
}
with patch.dict(btrfs.__states__, states_mock):
with patch("salt.states.btrfs._umount") as umount, patch(
"tempfile.mkdtemp", return_value="/tmp/xxx"
) as mkdtemp, patch.dict(btrfs.__states__, states_mock):
assert btrfs._mount("/dev/sda1", use_default=True) == "/tmp/xxx"
mkdtemp.assert_called_once()
states_mock["mount.mounted"].assert_called_with(
@ -193,20 +190,21 @@ def test__unset_cow():
)
@patch("salt.states.btrfs._umount")
@patch("salt.states.btrfs._mount")
def test_subvolume_created_exists(mount, umount):
def test_subvolume_created_exists():
"""
Test creating a subvolume.
"""
mount.return_value = "/tmp/xxx"
salt_mock = {
"btrfs.subvolume_exists": MagicMock(return_value=True),
}
opts_mock = {
"test": False,
}
with patch.dict(btrfs.__salt__, salt_mock), patch.dict(btrfs.__opts__, opts_mock):
with patch("salt.states.btrfs._umount") as umount, patch(
"salt.states.btrfs._mount", return_value="/tmp/xxx"
) as mount, patch.dict(btrfs.__salt__, salt_mock), patch.dict(
btrfs.__opts__, opts_mock
):
assert btrfs.subvolume_created(name="@/var", device="/dev/sda1") == {
"name": "@/var",
"result": True,
@ -218,20 +216,25 @@ def test_subvolume_created_exists(mount, umount):
umount.assert_called_once()
@patch("salt.states.btrfs._umount")
@patch("salt.states.btrfs._mount")
def test_subvolume_created_exists_decorator(mount, umount):
def test_subvolume_created_exists_decorator():
"""
Test creating a subvolume using a non-kwargs call
"""
mount.return_value = "/tmp/xxx"
salt_mock = {
"btrfs.subvolume_exists": MagicMock(return_value=True),
}
opts_mock = {
"test": False,
}
with patch.dict(btrfs.__salt__, salt_mock), patch.dict(btrfs.__opts__, opts_mock):
with patch("salt.states.btrfs._umount") as umount, patch(
"salt.states.btrfs._mount", return_value="/tmp/xxx"
) as mount, patch(
"tempfile.mkdtemp", return_value="/tmp/xxx"
) as mkdtemp, patch.dict(
btrfs.__salt__, salt_mock
), patch.dict(
btrfs.__opts__, opts_mock
):
assert btrfs.subvolume_created("@/var", "/dev/sda1") == {
"name": "@/var",
"result": True,
@ -243,20 +246,21 @@ def test_subvolume_created_exists_decorator(mount, umount):
umount.assert_called_once()
@patch("salt.states.btrfs._umount")
@patch("salt.states.btrfs._mount")
def test_subvolume_created_exists_test(mount, umount):
def test_subvolume_created_exists_test():
"""
Test creating a subvolume.
"""
mount.return_value = "/tmp/xxx"
salt_mock = {
"btrfs.subvolume_exists": MagicMock(return_value=True),
}
opts_mock = {
"test": True,
}
with patch.dict(btrfs.__salt__, salt_mock), patch.dict(btrfs.__opts__, opts_mock):
with patch("salt.states.btrfs._umount") as umount, patch(
"salt.states.btrfs._mount", return_value="/tmp/xxx"
) as mount, patch.dict(btrfs.__salt__, salt_mock), patch.dict(
btrfs.__opts__, opts_mock
):
assert btrfs.subvolume_created(name="@/var", device="/dev/sda1") == {
"name": "@/var",
"result": None,
@ -268,22 +272,25 @@ def test_subvolume_created_exists_test(mount, umount):
umount.assert_called_once()
@patch("salt.states.btrfs._is_default")
@patch("salt.states.btrfs._umount")
@patch("salt.states.btrfs._mount")
def test_subvolume_created_exists_was_default(mount, umount, is_default):
def test_subvolume_created_exists_was_default():
"""
Test creating a subvolume.
"""
mount.return_value = "/tmp/xxx"
is_default.return_value = True
salt_mock = {
"btrfs.subvolume_exists": MagicMock(return_value=True),
}
opts_mock = {
"test": False,
}
with patch.dict(btrfs.__salt__, salt_mock), patch.dict(btrfs.__opts__, opts_mock):
with patch("salt.states.btrfs._is_default", return_value=True), patch(
"salt.states.btrfs._umount"
) as umount, patch(
"salt.states.btrfs._mount", return_value="/tmp/xxx"
) as mount, patch.dict(
btrfs.__salt__, salt_mock
), patch.dict(
btrfs.__opts__, opts_mock
):
assert btrfs.subvolume_created(
name="@/var", device="/dev/sda1", set_default=True
) == {
@ -297,24 +304,25 @@ def test_subvolume_created_exists_was_default(mount, umount, is_default):
umount.assert_called_once()
@patch("salt.states.btrfs._set_default")
@patch("salt.states.btrfs._is_default")
@patch("salt.states.btrfs._umount")
@patch("salt.states.btrfs._mount")
def test_subvolume_created_exists_set_default(mount, umount, is_default, set_default):
def test_subvolume_created_exists_set_default():
"""
Test creating a subvolume.
"""
mount.return_value = "/tmp/xxx"
is_default.return_value = False
set_default.return_value = True
salt_mock = {
"btrfs.subvolume_exists": MagicMock(return_value=True),
}
opts_mock = {
"test": False,
}
with patch.dict(btrfs.__salt__, salt_mock), patch.dict(btrfs.__opts__, opts_mock):
with patch("salt.states.btrfs._is_default", return_value=False), patch(
"salt.states.btrfs._set_default", return_value=True
), patch("salt.states.btrfs._umount") as umount, patch(
"salt.states.btrfs._mount", return_value="/tmp/xxx"
) as mount, patch.dict(
btrfs.__salt__, salt_mock
), patch.dict(
btrfs.__opts__, opts_mock
):
assert btrfs.subvolume_created(
name="@/var", device="/dev/sda1", set_default=True
) == {
@ -328,26 +336,25 @@ def test_subvolume_created_exists_set_default(mount, umount, is_default, set_def
umount.assert_called_once()
@patch("salt.states.btrfs._set_default")
@patch("salt.states.btrfs._is_default")
@patch("salt.states.btrfs._umount")
@patch("salt.states.btrfs._mount")
def test_subvolume_created_exists_set_default_no_force(
mount, umount, is_default, set_default
):
def test_subvolume_created_exists_set_default_no_force():
"""
Test creating a subvolume.
"""
mount.return_value = "/tmp/xxx"
is_default.return_value = False
set_default.return_value = True
salt_mock = {
"btrfs.subvolume_exists": MagicMock(return_value=True),
}
opts_mock = {
"test": False,
}
with patch.dict(btrfs.__salt__, salt_mock), patch.dict(btrfs.__opts__, opts_mock):
with patch("salt.states.btrfs._is_default", return_value=False), patch(
"salt.states.btrfs._set_default", return_value=True
), patch("salt.states.btrfs._umount") as umount, patch(
"salt.states.btrfs._mount", return_value="/tmp/xxx"
) as mount, patch.dict(
btrfs.__salt__, salt_mock
), patch.dict(
btrfs.__opts__, opts_mock
):
assert btrfs.subvolume_created(
name="@/var",
device="/dev/sda1",
@ -364,22 +371,25 @@ def test_subvolume_created_exists_set_default_no_force(
umount.assert_called_once()
@patch("salt.states.btrfs._is_cow")
@patch("salt.states.btrfs._umount")
@patch("salt.states.btrfs._mount")
def test_subvolume_created_exists_no_cow(mount, umount, is_cow):
def test_subvolume_created_exists_no_cow():
"""
Test creating a subvolume.
"""
mount.return_value = "/tmp/xxx"
is_cow.return_value = False
salt_mock = {
"btrfs.subvolume_exists": MagicMock(return_value=True),
}
opts_mock = {
"test": False,
}
with patch.dict(btrfs.__salt__, salt_mock), patch.dict(btrfs.__opts__, opts_mock):
with patch("salt.states.btrfs._is_cow", return_value=False), patch(
"salt.states.btrfs._umount"
) as umount, patch(
"salt.states.btrfs._mount", return_value="/tmp/xxx"
) as mount, patch.dict(
btrfs.__salt__, salt_mock
), patch.dict(
btrfs.__opts__, opts_mock
):
assert btrfs.subvolume_created(
name="@/var", device="/dev/sda1", copy_on_write=False
) == {
@ -393,24 +403,27 @@ def test_subvolume_created_exists_no_cow(mount, umount, is_cow):
umount.assert_called_once()
@patch("salt.states.btrfs._unset_cow")
@patch("salt.states.btrfs._is_cow")
@patch("salt.states.btrfs._umount")
@patch("salt.states.btrfs._mount")
def test_subvolume_created_exists_unset_cow(mount, umount, is_cow, unset_cow):
def test_subvolume_created_exists_unset_cow():
"""
Test creating a subvolume.
"""
mount.return_value = "/tmp/xxx"
is_cow.return_value = True
unset_cow.return_value = True
salt_mock = {
"btrfs.subvolume_exists": MagicMock(return_value=True),
}
opts_mock = {
"test": False,
}
with patch.dict(btrfs.__salt__, salt_mock), patch.dict(btrfs.__opts__, opts_mock):
with patch("salt.states.btrfs._is_cow", return_value=True), patch(
"salt.states.btrfs._unset_cow", return_value=True
), patch("salt.states.btrfs._umount") as umount, patch(
"salt.states.btrfs._mount", return_value="/tmp/xxx"
) as mount, patch.dict(
btrfs.__salt__, salt_mock
), patch.dict(
btrfs.__salt__, salt_mock
), patch.dict(
btrfs.__opts__, opts_mock
):
assert btrfs.subvolume_created(
name="@/var", device="/dev/sda1", copy_on_write=False
) == {
@ -424,13 +437,10 @@ def test_subvolume_created_exists_unset_cow(mount, umount, is_cow, unset_cow):
umount.assert_called_once()
@patch("salt.states.btrfs._umount")
@patch("salt.states.btrfs._mount")
def test_subvolume_created(mount, umount):
def test_subvolume_created():
"""
Test creating a subvolume.
"""
mount.return_value = "/tmp/xxx"
salt_mock = {
"btrfs.subvolume_exists": MagicMock(return_value=False),
"btrfs.subvolume_create": MagicMock(),
@ -441,9 +451,13 @@ def test_subvolume_created(mount, umount):
opts_mock = {
"test": False,
}
with patch.dict(btrfs.__salt__, salt_mock), patch.dict(
with patch("salt.states.btrfs._umount") as umount, patch(
"salt.states.btrfs._mount", return_value="/tmp/xxx"
) as mount, patch.dict(btrfs.__salt__, salt_mock), patch.dict(
btrfs.__states__, states_mock
), patch.dict(btrfs.__opts__, opts_mock):
), patch.dict(
btrfs.__opts__, opts_mock
):
assert btrfs.subvolume_created(name="@/var", device="/dev/sda1") == {
"name": "@/var",
"result": True,
@ -456,13 +470,10 @@ def test_subvolume_created(mount, umount):
umount.assert_called_once()
@patch("salt.states.btrfs._umount")
@patch("salt.states.btrfs._mount")
def test_subvolume_created_fails_directory(mount, umount):
def test_subvolume_created_fails_directory():
"""
Test creating a subvolume.
"""
mount.return_value = "/tmp/xxx"
salt_mock = {
"btrfs.subvolume_exists": MagicMock(return_value=False),
}
@ -472,9 +483,13 @@ def test_subvolume_created_fails_directory(mount, umount):
opts_mock = {
"test": False,
}
with patch.dict(btrfs.__salt__, salt_mock), patch.dict(
with patch("salt.states.btrfs._umount") as umount, patch(
"salt.states.btrfs._mount", return_value="/tmp/xxx"
) as mount, patch.dict(btrfs.__salt__, salt_mock), patch.dict(
btrfs.__states__, states_mock
), patch.dict(btrfs.__opts__, opts_mock):
), patch.dict(
btrfs.__opts__, opts_mock
):
assert btrfs.subvolume_created(name="@/var", device="/dev/sda1") == {
"name": "@/var",
"result": False,
@ -486,13 +501,10 @@ def test_subvolume_created_fails_directory(mount, umount):
umount.assert_called_once()
@patch("salt.states.btrfs._umount")
@patch("salt.states.btrfs._mount")
def test_subvolume_created_fails(mount, umount):
def test_subvolume_created_fails():
"""
Test creating a subvolume.
"""
mount.return_value = "/tmp/xxx"
salt_mock = {
"btrfs.subvolume_exists": MagicMock(return_value=False),
"btrfs.subvolume_create": MagicMock(side_effect=CommandExecutionError),
@ -503,9 +515,13 @@ def test_subvolume_created_fails(mount, umount):
opts_mock = {
"test": False,
}
with patch.dict(btrfs.__salt__, salt_mock), patch.dict(
with patch("salt.states.btrfs._umount") as umount, patch(
"salt.states.btrfs._mount", return_value="/tmp/xxx"
) as mount, patch.dict(btrfs.__salt__, salt_mock), patch.dict(
btrfs.__states__, states_mock
), patch.dict(btrfs.__opts__, opts_mock):
), patch.dict(
btrfs.__opts__, opts_mock
):
assert btrfs.subvolume_created(name="@/var", device="/dev/sda1") == {
"name": "@/var",
"result": False,
@ -599,66 +615,56 @@ def test_diff_properties_emty_na():
assert btrfs._diff_properties(expected, current) == {}
@patch("salt.states.btrfs._umount")
@patch("salt.states.btrfs._mount")
@patch("os.path.exists")
def test_properties_subvolume_not_exists(exists, mount, umount):
def test_properties_subvolume_not_exists():
"""
Test when subvolume is not present
"""
exists.return_value = False
mount.return_value = "/tmp/xxx"
assert btrfs.properties(name="@/var", device="/dev/sda1") == {
"name": "@/var",
"result": False,
"changes": {},
"comment": ["Object @/var not found"],
}
with patch("os.path.exists", return_value=False), patch(
"salt.states.btrfs._umount"
) as umount, patch("salt.states.btrfs._mount", return_value="/tmp/xxx") as mount:
assert btrfs.properties(name="@/var", device="/dev/sda1") == {
"name": "@/var",
"result": False,
"changes": {},
"comment": ["Object @/var not found"],
}
mount.assert_called_once()
umount.assert_called_once()
@patch("salt.states.btrfs._umount")
@patch("salt.states.btrfs._mount")
@patch("os.path.exists")
def test_properties_default_root_subvolume(exists, mount, umount):
def test_properties_default_root_subvolume():
"""
Test when root subvolume resolves to another subvolume
"""
exists.return_value = False
mount.return_value = "/tmp/xxx"
assert btrfs.properties(name="/", device="/dev/sda1") == {
"name": "/",
"result": False,
"changes": {},
"comment": ["Object / not found"],
}
with patch("os.path.exists", return_value=False) as exists, patch(
"salt.states.btrfs._umount"
), patch("salt.states.btrfs._mount", return_value="/tmp/xxx") as mount:
assert btrfs.properties(name="/", device="/dev/sda1") == {
"name": "/",
"result": False,
"changes": {},
"comment": ["Object / not found"],
}
exists.assert_called_with("/tmp/xxx/.")
@patch("os.path.exists")
def test_properties_device_fail(exists):
def test_properties_device_fail():
"""
Test when we try to set a device that is not pressent
"""
exists.return_value = False
assert btrfs.properties(name="/dev/sda1", device=None) == {
"name": "/dev/sda1",
"result": False,
"changes": {},
"comment": ["Object /dev/sda1 not found"],
}
with patch("os.path.exists", return_value=False):
assert btrfs.properties(name="/dev/sda1", device=None) == {
"name": "/dev/sda1",
"result": False,
"changes": {},
"comment": ["Object /dev/sda1 not found"],
}
@patch("salt.states.btrfs._umount")
@patch("salt.states.btrfs._mount")
@patch("os.path.exists")
def test_properties_subvolume_fail(exists, mount, umount):
def test_properties_subvolume_fail():
"""
Test setting a wrong property in a subvolume
"""
exists.return_value = True
mount.return_value = "/tmp/xxx"
salt_mock = {
"btrfs.properties": MagicMock(
side_effect=[
@ -674,7 +680,15 @@ def test_properties_subvolume_fail(exists, mount, umount):
opts_mock = {
"test": False,
}
with patch.dict(btrfs.__salt__, salt_mock), patch.dict(btrfs.__opts__, opts_mock):
with patch("os.path.exists", return_value=True), patch(
"salt.states.btrfs._umount"
) as umount, patch(
"salt.states.btrfs._mount", return_value="/tmp/xxx"
) as mount, patch.dict(
btrfs.__salt__, salt_mock
), patch.dict(
btrfs.__opts__, opts_mock
):
assert btrfs.properties(
name="@/var", device="/dev/sda1", wrond_property=True
) == {
@ -688,15 +702,10 @@ def test_properties_subvolume_fail(exists, mount, umount):
umount.assert_called_once()
@patch("salt.states.btrfs._umount")
@patch("salt.states.btrfs._mount")
@patch("os.path.exists")
def test_properties_enable_ro_subvolume(exists, mount, umount):
def test_properties_enable_ro_subvolume():
"""
Test setting a ro property in a subvolume
"""
exists.return_value = True
mount.return_value = "/tmp/xxx"
salt_mock = {
"btrfs.properties": MagicMock(
side_effect=[
@ -719,7 +728,15 @@ def test_properties_enable_ro_subvolume(exists, mount, umount):
opts_mock = {
"test": False,
}
with patch.dict(btrfs.__salt__, salt_mock), patch.dict(btrfs.__opts__, opts_mock):
with patch("os.path.exists", return_value=True), patch(
"salt.states.btrfs._umount"
) as umount, patch(
"salt.states.btrfs._mount", return_value="/tmp/xxx"
) as mount, patch.dict(
btrfs.__salt__, salt_mock
), patch.dict(
btrfs.__opts__, opts_mock
):
assert btrfs.properties(name="@/var", device="/dev/sda1", ro=True) == {
"name": "@/var",
"result": True,
@ -732,15 +749,10 @@ def test_properties_enable_ro_subvolume(exists, mount, umount):
umount.assert_called_once()
@patch("salt.states.btrfs._umount")
@patch("salt.states.btrfs._mount")
@patch("os.path.exists")
def test_properties_test(exists, mount, umount):
def test_properties_test():
"""
Test setting a property in test mode.
"""
exists.return_value = True
mount.return_value = "/tmp/xxx"
salt_mock = {
"btrfs.properties": MagicMock(
side_effect=[
@ -756,7 +768,15 @@ def test_properties_test(exists, mount, umount):
opts_mock = {
"test": True,
}
with patch.dict(btrfs.__salt__, salt_mock), patch.dict(btrfs.__opts__, opts_mock):
with patch("os.path.exists", return_value=True), patch(
"salt.states.btrfs._umount"
) as umount, patch(
"salt.states.btrfs._mount", return_value="/tmp/xxx"
) as mount, patch.dict(
btrfs.__salt__, salt_mock
), patch.dict(
btrfs.__opts__, opts_mock
):
assert btrfs.properties(name="@/var", device="/dev/sda1", ro=True) == {
"name": "@/var",
"result": None,

8
tests/pytests/unit/test_minion.py
View file

@ -140,13 +140,11 @@ async def test_send_req_async_regression_62453(minion_opts):
assert rtn is False
@patch("salt.channel.client.ReqChannel.factory")
def test_mine_send_tries(req_channel_factory):
def test_mine_send_tries():
channel_enter = MagicMock()
channel_enter.send.side_effect = lambda load, timeout, tries: tries
channel = MagicMock()
channel.__enter__.return_value = channel_enter
req_channel_factory.return_value = channel
opts = {
"random_startup_delay": 0,
@ -154,7 +152,9 @@ def test_mine_send_tries(req_channel_factory):
"return_retry_tries": 20,
"minion_sign_messages": False,
}
with patch("salt.loader.grains"):
with patch("salt.channel.client.ReqChannel.factory", return_value=channel), patch(
"salt.loader.grains"
):
minion = salt.minion.Minion(opts)
minion.tok = "token"

17
tests/pytests/unit/utils/test_verify.py
View file

@ -7,9 +7,7 @@ from tests.support.mock import patch
@pytest.mark.skip_on_windows(reason="Not applicable for Windows.")
@patch("os.chown")
@patch("os.stat")
def test_verify_env_race_condition(mock_stat, mock_chown):
def test_verify_env_race_condition():
def _stat(path):
"""
Helper function for mock_stat, we want to raise errors for specific paths, but not until we get into the proper path.
@ -37,12 +35,15 @@ def test_verify_env_race_condition(mock_stat, mock_chown):
return
mock_stat.side_effect = _stat
mock_chown.side_effect = _chown
with patch("salt.utils.verify._get_pwnam", return_value=(None, None, 0, 0)), patch(
with patch("os.chown", side_effect=_chown) as mock_chown, patch(
"os.stat", side_effect=_stat
) as mock_stat, patch(
"salt.utils.verify._get_pwnam", return_value=(None, None, 0, 0)
), patch(
"os.getuid", return_value=0
), patch("os.listdir", return_value=["subdir"]), patch(
), patch(
"os.listdir", return_value=["subdir"]
), patch(
"os.path.isdir", return_value=True
), patch(
"salt.utils.path.os_walk",