Fixing various functions in the file state module that use user.info (#60502)

* Fixing various functions in the file state module that use user.info to get group information, certain hosts particularly proxy minions do not have the user.info function avaiable.  In those cases we fall back to using the username for the group name.  Also moving the existing file state module unit tests over to pytest and breaking them out into seperate files for easier management.

* Adding changelog.

* Fixing the failing tests for test_hardlink. We want to make sure that user.info is not in __salt__ so the changed logic takes effect.

* Skip the selinux tests unless running on Linux.

* Add a reason for the skip.

* Get results of file.check_perms into different variable then check result in that dictionary if the check_perms failed.  Update tests to mock file.check_perms with more accurate return.

* check_perms in file module returns objects while the file module for windows returns one.  Updating changes and tests

* Mock return from file.check_perms depending on the platform we are running tests on.

* requested changes.  round one.

* requested changes.  round two.

* requested changes.  round three.

* black changes.

* Converting things over to pytest.
This commit is contained in:
Gareth J. Greenaway 2021-08-24 08:43:50 -07:00 committed by GitHub
parent 4c8aad1c03
commit b9fef5dbd2
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
21 changed files with 3548 additions and 3190 deletions

1
changelog/57786.fixed Normal file
View file

@ -0,0 +1 @@
Fixing various functions in the file state module that use user.info to get group information, certain hosts particularly proxy minions do not have the user.info function avaiable.

View file

@ -1390,7 +1390,12 @@ def hardlink(
group = user
if group is None:
group = __salt__["file.gid_to_group"](__salt__["user.info"](user).get("gid", 0))
if "user.info" in __salt__:
group = __salt__["file.gid_to_group"](
__salt__["user.info"](user).get("gid", 0)
)
else:
group = user
preflight_errors = []
uid = __salt__["file.user_to_uid"](user)
@ -1650,7 +1655,12 @@ def symlink(
group = user
if group is None:
group = __salt__["file.gid_to_group"](__salt__["user.info"](user).get("gid", 0))
if "user.info" in __salt__:
group = __salt__["file.gid_to_group"](
__salt__["user.info"](user).get("gid", 0)
)
else:
group = user
preflight_errors = []
if salt.utils.platform.is_windows():
@ -7145,9 +7155,12 @@ def copy_(
group = user
if group is None:
group = __salt__["file.gid_to_group"](
__salt__["user.info"](user).get("gid", 0)
)
if "user.info" in __salt__:
group = __salt__["file.gid_to_group"](
__salt__["user.info"](user).get("gid", 0)
)
else:
group = user
u_check = _check_user(user, group)
if u_check:
@ -7231,9 +7244,14 @@ def copy_(
if not preserve:
if salt.utils.platform.is_windows():
# TODO: Add the other win_* parameters to this function
ret = __salt__["file.check_perms"](path=name, ret=ret, owner=user)
check_ret = __salt__["file.check_perms"](path=name, ret=ret, owner=user)
else:
__salt__["file.check_perms"](name, ret, user, group, mode)
check_ret, perms = __salt__["file.check_perms"](
name, ret, user, group, mode
)
if not check_ret["result"]:
ret["result"] = check_ret["result"]
ret["comment"] = check_ret["comment"]
except OSError:
return _error(ret, 'Failed to copy "{}" to "{}"'.format(source, name))
return ret

View file

@ -44,7 +44,7 @@ salt/modules/augeas_cfg.py:
salt/modules/cp.py:
- unit.modules.test_file
- unit.states.test_file
- pytests.unit.states.file.test_copy
- integration.modules.test_file
- integration.states.test_file

View file

@ -0,0 +1,123 @@
import logging
import os
import pytest
import salt.serializers.json as jsonserializer
import salt.serializers.msgpack as msgpackserializer
import salt.serializers.plist as plistserializer
import salt.serializers.python as pythonserializer
import salt.serializers.yaml as yamlserializer
import salt.states.file as filestate
from salt.exceptions import CommandExecutionError
from tests.support.mock import MagicMock, patch
log = logging.getLogger(__name__)
@pytest.fixture
def configure_loader_modules():
return {
filestate: {
"__env__": "base",
"__salt__": {"file.manage_file": False},
"__serializers__": {
"yaml.serialize": yamlserializer.serialize,
"yaml.seserialize": yamlserializer.serialize,
"python.serialize": pythonserializer.serialize,
"json.serialize": jsonserializer.serialize,
"plist.serialize": plistserializer.serialize,
"msgpack.serialize": msgpackserializer.serialize,
},
"__opts__": {"test": False, "cachedir": ""},
"__instance_id__": "",
"__low__": {},
"__utils__": {},
}
}
# 'absent' function tests: 1
def test_absent():
"""
Test to make sure that the named file or directory is absent.
"""
name = "/fake/file.conf"
ret = {"name": name, "result": False, "comment": "", "changes": {}}
mock_t = MagicMock(return_value=True)
mock_f = MagicMock(return_value=False)
mock_file = MagicMock(side_effect=[True, CommandExecutionError])
mock_tree = MagicMock(side_effect=[True, OSError])
comt = "Must provide name to file.absent"
ret.update({"comment": comt, "name": ""})
with patch.object(os.path, "islink", MagicMock(return_value=False)):
assert filestate.absent("") == ret
with patch.object(os.path, "isabs", mock_f):
comt = "Specified file {} is not an absolute path".format(name)
ret.update({"comment": comt, "name": name})
assert filestate.absent(name) == ret
with patch.object(os.path, "isabs", mock_t):
comt = 'Refusing to make "/" absent'
ret.update({"comment": comt, "name": "/"})
assert filestate.absent("/") == ret
with patch.object(os.path, "isfile", mock_t):
with patch.dict(filestate.__opts__, {"test": True}):
comt = "File {} is set for removal".format(name)
ret.update(
{
"comment": comt,
"name": name,
"result": None,
"changes": {"removed": "/fake/file.conf"},
}
)
assert filestate.absent(name) == ret
with patch.dict(filestate.__opts__, {"test": False}):
with patch.dict(filestate.__salt__, {"file.remove": mock_file}):
comt = "Removed file {}".format(name)
ret.update(
{"comment": comt, "result": True, "changes": {"removed": name}}
)
assert filestate.absent(name) == ret
comt = "Removed file {}".format(name)
ret.update({"comment": "", "result": False, "changes": {}})
assert filestate.absent(name) == ret
with patch.object(os.path, "isfile", mock_f):
with patch.object(os.path, "isdir", mock_t):
with patch.dict(filestate.__opts__, {"test": True}):
comt = "Directory {} is set for removal".format(name)
ret.update(
{"comment": comt, "changes": {"removed": name}, "result": None}
)
assert filestate.absent(name) == ret
with patch.dict(filestate.__opts__, {"test": False}):
with patch.dict(filestate.__salt__, {"file.remove": mock_tree}):
comt = "Removed directory {}".format(name)
ret.update(
{
"comment": comt,
"result": True,
"changes": {"removed": name},
}
)
assert filestate.absent(name) == ret
comt = "Failed to remove directory {}".format(name)
ret.update({"comment": comt, "result": False, "changes": {}})
assert filestate.absent(name) == ret
with patch.object(os.path, "isdir", mock_f):
with patch.dict(filestate.__opts__, {"test": True}):
comt = "File {} is not present".format(name)
ret.update({"comment": comt, "result": True})
assert filestate.absent(name) == ret

View file

@ -0,0 +1,155 @@
import logging
import os
import pytest
import salt.serializers.json as jsonserializer
import salt.serializers.msgpack as msgpackserializer
import salt.serializers.plist as plistserializer
import salt.serializers.python as pythonserializer
import salt.serializers.yaml as yamlserializer
import salt.states.file as filestate
import salt.utils.files
import salt.utils.json
import salt.utils.platform
import salt.utils.win_functions
import salt.utils.yaml
from tests.support.mock import MagicMock, mock_open, patch
log = logging.getLogger(__name__)
@pytest.fixture
def configure_loader_modules():
return {
filestate: {
"__env__": "base",
"__salt__": {"file.manage_file": False},
"__serializers__": {
"yaml.serialize": yamlserializer.serialize,
"yaml.seserialize": yamlserializer.serialize,
"python.serialize": pythonserializer.serialize,
"json.serialize": jsonserializer.serialize,
"plist.serialize": plistserializer.serialize,
"msgpack.serialize": msgpackserializer.serialize,
},
"__opts__": {"test": False, "cachedir": ""},
"__instance_id__": "",
"__low__": {},
"__utils__": {},
}
}
# 'comment' function tests: 1
def test_comment():
"""
Test to comment out specified lines in a file.
"""
with patch.object(os.path, "exists", MagicMock(return_value=True)):
name = "/etc/aliases" if salt.utils.platform.is_darwin() else "/etc/fstab"
regex = "bind 127.0.0.1"
ret = {"name": name, "result": False, "comment": "", "changes": {}}
comt = "Must provide name to file.comment"
ret.update({"comment": comt, "name": ""})
assert filestate.comment("", regex) == ret
mock_t = MagicMock(return_value=True)
mock_f = MagicMock(return_value=False)
with patch.object(os.path, "isabs", mock_f):
comt = "Specified file {} is not an absolute path".format(name)
ret.update({"comment": comt, "name": name})
assert filestate.comment(name, regex) == ret
with patch.object(os.path, "isabs", mock_t):
with patch.dict(
filestate.__salt__,
{"file.search": MagicMock(side_effect=[False, True, False, False])},
):
comt = "Pattern already commented"
ret.update({"comment": comt, "result": True})
assert filestate.comment(name, regex) == ret
comt = "{}: Pattern not found".format(regex)
ret.update({"comment": comt, "result": False})
assert filestate.comment(name, regex) == ret
with patch.dict(
filestate.__salt__,
{
"file.search": MagicMock(side_effect=[True, True, True]),
"file.comment": mock_t,
"file.comment_line": mock_t,
},
):
with patch.dict(filestate.__opts__, {"test": True}):
comt = "File {} is set to be updated".format(name)
ret.update(
{"comment": comt, "result": None, "changes": {name: "updated"}}
)
assert filestate.comment(name, regex) == ret
with patch.dict(filestate.__opts__, {"test": False}):
with patch.object(
salt.utils.files, "fopen", MagicMock(mock_open())
):
comt = "Commented lines successfully"
ret.update({"comment": comt, "result": True, "changes": {}})
assert filestate.comment(name, regex) == ret
# 'uncomment' function tests: 1
def test_uncomment():
"""
Test to uncomment specified commented lines in a file
"""
with patch.object(os.path, "exists", MagicMock(return_value=True)):
name = "/etc/aliases" if salt.utils.platform.is_darwin() else "/etc/fstab"
regex = "bind 127.0.0.1"
ret = {"name": name, "result": False, "comment": "", "changes": {}}
comt = "Must provide name to file.uncomment"
ret.update({"comment": comt, "name": ""})
assert filestate.uncomment("", regex) == ret
mock_t = MagicMock(return_value=True)
mock_f = MagicMock(return_value=False)
mock = MagicMock(side_effect=[False, True, False, False, True, True, True])
with patch.object(os.path, "isabs", mock_f):
comt = "Specified file {} is not an absolute path".format(name)
ret.update({"comment": comt, "name": name})
assert filestate.uncomment(name, regex) == ret
with patch.object(os.path, "isabs", mock_t):
with patch.dict(
filestate.__salt__,
{
"file.search": mock,
"file.uncomment": mock_t,
"file.comment_line": mock_t,
},
):
comt = "Pattern already uncommented"
ret.update({"comment": comt, "result": True})
assert filestate.uncomment(name, regex) == ret
comt = "{}: Pattern not found".format(regex)
ret.update({"comment": comt, "result": False})
assert filestate.uncomment(name, regex) == ret
with patch.dict(filestate.__opts__, {"test": True}):
comt = "File {} is set to be updated".format(name)
ret.update(
{"comment": comt, "result": None, "changes": {name: "updated"}}
)
assert filestate.uncomment(name, regex) == ret
with patch.dict(filestate.__opts__, {"test": False}):
with patch.object(
salt.utils.files, "fopen", MagicMock(mock_open())
):
comt = "Uncommented lines successfully"
ret.update({"comment": comt, "result": True, "changes": {}})
assert filestate.uncomment(name, regex) == ret

View file

@ -0,0 +1,205 @@
import logging
import os
import shutil
import pytest
import salt.serializers.json as jsonserializer
import salt.serializers.msgpack as msgpackserializer
import salt.serializers.plist as plistserializer
import salt.serializers.python as pythonserializer
import salt.serializers.yaml as yamlserializer
import salt.states.file as filestate
import salt.utils.files
import salt.utils.json
import salt.utils.platform
import salt.utils.win_functions
import salt.utils.yaml
from tests.support.mock import MagicMock, patch
log = logging.getLogger(__name__)
@pytest.fixture
def configure_loader_modules():
return {
filestate: {
"__env__": "base",
"__salt__": {"file.manage_file": False},
"__serializers__": {
"yaml.serialize": yamlserializer.serialize,
"yaml.seserialize": yamlserializer.serialize,
"python.serialize": pythonserializer.serialize,
"json.serialize": jsonserializer.serialize,
"plist.serialize": plistserializer.serialize,
"msgpack.serialize": msgpackserializer.serialize,
},
"__opts__": {"test": False, "cachedir": ""},
"__instance_id__": "",
"__low__": {},
"__utils__": {},
}
}
# 'copy' function tests: 1
def test_copy():
"""
Test if the source file exists on the system, copy it to the named file.
"""
name = "/tmp/salt"
source = "/tmp/salt/salt"
user = "salt"
group = "saltstack"
ret = {"name": name, "result": False, "comment": "", "changes": {}}
comt = "Must provide name to file.copy"
ret.update({"comment": comt, "name": ""})
assert filestate.copy_("", source) == ret
mock_t = MagicMock(return_value=True)
mock_f = MagicMock(return_value=False)
mock_uid = MagicMock(side_effect=["", "1000", "1000"])
mock_gid = MagicMock(side_effect=["", "1000", "1000"])
mock_user = MagicMock(return_value=user)
mock_grp = MagicMock(return_value=group)
mock_io = MagicMock(side_effect=IOError)
with patch.object(os.path, "isabs", mock_f):
comt = "Specified file {} is not an absolute path".format(name)
ret.update({"comment": comt, "name": name})
assert filestate.copy_(name, source) == ret
with patch.object(os.path, "isabs", mock_t):
with patch.object(os.path, "exists", mock_f):
comt = 'Source file "{}" is not present'.format(source)
ret.update({"comment": comt, "result": False})
assert filestate.copy_(name, source) == ret
with patch.object(os.path, "exists", mock_t):
with patch.dict(
filestate.__salt__,
{
"file.user_to_uid": mock_uid,
"file.group_to_gid": mock_gid,
"file.get_user": mock_user,
"file.get_group": mock_grp,
"file.get_mode": mock_grp,
"file.check_perms": mock_t,
},
):
# Group argument is ignored on Windows systems. Group is set
# to user
if salt.utils.platform.is_windows():
comt = "User salt is not available Group salt is not available"
else:
comt = "User salt is not available Group saltstack is not available"
ret.update({"comment": comt, "result": False})
assert filestate.copy_(name, source, user=user, group=group) == ret
comt1 = (
'Failed to delete "{}" in preparation for'
" forced move".format(name)
)
comt2 = (
'The target file "{}" exists and will not be '
"overwritten".format(name)
)
comt3 = 'File "{}" is set to be copied to "{}"'.format(source, name)
with patch.object(os.path, "isdir", mock_f):
with patch.object(os.path, "lexists", mock_t):
with patch.dict(filestate.__opts__, {"test": False}):
with patch.dict(
filestate.__salt__, {"file.remove": mock_io}
):
ret.update({"comment": comt1, "result": False})
assert (
filestate.copy_(
name, source, preserve=True, force=True
)
== ret
)
with patch.object(os.path, "isfile", mock_t):
ret.update({"comment": comt2, "result": True})
assert (
filestate.copy_(name, source, preserve=True) == ret
)
with patch.object(os.path, "lexists", mock_f):
with patch.dict(filestate.__opts__, {"test": True}):
ret.update({"comment": comt3, "result": None})
assert filestate.copy_(name, source, preserve=True) == ret
with patch.dict(filestate.__opts__, {"test": False}):
comt = "The target directory /tmp is not present"
ret.update({"comment": comt, "result": False})
assert filestate.copy_(name, source, preserve=True) == ret
check_perms_ret = {
"name": name,
"changes": {},
"comment": [],
"result": True,
}
check_perms_perms = {}
if salt.utils.platform.is_windows():
mock_check_perms = MagicMock(return_value=check_perms_ret)
else:
mock_check_perms = MagicMock(
return_value=(check_perms_ret, check_perms_perms)
)
with patch.dict(
filestate.__salt__,
{
"file.user_to_uid": mock_uid,
"file.group_to_gid": mock_gid,
"file.get_user": mock_user,
"file.get_group": mock_grp,
"file.get_mode": mock_grp,
"file.check_perms": mock_check_perms,
},
):
comt = 'Copied "{}" to "{}"'.format(source, name)
with patch.dict(filestate.__opts__, {"user": "salt"}), patch.object(
os.path, "isdir", mock_t
), patch.object(os.path, "lexists", mock_f), patch.dict(
filestate.__opts__, {"test": False}
), patch.dict(
filestate.__salt__, {"file.remove": mock_io}
), patch.object(
shutil, "copytree", MagicMock()
):
group = None
ret.update(
{
"comment": comt,
"result": True,
"changes": {"/tmp/salt": "/tmp/salt/salt"},
}
)
res = filestate.copy_(name, source, group=group, preserve=False)
assert res == ret
comt = 'Copied "{}" to "{}"'.format(source, name)
with patch.dict(filestate.__opts__, {"user": "salt"}), patch.object(
os.path, "isdir", MagicMock(side_effect=[False, True, False])
), patch.object(os.path, "lexists", mock_f), patch.dict(
filestate.__opts__, {"test": False}
), patch.dict(
filestate.__salt__, {"file.remove": mock_io}
), patch.object(
shutil, "copy", MagicMock()
):
group = None
ret.update(
{
"comment": comt,
"result": True,
"changes": {"/tmp/salt": "/tmp/salt/salt"},
}
)
res = filestate.copy_(name, source, group=group, preserve=False)
assert res == ret

View file

@ -0,0 +1,293 @@
import logging
import os
import pytest
import salt.serializers.json as jsonserializer
import salt.serializers.msgpack as msgpackserializer
import salt.serializers.plist as plistserializer
import salt.serializers.python as pythonserializer
import salt.serializers.yaml as yamlserializer
import salt.states.file as filestate
import salt.utils.files
import salt.utils.json
import salt.utils.platform
import salt.utils.win_functions
import salt.utils.yaml
from salt.exceptions import CommandExecutionError
from tests.support.mock import MagicMock, patch
log = logging.getLogger(__name__)
@pytest.fixture
def configure_loader_modules():
return {
filestate: {
"__env__": "base",
"__salt__": {"file.manage_file": False},
"__serializers__": {
"yaml.serialize": yamlserializer.serialize,
"yaml.seserialize": yamlserializer.serialize,
"python.serialize": pythonserializer.serialize,
"json.serialize": jsonserializer.serialize,
"plist.serialize": plistserializer.serialize,
"msgpack.serialize": msgpackserializer.serialize,
},
"__opts__": {"test": False, "cachedir": ""},
"__instance_id__": "",
"__low__": {},
"__utils__": {},
}
}
# 'directory' function tests: 1
def test_directory():
"""
Test to ensure that a named directory is present and has the right perms
"""
name = "/etc/testdir"
user = "salt"
group = "saltstack"
if salt.utils.platform.is_windows():
name = name.replace("/", "\\")
ret = {"name": name, "result": False, "comment": "", "changes": {}}
check_perms_ret = {"name": name, "result": False, "comment": "", "changes": {}}
comt = "Must provide name to file.directory"
ret.update({"comment": comt, "name": ""})
assert filestate.directory("") == ret
comt = "Cannot specify both max_depth and clean"
ret.update({"comment": comt, "name": name})
assert filestate.directory(name, clean=True, max_depth=2) == ret
mock_t = MagicMock(return_value=True)
mock_f = MagicMock(return_value=False)
if salt.utils.platform.is_windows():
mock_perms = MagicMock(return_value=check_perms_ret)
else:
mock_perms = MagicMock(return_value=(check_perms_ret, ""))
mock_uid = MagicMock(
side_effect=[
"",
"U12",
"U12",
"U12",
"U12",
"U12",
"U12",
"U12",
"U12",
"U12",
"U12",
]
)
mock_gid = MagicMock(
side_effect=[
"",
"G12",
"G12",
"G12",
"G12",
"G12",
"G12",
"G12",
"G12",
"G12",
"G12",
]
)
mock_check = MagicMock(
return_value=(
None,
'The directory "{}" will be changed'.format(name),
{name: {"directory": "new"}},
)
)
mock_error = CommandExecutionError
with patch.dict(
filestate.__salt__,
{
"config.manage_mode": mock_t,
"file.user_to_uid": mock_uid,
"file.group_to_gid": mock_gid,
"file.stats": mock_f,
"file.check_perms": mock_perms,
"file.mkdir": mock_t,
},
), patch("salt.utils.win_dacl.get_sid", mock_error), patch(
"os.path.isdir", mock_t
), patch(
"salt.states.file._check_directory_win", mock_check
):
if salt.utils.platform.is_windows():
comt = ""
else:
comt = "User salt is not available Group saltstack is not available"
ret.update({"comment": comt, "name": name})
assert filestate.directory(name, user=user, group=group) == ret
with patch.object(os.path, "isabs", mock_f):
comt = "Specified file {} is not an absolute path".format(name)
ret.update({"comment": comt})
assert filestate.directory(name, user=user, group=group) == ret
with patch.object(os.path, "isabs", mock_t):
with patch.object(
os.path,
"isfile",
MagicMock(side_effect=[True, True, False, True, True, True, False]),
):
with patch.object(os.path, "lexists", mock_t):
comt = "File exists where the backup target A should go"
ret.update({"comment": comt})
assert (
filestate.directory(
name, user=user, group=group, backupname="A"
)
== ret
)
with patch.object(os.path, "isfile", mock_t):
comt = "Specified location {} exists and is a file".format(name)
ret.update({"comment": comt})
assert filestate.directory(name, user=user, group=group) == ret
with patch.object(os.path, "islink", mock_t):
comt = "Specified location {} exists and is a symlink".format(name)
ret.update({"comment": comt})
assert filestate.directory(name, user=user, group=group) == ret
with patch.object(os.path, "isdir", mock_f):
with patch.dict(filestate.__opts__, {"test": True}):
if salt.utils.platform.is_windows():
comt = 'The directory "{}" will be changed' "".format(name)
else:
comt = (
"The following files will be changed:\n{}:"
" directory - new\n".format(name)
)
ret.update(
{
"comment": comt,
"result": None,
"changes": {name: {"directory": "new"}},
}
)
assert filestate.directory(name, user=user, group=group) == ret
with patch.dict(filestate.__opts__, {"test": False}):
with patch.object(os.path, "isdir", mock_f):
comt = "No directory to create {} in".format(name)
ret.update({"comment": comt, "result": False})
assert filestate.directory(name, user=user, group=group) == ret
if salt.utils.platform.is_windows():
isdir_side_effect = [False, True, False]
else:
isdir_side_effect = [True, False, True, False]
with patch.object(
os.path, "isdir", MagicMock(side_effect=isdir_side_effect)
):
comt = "Failed to create directory {}".format(name)
ret.update(
{
"comment": comt,
"result": False,
"changes": {name: {"directory": "new"}},
}
)
assert filestate.directory(name, user=user, group=group) == ret
check_perms_ret = {
"name": name,
"result": False,
"comment": "",
"changes": {},
}
if salt.utils.platform.is_windows():
mock_perms = MagicMock(return_value=check_perms_ret)
else:
mock_perms = MagicMock(return_value=(check_perms_ret, ""))
recurse = ["silent"]
ret = {
"name": name,
"result": False,
"comment": "Directory /etc/testdir updated",
"changes": {"recursion": "Changes silenced"},
}
if salt.utils.platform.is_windows():
ret["comment"] = ret["comment"].replace("/", "\\")
with patch.dict(
filestate.__salt__, {"file.check_perms": mock_perms}
):
with patch.object(os.path, "isdir", mock_t):
assert (
filestate.directory(
name, user=user, recurse=recurse, group=group
)
== ret
)
check_perms_ret = {
"name": name,
"result": False,
"comment": "",
"changes": {},
}
if salt.utils.platform.is_windows():
mock_perms = MagicMock(return_value=check_perms_ret)
else:
mock_perms = MagicMock(return_value=(check_perms_ret, ""))
recurse = ["ignore_files", "ignore_dirs"]
ret = {
"name": name,
"result": False,
"comment": 'Must not specify "recurse" '
'options "ignore_files" and '
'"ignore_dirs" at the same '
"time.",
"changes": {},
}
with patch.dict(
filestate.__salt__, {"file.check_perms": mock_perms}
):
with patch.object(os.path, "isdir", mock_t):
assert (
filestate.directory(
name, user=user, recurse=recurse, group=group
)
== ret
)
comt = "Directory {} updated".format(name)
ret = {
"name": name,
"result": True,
"comment": comt,
"changes": {"group": "group", "mode": "0777", "user": "user"},
}
check_perms_ret = {
"name": name,
"result": True,
"comment": "",
"changes": {"group": "group", "mode": "0777", "user": "user"},
}
if salt.utils.platform.is_windows():
_mock_perms = MagicMock(return_value=check_perms_ret)
else:
_mock_perms = MagicMock(return_value=(check_perms_ret, ""))
with patch.object(os.path, "isdir", mock_t):
with patch.dict(
filestate.__salt__, {"file.check_perms": _mock_perms}
):
assert (
filestate.directory(name, user=user, group=group) == ret
)

View file

@ -0,0 +1,579 @@
import logging
import os
import plistlib
import pprint
import msgpack
import pytest
import salt.serializers.json as jsonserializer
import salt.serializers.msgpack as msgpackserializer
import salt.serializers.plist as plistserializer
import salt.serializers.python as pythonserializer
import salt.serializers.yaml as yamlserializer
import salt.states.file as filestate
import salt.utils.files
import salt.utils.json
import salt.utils.platform
import salt.utils.win_functions
import salt.utils.yaml
from salt.exceptions import CommandExecutionError
from tests.support.mock import MagicMock, Mock, patch
log = logging.getLogger(__name__)
@pytest.fixture
def configure_loader_modules():
return {
filestate: {
"__env__": "base",
"__salt__": {"file.manage_file": False},
"__serializers__": {
"yaml.serialize": yamlserializer.serialize,
"yaml.seserialize": yamlserializer.serialize,
"python.serialize": pythonserializer.serialize,
"json.serialize": jsonserializer.serialize,
"plist.serialize": plistserializer.serialize,
"msgpack.serialize": msgpackserializer.serialize,
},
"__opts__": {"test": False, "cachedir": ""},
"__instance_id__": "",
"__low__": {},
"__utils__": {},
}
}
def test_serialize():
def returner(contents, *args, **kwargs):
returner.returned = contents
returner.returned = None
with patch.dict(filestate.__salt__, {"file.manage_file": returner}):
dataset = {"foo": True, "bar": 42, "baz": [1, 2, 3], "qux": 2.0}
# If no serializer passed, result should be serialized as YAML
filestate.serialize("/tmp", dataset)
assert salt.utils.yaml.safe_load(returner.returned) == dataset
# If serializer and formatter passed, state should not proceed.
ret = filestate.serialize("/tmp", dataset, serializer="yaml", formatter="json")
assert ret["result"] is False
assert ret["comment"] == "Only one of serializer and formatter are allowed", ret
# YAML
filestate.serialize("/tmp", dataset, serializer="yaml")
assert salt.utils.yaml.safe_load(returner.returned) == dataset
filestate.serialize("/tmp", dataset, formatter="yaml")
assert salt.utils.yaml.safe_load(returner.returned) == dataset
# JSON
filestate.serialize("/tmp", dataset, serializer="json")
assert salt.utils.json.loads(returner.returned) == dataset
filestate.serialize("/tmp", dataset, formatter="json")
assert salt.utils.json.loads(returner.returned) == dataset
# plist
filestate.serialize("/tmp", dataset, serializer="plist")
assert plistlib.loads(returner.returned) == dataset
filestate.serialize("/tmp", dataset, formatter="plist")
assert plistlib.loads(returner.returned) == dataset
# Python
filestate.serialize("/tmp", dataset, serializer="python")
assert returner.returned == pprint.pformat(dataset) + "\n"
filestate.serialize("/tmp", dataset, formatter="python")
assert returner.returned == pprint.pformat(dataset) + "\n"
# msgpack
filestate.serialize("/tmp", dataset, serializer="msgpack")
assert returner.returned == msgpack.packb(dataset)
filestate.serialize("/tmp", dataset, formatter="msgpack")
assert returner.returned == msgpack.packb(dataset)
mock_serializer = Mock(return_value="")
with patch.dict(filestate.__serializers__, {"json.serialize": mock_serializer}):
# Test with "serializer" arg
filestate.serialize(
"/tmp", dataset, formatter="json", serializer_opts=[{"indent": 8}]
)
mock_serializer.assert_called_with(
dataset, indent=8, separators=(",", ": "), sort_keys=True
)
# Test with "formatter" arg
mock_serializer.reset_mock()
filestate.serialize(
"/tmp", dataset, formatter="json", serializer_opts=[{"indent": 8}]
)
mock_serializer.assert_called_with(
dataset, indent=8, separators=(",", ": "), sort_keys=True
)
def test_contents_and_contents_pillar():
def returner(contents, *args, **kwargs):
returner.returned = contents
returner.returned = None
manage_mode_mock = MagicMock()
with patch.dict(
filestate.__salt__,
{"file.manage_file": returner, "config.manage_mode": manage_mode_mock},
):
ret = filestate.managed("/tmp/foo", contents="hi", contents_pillar="foo:bar")
assert not ret["result"]
def test_contents_pillar_doesnt_add_more_newlines():
# make sure the newline
pillar_value = "i am the pillar value{}".format(os.linesep)
returner = MagicMock(return_value=None)
path = "/tmp/foo"
pillar_path = "foo:bar"
# the values don't matter here
pillar_mock = MagicMock(return_value=pillar_value)
with patch.dict(
filestate.__salt__,
{
"file.manage_file": returner,
"config.manage_mode": MagicMock(),
"file.source_list": MagicMock(return_value=[None, None]),
"file.get_managed": MagicMock(return_value=[None, None, None]),
"pillar.get": pillar_mock,
},
):
ret = filestate.managed(path, contents_pillar=pillar_path)
# make sure no errors are returned
assert ret is None
# Make sure the contents value matches the expected value.
# returner.call_args[0] will be an args tuple containing all the args
# passed to the mocked returner for file.manage_file. Any changes to
# the arguments for file.manage_file may make this assertion fail.
# If the test is failing, check the position of the "contents" param
# in the manage_file() function in salt/modules/file.py, the fix is
# likely as simple as updating the 2nd index below.
assert returner.call_args[0][-5] == pillar_value
# 'exists' function tests: 1
def test_exists():
"""
Test to verify that the named file or directory is present or exists.
"""
name = "/etc/grub.conf"
ret = {"name": name, "result": False, "comment": "", "changes": {}}
mock_t = MagicMock(return_value=True)
mock_f = MagicMock(return_value=False)
comt = "Must provide name to file.exists"
ret.update({"comment": comt, "name": ""})
assert filestate.exists("") == ret
with patch.object(os.path, "exists", mock_f):
comt = "Specified path {} does not exist".format(name)
ret.update({"comment": comt, "name": name})
assert filestate.exists(name) == ret
with patch.object(os.path, "exists", mock_t):
comt = "Path {} exists".format(name)
ret.update({"comment": comt, "result": True})
assert filestate.exists(name) == ret
# 'missing' function tests: 1
def test_missing():
"""
Test to verify that the named file or directory is missing.
"""
name = "/etc/grub.conf"
ret = {"name": name, "result": False, "comment": "", "changes": {}}
mock_t = MagicMock(return_value=True)
mock_f = MagicMock(return_value=False)
comt = "Must provide name to file.missing"
ret.update({"comment": comt, "name": "", "changes": {}})
assert filestate.missing("") == ret
with patch.object(os.path, "exists", mock_t):
comt = "Specified path {} exists".format(name)
ret.update({"comment": comt, "name": name})
assert filestate.missing(name) == ret
with patch.object(os.path, "exists", mock_f):
comt = "Path {} is missing".format(name)
ret.update({"comment": comt, "result": True})
assert filestate.missing(name) == ret
# 'recurse' function tests: 1
def test_recurse():
"""
Test to recurse through a subdirectory on the master
and copy said subdirectory over to the specified path.
"""
name = "/opt/code/flask"
source = "salt://code/flask"
user = "salt"
group = "saltstack"
if salt.utils.platform.is_windows():
name = name.replace("/", "\\")
ret = {"name": name, "result": False, "comment": "", "changes": {}}
comt = (
"'mode' is not allowed in 'file.recurse'."
" Please use 'file_mode' and 'dir_mode'."
)
ret.update({"comment": comt})
assert filestate.recurse(name, source, mode="W") == ret
mock_t = MagicMock(return_value=True)
mock_f = MagicMock(return_value=False)
mock_uid = MagicMock(return_value="")
mock_gid = MagicMock(return_value="")
mock_l = MagicMock(return_value=[])
mock_emt = MagicMock(side_effect=[[], ["code/flask"], ["code/flask"]])
mock_lst = MagicMock(
side_effect=[CommandExecutionError, (source, ""), (source, ""), (source, "")]
)
with patch.dict(
filestate.__salt__,
{
"config.manage_mode": mock_t,
"file.user_to_uid": mock_uid,
"file.group_to_gid": mock_gid,
"file.source_list": mock_lst,
"cp.list_master_dirs": mock_emt,
"cp.list_master": mock_l,
},
):
# Group argument is ignored on Windows systems. Group is set to user
if salt.utils.platform.is_windows():
comt = "User salt is not available Group salt is not available"
else:
comt = "User salt is not available Group saltstack is not available"
ret.update({"comment": comt})
assert filestate.recurse(name, source, user=user, group=group) == ret
with patch.object(os.path, "isabs", mock_f):
comt = "Specified file {} is not an absolute path".format(name)
ret.update({"comment": comt})
assert filestate.recurse(name, source) == ret
with patch.object(os.path, "isabs", mock_t):
comt = "Invalid source '1' (must be a salt:// URI)"
ret.update({"comment": comt})
assert filestate.recurse(name, 1) == ret
comt = "Invalid source '//code/flask' (must be a salt:// URI)"
ret.update({"comment": comt})
assert filestate.recurse(name, "//code/flask") == ret
comt = "Recurse failed: "
ret.update({"comment": comt})
assert filestate.recurse(name, source) == ret
comt = (
"The directory 'code/flask' does not exist"
" on the salt fileserver in saltenv 'base'"
)
ret.update({"comment": comt})
assert filestate.recurse(name, source) == ret
with patch.object(os.path, "isdir", mock_f):
with patch.object(os.path, "exists", mock_t):
comt = "The path {} exists and is not a directory".format(name)
ret.update({"comment": comt})
assert filestate.recurse(name, source) == ret
with patch.object(os.path, "isdir", mock_t):
comt = "The directory {} is in the correct state".format(name)
ret.update({"comment": comt, "result": True})
assert filestate.recurse(name, source) == ret
# 'replace' function tests: 1
def test_replace():
"""
Test to maintain an edit in a file.
"""
name = "/etc/grub.conf"
pattern = "CentOS +"
repl = "salt"
ret = {"name": name, "result": False, "comment": "", "changes": {}}
comt = "Must provide name to file.replace"
ret.update({"comment": comt, "name": "", "changes": {}})
assert filestate.replace("", pattern, repl) == ret
mock_t = MagicMock(return_value=True)
mock_f = MagicMock(return_value=False)
with patch.object(os.path, "isabs", mock_f):
comt = "Specified file {} is not an absolute path".format(name)
ret.update({"comment": comt, "name": name})
assert filestate.replace(name, pattern, repl) == ret
with patch.object(os.path, "isabs", mock_t):
with patch.object(os.path, "exists", mock_t):
with patch.dict(filestate.__salt__, {"file.replace": mock_f}):
with patch.dict(filestate.__opts__, {"test": False}):
comt = "No changes needed to be made"
ret.update({"comment": comt, "name": name, "result": True})
assert filestate.replace(name, pattern, repl) == ret
# 'blockreplace' function tests: 1
def test_blockreplace():
"""
Test to maintain an edit in a file in a zone
delimited by two line markers.
"""
with patch("salt.states.file._load_accumulators", MagicMock(return_value=([], []))):
name = "/etc/hosts"
ret = {"name": name, "result": False, "comment": "", "changes": {}}
comt = "Must provide name to file.blockreplace"
ret.update({"comment": comt, "name": ""})
assert filestate.blockreplace("") == ret
mock_t = MagicMock(return_value=True)
mock_f = MagicMock(return_value=False)
with patch.object(os.path, "isabs", mock_f):
comt = "Specified file {} is not an absolute path".format(name)
ret.update({"comment": comt, "name": name})
assert filestate.blockreplace(name) == ret
with patch.object(os.path, "isabs", mock_t), patch.object(
os.path, "exists", mock_t
):
with patch.dict(filestate.__salt__, {"file.blockreplace": mock_t}):
with patch.dict(filestate.__opts__, {"test": True}):
comt = "Changes would be made"
ret.update(
{"comment": comt, "result": None, "changes": {"diff": True}}
)
assert filestate.blockreplace(name) == ret
# 'touch' function tests: 1
def test_touch():
"""
Test to replicate the 'nix "touch" command to create a new empty
file or update the atime and mtime of an existing file.
"""
name = "/var/log/httpd/logrotate.empty"
ret = {"name": name, "result": False, "comment": "", "changes": {}}
comt = "Must provide name to file.touch"
ret.update({"comment": comt, "name": ""})
assert filestate.touch("") == ret
mock_t = MagicMock(return_value=True)
mock_f = MagicMock(return_value=False)
with patch.object(os.path, "isabs", mock_f):
comt = "Specified file {} is not an absolute path".format(name)
ret.update({"comment": comt, "name": name})
assert filestate.touch(name) == ret
with patch.object(os.path, "isabs", mock_t):
with patch.object(os.path, "exists", mock_f):
with patch.dict(filestate.__opts__, {"test": True}):
comt = "File {} is set to be created".format(name)
ret.update({"comment": comt, "result": None, "changes": {"new": name}})
assert filestate.touch(name) == ret
with patch.dict(filestate.__opts__, {"test": False}):
with patch.object(os.path, "isdir", mock_f):
comt = "Directory not present to touch file {}".format(name)
ret.update({"comment": comt, "result": False, "changes": {}})
assert filestate.touch(name) == ret
with patch.object(os.path, "isdir", mock_t):
with patch.dict(filestate.__salt__, {"file.touch": mock_t}):
comt = "Created empty file {}".format(name)
ret.update(
{"comment": comt, "result": True, "changes": {"new": name}}
)
assert filestate.touch(name) == ret
# 'accumulated' function tests: 1
def test_accumulated():
"""
Test to prepare accumulator which can be used in template in file.
"""
with patch(
"salt.states.file._load_accumulators", MagicMock(return_value=({}, {}))
), patch("salt.states.file._persist_accummulators", MagicMock(return_value=True)):
name = "animals_doing_things"
filename = "/tmp/animal_file.txt"
text = " jumps over the lazy dog."
ret = {"name": name, "result": False, "comment": "", "changes": {}}
comt = "Must provide name to file.accumulated"
ret.update({"comment": comt, "name": ""})
assert filestate.accumulated("", filename, text) == ret
comt = "No text supplied for accumulator"
ret.update({"comment": comt, "name": name})
assert filestate.accumulated(name, filename, None) == ret
with patch.dict(
filestate.__low__,
{
"require_in": "file",
"watch_in": "salt",
"__sls__": "SLS",
"__id__": "ID",
},
):
comt = "Orphaned accumulator animals_doing_things in SLS:ID"
ret.update({"comment": comt, "name": name})
assert filestate.accumulated(name, filename, text) == ret
with patch.dict(
filestate.__low__,
{
"require_in": [{"file": "A"}],
"watch_in": [{"B": "C"}],
"__sls__": "SLS",
"__id__": "ID",
},
):
comt = "Accumulator {} for file {} was charged by text".format(
name, filename
)
ret.update({"comment": comt, "name": name, "result": True})
assert filestate.accumulated(name, filename, text) == ret
# 'serialize' function tests: 1
def test_serialize_into_managed_file():
"""
Test to serializes dataset and store it into managed file.
"""
name = "/etc/dummy/package.json"
ret = {"name": name, "result": False, "comment": "", "changes": {}}
comt = "Must provide name to file.serialize"
ret.update({"comment": comt, "name": ""})
assert filestate.serialize("") == ret
mock_t = MagicMock(return_value=True)
mock_f = MagicMock(return_value=False)
with patch.object(os.path, "isfile", mock_f):
comt = "File {} is not present and is not set for creation".format(name)
ret.update({"comment": comt, "name": name, "result": True})
assert filestate.serialize(name, create=False) == ret
comt = "Only one of 'dataset' and 'dataset_pillar' is permitted"
ret.update({"comment": comt, "result": False})
assert filestate.serialize(name, dataset=True, dataset_pillar=True) == ret
comt = "Neither 'dataset' nor 'dataset_pillar' was defined"
ret.update({"comment": comt, "result": False})
assert filestate.serialize(name) == ret
with patch.object(os.path, "isfile", mock_t):
comt = "merge_if_exists is not supported for the python serializer"
ret.update({"comment": comt, "result": False})
assert (
filestate.serialize(
name, dataset=True, merge_if_exists=True, formatter="python"
)
== ret
)
comt = (
"The a serializer could not be found. "
"It either does not exist or its prerequisites are not installed."
)
ret.update({"comment": comt, "result": False})
assert filestate.serialize(name, dataset=True, formatter="A") == ret
mock_changes = MagicMock(return_value=True)
mock_no_changes = MagicMock(return_value=False)
# __opts__['test']=True with changes
with patch.dict(filestate.__salt__, {"file.check_managed_changes": mock_changes}):
with patch.dict(filestate.__opts__, {"test": True}):
comt = "Dataset will be serialized and stored into {}".format(name)
ret.update({"comment": comt, "result": None, "changes": True})
assert filestate.serialize(name, dataset=True, formatter="python") == ret
# __opts__['test']=True without changes
with patch.dict(
filestate.__salt__, {"file.check_managed_changes": mock_no_changes}
):
with patch.dict(filestate.__opts__, {"test": True}):
comt = "The file {} is in the correct state".format(name)
ret.update({"comment": comt, "result": True, "changes": False})
assert filestate.serialize(name, dataset=True, formatter="python") == ret
mock = MagicMock(return_value=ret)
with patch.dict(filestate.__opts__, {"test": False}):
with patch.dict(filestate.__salt__, {"file.manage_file": mock}):
comt = "Dataset will be serialized and stored into {}".format(name)
ret.update({"comment": comt, "result": None})
assert filestate.serialize(name, dataset=True, formatter="python") == ret
# 'mknod' function tests: 1
def test_mknod():
"""
Test to create a special file similar to the 'nix mknod command.
"""
name = "/dev/AA"
ntype = "a"
ret = {"name": name, "result": False, "comment": "", "changes": {}}
comt = "Must provide name to file.mknod"
ret.update({"comment": comt, "name": ""})
assert filestate.mknod("", ntype) == ret
comt = (
"Node type unavailable: 'a'. Available node types are "
"character ('c'), block ('b'), and pipe ('p')"
)
ret.update({"comment": comt, "name": name})
assert filestate.mknod(name, ntype) == ret
# 'mod_run_check_cmd' function tests: 1
def test_mod_run_check_cmd():
"""
Test to execute the check_cmd logic.
"""
cmd = "A"
filename = "B"
ret = {
"comment": "check_cmd execution failed",
"result": False,
"skip_watch": True,
}
mock = MagicMock(side_effect=[{"retcode": 1}, {"retcode": 0}])
with patch.dict(filestate.__salt__, {"cmd.run_all": mock}):
assert filestate.mod_run_check_cmd(cmd, filename) == ret
assert filestate.mod_run_check_cmd(cmd, filename)

View file

@ -0,0 +1,82 @@
import logging
import pytest
import salt.states.file as filestate
log = logging.getLogger(__name__)
@pytest.mark.skip_on_windows(reason="Do not run on Windows")
def test__find_keep_files_unix():
keep = filestate._find_keep_files(
"/test/parent_folder", ["/test/parent_folder/meh.txt"]
)
expected = [
"/",
"/test",
"/test/parent_folder",
"/test/parent_folder/meh.txt",
]
actual = sorted(list(keep))
assert actual == expected, actual
@pytest.mark.skip_unless_on_windows(reason="Do not run except on Windows")
def test__find_keep_files_win32():
"""
Test _find_keep_files. The `_find_keep_files` function is only called by
_clean_dir.
"""
keep = filestate._find_keep_files(
"c:\\test\\parent_folder",
["C:\\test\\parent_folder\\meh-1.txt", "C:\\Test\\Parent_folder\\Meh-2.txt"],
)
expected = [
"c:\\",
"c:\\test",
"c:\\test\\parent_folder",
"c:\\test\\parent_folder\\meh-1.txt",
"c:\\test\\parent_folder\\meh-2.txt",
]
actual = sorted(list(keep))
assert actual == expected
@pytest.mark.skip_unless_on_windows(reason="Do not run except on Windows")
def test__clean_dir_win32():
"""
Test _clean_dir to ensure that regardless of case, we keep all files
requested and do not delete any. Therefore, the expected list should
be empty for this test.
"""
keep = filestate._clean_dir(
"c:\\test\\parent_folder",
[
"C:\\test\\parent_folder\\meh-1.txt",
"C:\\Test\\Parent_folder\\Meh-2.txt",
],
exclude_pat=None,
)
actual = sorted(list(keep))
expected = []
assert actual == expected
@pytest.mark.skip_unless_on_darwin(reason="Do not run except on OS X")
def test__find_keep_files_darwin():
"""
Test _clean_dir to ensure that regardless of case, we keep all files
requested and do not delete any. Therefore, the expected list should
be empty for this test.
"""
keep = filestate._clean_dir(
"/test/parent_folder",
[
"/test/folder/parent_folder/meh-1.txt",
"/Test/folder/Parent_Folder/Meh-2.txt",
],
exclude_pat=None,
)
actual = sorted(list(keep))
expected = []
assert actual == expected

View file

@ -0,0 +1,442 @@
import logging
import os
import pytest
import salt.serializers.json as jsonserializer
import salt.serializers.msgpack as msgpackserializer
import salt.serializers.plist as plistserializer
import salt.serializers.python as pythonserializer
import salt.serializers.yaml as yamlserializer
import salt.states.file as filestate
import salt.utils.files
import salt.utils.json
import salt.utils.platform
import salt.utils.win_functions
import salt.utils.yaml
from salt.exceptions import CommandExecutionError
from tests.support.mock import MagicMock, patch
log = logging.getLogger(__name__)
@pytest.fixture
def configure_loader_modules():
return {
filestate: {
"__env__": "base",
"__salt__": {"file.manage_file": False},
"__serializers__": {
"yaml.serialize": yamlserializer.serialize,
"yaml.seserialize": yamlserializer.serialize,
"python.serialize": pythonserializer.serialize,
"json.serialize": jsonserializer.serialize,
"plist.serialize": plistserializer.serialize,
"msgpack.serialize": msgpackserializer.serialize,
},
"__opts__": {"test": False, "cachedir": ""},
"__instance_id__": "",
"__low__": {},
"__utils__": {},
}
}
@pytest.mark.skip_on_windows(reason="Do not run on Windows")
def test_hardlink(tmp_path):
"""
Test to create a hardlink.
"""
name = str(tmp_path / "testfile.txt")
target = str(tmp_path / "target.txt")
with salt.utils.files.fopen(target, "w") as fp:
fp.write("")
test_dir = str(tmp_path)
user, group = "salt", "saltstack"
def return_val(**kwargs):
res = {
"name": name,
"result": False,
"comment": "",
"changes": {},
}
res.update(kwargs)
return res
mock_t = MagicMock(return_value=True)
mock_f = MagicMock(return_value=False)
mock_empty = MagicMock(return_value="")
mock_uid = MagicMock(return_value="U1001")
mock_gid = MagicMock(return_value="g1001")
mock_nothing = MagicMock(return_value={})
mock_stats = MagicMock(return_value={"inode": 1})
mock_execerror = MagicMock(side_effect=CommandExecutionError)
patches = {}
patches["file.user_to_uid"] = mock_empty
patches["file.group_to_gid"] = mock_empty
patches["user.info"] = mock_empty
patches["file.is_hardlink"] = mock_t
patches["file.stats"] = mock_empty
# Argument validation
with patch.dict(filestate.__salt__, patches):
expected = "Must provide name to file.hardlink"
ret = return_val(comment=expected, name="")
assert filestate.hardlink("", target) == ret
# User validation for dir_mode
with patch.dict(filestate.__salt__, patches), patch.dict(
filestate.__salt__, {"file.user_to_uid": mock_empty}
), patch.dict(filestate.__salt__, {"file.group_to_gid": mock_gid}), patch.object(
os.path, "isabs", mock_t
):
expected = "User {} does not exist".format(user)
ret = return_val(comment=expected, name=name)
assert filestate.hardlink(name, target, user=user, group=group) == ret
# Group validation for dir_mode
with patch.dict(filestate.__salt__, patches), patch.dict(
filestate.__salt__, {"file.user_to_uid": mock_uid}
), patch.dict(filestate.__salt__, {"file.group_to_gid": mock_empty}), patch.object(
os.path, "isabs", mock_t
):
expected = "Group {} does not exist".format(group)
ret = return_val(comment=expected, name=name)
assert filestate.hardlink(name, target, user=user, group=group) == ret
# Absolute path for name
nonabs = "./non-existent-path/to/non-existent-file"
with patch.dict(filestate.__salt__, patches), patch.dict(
filestate.__salt__, {"file.user_to_uid": mock_uid}
), patch.dict(filestate.__salt__, {"file.group_to_gid": mock_gid}):
expected = "Specified file {} is not an absolute path".format(nonabs)
ret = return_val(comment=expected, name=nonabs)
assert filestate.hardlink(nonabs, target, user=user, group=group) == ret
# Absolute path for target
with patch.dict(filestate.__salt__, patches), patch.dict(
filestate.__salt__, {"file.user_to_uid": mock_uid}
), patch.dict(filestate.__salt__, {"file.group_to_gid": mock_gid}):
expected = "Specified target {} is not an absolute path".format(nonabs)
ret = return_val(comment=expected, name=name)
assert filestate.hardlink(name, nonabs, user=user, group=group) == ret
# Test option -- nonexistent target
with patch.dict(filestate.__salt__, patches), patch.dict(
filestate.__salt__, {"file.user_to_uid": mock_uid}
), patch.dict(filestate.__salt__, {"file.group_to_gid": mock_gid}), patch.object(
os.path, "exists", mock_f
), patch.dict(
filestate.__opts__, {"test": True}
):
expected = "Target {} for hard link does not exist".format(target)
ret = return_val(comment=expected, name=name)
assert filestate.hardlink(name, target, user=user, group=group) == ret
# Test option -- target is a directory
with patch.dict(filestate.__salt__, patches), patch.dict(
filestate.__salt__, {"file.user_to_uid": mock_uid}
), patch.dict(filestate.__salt__, {"file.group_to_gid": mock_gid}), patch.object(
os.path, "exists", mock_t
), patch.dict(
filestate.__opts__, {"test": True}
):
expected = "Unable to hard link from directory {}".format(test_dir)
ret = return_val(comment=expected, name=name)
assert filestate.hardlink(name, test_dir, user=user, group=group) == ret
# Test option -- name is a directory
with patch.dict(filestate.__salt__, patches), patch.dict(
filestate.__salt__, {"file.user_to_uid": mock_uid}
), patch.dict(filestate.__salt__, {"file.group_to_gid": mock_gid}), patch.dict(
filestate.__opts__, {"test": True}
):
expected = "Unable to hard link to directory {}".format(test_dir)
ret = return_val(comment=expected, name=test_dir)
assert filestate.hardlink(test_dir, target, user=user, group=group) == ret
# Test option -- name does not exist
with patch.dict(filestate.__salt__, patches), patch.dict(
filestate.__salt__, {"file.user_to_uid": mock_uid}
), patch.dict(filestate.__salt__, {"file.group_to_gid": mock_gid}), patch.dict(
filestate.__opts__, {"test": True}
):
expected = "Hard link {} to {} is set for creation".format(name, target)
changes = dict(new=name)
ret = return_val(result=None, comment=expected, name=name, changes=changes)
assert filestate.hardlink(name, target, user=user, group=group) == ret
# Test option -- hardlink matches
with patch.dict(filestate.__salt__, patches), patch.dict(
filestate.__salt__, {"file.user_to_uid": mock_uid}
), patch.dict(filestate.__salt__, {"file.group_to_gid": mock_gid}), patch.dict(
filestate.__salt__, {"file.is_hardlink": mock_t}
), patch.dict(
filestate.__salt__, {"file.stats": mock_stats}
), patch.object(
os.path, "exists", mock_t
), patch.dict(
filestate.__opts__, {"test": True}
):
expected = "The hard link {} is presently targetting {}".format(name, target)
ret = return_val(result=True, comment=expected, name=name)
assert filestate.hardlink(name, target, user=user, group=group) == ret
# Test option -- hardlink does not match
with patch.dict(filestate.__salt__, patches), patch.dict(
filestate.__salt__, {"file.user_to_uid": mock_uid}
), patch.dict(filestate.__salt__, {"file.group_to_gid": mock_gid}), patch.dict(
filestate.__salt__, {"file.is_hardlink": mock_t}
), patch.dict(
filestate.__salt__, {"file.stats": mock_nothing}
), patch.object(
os.path, "exists", mock_t
), patch.dict(
filestate.__opts__, {"test": True}
):
expected = "Link {} target is set to be changed to {}".format(name, target)
changes = dict(change=name)
ret = return_val(result=None, comment=expected, name=name, changes=changes)
assert filestate.hardlink(name, target, user=user, group=group) == ret
# Test option -- force removal
with patch.dict(filestate.__salt__, patches), patch.dict(
filestate.__salt__, {"file.user_to_uid": mock_uid}
), patch.dict(filestate.__salt__, {"file.group_to_gid": mock_gid}), patch.dict(
filestate.__salt__, {"file.is_hardlink": mock_f}
), patch.object(
os.path, "exists", mock_t
), patch.dict(
filestate.__opts__, {"test": True}
):
expected = (
"The file or directory {} is set for removal to "
"make way for a new hard link targeting {}".format(name, target)
)
ret = return_val(result=None, comment=expected, name=name)
assert (
filestate.hardlink(name, target, force=True, user=user, group=group) == ret
)
# Test option -- without force removal
with patch.dict(filestate.__salt__, patches), patch.dict(
filestate.__salt__, {"file.user_to_uid": mock_uid}
), patch.dict(filestate.__salt__, {"file.group_to_gid": mock_gid}), patch.dict(
filestate.__salt__, {"file.is_hardlink": mock_f}
), patch.object(
os.path, "exists", mock_t
), patch.dict(
filestate.__opts__, {"test": True}
):
expected = (
"File or directory exists where the hard link {} "
"should be. Did you mean to use force?".format(name)
)
ret = return_val(result=False, comment=expected, name=name)
assert (
filestate.hardlink(name, target, force=False, user=user, group=group) == ret
)
# Target is a directory
with patch.dict(filestate.__salt__, patches), patch.dict(
filestate.__salt__, {"file.user_to_uid": mock_uid}
), patch.dict(filestate.__salt__, {"file.group_to_gid": mock_gid}):
expected = "Unable to hard link from directory {}".format(test_dir)
ret = return_val(comment=expected, name=name)
assert filestate.hardlink(name, test_dir, user=user, group=group) == ret
# Name is a directory
with patch.dict(filestate.__salt__, patches), patch.dict(
filestate.__salt__, {"file.user_to_uid": mock_uid}
), patch.dict(filestate.__salt__, {"file.group_to_gid": mock_gid}):
expected = "Unable to hard link to directory {}".format(test_dir)
ret = return_val(comment=expected, name=test_dir)
assert filestate.hardlink(test_dir, target, user=user, group=group) == ret
# Try overwrite file with link
with patch.dict(filestate.__salt__, patches), patch.dict(
filestate.__salt__, {"file.user_to_uid": mock_uid}
), patch.dict(filestate.__salt__, {"file.group_to_gid": mock_gid}), patch.dict(
filestate.__salt__, {"file.is_hardlink": mock_f}
), patch.object(
os.path, "isfile", mock_t
):
expected = "File exists where the hard link {} should be".format(name)
ret = return_val(comment=expected, name=name)
assert filestate.hardlink(name, target, user=user, group=group) == ret
# Try overwrite link with same
with patch.dict(filestate.__salt__, patches), patch.dict(
filestate.__salt__, {"file.user_to_uid": mock_uid}
), patch.dict(filestate.__salt__, {"file.group_to_gid": mock_gid}), patch.dict(
filestate.__salt__, {"file.is_hardlink": mock_t}
), patch.dict(
filestate.__salt__, {"file.stats": mock_stats}
), patch.object(
os.path, "isfile", mock_f
):
expected = "Target of hard link {} is already pointing to {}".format(
name, target
)
ret = return_val(result=True, comment=expected, name=name)
assert filestate.hardlink(name, target, user=user, group=group) == ret
# Really overwrite link with same
with patch.dict(filestate.__salt__, patches), patch.dict(
filestate.__salt__, {"file.user_to_uid": mock_uid}
), patch.dict(filestate.__salt__, {"file.group_to_gid": mock_gid}), patch.dict(
filestate.__salt__, {"file.is_hardlink": mock_t}
), patch.dict(
filestate.__salt__, {"file.link": mock_t}
), patch.dict(
filestate.__salt__, {"file.stats": mock_nothing}
), patch.object(
os, "remove", mock_t
), patch.object(
os.path, "isfile", mock_f
):
expected = "Set target of hard link {} -> {}".format(name, target)
changes = dict(new=name)
ret = return_val(result=True, comment=expected, name=name, changes=changes)
assert filestate.hardlink(name, target, user=user, group=group) == ret
# Fail at overwriting link with same
with patch.dict(filestate.__salt__, patches), patch.dict(
filestate.__salt__, {"file.user_to_uid": mock_uid}
), patch.dict(filestate.__salt__, {"file.group_to_gid": mock_gid}), patch.dict(
filestate.__salt__, {"file.is_hardlink": mock_t}
), patch.dict(
filestate.__salt__, {"file.link": mock_execerror}
), patch.dict(
filestate.__salt__, {"file.stats": mock_nothing}
), patch.object(
os, "remove", mock_t
), patch.object(
os.path, "isfile", mock_f
):
expected = "Unable to set target of hard link {} -> {}: {}".format(
name, target, ""
)
ret = return_val(result=False, comment=expected, name=name)
assert filestate.hardlink(name, target, user=user, group=group) == ret
# Make new link
with patch.dict(filestate.__salt__, patches), patch.dict(
filestate.__salt__, {"file.user_to_uid": mock_uid}
), patch.dict(filestate.__salt__, {"file.group_to_gid": mock_gid}), patch.dict(
filestate.__salt__, {"file.is_hardlink": mock_f}
), patch.dict(
filestate.__salt__, {"file.link": mock_f}
), patch.dict(
filestate.__salt__, {"file.stats": mock_nothing}
), patch.object(
os, "remove", mock_t
), patch.object(
os.path, "isfile", mock_f
):
expected = "Created new hard link {} -> {}".format(name, target)
changes = dict(new=name)
ret = return_val(result=True, comment=expected, name=name, changes=changes)
assert filestate.hardlink(name, target, user=user, group=group) == ret
# Fail while making new link
with patch.dict(filestate.__salt__, patches), patch.dict(
filestate.__salt__, {"file.user_to_uid": mock_uid}
), patch.dict(filestate.__salt__, {"file.group_to_gid": mock_gid}), patch.dict(
filestate.__salt__, {"file.is_hardlink": mock_f}
), patch.dict(
filestate.__salt__, {"file.link": mock_execerror}
), patch.dict(
filestate.__salt__, {"file.stats": mock_nothing}
), patch.object(
os, "remove", mock_t
), patch.object(
os.path, "isfile", mock_f
):
expected = "Unable to create new hard link {} -> {}: {}".format(
name, target, ""
)
ret = return_val(result=False, comment=expected, name=name)
assert filestate.hardlink(name, target, user=user, group=group) == ret
# Force making new link over file
with patch.dict(filestate.__salt__, patches), patch.dict(
filestate.__salt__, {"file.user_to_uid": mock_uid}
), patch.dict(filestate.__salt__, {"file.group_to_gid": mock_gid}), patch.dict(
filestate.__salt__, {"file.is_hardlink": mock_f}
), patch.dict(
filestate.__salt__, {"file.link": mock_t}
), patch.dict(
filestate.__salt__, {"file.stats": mock_nothing}
), patch.object(
os, "remove", mock_t
), patch.object(
os.path, "isfile", mock_t
):
expected = "Created new hard link {} -> {}".format(name, target)
changes = dict(new=name)
changes["forced"] = "File for hard link was forcibly replaced"
ret = return_val(result=True, comment=expected, name=name, changes=changes)
assert (
filestate.hardlink(name, target, user=user, force=True, group=group) == ret
)
# Force making new link over file but error out
with patch.dict(filestate.__salt__, patches), patch.dict(
filestate.__salt__, {"file.user_to_uid": mock_uid}
), patch.dict(filestate.__salt__, {"file.group_to_gid": mock_gid}), patch.dict(
filestate.__salt__, {"file.is_hardlink": mock_f}
), patch.dict(
filestate.__salt__, {"file.link": mock_execerror}
), patch.dict(
filestate.__salt__, {"file.stats": mock_nothing}
), patch.object(
os, "remove", mock_t
), patch.object(
os.path, "isfile", mock_t
):
expected = "Unable to create new hard link {} -> {}: {}".format(
name, target, ""
)
changes = dict(forced="File for hard link was forcibly replaced")
ret = return_val(result=False, comment=expected, name=name, changes=changes)
assert (
filestate.hardlink(name, target, user=user, force=True, group=group) == ret
)
patches = {}
patches["file.user_to_uid"] = mock_empty
patches["file.group_to_gid"] = mock_empty
patches["file.is_hardlink"] = mock_t
patches["file.stats"] = mock_empty
# Make new link when group is None and file.gid_to_group is unavailable
with patch.dict(filestate.__salt__, patches), patch.dict(
filestate.__salt__, {"file.user_to_uid": mock_uid}
), patch.dict(filestate.__salt__, {"file.group_to_gid": mock_gid}), patch.dict(
filestate.__salt__, {"file.is_hardlink": mock_f}
), patch.dict(
filestate.__salt__, {"file.link": mock_f}
), patch.dict(
filestate.__salt__, {"file.stats": mock_nothing}
), patch.object(
os, "remove", mock_t
), patch.object(
os.path, "isfile", mock_f
):
group = None
expected = "Created new hard link {} -> {}".format(name, target)
changes = dict(new=name)
ret = return_val(result=True, comment=expected, name=name, changes=changes)
assert filestate.hardlink(name, target, user=user, group=group) == ret

View file

@ -0,0 +1,121 @@
import collections
import logging
import pytest
import salt.serializers.json as jsonserializer
import salt.serializers.msgpack as msgpackserializer
import salt.serializers.plist as plistserializer
import salt.serializers.python as pythonserializer
import salt.serializers.yaml as yamlserializer
import salt.states.file as filestate
from tests.support.helpers import dedent
log = logging.getLogger(__name__)
@pytest.fixture
def configure_loader_modules():
return {
filestate: {
"__env__": "base",
"__salt__": {"file.manage_file": False},
"__serializers__": {
"yaml.serialize": yamlserializer.serialize,
"yaml.seserialize": yamlserializer.serialize,
"python.serialize": pythonserializer.serialize,
"json.serialize": jsonserializer.serialize,
"plist.serialize": plistserializer.serialize,
"msgpack.serialize": msgpackserializer.serialize,
},
"__opts__": {"test": False, "cachedir": ""},
"__instance_id__": "",
"__low__": {},
"__utils__": {},
}
}
def test_file_keyvalue_key_values(tmp_path):
"""
test file.keyvalue when using key_values kwarg
"""
contents = dedent(
"""\
#PermitRootLogin prohibit-password
#StrictMode yes
"""
)
with pytest.helpers.temp_file(
"tempfile", contents=contents, directory=tmp_path
) as tempfile:
ret = filestate.keyvalue(
name=str(tempfile),
key_values=collections.OrderedDict(PermitRootLogin="yes"),
separator=" ",
uncomment="#",
key_ignore_case=True,
)
f_contents = tempfile.read_text()
assert "PermitRootLogin yes" in f_contents
assert "#StrictMode yes" in f_contents
def test_file_keyvalue_empty(tmp_path):
"""
test file.keyvalue when key_values is empty
"""
contents = dedent(
"""\
#PermitRootLogin prohibit-password
#StrictMode yes
"""
)
with pytest.helpers.temp_file(
"tempfile", contents=contents, directory=tmp_path
) as tempfile:
ret = filestate.keyvalue(
name=str(tempfile),
key_values={},
separator=" ",
uncomment="#",
key_ignore_case=True,
)
assert (
ret["comment"]
== "file.keyvalue key and value not supplied and key_values is empty"
)
f_contents = tempfile.read_text()
assert "PermitRootLogin yes" not in f_contents
assert "#StrictMode yes" in f_contents
def test_file_keyvalue_not_dict(tmp_path):
"""
test file.keyvalue when key_values not a dict
"""
contents = dedent(
"""\
#PermitRootLogin prohibit-password
#StrictMode yes
"""
)
with pytest.helpers.temp_file(
"tempfile", contents=contents, directory=tmp_path
) as tempfile:
ret = filestate.keyvalue(
name=str(tempfile),
key_values=["PermiteRootLogin", "yes"],
separator=" ",
uncomment="#",
key_ignore_case=True,
)
assert (
ret["comment"]
== "file.keyvalue key and value not supplied and key_values is not a dictionary"
)
f_contents = tempfile.read_text()
assert "PermitRootLogin yes" not in f_contents
assert "#StrictMode yes" in f_contents

View file

@ -0,0 +1,375 @@
import logging
import os
import pytest
import salt.serializers.json as jsonserializer
import salt.serializers.msgpack as msgpackserializer
import salt.serializers.plist as plistserializer
import salt.serializers.python as pythonserializer
import salt.serializers.yaml as yamlserializer
import salt.states.file as filestate
import salt.utils.files
import salt.utils.json
import salt.utils.platform
import salt.utils.win_functions
import salt.utils.yaml
from salt.exceptions import CommandExecutionError
from tests.support.mock import MagicMock, patch
log = logging.getLogger(__name__)
@pytest.fixture
def configure_loader_modules():
return {
filestate: {
"__env__": "base",
"__salt__": {"file.manage_file": False},
"__serializers__": {
"yaml.serialize": yamlserializer.serialize,
"yaml.seserialize": yamlserializer.serialize,
"python.serialize": pythonserializer.serialize,
"json.serialize": jsonserializer.serialize,
"plist.serialize": plistserializer.serialize,
"msgpack.serialize": msgpackserializer.serialize,
},
"__opts__": {"test": False, "cachedir": ""},
"__instance_id__": "",
"__low__": {},
"__utils__": {},
}
}
# 'managed' function tests: 1
def test_file_managed_should_fall_back_to_binary():
expected_contents = b"\x8b"
filename = "/tmp/blarg"
mock_manage = MagicMock(return_value={"fnord": "fnords"})
with patch("salt.states.file._load_accumulators", MagicMock(return_value=([], []))):
with patch.dict(
filestate.__salt__,
{
"file.get_managed": MagicMock(return_value=["", "", ""]),
"file.source_list": MagicMock(return_value=["", ""]),
"file.manage_file": mock_manage,
"pillar.get": MagicMock(return_value=expected_contents),
},
):
ret = filestate.managed(filename, contents_pillar="fnord", encoding="utf-8")
actual_contents = mock_manage.call_args[0][14]
assert actual_contents == expected_contents
def test_managed():
"""
Test to manage a given file, this function allows for a file to be
downloaded from the salt master and potentially run through a templating
system.
"""
with patch("salt.states.file._load_accumulators", MagicMock(return_value=([], []))):
name = "/etc/grub.conf"
user = "salt"
group = "saltstack"
ret = {"name": name, "result": False, "comment": "", "changes": {}}
mock_t = MagicMock(return_value=True)
mock_f = MagicMock(return_value=False)
mock_cmd_fail = MagicMock(return_value={"retcode": 1})
mock_uid = MagicMock(
side_effect=[
"",
"U12",
"U12",
"U12",
"U12",
"U12",
"U12",
"U12",
"U12",
"U12",
"U12",
"U12",
"U12",
"U12",
"U12",
"U12",
"U12",
]
)
mock_gid = MagicMock(
side_effect=[
"",
"G12",
"G12",
"G12",
"G12",
"G12",
"G12",
"G12",
"G12",
"G12",
"G12",
"G12",
"G12",
"G12",
"G12",
"G12",
"G12",
]
)
mock_if = MagicMock(
side_effect=[True, False, False, False, False, False, False, False, False]
)
if salt.utils.platform.is_windows():
mock_ret = MagicMock(return_value=ret)
else:
mock_ret = MagicMock(return_value=(ret, None))
mock_dict = MagicMock(return_value={})
mock_cp = MagicMock(side_effect=[Exception, True])
mock_ex = MagicMock(
side_effect=[Exception, {"changes": {name: name}}, True, Exception]
)
mock_mng = MagicMock(
side_effect=[
Exception,
("", "", ""),
("", "", ""),
("", "", True),
("", "", True),
("", "", ""),
("", "", ""),
("", "", ""),
]
)
mock_file = MagicMock(
side_effect=[
CommandExecutionError,
("", ""),
("", ""),
("", ""),
("", ""),
("", ""),
("", ""),
("", ""),
("", ""),
("", ""),
]
)
with patch.dict(
filestate.__salt__,
{
"config.manage_mode": mock_t,
"file.user_to_uid": mock_uid,
"file.group_to_gid": mock_gid,
"file.file_exists": mock_if,
"file.check_perms": mock_ret,
"file.check_managed_changes": mock_dict,
"file.get_managed": mock_mng,
"file.source_list": mock_file,
"file.copy": mock_cp,
"file.manage_file": mock_ex,
"cmd.run_all": mock_cmd_fail,
},
):
comt = "Destination file name is required"
ret.update({"comment": comt, "name": "", "changes": {}})
assert filestate.managed("") == ret
with patch.object(os.path, "isfile", mock_f):
comt = "File {} is not present and is not set for creation".format(name)
ret.update({"comment": comt, "name": name, "result": True})
assert filestate.managed(name, create=False) == ret
# Group argument is ignored on Windows systems. Group is set to
# user
if salt.utils.platform.is_windows():
comt = "User salt is not available Group salt is not available"
else:
comt = "User salt is not available Group saltstack is not available"
ret.update({"comment": comt, "result": False})
assert filestate.managed(name, user=user, group=group) == ret
with patch.object(os.path, "isabs", mock_f):
comt = "Specified file {} is not an absolute path".format(name)
ret.update({"comment": comt, "result": False})
assert filestate.managed(name, user=user, group=group) == ret
with patch.object(os.path, "isabs", mock_t):
with patch.object(os.path, "isdir", mock_t):
comt = "Specified target {} is a directory".format(name)
ret.update({"comment": comt})
assert filestate.managed(name, user=user, group=group) == ret
with patch.object(os.path, "isdir", mock_f):
comt = "Context must be formed as a dict"
ret.update({"comment": comt})
assert (
filestate.managed(name, user=user, group=group, context=True)
== ret
)
comt = "Defaults must be formed as a dict"
ret.update({"comment": comt})
assert (
filestate.managed(name, user=user, group=group, defaults=True)
== ret
)
comt = (
"Only one of 'contents', 'contents_pillar', "
"and 'contents_grains' is permitted"
)
ret.update({"comment": comt})
assert (
filestate.managed(
name,
user=user,
group=group,
contents="A",
contents_grains="B",
contents_pillar="C",
)
== ret
)
with patch.object(os.path, "exists", mock_t):
with patch.dict(filestate.__opts__, {"test": True}):
comt = "File {} not updated".format(name)
ret.update({"comment": comt})
assert (
filestate.managed(
name, user=user, group=group, replace=False
)
== ret
)
comt = "The file {} is in the correct state".format(name)
ret.update({"comment": comt, "result": True})
assert (
filestate.managed(
name, user=user, contents="A", group=group
)
== ret
)
with patch.object(os.path, "exists", mock_f):
with patch.dict(filestate.__opts__, {"test": False}):
comt = "Unable to manage file: "
ret.update({"comment": comt, "result": False})
assert (
filestate.managed(
name, user=user, group=group, contents="A"
)
== ret
)
comt = "Unable to manage file: "
ret.update({"comment": comt, "result": False})
assert (
filestate.managed(
name, user=user, group=group, contents="A"
)
== ret
)
with patch.object(
salt.utils.files, "mkstemp", return_value=name
):
comt = "Unable to copy file {0} to {0}: ".format(name)
ret.update({"comment": comt, "result": False})
assert (
filestate.managed(
name, user=user, group=group, check_cmd="A"
)
== ret
)
comt = "Unable to check_cmd file: "
ret.update({"comment": comt, "result": False})
assert (
filestate.managed(
name, user=user, group=group, check_cmd="A"
)
== ret
)
comt = "check_cmd execution failed"
ret.update(
{"comment": comt, "result": False, "skip_watch": True}
)
assert (
filestate.managed(
name, user=user, group=group, check_cmd="A"
)
== ret
)
comt = "check_cmd execution failed"
ret.update({"comment": True, "changes": {}})
ret.pop("skip_watch", None)
assert (
filestate.managed(name, user=user, group=group) == ret
)
assert filestate.managed(name, user=user, group=group)
comt = "Unable to manage file: "
ret.update({"comment": comt})
assert (
filestate.managed(name, user=user, group=group) == ret
)
if salt.utils.platform.is_windows():
mock_ret = MagicMock(return_value=ret)
comt = "File {} not updated".format(name)
else:
perms = {"luser": user, "lmode": "0644", "lgroup": group}
mock_ret = MagicMock(return_value=(ret, perms))
comt = (
"File {} will be updated with "
"permissions 0400 from its current "
"state of 0644".format(name)
)
with patch.dict(filestate.__salt__, {"file.check_perms": mock_ret}):
with patch.object(os.path, "exists", mock_t):
with patch.dict(filestate.__opts__, {"test": True}):
ret.update({"comment": comt})
if salt.utils.platform.is_windows():
assert (
filestate.managed(name, user=user, group=group)
== ret
)
else:
assert (
filestate.managed(
name, user=user, group=group, mode=400
)
== ret
)
# Replace is False, test is True, mode is empty
# should return "File not updated"
# https://github.com/saltstack/salt/issues/59276
if salt.utils.platform.is_windows():
mock_ret = MagicMock(return_value=ret)
else:
perms = {"luser": user, "lmode": "0644", "lgroup": group}
mock_ret = MagicMock(return_value=(ret, perms))
comt = "File {} not updated".format(name)
with patch.dict(filestate.__salt__, {"file.check_perms": mock_ret}):
with patch.object(os.path, "exists", mock_t):
with patch.dict(filestate.__opts__, {"test": True}):
ret.update({"comment": comt})
if salt.utils.platform.is_windows():
assert (
filestate.managed(name, user=user, group=group)
== ret
)
else:
assert (
filestate.managed(name, user=user, group=group)
== ret
)

View file

@ -0,0 +1,115 @@
import logging
import os
import pytest
import salt.serializers.json as jsonserializer
import salt.serializers.msgpack as msgpackserializer
import salt.serializers.plist as plistserializer
import salt.serializers.python as pythonserializer
import salt.serializers.yaml as yamlserializer
import salt.states.file as filestate
import salt.utils.files
import salt.utils.json
import salt.utils.platform
import salt.utils.win_functions
import salt.utils.yaml
from tests.support.mock import MagicMock, mock_open, patch
log = logging.getLogger(__name__)
@pytest.fixture
def configure_loader_modules():
return {
filestate: {
"__env__": "base",
"__salt__": {"file.manage_file": False},
"__serializers__": {
"yaml.serialize": yamlserializer.serialize,
"yaml.seserialize": yamlserializer.serialize,
"python.serialize": pythonserializer.serialize,
"json.serialize": jsonserializer.serialize,
"plist.serialize": plistserializer.serialize,
"msgpack.serialize": msgpackserializer.serialize,
},
"__opts__": {"test": False, "cachedir": ""},
"__instance_id__": "",
"__low__": {},
"__utils__": {},
}
}
# 'prepend' function tests: 1
def test_prepend():
"""
Test to ensure that some text appears at the beginning of a file.
"""
name = "/tmp/etc/motd"
if salt.utils.platform.is_windows():
name = "c:\\tmp\\etc\\motd"
assert not os.path.exists(os.path.split(name)[0])
source = ["salt://motd/hr-messages.tmpl"]
sources = ["salt://motd/devops-messages.tmpl"]
text = ["Trust no one unless you have eaten much salt with him."]
ret = {"name": name, "result": False, "comment": "", "changes": {}}
comt = "Must provide name to file.prepend"
ret.update({"comment": comt, "name": ""})
assert filestate.prepend("") == ret
comt = "source and sources are mutually exclusive"
ret.update({"comment": comt, "name": name})
assert filestate.prepend(name, source=source, sources=sources) == ret
mock_t = MagicMock(return_value=True)
mock_f = MagicMock(return_value=False)
with patch.dict(
filestate.__salt__,
{
"file.directory_exists": mock_f,
"file.makedirs": mock_t,
"file.stats": mock_f,
"cp.get_template": mock_f,
"file.search": mock_f,
"file.prepend": mock_t,
},
):
comt = "The following files will be changed:\n/tmp/etc: directory - new\n"
changes = {"/tmp/etc": {"directory": "new"}}
if salt.utils.platform.is_windows():
comt = 'The directory "c:\\tmp\\etc" will be changed'
changes = {"c:\\tmp\\etc": {"directory": "new"}}
ret.update({"comment": comt, "name": name, "changes": changes})
assert filestate.prepend(name, makedirs=True) == ret
with patch.object(os.path, "isabs", mock_f):
comt = "Specified file {} is not an absolute path".format(name)
ret.update({"comment": comt, "changes": {}})
assert filestate.prepend(name) == ret
with patch.object(os.path, "isabs", mock_t):
with patch.object(os.path, "exists", mock_t):
comt = "Failed to load template file {}".format(source)
ret.update({"comment": comt, "name": source, "data": []})
assert filestate.prepend(name, source=source) == ret
ret.pop("data", None)
ret.update({"name": name})
with patch.object(
salt.utils.files, "fopen", MagicMock(mock_open(read_data=""))
):
with patch.dict(filestate.__utils__, {"files.is_text": mock_f}):
with patch.dict(filestate.__opts__, {"test": True}):
change = {"diff": "Replace binary file"}
comt = "File {} is set to be updated".format(name)
ret.update(
{"comment": comt, "result": None, "changes": change}
)
assert filestate.prepend(name, text=text) == ret
with patch.dict(filestate.__opts__, {"test": False}):
comt = "Prepended 1 lines"
ret.update({"comment": comt, "result": True, "changes": {}})
assert filestate.prepend(name, text=text) == ret

View file

@ -0,0 +1,96 @@
import logging
import os
import pytest
import salt.modules.file as filemod
import salt.states.file as filestate
import salt.utils.files
import salt.utils.json
import salt.utils.platform
import salt.utils.win_functions
import salt.utils.yaml
log = logging.getLogger(__name__)
@pytest.fixture
def configure_loader_modules():
return {filestate: {"__salt__": {"file.stats": filemod.stats}}}
@pytest.mark.destructive_test
@pytest.mark.skip_on_windows(reason="File modes do not exist on windows")
def test__check_directory(tmp_path):
"""
Test the _check_directory function
Make sure that recursive file permission checks return correctly
"""
# set file permissions
# Run _check_directory function
# Verify that it returns correctly
# Delete tmp directory structure
root_tmp_dir = str(tmp_path / "test__check_dir")
expected_mode = 0o770
changed_mode = 0o755
depth = 3
def create_files(tmp_dir):
for f in range(depth):
path = os.path.join(tmp_dir, "file_{:03}.txt".format(f))
with salt.utils.files.fopen(path, "w+"):
os.chmod(path, expected_mode)
# Create tmp directory structure
os.mkdir(root_tmp_dir)
os.chmod(root_tmp_dir, expected_mode)
create_files(root_tmp_dir)
for d in range(depth):
dir_name = os.path.join(root_tmp_dir, "dir{:03}".format(d))
os.mkdir(dir_name)
os.chmod(dir_name, expected_mode)
create_files(dir_name)
for s in range(depth):
sub_dir_name = os.path.join(dir_name, "dir{:03}".format(s))
os.mkdir(sub_dir_name)
os.chmod(sub_dir_name, expected_mode)
create_files(sub_dir_name)
# Symlinks on linux systems always have 0o777 permissions.
# Ensure we are not treating them as modified files.
target_dir = os.path.join(root_tmp_dir, "link_target_dir")
target_file = os.path.join(target_dir, "link_target_file")
link_dir = os.path.join(root_tmp_dir, "link_dir")
link_to_dir = os.path.join(link_dir, "link_to_dir")
link_to_file = os.path.join(link_dir, "link_to_file")
os.mkdir(target_dir)
os.mkdir(link_dir)
with salt.utils.files.fopen(target_file, "w+"):
pass
os.symlink(target_dir, link_to_dir)
os.symlink(target_file, link_to_file)
for path in (target_dir, target_file, link_dir, link_to_dir, link_to_file):
try:
os.chmod(path, expected_mode, follow_symlinks=False)
except (NotImplementedError, SystemError, OSError):
os.chmod(path, expected_mode)
# Set some bad permissions
changed_files = {
os.path.join(root_tmp_dir, "file_000.txt"),
os.path.join(root_tmp_dir, "dir002", "file_000.txt"),
os.path.join(root_tmp_dir, "dir000", "dir001", "file_002.txt"),
os.path.join(root_tmp_dir, "dir001", "dir002"),
os.path.join(root_tmp_dir, "dir002", "dir000"),
os.path.join(root_tmp_dir, "dir001"),
}
for c in changed_files:
os.chmod(c, changed_mode)
ret = filestate._check_directory(
root_tmp_dir,
dir_mode=oct(expected_mode),
file_mode=oct(expected_mode),
recurse=["mode"],
)
assert changed_files == set(ret[-1].keys())

View file

@ -0,0 +1,137 @@
import logging
import os
import shutil
import pytest
import salt.serializers.json as jsonserializer
import salt.serializers.msgpack as msgpackserializer
import salt.serializers.plist as plistserializer
import salt.serializers.python as pythonserializer
import salt.serializers.yaml as yamlserializer
import salt.states.file as filestate
from tests.support.mock import MagicMock, patch
log = logging.getLogger(__name__)
@pytest.fixture
def configure_loader_modules():
return {
filestate: {
"__env__": "base",
"__salt__": {"file.manage_file": False},
"__serializers__": {
"yaml.serialize": yamlserializer.serialize,
"yaml.seserialize": yamlserializer.serialize,
"python.serialize": pythonserializer.serialize,
"json.serialize": jsonserializer.serialize,
"plist.serialize": plistserializer.serialize,
"msgpack.serialize": msgpackserializer.serialize,
},
"__opts__": {"test": False, "cachedir": ""},
"__instance_id__": "",
"__low__": {},
"__utils__": {},
}
}
# 'rename' function tests: 1
def test_rename():
"""
Test if the source file exists on the system,
rename it to the named file.
"""
name = "/tmp/salt"
source = "/tmp/salt/salt"
ret = {"name": name, "result": False, "comment": "", "changes": {}}
comt = "Must provide name to file.rename"
ret.update({"comment": comt, "name": ""})
assert filestate.rename("", source) == ret
mock_t = MagicMock(return_value=True)
mock_f = MagicMock(return_value=False)
mock_lex = MagicMock(side_effect=[False, True, True])
with patch.object(os.path, "isabs", mock_f):
comt = "Specified file {} is not an absolute path".format(name)
ret.update({"comment": comt, "name": name})
assert filestate.rename(name, source) == ret
mock_lex = MagicMock(return_value=False)
with patch.object(os.path, "isabs", mock_t):
with patch.object(os.path, "lexists", mock_lex):
comt = 'Source file "{}" has already been moved out of place'.format(source)
ret.update({"comment": comt, "result": True})
assert filestate.rename(name, source) == ret
mock_lex = MagicMock(side_effect=[True, True, True])
with patch.object(os.path, "isabs", mock_t):
with patch.object(os.path, "lexists", mock_lex):
comt = 'The target file "{}" exists and will not be overwritten'.format(
name
)
ret.update({"comment": comt, "result": True})
assert filestate.rename(name, source) == ret
mock_lex = MagicMock(side_effect=[True, True, True])
mock_rem = MagicMock(side_effect=IOError)
with patch.object(os.path, "isabs", mock_t):
with patch.object(os.path, "lexists", mock_lex):
with patch.dict(filestate.__opts__, {"test": False}):
comt = 'Failed to delete "{}" in preparation for forced move'.format(
name
)
with patch.dict(filestate.__salt__, {"file.remove": mock_rem}):
ret.update({"name": name, "comment": comt, "result": False})
assert filestate.rename(name, source, force=True) == ret
mock_lex = MagicMock(side_effect=[True, False, False])
with patch.object(os.path, "isabs", mock_t):
with patch.object(os.path, "lexists", mock_lex):
with patch.dict(filestate.__opts__, {"test": True}):
comt = 'File "{}" is set to be moved to "{}"'.format(source, name)
ret.update({"name": name, "comment": comt, "result": None})
assert filestate.rename(name, source) == ret
mock_lex = MagicMock(side_effect=[True, False, False])
with patch.object(os.path, "isabs", mock_t):
with patch.object(os.path, "lexists", mock_lex):
with patch.object(os.path, "isdir", mock_f):
with patch.dict(filestate.__opts__, {"test": False}):
comt = "The target directory /tmp is not present"
ret.update({"name": name, "comment": comt, "result": False})
assert filestate.rename(name, source) == ret
mock_lex = MagicMock(side_effect=[True, False, False])
with patch.object(os.path, "isabs", mock_t):
with patch.object(os.path, "lexists", mock_lex):
with patch.object(os.path, "isdir", mock_t):
with patch.object(os.path, "islink", mock_f):
with patch.dict(filestate.__opts__, {"test": False}):
with patch.object(
shutil, "move", MagicMock(side_effect=IOError)
):
comt = 'Failed to move "{}" to "{}"'.format(source, name)
ret.update({"name": name, "comment": comt, "result": False})
assert filestate.rename(name, source) == ret
mock_lex = MagicMock(side_effect=[True, False, False])
with patch.object(os.path, "isabs", mock_t):
with patch.object(os.path, "lexists", mock_lex):
with patch.object(os.path, "isdir", mock_t):
with patch.object(os.path, "islink", mock_f):
with patch.dict(filestate.__opts__, {"test": False}):
with patch.object(shutil, "move", MagicMock()):
comt = 'Moved "{}" to "{}"'.format(source, name)
ret.update(
{
"name": name,
"comment": comt,
"result": True,
"changes": {name: source},
}
)
assert filestate.rename(name, source) == ret

View file

@ -0,0 +1,233 @@
import logging
import os
from datetime import datetime
import pytest
import salt.serializers.json as jsonserializer
import salt.serializers.msgpack as msgpackserializer
import salt.serializers.plist as plistserializer
import salt.serializers.python as pythonserializer
import salt.serializers.yaml as yamlserializer
import salt.states.file as filestate
from tests.support.mock import MagicMock, call, patch
try:
from dateutil.relativedelta import relativedelta
HAS_DATEUTIL = True
except ImportError:
HAS_DATEUTIL = False
NO_DATEUTIL_REASON = "python-dateutil is not installed"
log = logging.getLogger(__name__)
@pytest.fixture
def configure_loader_modules():
return {
filestate: {
"__env__": "base",
"__salt__": {"file.manage_file": False},
"__serializers__": {
"yaml.serialize": yamlserializer.serialize,
"yaml.seserialize": yamlserializer.serialize,
"python.serialize": pythonserializer.serialize,
"json.serialize": jsonserializer.serialize,
"plist.serialize": plistserializer.serialize,
"msgpack.serialize": msgpackserializer.serialize,
},
"__opts__": {"test": False, "cachedir": ""},
"__instance_id__": "",
"__low__": {},
"__utils__": {},
}
}
@pytest.mark.skipif(not HAS_DATEUTIL, reason=NO_DATEUTIL_REASON)
@pytest.mark.slow_test
def test_retention_schedule():
"""
Test to execute the retention_schedule logic.
This test takes advantage of knowing which files it is generating,
which means it can easily generate list of which files it should keep.
"""
def generate_fake_files(
format="example_name_%Y%m%dT%H%M%S.tar.bz2",
starting=datetime(2016, 2, 8, 9),
every=relativedelta(minutes=30),
ending=datetime(2015, 12, 25),
maxfiles=None,
):
"""
For starting, make sure that it's over a week from the beginning of the month
For every, pick only one of minutes, hours, days, weeks, months or years
For ending, the further away it is from starting, the slower the tests run
Full coverage requires over a year of separation, but that's painfully slow.
"""
if every.years:
ts = datetime(starting.year, 1, 1)
elif every.months:
ts = datetime(starting.year, starting.month, 1)
elif every.days:
ts = datetime(starting.year, starting.month, starting.day)
elif every.hours:
ts = datetime(starting.year, starting.month, starting.day, starting.hour)
elif every.minutes:
ts = datetime(starting.year, starting.month, starting.day, starting.hour, 0)
else:
raise NotImplementedError("not sure what you're trying to do here")
fake_files = []
count = 0
while ending < ts:
fake_files.append(ts.strftime(format=format))
count += 1
if maxfiles and maxfiles == "all" or maxfiles and count >= maxfiles:
break
ts -= every
return fake_files
fake_name = "/some/dir/name"
fake_retain = {
"most_recent": 2,
"first_of_hour": 4,
"first_of_day": 7,
"first_of_week": 6,
"first_of_month": 6,
"first_of_year": "all",
}
fake_strptime_format = "example_name_%Y%m%dT%H%M%S.tar.bz2"
fake_matching_file_list = generate_fake_files()
# Add some files which do not match fake_strptime_format
fake_no_match_file_list = generate_fake_files(
format="no_match_%Y%m%dT%H%M%S.tar.bz2", every=relativedelta(days=1)
)
def lstat_side_effect(path):
import re
from time import mktime
x = re.match(r"^[^\d]*(\d{8}T\d{6})\.tar\.bz2$", path).group(1)
ts = mktime(datetime.strptime(x, "%Y%m%dT%H%M%S").timetuple())
return {
"st_atime": 0.0,
"st_ctime": 0.0,
"st_gid": 0,
"st_mode": 33188,
"st_mtime": ts,
"st_nlink": 1,
"st_size": 0,
"st_uid": 0,
}
mock_t = MagicMock(return_value=True)
mock_f = MagicMock(return_value=False)
mock_lstat = MagicMock(side_effect=lstat_side_effect)
mock_remove = MagicMock()
def run_checks(isdir=mock_t, strptime_format=None, test=False):
expected_ret = {
"name": fake_name,
"changes": {"retained": [], "deleted": [], "ignored": []},
"result": True,
"comment": "Name provided to file.retention must be a directory",
}
if strptime_format:
fake_file_list = sorted(fake_matching_file_list + fake_no_match_file_list)
else:
fake_file_list = sorted(fake_matching_file_list)
mock_readdir = MagicMock(return_value=fake_file_list)
with patch.dict(filestate.__opts__, {"test": test}):
with patch.object(os.path, "isdir", isdir):
mock_readdir.reset_mock()
with patch.dict(filestate.__salt__, {"file.readdir": mock_readdir}):
with patch.dict(filestate.__salt__, {"file.lstat": mock_lstat}):
mock_remove.reset_mock()
with patch.dict(
filestate.__salt__, {"file.remove": mock_remove}
):
if strptime_format:
actual_ret = filestate.retention_schedule(
fake_name,
fake_retain,
strptime_format=fake_strptime_format,
)
else:
actual_ret = filestate.retention_schedule(
fake_name, fake_retain
)
if not isdir():
mock_readdir.assert_has_calls([])
expected_ret["result"] = False
else:
mock_readdir.assert_called_once_with(fake_name)
ignored_files = fake_no_match_file_list if strptime_format else []
retained_files = set(
generate_fake_files(maxfiles=fake_retain["most_recent"])
)
junk_list = [
("first_of_hour", relativedelta(hours=1)),
("first_of_day", relativedelta(days=1)),
("first_of_week", relativedelta(weeks=1)),
("first_of_month", relativedelta(months=1)),
("first_of_year", relativedelta(years=1)),
]
for retainable, retain_interval in junk_list:
new_retains = set(
generate_fake_files(
maxfiles=fake_retain[retainable], every=retain_interval
)
)
# if we generate less than the number of files expected,
# then the oldest file will also be retained
# (correctly, since its the first in it's category)
if (
fake_retain[retainable] == "all"
or len(new_retains) < fake_retain[retainable]
):
new_retains.add(fake_file_list[0])
retained_files |= new_retains
deleted_files = sorted(
list(set(fake_file_list) - retained_files - set(ignored_files)),
reverse=True,
)
retained_files = sorted(list(retained_files), reverse=True)
expected_ret["changes"] = {
"retained": retained_files,
"deleted": deleted_files,
"ignored": ignored_files,
}
if test:
expected_ret["result"] = None
expected_ret[
"comment"
] = "{} backups would have been removed from {}.\n" "".format(
len(deleted_files), fake_name
)
else:
expected_ret[
"comment"
] = "{} backups were removed from {}.\n" "".format(
len(deleted_files), fake_name
)
mock_remove.assert_has_calls(
[call(os.path.join(fake_name, x)) for x in deleted_files],
any_order=True,
)
assert actual_ret == expected_ret
run_checks(isdir=mock_f)
run_checks()
run_checks(test=True)
run_checks(strptime_format=fake_strptime_format)
run_checks(strptime_format=fake_strptime_format, test=True)

View file

@ -0,0 +1,59 @@
import logging
import os
import pytest
import salt.states.file as filestate
from tests.support.mock import MagicMock, patch
log = logging.getLogger(__name__)
pytestmark = pytest.mark.skipif(
pytest.mark.skip_unless_on_linux, reason="Only run on Linux"
)
@pytest.fixture
def configure_loader_modules():
return {
filestate: {
"__env__": "base",
"__salt__": {"file.manage_file": False},
"__opts__": {"test": False, "cachedir": ""},
"__instance_id__": "",
"__low__": {},
"__utils__": {},
}
}
def test_selinux_change():
file_name = "/tmp/some-test-file"
check_perms_result = [
{
"comment": "The file {} is set to be changed".format(file_name),
"changes": {
"selinux": {
"New": "User: unconfined_u Type: lost_found_t",
"Old": "User: system_u Type: user_tmp_t",
}
},
"name": file_name,
"result": True,
},
{"luser": "root", "lmode": "0644", "lgroup": "root"},
]
with patch.object(os.path, "exists", MagicMock(return_value=True)):
with patch.dict(
filestate.__salt__,
{
"file.source_list": MagicMock(return_value=[file_name, None]),
"file.check_perms": MagicMock(return_value=check_perms_result),
},
):
ret = filestate.managed(
file_name,
selinux={"seuser": "unconfined_u", "setype": "user_tmp_t"},
)
assert ret["result"]

View file

@ -0,0 +1,399 @@
import logging
import os
import pytest
import salt.serializers.json as jsonserializer
import salt.serializers.msgpack as msgpackserializer
import salt.serializers.plist as plistserializer
import salt.serializers.python as pythonserializer
import salt.serializers.yaml as yamlserializer
import salt.states.file as filestate
import salt.utils.files
import salt.utils.json
import salt.utils.platform
import salt.utils.win_functions
import salt.utils.yaml
from tests.support.mock import MagicMock, patch
log = logging.getLogger(__name__)
@pytest.fixture
def configure_loader_modules():
return {
filestate: {
"__env__": "base",
"__salt__": {"file.manage_file": False},
"__serializers__": {
"yaml.serialize": yamlserializer.serialize,
"yaml.seserialize": yamlserializer.serialize,
"python.serialize": pythonserializer.serialize,
"json.serialize": jsonserializer.serialize,
"plist.serialize": plistserializer.serialize,
"msgpack.serialize": msgpackserializer.serialize,
},
"__opts__": {"test": False, "cachedir": ""},
"__instance_id__": "",
"__low__": {},
"__utils__": {},
}
}
def test_symlink():
"""
Test to create a symlink.
"""
name = os.sep + os.path.join("tmp", "testfile.txt")
target = salt.utils.files.mkstemp()
test_dir = os.sep + "tmp"
user = "salt"
if salt.utils.platform.is_windows():
group = "salt"
else:
group = "saltstack"
def return_val(kwargs):
val = {
"name": name,
"result": False,
"comment": "",
"changes": {},
}
val.update(kwargs)
return val
mock_t = MagicMock(return_value=True)
mock_f = MagicMock(return_value=False)
mock_empty = MagicMock(return_value="")
mock_uid = MagicMock(return_value="U1001")
mock_gid = MagicMock(return_value="g1001")
mock_target = MagicMock(return_value=target)
mock_user = MagicMock(return_value=user)
mock_grp = MagicMock(return_value=group)
mock_os_error = MagicMock(side_effect=OSError)
with patch.dict(filestate.__salt__, {"config.manage_mode": mock_t}):
comt = "Must provide name to file.symlink"
ret = return_val({"comment": comt, "name": ""})
assert filestate.symlink("", target) == ret
with patch.dict(
filestate.__salt__,
{
"config.manage_mode": mock_t,
"file.user_to_uid": mock_empty,
"file.group_to_gid": mock_empty,
"user.info": mock_empty,
"user.current": mock_user,
},
):
if salt.utils.platform.is_windows():
comt = "User {} does not exist".format(user)
ret = return_val({"comment": comt, "name": name})
else:
comt = "User {} does not exist. Group {} does not exist.".format(
user, group
)
ret = return_val({"comment": comt, "name": name})
assert filestate.symlink(name, target, user=user, group=group) == ret
with patch.dict(
filestate.__salt__,
{
"config.manage_mode": mock_t,
"file.user_to_uid": mock_uid,
"file.group_to_gid": mock_gid,
"file.is_link": mock_f,
"user.info": mock_empty,
"user.current": mock_user,
},
), patch.dict(filestate.__opts__, {"test": True}), patch.object(
os.path, "exists", mock_f
):
if salt.utils.platform.is_windows():
comt = "User {} does not exist".format(user)
ret = return_val(
{"comment": comt, "result": False, "name": name, "changes": {}}
)
else:
comt = "Symlink {} to {} is set for creation".format(name, target)
ret = return_val(
{"comment": comt, "result": None, "changes": {"new": name}}
)
assert filestate.symlink(name, target, user=user, group=group) == ret
with patch.dict(
filestate.__salt__,
{
"config.manage_mode": mock_t,
"file.user_to_uid": mock_uid,
"file.group_to_gid": mock_gid,
"file.is_link": mock_f,
"user.info": mock_empty,
"user.current": mock_user,
},
), patch.dict(filestate.__opts__, {"test": False}), patch.object(
os.path, "isdir", mock_f
), patch.object(
os.path, "exists", mock_f
):
if salt.utils.platform.is_windows():
comt = "User {} does not exist".format(user)
ret = return_val(
{"comment": comt, "result": False, "name": name, "changes": {}}
)
else:
comt = "Directory {} for symlink is not present".format(test_dir)
ret = return_val({"comment": comt, "result": False, "changes": {}})
assert filestate.symlink(name, target, user=user, group=group) == ret
with patch.dict(
filestate.__salt__,
{
"config.manage_mode": mock_t,
"file.user_to_uid": mock_uid,
"file.group_to_gid": mock_gid,
"file.is_link": mock_t,
"file.readlink": mock_target,
"user.info": mock_empty,
"user.current": mock_user,
},
), patch.dict(filestate.__opts__, {"test": False}), patch.object(
os.path, "isdir", mock_t
), patch.object(
salt.states.file, "_check_symlink_ownership", mock_t
), patch(
"salt.utils.win_functions.get_sid_from_name", return_value="test-sid"
):
if salt.utils.platform.is_windows():
comt = "Symlink {} is present and owned by {}".format(name, user)
else:
comt = "Symlink {} is present and owned by {}:{}".format(name, user, group)
ret = return_val({"comment": comt, "result": True, "changes": {}})
assert filestate.symlink(name, target, user=user, group=group) == ret
with patch.dict(
filestate.__salt__,
{
"config.manage_mode": mock_t,
"file.user_to_uid": mock_uid,
"file.group_to_gid": mock_gid,
"file.is_link": mock_f,
"file.readlink": mock_target,
"user.info": mock_empty,
"user.current": mock_user,
},
), patch.dict(filestate.__opts__, {"test": False}), patch.object(
os.path, "isdir", mock_t
), patch.object(
os.path, "exists", mock_t
), patch.object(
os.path, "lexists", mock_t
), patch(
"salt.utils.win_functions.get_sid_from_name", return_value="test-sid"
):
comt = (
"Symlink & backup dest exists and Force not set. {} -> "
"{} - backup: {}".format(name, target, os.path.join(test_dir, "SALT"))
)
ret.update({"comment": comt, "result": False, "changes": {}})
assert (
filestate.symlink(name, target, user=user, group=group, backupname="SALT")
== ret
)
with patch.dict(
filestate.__salt__,
{
"config.manage_mode": mock_t,
"file.user_to_uid": mock_uid,
"file.group_to_gid": mock_gid,
"file.is_link": mock_f,
"file.readlink": mock_target,
"user.info": mock_empty,
"user.current": mock_user,
},
), patch.dict(filestate.__opts__, {"test": False}), patch.object(
os.path, "exists", mock_t
), patch.object(
os.path, "isfile", mock_t
), patch.object(
os.path, "isdir", mock_t
), patch(
"salt.utils.win_functions.get_sid_from_name", return_value="test-sid"
):
comt = "Backupname must be an absolute path or a file name: {}".format(
"tmp/SALT"
)
ret.update({"comment": comt, "result": False, "changes": {}})
assert (
filestate.symlink(
name, target, user=user, group=group, backupname="tmp/SALT"
)
== ret
)
with patch.dict(
filestate.__salt__,
{
"config.manage_mode": mock_t,
"file.user_to_uid": mock_uid,
"file.group_to_gid": mock_gid,
"file.is_link": mock_f,
"file.readlink": mock_target,
"user.info": mock_empty,
"user.current": mock_user,
},
), patch.dict(filestate.__opts__, {"test": False}), patch.object(
os.path, "isdir", mock_t
), patch.object(
os.path, "exists", mock_t
), patch.object(
os.path, "isfile", mock_t
), patch(
"salt.utils.win_functions.get_sid_from_name", return_value="test-sid"
):
comt = "File exists where the symlink {} should be".format(name)
ret = return_val({"comment": comt, "changes": {}, "result": False})
assert filestate.symlink(name, target, user=user, group=group) == ret
with patch.dict(
filestate.__salt__,
{
"config.manage_mode": mock_t,
"file.user_to_uid": mock_uid,
"file.group_to_gid": mock_gid,
"file.is_link": mock_f,
"file.readlink": mock_target,
"file.symlink": mock_t,
"user.info": mock_t,
"file.lchown": mock_f,
},
), patch.dict(filestate.__opts__, {"test": False}), patch.object(
os.path, "isdir", MagicMock(side_effect=[True, False])
), patch.object(
os.path, "isdir", mock_t
), patch.object(
os.path, "exists", mock_t
), patch(
"salt.utils.win_functions.get_sid_from_name", return_value="test-sid"
):
comt = "Directory exists where the symlink {} should be".format(name)
ret = return_val({"comment": comt, "result": False, "changes": {}})
assert filestate.symlink(name, target, user=user, group=group) == ret
with patch.dict(
filestate.__salt__,
{
"config.manage_mode": mock_t,
"file.user_to_uid": mock_uid,
"file.group_to_gid": mock_gid,
"file.is_link": mock_f,
"file.readlink": mock_target,
"file.symlink": mock_os_error,
"user.info": mock_t,
"file.lchown": mock_f,
},
), patch.dict(filestate.__opts__, {"test": False}), patch.object(
os.path, "isdir", MagicMock(side_effect=[True, False])
), patch.object(
os.path, "isfile", mock_f
), patch(
"salt.utils.win_functions.get_sid_from_name", return_value="test-sid"
):
comt = "Unable to create new symlink {} -> {}: ".format(name, target)
ret = return_val({"comment": comt, "result": False, "changes": {}})
assert filestate.symlink(name, target, user=user, group=group) == ret
with patch.dict(
filestate.__salt__,
{
"config.manage_mode": mock_t,
"file.user_to_uid": mock_uid,
"file.group_to_gid": mock_gid,
"file.is_link": mock_f,
"file.readlink": mock_target,
"file.symlink": mock_t,
"user.info": mock_t,
"file.lchown": mock_f,
"file.get_user": mock_user,
"file.get_group": mock_grp,
},
), patch.dict(filestate.__opts__, {"test": False}), patch.object(
os.path, "isdir", MagicMock(side_effect=[True, False])
), patch.object(
os.path, "isfile", mock_f
), patch(
"salt.states.file._check_symlink_ownership", return_value=True
), patch(
"salt.utils.win_functions.get_sid_from_name", return_value="test-sid"
):
comt = "Created new symlink {} -> {}".format(name, target)
ret = return_val({"comment": comt, "result": True, "changes": {"new": name}})
assert filestate.symlink(name, target, user=user, group=group) == ret
with patch.dict(
filestate.__salt__,
{
"config.manage_mode": mock_t,
"file.user_to_uid": mock_uid,
"file.group_to_gid": mock_gid,
"file.is_link": mock_f,
"file.readlink": mock_target,
"file.symlink": mock_t,
"user.info": mock_t,
"file.lchown": mock_f,
"file.get_user": mock_empty,
"file.get_group": mock_empty,
},
), patch.dict(filestate.__opts__, {"test": False}), patch.object(
os.path, "isdir", MagicMock(side_effect=[True, False])
), patch.object(
os.path, "isfile", mock_f
), patch(
"salt.utils.win_functions.get_sid_from_name", return_value="test-sid"
), patch(
"salt.states.file._set_symlink_ownership", return_value=False
), patch(
"salt.states.file._check_symlink_ownership", return_value=False
):
comt = (
"Created new symlink {} -> {}, but was unable to set "
"ownership to {}:{}".format(name, target, user, group)
)
ret = return_val({"comment": comt, "result": False, "changes": {"new": name}})
assert filestate.symlink(name, target, user=user, group=group) == ret
with patch.dict(
filestate.__salt__,
{
"config.manage_mode": mock_t,
"file.user_to_uid": mock_uid,
"file.group_to_gid": mock_gid,
"file.is_link": mock_f,
"file.readlink": mock_target,
"file.symlink": mock_t,
"user.info": mock_t,
"file.lchown": mock_f,
"file.get_user": mock_empty,
"file.get_group": mock_empty,
},
), patch.dict(filestate.__opts__, {"test": False}), patch.object(
os.path, "isdir", MagicMock(side_effect=[True, False])
), patch.object(
os.path, "isfile", mock_f
), patch(
"salt.utils.win_functions.get_sid_from_name", return_value="test-sid"
), patch(
"salt.states.file._set_symlink_ownership", return_value=True
), patch(
"salt.states.file._check_symlink_ownership", return_value=True
):
group = None
comt = "Created new symlink {} -> {}".format(name, target)
ret = return_val({"comment": comt, "result": True, "changes": {"new": name}})
res = filestate.symlink(name, target, user=user, group=user)
assert res == ret

View file

@ -0,0 +1,107 @@
import logging
import os
from datetime import datetime
import pytest
import salt.states.file as filestate
import salt.utils.files
import salt.utils.json
import salt.utils.platform
import salt.utils.win_functions
import salt.utils.yaml
from tests.support.mock import MagicMock, patch
log = logging.getLogger(__name__)
@pytest.fixture
def configure_loader_modules():
return {filestate: {"__salt__": {}, "__opts__": {}}}
def test__tidied():
name = os.sep + "test"
if salt.utils.platform.is_windows():
name = "c:" + name
walker = [
(os.path.join("test", "test1"), [], ["file1"]),
(os.path.join("test", "test2", "test3"), [], []),
(os.path.join("test", "test2"), ["test3"], ["file2"]),
("test", ["test1", "test2"], ["file3"]),
]
today_delta = datetime.today() - datetime.utcfromtimestamp(0)
remove = MagicMock(name="file.remove")
with patch("os.walk", return_value=walker), patch(
"os.path.islink", return_value=False
), patch("os.path.getatime", return_value=today_delta.total_seconds()), patch(
"os.path.getsize", return_value=10
), patch.dict(
filestate.__opts__, {"test": False}
), patch.dict(
filestate.__salt__, {"file.remove": remove}
), patch(
"os.path.isdir", return_value=True
):
ret = filestate.tidied(name=name)
exp = {
"name": name,
"changes": {
"removed": [
os.path.join("test", "test1", "file1"),
os.path.join("test", "test2", "file2"),
os.path.join("test", "file3"),
]
},
"result": True,
"comment": "Removed 3 files or directories from directory {}".format(name),
}
assert exp == ret
assert remove.call_count == 3
remove.reset_mock()
with patch("os.walk", return_value=walker), patch(
"os.path.islink", return_value=False
), patch("os.path.getatime", return_value=today_delta.total_seconds()), patch(
"os.path.getsize", return_value=10
), patch.dict(
filestate.__opts__, {"test": False}
), patch.dict(
filestate.__salt__, {"file.remove": remove}
), patch(
"os.path.isdir", return_value=True
):
ret = filestate.tidied(name=name, rmdirs=True)
exp = {
"name": name,
"changes": {
"removed": [
os.path.join("test", "test1", "file1"),
os.path.join("test", "test2", "file2"),
os.path.join("test", "test2", "test3"),
os.path.join("test", "file3"),
os.path.join("test", "test1"),
os.path.join("test", "test2"),
]
},
"result": True,
"comment": "Removed 6 files or directories from directory {}".format(name),
}
assert exp == ret
assert remove.call_count == 6
def test__bad_input():
exp = {
"name": "test/",
"changes": {},
"result": False,
"comment": "Specified file test/ is not an absolute path",
}
assert filestate.tidied(name="test/") == exp
exp = {
"name": "/bad-directory-name/",
"changes": {},
"result": False,
"comment": "/bad-directory-name/ does not exist or is not a directory.",
}
assert filestate.tidied(name="/bad-directory-name/") == exp

File diff suppressed because it is too large Load diff