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

Merge pull request #65457 from s0undt3ch/hotfix/merge-forward

Browse source 
[master] Merge 3006.x into master
This commit is contained in:
Pedro Algarvio 2023-10-27 13:59:34 +01:00 • committed by GitHub
commit 93b640543c
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
20 changed files with 528 additions and 249 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
changelog/65411.fixed.md Normal file
View file

@ -0,0 +1,2 @@
Fixes an issue setting user or machine policy on Windows when the Group Policy
directory is missing

2
salt/_compat.py
View file

@ -4,6 +4,8 @@ Salt compatibility code
# pylint: disable=unused-import
import sys
# pragma: no cover
# The ipaddress module included in Salt is from Python 3.9.5.
# When running from Py3.9.5+ use the standard library module, use ours otherwise
if sys.version_info >= (3, 9, 5):

52
salt/modules/win_file.py
View file

@ -16,8 +16,10 @@ import stat
import sys
import tempfile
import salt.utils.files
import salt.utils.path
import salt.utils.platform
import salt.utils.user
from salt.exceptions import CommandExecutionError, SaltInvocationError
from salt.modules.file import (
__clean_tmp,
@ -107,6 +109,15 @@ try:
except ImportError:
HAS_WINDOWS_MODULES = False
HAS_WIN_DACL = False
try:
if salt.utils.platform.is_windows():
import salt.utils.win_dacl
HAS_WIN_DACL = True
except ImportError:
HAS_WIN_DACL = False
if salt.utils.platform.is_windows():
if HAS_WINDOWS_MODULES:
# namespace functions from file.py
@ -194,6 +205,8 @@ def __virtual__():
"""
if not salt.utils.platform.is_windows() or not HAS_WINDOWS_MODULES:
return False, "Module win_file: Missing Win32 modules"
if not HAS_WIN_DACL:
return False, "Module win_file: Unable to load salt.utils.win_dacl"
return __virtualname__
@ -305,7 +318,7 @@ def group_to_gid(group):
if group is None:
return ""
return __utils__["dacl.get_sid_string"](group)
return salt.utils.win_dacl.get_sid_string(group)
def get_pgid(path, follow_symlinks=True):
@ -346,8 +359,8 @@ def get_pgid(path, follow_symlinks=True):
if follow_symlinks and sys.getwindowsversion().major >= 6:
path = _resolve_symlink(path)
group_name = __utils__["dacl.get_primary_group"](path)
return __utils__["dacl.get_sid_string"](group_name)
group_name = salt.utils.win_dacl.get_primary_group(path)
return salt.utils.win_dacl.get_sid_string(group_name)
def get_pgroup(path, follow_symlinks=True):
@ -498,7 +511,7 @@ def uid_to_user(uid):
if uid is None or uid == "":
return ""
return __utils__["dacl.get_name"](uid)
return salt.utils.win_dacl.get_name(uid)
def user_to_uid(user):
@ -518,9 +531,9 @@ def user_to_uid(user):
salt '*' file.user_to_uid myusername
"""
if user is None:
user = __utils__["user.get_user"]()
user = salt.utils.user.get_user()
return __utils__["dacl.get_sid_string"](user)
return salt.utils.win_dacl.get_sid_string(user)
def get_uid(path, follow_symlinks=True):
@ -558,8 +571,8 @@ def get_uid(path, follow_symlinks=True):
if follow_symlinks and sys.getwindowsversion().major >= 6:
path = _resolve_symlink(path)
owner_sid = __utils__["dacl.get_owner"](path)
return __utils__["dacl.get_sid_string"](owner_sid)
owner_sid = salt.utils.win_dacl.get_owner(path)
return salt.utils.win_dacl.get_sid_string(owner_sid)
def get_user(path, follow_symlinks=True):
@ -597,7 +610,7 @@ def get_user(path, follow_symlinks=True):
if follow_symlinks and sys.getwindowsversion().major >= 6:
path = _resolve_symlink(path)
return __utils__["dacl.get_owner"](path)
return salt.utils.win_dacl.get_owner(path)
def get_mode(path):
@ -735,9 +748,9 @@ def chown(path, user, group=None, pgroup=None, follow_symlinks=True):
if not os.path.exists(path):
raise CommandExecutionError(f"Path not found: {path}")
__utils__["dacl.set_owner"](path, user)
salt.utils.win_dacl.set_owner(path, user)
if pgroup:
__utils__["dacl.set_primary_group"](path, pgroup)
salt.utils.win_dacl.set_primary_group(path, pgroup)
return True
@ -767,7 +780,7 @@ def chpgrp(path, group):
salt '*' file.chpgrp c:\\temp\\test.txt Administrators
salt '*' file.chpgrp c:\\temp\\test.txt "'None'"
"""
return __utils__["dacl.set_primary_group"](path, group)
return salt.utils.win_dacl.set_primary_group(path, group)
def chgrp(path, group):
@ -802,7 +815,7 @@ def chgrp(path, group):
.. code-block:: bash
salt '*' file.chpgrp c:\\temp\\test.txt administrators
salt '*' file.chgrp c:\\temp\\test.txt administrators
"""
func_name = f"{__virtualname__}.chgrp"
if __opts__.get("fun", "") == func_name:
@ -871,7 +884,7 @@ def stats(path, hash_type="sha256", follow_symlinks=True):
ret["mtime"] = pstat.st_mtime
ret["ctime"] = pstat.st_ctime
ret["size"] = pstat.st_size
ret["mode"] = __utils__["files.normalize_mode"](oct(stat.S_IMODE(pstat.st_mode)))
ret["mode"] = salt.utils.files.normalize_mode(oct(stat.S_IMODE(pstat.st_mode)))
if hash_type:
ret["sum"] = get_sum(path, hash_type)
ret["type"] = "file"
@ -1513,7 +1526,7 @@ def is_link(path):
)
try:
return __utils__["path.islink"](path)
return salt.utils.path.islink(path)
except Exception as exc: # pylint: disable=broad-except
raise CommandExecutionError(exc)
@ -1604,10 +1617,10 @@ def mkdir(
# Set owner
if owner:
__utils__["dacl.set_owner"](obj_name=path, principal=owner)
salt.utils.win_dacl.set_owner(obj_name=path, principal=owner)
# Set permissions
__utils__["dacl.set_perms"](
salt.utils.win_dacl.set_perms(
obj_name=path,
obj_type="file",
grant_perms=grant_perms,
@ -1926,7 +1939,7 @@ def check_perms(
path = os.path.expanduser(path)
return __utils__["dacl.check_perms"](
return salt.utils.win_dacl.check_perms(
obj_name=path,
obj_type="file",
ret=ret,
@ -1935,6 +1948,7 @@ def check_perms(
deny_perms=deny_perms,
inheritance=inheritance,
reset=reset,
test_mode=__opts__["test"],
)
@ -2012,7 +2026,7 @@ def set_perms(path, grant_perms=None, deny_perms=None, inheritance=True, reset=F
# Specify advanced attributes with a list
salt '*' file.set_perms C:\\Temp\\ "{'jsnuffy': {'perms': ['read_attributes', 'read_ea'], 'applies_to': 'this_folder_only'}}"
"""
return __utils__["dacl.set_perms"](
return salt.utils.win_dacl.set_perms(
obj_name=path,
obj_type="file",
grant_perms=grant_perms,

93
salt/utils/win_dacl.py
View file

@ -496,9 +496,7 @@ def dacl(obj_name=None, obj_type="file"):
"""
# Validate obj_type
if obj_type.lower() not in self.obj_type:
raise SaltInvocationError(
'Invalid "obj_type" passed: {}'.format(obj_type)
)
raise SaltInvocationError(f'Invalid "obj_type" passed: {obj_type}')
self.dacl_type = obj_type.lower()
@ -514,7 +512,7 @@ def dacl(obj_name=None, obj_type="file"):
)
except pywintypes.error as exc:
if "The system cannot find" in exc.strerror:
msg = "System cannot find {}".format(obj_name)
msg = f"System cannot find {obj_name}"
log.exception(msg)
raise CommandExecutionError(msg)
raise
@ -588,9 +586,7 @@ def dacl(obj_name=None, obj_type="file"):
valid_hive = hives[passed_hive.upper()]
except KeyError:
log.exception("Invalid Registry Hive: %s", passed_hive)
raise CommandExecutionError(
"Invalid Registry Hive: {}".format(passed_hive)
)
raise CommandExecutionError(f"Invalid Registry Hive: {passed_hive}")
reg.insert(0, valid_hive)
@ -632,7 +628,7 @@ def dacl(obj_name=None, obj_type="file"):
if applies_to not in self.ace_prop[self.dacl_type]:
raise SaltInvocationError(
"Invalid 'applies_to' for type {}".format(self.dacl_type)
f"Invalid 'applies_to' for type {self.dacl_type}"
)
if self.dacl is None:
@ -644,7 +640,7 @@ def dacl(obj_name=None, obj_type="file"):
try:
perm_flag = self.ace_perms[self.dacl_type]["basic"][permissions]
except KeyError as exc:
msg = "Invalid permission specified: {}".format(permissions)
msg = f"Invalid permission specified: {permissions}"
log.exception(msg)
raise CommandExecutionError(msg, exc)
else:
@ -652,12 +648,12 @@ def dacl(obj_name=None, obj_type="file"):
try:
perm_flag |= self.ace_perms[self.dacl_type]["advanced"][perm]
except KeyError as exc:
msg = "Invalid permission specified: {}".format(perm)
msg = f"Invalid permission specified: {perm}"
log.exception(msg)
raise CommandExecutionError(msg, exc)
if access_mode.lower() not in ["grant", "deny"]:
raise SaltInvocationError("Invalid Access Mode: {}".format(access_mode))
raise SaltInvocationError(f"Invalid Access Mode: {access_mode}")
# Add ACE to the DACL
# Grant or Deny
@ -686,11 +682,9 @@ def dacl(obj_name=None, obj_type="file"):
)
else:
log.exception("Invalid access mode: %s", access_mode)
raise SaltInvocationError(
"Invalid access mode: {}".format(access_mode)
)
raise SaltInvocationError(f"Invalid access mode: {access_mode}")
except Exception as exc: # pylint: disable=broad-except
return False, "Error: {}".format(exc)
return False, f"Error: {exc}"
return True
@ -945,7 +939,7 @@ def dacl(obj_name=None, obj_type="file"):
# If still nothing, it must be undefined
if not ace_perms:
ace_perms = ["Undefined Permission: {}".format(ace[1])]
ace_perms = [f"Undefined Permission: {ace[1]}"]
return (
principal,
@ -1001,7 +995,7 @@ def dacl(obj_name=None, obj_type="file"):
offset += 1
if not ret:
ret = ["ACE not found for {}".format(principal)]
ret = [f"ACE not found for {principal}"]
return ret
@ -1100,7 +1094,7 @@ def dacl(obj_name=None, obj_type="file"):
)
except pywintypes.error as exc:
raise CommandExecutionError(
"Failed to set permissions: {}".format(obj_name), exc.strerror
f"Failed to set permissions: {obj_name}", exc.strerror
)
return True
@ -1146,7 +1140,7 @@ def get_sid(principal):
sid = win32security.ConvertStringSidToSid(sid)
except pywintypes.error:
log.exception("Invalid user/group or sid: %s", principal)
raise CommandExecutionError("Invalid user/group or sid: {}".format(principal))
raise CommandExecutionError(f"Invalid user/group or sid: {principal}")
except TypeError:
raise CommandExecutionError
@ -1189,7 +1183,7 @@ def get_sid_string(principal):
return win32security.ConvertSidToStringSid(principal)
except pywintypes.error:
log.exception("Invalid principal %s", principal)
raise CommandExecutionError("Invalid principal {}".format(principal))
raise CommandExecutionError(f"Invalid principal {principal}")
def get_name(principal):
@ -1249,7 +1243,7 @@ def get_name(principal):
# https://docs.microsoft.com/en-us/previous-versions/technet-magazine/cc138011(v=msdn.10)
# https://docs.microsoft.com/en-us/previous-versions/windows/it-pro/windows-server-2008-R2-and-2008/dd548356(v=ws.10)
if str_sid.startswith("S-1-5-80"):
name = "NT Service\\{}".format(name)
name = f"NT Service\\{name}"
return name
except (pywintypes.error, TypeError) as exc:
@ -1261,10 +1255,10 @@ def get_name(principal):
# All capability SIDs begin with `S-1-15-3`, so we'll only throw an
# error when the sid does not begin with `S-1-15-3`
if not str_sid.startswith("S-1-15-3"):
message = 'Error resolving "{}"'.format(principal)
message = f'Error resolving "{principal}"'
if type(exc) == pywintypes.error:
win_error = win32api.FormatMessage(exc.winerror).rstrip("\n")
message = "{}: {}".format(message, win_error)
message = f"{message}: {win_error}"
log.exception(message)
raise CommandExecutionError(message, exc)
@ -1326,7 +1320,7 @@ def get_owner(obj_name, obj_type="file"):
try:
obj_type_flag = flags().obj_type[obj_type.lower()]
except KeyError:
raise SaltInvocationError('Invalid "obj_type" passed: {}'.format(obj_type))
raise SaltInvocationError(f'Invalid "obj_type" passed: {obj_type}')
if obj_type in ["registry", "registry32"]:
obj_name = dacl().get_reg_name(obj_name)
@ -1348,7 +1342,7 @@ def get_owner(obj_name, obj_type="file"):
else:
log.exception("Failed to get the owner: %s", obj_name)
raise CommandExecutionError(
"Failed to get owner: {}".format(obj_name), exc.strerror
f"Failed to get owner: {obj_name}", exc.strerror
)
return get_name(owner_sid)
@ -1411,7 +1405,7 @@ def get_primary_group(obj_name, obj_type="file"):
try:
obj_type_flag = flags().obj_type[obj_type.lower()]
except KeyError:
raise SaltInvocationError('Invalid "obj_type" passed: {}'.format(obj_type))
raise SaltInvocationError(f'Invalid "obj_type" passed: {obj_type}')
if "registry" in obj_type.lower():
obj_name = dacl().get_reg_name(obj_name)
@ -1434,7 +1428,7 @@ def get_primary_group(obj_name, obj_type="file"):
else:
log.exception("Failed to get the primary group: %s", obj_name)
raise CommandExecutionError(
"Failed to get primary group: {}".format(obj_name), exc.strerror
f"Failed to get primary group: {obj_name}", exc.strerror
)
return get_name(win32security.ConvertSidToStringSid(primary_group_gid))
@ -1476,7 +1470,7 @@ def set_owner(obj_name, principal, obj_type="file"):
# Validate obj_type
if obj_type.lower() not in obj_flags.obj_type:
raise SaltInvocationError('Invalid "obj_type" passed: {}'.format(obj_type))
raise SaltInvocationError(f'Invalid "obj_type" passed: {obj_type}')
if "registry" in obj_type.lower():
obj_name = dacl().get_reg_name(obj_name)
@ -1513,9 +1507,7 @@ def set_owner(obj_name, principal, obj_type="file"):
)
except pywintypes.error as exc:
log.exception("Failed to make %s the owner: %s", principal, exc)
raise CommandExecutionError(
"Failed to set owner: {}".format(obj_name), exc.strerror
)
raise CommandExecutionError(f"Failed to set owner: {obj_name}", exc.strerror)
return True
@ -1561,7 +1553,7 @@ def set_primary_group(obj_name, principal, obj_type="file"):
# Validate obj_type
if obj_type.lower() not in obj_flags.obj_type:
raise SaltInvocationError('Invalid "obj_type" passed: {}'.format(obj_type))
raise SaltInvocationError(f'Invalid "obj_type" passed: {obj_type}')
if "registry" in obj_type.lower():
obj_name = dacl().get_reg_name(obj_name)
@ -1599,7 +1591,7 @@ def set_primary_group(obj_name, principal, obj_type="file"):
except pywintypes.error as exc:
log.exception("Failed to make %s the primary group: %s", principal, exc)
raise CommandExecutionError(
"Failed to set primary group: {}".format(obj_name), exc.strerror
f"Failed to set primary group: {obj_name}", exc.strerror
)
return True
@ -1819,9 +1811,7 @@ def has_permission(
"""
# Validate access_mode
if access_mode.lower() not in ["grant", "deny"]:
raise SaltInvocationError(
'Invalid "access_mode" passed: {}'.format(access_mode)
)
raise SaltInvocationError(f'Invalid "access_mode" passed: {access_mode}')
access_mode = access_mode.lower()
# Get the DACL
@ -1838,7 +1828,7 @@ def has_permission(
obj_dacl.ace_perms[obj_type]["advanced"].get(permission.lower(), False),
)
if not chk_flag:
raise SaltInvocationError('Invalid "permission" passed: {}'.format(permission))
raise SaltInvocationError(f'Invalid "permission" passed: {permission}')
# Check each ace for sid and type
cur_flag = None
@ -1919,9 +1909,7 @@ def has_permissions(
# Validate access_mode
if access_mode.lower() not in ["grant", "deny"]:
raise SaltInvocationError(
'Invalid "access_mode" passed: {}'.format(access_mode)
)
raise SaltInvocationError(f'Invalid "access_mode" passed: {access_mode}')
access_mode = access_mode.lower()
# Get the DACL
@ -1940,9 +1928,7 @@ def has_permissions(
obj_dacl.ace_perms[obj_type]["advanced"].get(permission.lower(), False),
)
if not chk_flag:
raise SaltInvocationError(
'Invalid "permission" passed: {}'.format(permission)
)
raise SaltInvocationError(f'Invalid "permission" passed: {permission}')
# Check each ace for sid and type
cur_flag = None
@ -1998,7 +1984,7 @@ def set_inheritance(obj_name, enabled, obj_type="file", clear=False):
"""
if obj_type not in ["file", "registry", "registry32"]:
raise SaltInvocationError(
"obj_type called with incorrect parameter: {}".format(obj_name)
f"obj_type called with incorrect parameter: {obj_name}"
)
if clear:
@ -2161,7 +2147,7 @@ def copy_security(
try:
obj_type_flag = flags().obj_type[obj_type.lower()]
except KeyError:
raise SaltInvocationError('Invalid "obj_type" passed: {}'.format(obj_type))
raise SaltInvocationError(f'Invalid "obj_type" passed: {obj_type}')
security_flags = 0
if copy_owner:
@ -2214,9 +2200,7 @@ def copy_security(
target, obj_type_flag, security_flags, sd_sid, sd_gid, sd_dacl, sd_sacl
)
except pywintypes.error as exc:
raise CommandExecutionError(
"Failed to set security info: {}".format(exc.strerror)
)
raise CommandExecutionError(f"Failed to set security info: {exc.strerror}")
return True
@ -2254,7 +2238,7 @@ def _check_perms(obj_name, obj_type, new_perms, access_mode, ret, test_mode=Fals
dict: A dictionary of return data as expected by the state system
"""
access_mode = access_mode.lower()
perms_label = "{}_perms".format(access_mode)
perms_label = f"{access_mode}_perms"
cur_perms = get_permissions(obj_name=obj_name, obj_type=obj_type)
changes = {}
for user in new_perms:
@ -2349,7 +2333,7 @@ def check_perms(
deny_perms=None,
inheritance=True,
reset=False,
test_mode=None,
test_mode=False,
):
"""
Check owner and permissions for the passed directory. This function checks
@ -2429,12 +2413,9 @@ def check_perms(
}
})
"""
if test_mode is None:
test_mode = __opts__["test"]
# Validate obj_type
if obj_type.lower() not in flags().obj_type:
raise SaltInvocationError('Invalid "obj_type" passed: {}'.format(obj_type))
raise SaltInvocationError(f'Invalid "obj_type" passed: {obj_type}')
obj_type = obj_type.lower()
@ -2460,9 +2441,7 @@ def check_perms(
ret["changes"]["owner"] = owner
except CommandExecutionError:
ret["result"] = False
ret["comment"].append(
'Failed to change owner to "{}"'.format(owner)
)
ret["comment"].append(f'Failed to change owner to "{owner}"')
# Check inheritance
if inheritance is not None:

12
tests/conftest.py
View file

@ -813,6 +813,17 @@ def groups_collection_modifyitems(config, items):
# Just one group, don't do any filtering
return
terminal_reporter = config.pluginmanager.get_plugin("terminalreporter")
if config.getoption("--last-failed") or config.getoption("--failed-first"):
# This is a test failure rerun, applying test groups would break this
terminal_reporter.write(
"\nNot splitting collected tests into chunks since --lf/--last-failed or "
"-ff/--failed-first was passed on the CLI.\n",
yellow=True,
)
return
total_items = len(items)
# Devide into test groups
@ -828,7 +839,6 @@ def groups_collection_modifyitems(config, items):
if deselected:
config.hook.pytest_deselected(items=deselected)
terminal_reporter = config.pluginmanager.get_plugin("terminalreporter")
terminal_reporter.write(
f"Running test group #{group_id}(out of #{group_count}) ({len(items)} out of {total_items} tests)\n",
yellow=True,

2
tests/integration/modules/test_mac_power.py
View file

@ -230,7 +230,7 @@ class MacPowerModuleTestRestartPowerFailure(ModuleCase):
Reset to original value
"""
if self.RESTART_POWER is not None:
self.run_function("power.set_sleep_on_power_button", [self.SLEEP_ON_BUTTON])
self.run_function("power.set_sleep_on_power_button", [self.RESTART_POWER])
def test_restart_power_failure(self):
"""

2
tests/pytests/functional/modules/test_mac_pkgutil.py
View file

@ -56,6 +56,8 @@ def macports_package_url(macports_package_filename):
@pytest.fixture(scope="module")
def pkg_name(grains):
if grains["osrelease_info"][0] >= 12:
return "com.apple.pkg.XcodeSystemResources"
if grains["osrelease_info"][0] >= 11:
return "com.apple.pkg.InstallAssistant.macOSBigSur"
if grains["osrelease_info"][:2] == (10, 15):

14
tests/pytests/functional/modules/win_file/test_check_perms.py
View file

@ -21,9 +21,9 @@ def configure_loader_modules():
"__utils__": {
"dacl.check_perms": win_dacl.check_perms,
"dacl.set_perms": win_dacl.set_perms,
}
},
"__opts__": {"test": False},
},
win_dacl: {"__opts__": {"test": False}},
}
@ -43,7 +43,7 @@ def test_check_perms_set_owner_test_true(test_file):
"name": str(test_file),
"result": None,
}
with patch.dict(win_dacl.__opts__, {"test": True}):
with patch.dict(win_file.__opts__, {"test": True}):
result = win_file.check_perms(
path=str(test_file), owner="Backup Operators", inheritance=None
)
@ -76,7 +76,7 @@ def test_check_perms_deny_test_true(test_file):
"name": str(test_file),
"result": None,
}
with patch.dict(win_dacl.__opts__, {"test": True}):
with patch.dict(win_file.__opts__, {"test": True}):
result = win_file.check_perms(
path=str(test_file),
deny_perms={"Users": {"perms": "read_execute"}},
@ -113,7 +113,7 @@ def test_check_perms_grant_test_true(test_file):
"name": str(test_file),
"result": None,
}
with patch.dict(win_dacl.__opts__, {"test": True}):
with patch.dict(win_file.__opts__, {"test": True}):
result = win_file.check_perms(
path=str(test_file),
grant_perms={"Users": {"perms": "read_execute"}},
@ -150,7 +150,7 @@ def test_check_perms_inheritance_false_test_true(test_file):
"name": str(test_file),
"result": None,
}
with patch.dict(win_dacl.__opts__, {"test": True}):
with patch.dict(win_file.__opts__, {"test": True}):
result = win_file.check_perms(path=str(test_file), inheritance=False)
assert result == expected
@ -214,7 +214,7 @@ def test_check_perms_reset_test_true(test_file):
"name": str(test_file),
"result": None,
}
with patch.dict(win_dacl.__opts__, {"test": True}):
with patch.dict(win_file.__opts__, {"test": True}):
result = win_file.check_perms(
path=str(test_file),
grant_perms={

41
tests/pytests/functional/states/file/test_directory_win.py
View file

@ -2,8 +2,6 @@ import os
import pytest
import salt.modules.win_file as win_file
import salt.states.file as file
import salt.utils.win_dacl as win_dacl
import salt.utils.win_functions as win_functions
@ -20,28 +18,7 @@ pytestmark = [
]
@pytest.fixture(scope="module")
def configure_loader_modules():
return {
file: {
"__opts__": {"test": False},
"__salt__": {
"file.mkdir": win_file.mkdir,
"file.check_perms": win_file.check_perms,
},
},
win_file: {
"__utils__": {
"dacl.check_perms": win_dacl.check_perms,
"dacl.set_owner": win_dacl.set_owner,
"dacl.set_perms": win_dacl.set_perms,
},
},
win_dacl: {"__opts__": {"test": False}},
}
def test_directory_new(tmp_path):
def test_directory_new(file, tmp_path):
"""
Test file.directory when the directory does not exist
Should just return "New Dir"
@ -107,7 +84,7 @@ def test_directory_new(tmp_path):
assert permissions == expected
def test_directory_new_no_inherit(tmp_path):
def test_directory_new_no_inherit(file, tmp_path):
"""
Test file.directory when the directory does not exist
Should just return "New Dir"
@ -127,7 +104,7 @@ def test_directory_new_no_inherit(tmp_path):
assert permissions["Inherited"] == {}
def test_directory_new_reset(tmp_path):
def test_directory_new_reset(file, tmp_path):
"""
Test file.directory when the directory does not exist
Should just return "New Dir"
@ -182,7 +159,7 @@ def test_directory_new_reset(tmp_path):
assert permissions == expected
def test_directory_new_reset_no_inherit(tmp_path):
def test_directory_new_reset_no_inherit(file, tmp_path):
"""
Test file.directory when the directory does not exist
Should just return "New Dir"
@ -219,7 +196,7 @@ def test_directory_new_reset_no_inherit(tmp_path):
assert permissions == expected
def test_directory_existing(tmp_path):
def test_directory_existing(file, tmp_path):
path = str(tmp_path)
ret = file.directory(
name=path,
@ -293,7 +270,7 @@ def test_directory_existing(tmp_path):
assert permissions == expected
def test_directory_existing_existing_user(tmp_path):
def test_directory_existing_existing_user(file, tmp_path):
path = str(tmp_path)
win_dacl.set_permissions(
obj_name=path,
@ -374,7 +351,7 @@ def test_directory_existing_existing_user(tmp_path):
assert permissions == expected
def test_directory_existing_no_inherit(tmp_path):
def test_directory_existing_no_inherit(file, tmp_path):
path = str(tmp_path)
ret = file.directory(
name=path,
@ -398,7 +375,7 @@ def test_directory_existing_no_inherit(tmp_path):
assert permissions["Inherited"] == {}
def test_directory_existing_reset(tmp_path):
def test_directory_existing_reset(file, tmp_path):
path = str(tmp_path)
win_dacl.set_permissions(
obj_name=path,
@ -462,7 +439,7 @@ def test_directory_existing_reset(tmp_path):
assert permissions == expected
def test_directory_existing_reset_no_inherit(tmp_path):
def test_directory_existing_reset_no_inherit(file, tmp_path):
path = str(tmp_path)
ret = file.directory(
name=path,

21
tests/pytests/functional/states/test_reg.py
View file

@ -29,7 +29,7 @@ class RegVars:
self.key = "SOFTWARE\\Salt-Testing"
self.hive = self.hive
self.key = self.key
self.name = "{}\\{}".format(self.hive, self.key)
self.name = f"{self.hive}\\{self.key}"
self.vname = "version"
self.vdata = "0.15.3"
self.current_user = win_functions.get_current_user(with_domain=False)
@ -49,7 +49,6 @@ def configure_loader_modules():
"dacl.check_perms": win_dacl.check_perms,
},
},
win_dacl: {"__opts__": {"test": False}},
}
@ -98,7 +97,7 @@ def test_present(reg_vars):
Test reg.present
"""
expected = {
"comment": "Added {} to {}".format(reg_vars.vname, reg_vars.name),
"comment": f"Added {reg_vars.vname} to {reg_vars.name}",
"changes": {
"reg": {
"Added": {
@ -235,7 +234,7 @@ def test_present_string_dword(reg_vars, clean):
vtype = "REG_DWORD"
expected_vdata = 1
expected = {
"comment": "Added {} to {}".format(vname, reg_vars.name),
"comment": f"Added {vname} to {reg_vars.name}",
"changes": {
"reg": {
"Added": {
@ -269,7 +268,7 @@ def test_present_string_dword_existing(reg_vars, clean):
hive=reg_vars.hive, key=reg_vars.key, vname=vname, vdata=vdata, vtype=vtype
)
expected = {
"comment": "{} in {} is already present".format(vname, reg_vars.name),
"comment": f"{vname} in {reg_vars.name} is already present",
"changes": {},
"name": reg_vars.name,
"result": True,
@ -305,7 +304,7 @@ def test_present_test_true(reg_vars, clean):
def test_present_existing(reg_vars, reset):
expected = {
"comment": "{} in {} is already present".format(reg_vars.vname, reg_vars.name),
"comment": f"{reg_vars.vname} in {reg_vars.name} is already present",
"changes": {},
"name": reg_vars.name,
"result": True,
@ -324,7 +323,7 @@ def test_present_existing_key_only(reg_vars, clean):
reg_util.set_value(hive=reg_vars.hive, key=reg_vars.key)
expected = {
"comment": "(Default) in {} is already present".format(reg_vars.name),
"comment": f"(Default) in {reg_vars.name} is already present",
"changes": {},
"name": reg_vars.name,
"result": True,
@ -334,7 +333,7 @@ def test_present_existing_key_only(reg_vars, clean):
def test_present_existing_test_true(reg_vars, reset):
expected = {
"comment": "{} in {} is already present".format(reg_vars.vname, reg_vars.name),
"comment": f"{reg_vars.vname} in {reg_vars.name} is already present",
"changes": {},
"name": reg_vars.name,
"result": True,
@ -351,7 +350,7 @@ def test_absent(reg_vars, reset):
Test to remove a registry entry.
"""
expected = {
"comment": "Removed {} from {}".format(reg_vars.key, reg_vars.hive),
"comment": f"Removed {reg_vars.key} from {reg_vars.hive}",
"changes": {
"reg": {"Removed": {"Entry": reg_vars.vname, "Key": reg_vars.name}}
},
@ -380,7 +379,7 @@ def test_absent_already_absent(reg_vars, clean):
Test to remove a registry entry.
"""
expected = {
"comment": "{} is already absent".format(reg_vars.name),
"comment": f"{reg_vars.name} is already absent",
"changes": {},
"name": reg_vars.name,
"result": True,
@ -393,7 +392,7 @@ def test_absent_already_absent_test_true(reg_vars, clean):
Test to remove a registry entry.
"""
expected = {
"comment": "{} is already absent".format(reg_vars.name),
"comment": f"{reg_vars.name} is already absent",
"changes": {},
"name": reg_vars.name,
"result": True,

17
tests/pytests/functional/test_payload.py
View file

@ -43,7 +43,7 @@ class EchoServer:
"""
context = zmq.Context()
socket = context.socket(zmq.REP)
socket.bind("tcp://*:{}".format(port))
socket.bind(f"tcp://*:{port}")
while event.is_set():
try:
# Wait for next request from client
@ -77,7 +77,7 @@ def echo_server(echo_port):
@pytest.fixture
def sreq(echo_port):
yield salt.payload.SREQ("tcp://127.0.0.1:{}".format(echo_port))
yield salt.payload.SREQ(f"tcp://127.0.0.1:{echo_port}")
@pytest.mark.slow_test
@ -138,6 +138,19 @@ def test_destroy(sreq, echo_server):
"""
Test the __del__ capabilities
"""
# ensure we actually have an open socket and not just testing against
# no actual sockets created.
assert sreq.send("clear", "foo") == {"enc": "clear", "load": "foo"}
# ensure no exceptions when we go to destroy the sreq, since __del__
# swallows exceptions, we have to call destroy directly
sreq.destroy()
@pytest.mark.slow_test
def test_clear_socket(sreq, echo_server):
# ensure we actually have an open socket and not just testing against
# no actual sockets created.
assert sreq.send("clear", "foo") == {"enc": "clear", "load": "foo"}
assert hasattr(sreq, "_socket")
sreq.clear_socket()
assert hasattr(sreq, "_socket") is False

31
tests/pytests/functional/utils/win_dacl/test_file.py
View file

@ -1,7 +1,6 @@
import pytest
import salt.utils.win_dacl as win_dacl
from tests.support.mock import patch
pytestmark = [
pytest.mark.windows_whitelisted,
@ -819,22 +818,22 @@ def test_check_perms(test_file):
def test_check_perms_test_true(test_file):
with patch.dict(win_dacl.__opts__, {"test": True}):
result = win_dacl.check_perms(
obj_name=str(test_file),
obj_type="file",
ret=None,
owner="Users",
grant_perms={"Backup Operators": {"perms": "read"}},
deny_perms={
"NETWORK SERVICE": {
"perms": ["delete", "set_value", "write_dac", "write_owner"]
},
"Backup Operators": {"perms": ["delete"]},
result = win_dacl.check_perms(
obj_name=str(test_file),
obj_type="file",
ret=None,
owner="Users",
grant_perms={"Backup Operators": {"perms": "read"}},
deny_perms={
"NETWORK SERVICE": {
"perms": ["delete", "set_value", "write_dac", "write_owner"]
},
inheritance=True,
reset=False,
)
"Backup Operators": {"perms": ["delete"]},
},
inheritance=True,
reset=False,
test_mode=True,
)
expected = {
"changes": {

42
tests/pytests/functional/utils/win_dacl/test_reg.py
View file

@ -3,7 +3,6 @@ from saltfactories.utils import random_string
import salt.utils.win_dacl as win_dacl
import salt.utils.win_reg as win_reg
from tests.support.mock import patch
pytestmark = [
pytest.mark.windows_whitelisted,
@ -12,15 +11,6 @@ pytestmark = [
]
@pytest.fixture
def configure_loader_modules(minion_opts):
return {
win_dacl: {
"__opts__": minion_opts,
},
}
@pytest.fixture(scope="module")
def fake_key():
return "SOFTWARE\\{}".format(random_string("SaltTesting-", lowercase=False))
@ -29,7 +19,7 @@ def fake_key():
@pytest.fixture(scope="function")
def reg_key(fake_key):
win_reg.set_value(hive="HKLM", key=fake_key, vname="fake_name", vdata="fake_data")
yield "HKLM\\{}".format(fake_key)
yield f"HKLM\\{fake_key}"
win_reg.delete_key_recursive(hive="HKLM", key=fake_key)
@ -433,22 +423,22 @@ def test_check_perms(reg_key):
def test_check_perms_test_true(reg_key):
with patch.dict(win_dacl.__opts__, {"test": True}):
result = win_dacl.check_perms(
obj_name=reg_key,
obj_type="registry",
ret=None,
owner="Users",
grant_perms={"Backup Operators": {"perms": "read"}},
deny_perms={
"NETWORK SERVICE": {
"perms": ["delete", "set_value", "write_dac", "write_owner"]
},
"Backup Operators": {"perms": ["delete"]},
result = win_dacl.check_perms(
obj_name=reg_key,
obj_type="registry",
ret=None,
owner="Users",
grant_perms={"Backup Operators": {"perms": "read"}},
deny_perms={
"NETWORK SERVICE": {
"perms": ["delete", "set_value", "write_dac", "write_owner"]
},
inheritance=True,
reset=False,
)
"Backup Operators": {"perms": ["delete"]},
},
inheritance=True,
reset=False,
test_mode=True,
)
expected = {
"changes": {

11
tests/pytests/integration/client/test_runner.py
View file

@ -150,3 +150,14 @@ def test_invalid_kwargs_are_ignored(client, auth_creds):
ret = client.cmd_sync(low.copy())
assert ret
assert ret[0] == "foo"
def test_get_docs(client):
ret = client.get_docs(arg="*")
assert "auth.del_token" in ret
assert "auth.mk_token" in ret
assert "cache.clear_pillar" in ret
assert "cache.grains" in ret
assert "state.soft_kill" in ret
assert "virt.start" in ret
assert "test.arg" in ret

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

@ -57,7 +57,10 @@ def configure_loader_modules():
ret.update(
{
win_dacl: {"__opts__": opts},
win_file: {"__utils__": {"dacl.check_perms": win_dacl.check_perms}},
win_file: {
"__utils__": {"dacl.check_perms": win_dacl.check_perms},
"__opts__": opts,
},
}
)
@ -203,9 +206,7 @@ def test_replace_append(multiline_file):
with salt.utils.files.fopen(multiline_file, "rb") as fp:
assert (
salt.utils.stringutils.to_bytes(
os.linesep.join(
["#-- START BLOCK 2", "{}#-- END BLOCK 2".format(new_content)]
)
os.linesep.join(["#-- START BLOCK 2", f"{new_content}#-- END BLOCK 2"])
)
in fp.read()
)
@ -246,9 +247,7 @@ def test_replace_insert_after(multiline_file):
with salt.utils.files.fopen(multiline_file, "rb") as fp:
assert (
salt.utils.stringutils.to_bytes(
os.linesep.join(
["#-- START BLOCK 2", "{}#-- END BLOCK 2".format(new_content)]
)
os.linesep.join(["#-- START BLOCK 2", f"{new_content}#-- END BLOCK 2"])
)
in fp.read()
)
@ -325,9 +324,7 @@ def test_replace_prepend(multiline_file):
with salt.utils.files.fopen(multiline_file, "rb") as fp:
assert (
salt.utils.stringutils.to_bytes(
os.linesep.join(
["#-- START BLOCK 2", "{}#-- END BLOCK 2".format(new_content)]
)
os.linesep.join(["#-- START BLOCK 2", f"{new_content}#-- END BLOCK 2"])
)
not in fp.read()
)
@ -352,7 +349,7 @@ def test_replace_prepend(multiline_file):
os.linesep.join(
[
"#-- START BLOCK 2",
"{}#-- END BLOCK 2".format(new_content),
f"{new_content}#-- END BLOCK 2",
]
)
)
@ -394,9 +391,7 @@ def test_replace_insert_before(multiline_file):
with salt.utils.files.fopen(multiline_file, "rb") as fp:
assert (
salt.utils.stringutils.to_bytes(
os.linesep.join(
["#-- START BLOCK 2", "{}#-- END BLOCK 2".format(new_content)]
)
os.linesep.join(["#-- START BLOCK 2", f"{new_content}#-- END BLOCK 2"])
)
in fp.read()
)
@ -428,7 +423,7 @@ def test_replace_partial_marked_lines(multiline_file):
def test_backup(multiline_file):
fext = ".bak"
bak_file = "{}{}".format(multiline_file, fext)
bak_file = f"{multiline_file}{fext}"
if salt.utils.platform.is_windows():
check_perms_patch = win_file.check_perms
@ -448,7 +443,7 @@ def test_backup(multiline_file):
assert not os.path.exists(bak_file)
fext = ".bak"
bak_file = "{}{}".format(multiline_file, fext)
bak_file = f"{multiline_file}{fext}"
if salt.utils.platform.is_windows():
check_perms_patch = win_file.check_perms

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

@ -1711,37 +1711,39 @@ def test_latest_version_fromrepo_multiple_names():
"linux-cloud-tools-virtual": ["5.15.0.69.67"],
"linux-generic": ["5.15.0.69.67"],
}
apt_ret_cloud = {
apt_ret = {
"pid": 4361,
"retcode": 0,
"stdout": "linux-cloud-tools-virtual:\n"
f"Installed: 5.15.0.69.67\n Candidate: {version}\n Version"
f"table:\n {version} 990\n 990"
f"https://mirrors.edge.kernel.org/ubuntu {fromrepo}/main amd64"
"Packages\n 500 https://mirrors.edge.kernel.org/ubuntu"
"jammy-security/main amd64 Packages\n ***5.15.0.69.67 100\n"
"100 /var/lib/dpkg/status\n 5.15.0.25.27 500\n 500"
"https://mirrors.edge.kernel.org/ubuntu jammy/main amd64 Packages",
"stderr": "",
}
apt_ret_generic = {
"pid": 4821,
"retcode": 0,
"stdout": "linux-generic:\n"
f"Installed: 5.15.0.69.67\n Candidate: {version}\n"
f"Version table:\n {version} 990\n 990"
"https://mirrors.edge.kernel.org/ubuntu"
"jammy-updates/main amd64 Packages\n 500"
"https://mirrors.edge.kernel.org/ubuntu"
"jammy-security/main amd64 Packages\n *** 5.15.0.69.67"
"100\n 100 /var/lib/dpkg/status\n 5.15.0.25.27"
"500\n 500 https://mirrors.edge.kernel.org/ubuntu"
"jammy/main amd64 Packages",
"stdout": textwrap.dedent(
f"""\
linux-cloud-tools-virtual:
Installed: 5.15.0.69.67
Candidate: {version}
Versiontable:
{version} 990
990https://mirrors.edge.kernel.org/ubuntu {fromrepo}/main amd64Packages
500 https://mirrors.edge.kernel.org/ubuntujammy-security/main amd64 Packages
***5.15.0.69.67 100
100 /var/lib/dpkg/status
5.15.0.25.27 500
500https://mirrors.edge.kernel.org/ubuntu jammy/main amd64 Packages
linux-generic:
Installed: 5.15.0.69.67
Candidate: {version}
Version table:
{version} 990
990https://mirrors.edge.kernel.org/ubuntujammy-updates/main amd64 Packages
500https://mirrors.edge.kernel.org/ubuntujammy-security/main amd64 Packages
*** 5.15.0.69.67100
100 /var/lib/dpkg/status
5.15.0.25.27500
500 https://mirrors.edge.kernel.org/ubuntujammy/main amd64 Packages
"""
),
"stderr": "",
}
mock_apt = MagicMock()
mock_apt.side_effect = [apt_ret_cloud, apt_ret_generic]
mock_apt = MagicMock(return_value=apt_ret)
patch_apt = patch("salt.modules.aptpkg._call_apt", mock_apt)
mock_list_pkgs = MagicMock(return_value=list_ret)
patch_list_pkgs = patch("salt.modules.aptpkg.list_pkgs", mock_list_pkgs)
@ -1754,30 +1756,18 @@ def test_latest_version_fromrepo_multiple_names():
show_installed=True,
)
assert ret == {"linux-cloud-tools-virtual": version, "linux-generic": version}
assert mock_apt.call_args_list == [
call(
[
"apt-cache",
"-q",
"policy",
"linux-cloud-tools-virtual",
"-o",
"APT::Default-Release=jammy-updates",
],
scope=False,
),
call(
[
"apt-cache",
"-q",
"policy",
"linux-generic",
"-o",
"APT::Default-Release=jammy-updates",
],
scope=False,
),
]
mock_apt.assert_called_once_with(
[
"apt-cache",
"-q",
"policy",
"linux-cloud-tools-virtual",
"linux-generic",
"-o",
"APT::Default-Release=jammy-updates",
],
scope=False,
)
def test_hold():

211
tests/pytests/unit/modules/test_win_file.py
View file

@ -1,13 +1,43 @@
import os
import re
import pytest
import salt.modules.win_file as win_file
import salt.utils.user
import salt.utils.win_dacl
from salt.exceptions import CommandExecutionError
from tests.support.mock import patch
pytestmark = [pytest.mark.windows_whitelisted, pytest.mark.skip_unless_on_windows]
@pytest.fixture
def configure_loader_modules():
return {
win_file: {},
salt.utils.win_dacl: {},
}
def test__virtual__not_windows():
with patch("salt.utils.platform.is_windows", autospec=True, return_value=False):
expected = (False, "Module win_file: Missing Win32 modules")
result = win_file.__virtual__()
assert result == expected
with patch("salt.modules.win_file.HAS_WINDOWS_MODULES", False):
expected = (False, "Module win_file: Missing Win32 modules")
result = win_file.__virtual__()
assert result == expected
def test__virtual__no_dacl():
with patch("salt.modules.win_file.HAS_WIN_DACL", False):
expected = (False, "Module win_file: Unable to load salt.utils.win_dacl")
result = win_file.__virtual__()
assert result == expected
def test__get_version_os():
expected = ["32-bit Windows", "Windows NT"]
result = win_file._get_version_os(0x00040004)
@ -56,6 +86,187 @@ def test__get_version_sys():
assert regex.search(result)
def test_get_pgid_error():
with pytest.raises(CommandExecutionError):
win_file.get_pgid("C:\\Path\\That\\Does\\Not\\Exist.txt")
def test_get_pgid():
"""
We can't know what this value is, so we're just making sure it found
something
"""
result = win_file.get_pgid(os.getenv("COMSPEC"))
assert result != ""
def test_group_to_gid():
with patch.dict(win_file.__opts__, {}):
result = win_file.group_to_gid("Administrators")
expected = "S-1-5-32-544"
assert result == expected
def test_group_to_gid_empty():
with patch.dict(win_file.__opts__, {}):
result = win_file.group_to_gid("")
expected = "S-1-5-32"
assert result == expected
def test_uid_to_user():
result = win_file.uid_to_user("S-1-5-32-544")
expected = "Administrators"
assert result == expected
def test_uid_to_user_empty():
result = win_file.uid_to_user("")
expected = ""
assert result == expected
def test_user_to_uid():
result = win_file.user_to_uid("Administrator")
expected = salt.utils.win_dacl.get_sid_string("Administrator")
assert result == expected
def test_user_to_uid_none():
result = win_file.user_to_uid(None)
expected = salt.utils.win_dacl.get_sid_string(salt.utils.user.get_user())
assert result == expected
def test_get_uid():
"""
We can't know what this value is, so we're just making sure it found
something
"""
result = win_file.get_uid(os.getenv("WINDIR"))
assert result != ""
def test_get_uid_error():
with pytest.raises(CommandExecutionError):
win_file.get_uid("C:\\fake\\path")
def test_chown(tmp_path):
test_file = tmp_path / "test_file.txt"
test_file.touch()
win_file.chown(path=str(test_file), user="Administrators", pgroup="Guests")
assert win_file.get_user(str(test_file)) == "Administrators"
assert win_file.get_pgroup(str(test_file)) == "Guests"
def test_chpgrp(tmp_path):
test_file = tmp_path / "test_file.txt"
test_file.touch()
win_file.chown(path=str(test_file), user="Administrators", pgroup="Guests")
win_file.chpgrp(path=str(test_file), group="Administrators")
assert win_file.get_pgroup(str(test_file)) == "Administrators"
def test_stats_mode(tmp_path):
test_file = tmp_path / "test_file.txt"
test_file.touch()
results = win_file.stats(str(test_file))
assert results["mode"] == "0666"
def test_is_link_true(tmp_path):
test_source = tmp_path / "test_source.txt"
test_link = tmp_path / "test_link.txt"
test_source.touch()
test_link.symlink_to(test_source)
results = win_file.is_link(str(test_link))
expected = True
assert results == expected
def test_is_link_false(tmp_path):
test_file = tmp_path / "test_not_link.txt"
test_file.touch()
results = win_file.is_link(str(test_file))
expected = False
assert results == expected
def test_mkdir(tmp_path):
test_dir = tmp_path / "test_dir"
grant_perms = {"Guests": {"perms": "full_control"}}
win_file.mkdir(
path=str(test_dir),
owner="Administrators",
grant_perms=grant_perms,
)
owner = win_file.get_user(str(test_dir))
assert owner == "Administrators"
perms = salt.utils.win_dacl.get_permissions(str(test_dir))
assert perms["Not Inherited"]["Guests"]["grant"]["permissions"] == "Full control"
def test_check_perms(tmp_path):
test_dir = tmp_path / "test_dir"
test_dir.mkdir()
grant_perms = {"Guests": {"perms": "full_control"}}
ret = {}
with patch.dict(salt.utils.win_dacl.__opts__, {"test": False}):
result = win_file.check_perms(
path=str(test_dir),
ret=ret,
owner="Guests",
grant_perms=grant_perms,
)
expected = {
"changes": {
"grant_perms": {
"Guests": {
"permissions": "full_control",
},
},
"owner": "Guests",
},
"comment": "",
"name": str(test_dir),
"result": True,
}
assert result == expected
owner = win_file.get_user(str(test_dir))
assert owner == "Guests"
perms = salt.utils.win_dacl.get_permissions(str(test_dir))
assert perms["Not Inherited"]["Guests"]["grant"]["permissions"] == "Full control"
def test_set_perms(tmp_path):
test_dir = tmp_path / "test_dir"
test_dir.mkdir()
grant_perms = {"Guests": {"perms": "full_control"}}
win_file.set_perms(
path=str(test_dir),
grant_perms=grant_perms,
)
perms = salt.utils.win_dacl.get_permissions(str(test_dir))
assert perms["Not Inherited"]["Guests"]["grant"]["permissions"] == "Full control"
def test_get_user():
"""
We can't know what this value is, so we're just making sure it found
something
"""
result = win_file.get_user(os.getenv("WINDIR"))
assert result != ""
def test_get_user_error():
with pytest.raises(CommandExecutionError):
win_file.get_user("C:\\fake\\path")
def test_version_missing_file():
with pytest.raises(CommandExecutionError):
win_file.version("C:\\Windows\\bogus.exe")

93
tests/pytests/unit/test_payload.py
View file

@ -7,8 +7,11 @@ import copy
import datetime
import logging
import zmq
import salt.exceptions
import salt.payload
import salt.utils.msgpack
from salt.defaults import _Constant
from salt.utils import immutabletypes
from salt.utils.odict import OrderedDict
@ -210,3 +213,93 @@ def test_constants():
sdata = salt.payload.dumps(constant)
odata = salt.payload.loads(sdata)
assert odata == constant
def test_package():
value = salt.utils.msgpack.dumps("test")
assert salt.payload.package("test") == value
def test_unpackage():
value = [b"test"]
packed = salt.utils.msgpack.dumps(value)
assert salt.payload.unpackage(packed) == value
def test_format_payload():
expected = salt.utils.msgpack.dumps(
{"enc": [b"test"], "load": {"kwargs": {"foo": "bar"}}}
)
enc = [b"test"]
kwargs = {"foo": "bar"}
payload = salt.payload.format_payload(enc=enc, kwargs=kwargs)
assert payload == expected
def test_SREQ_init():
req = salt.payload.SREQ(
"tcp://salt:3434", id_=b"id", serial="msgpack", linger=1, opts=None
)
assert req.master == "tcp://salt:3434"
assert req.id_ == b"id"
assert req.linger == 1
assert req.opts is None
assert isinstance(req.context, zmq.Context)
assert isinstance(req.poller, zmq.Poller)
def test_SREQ_socket():
req = salt.payload.SREQ(
"tcp://salt:3434", id_=b"id", serial="msgpack", linger=1, opts=None
)
# socket() is a property that auto creates a socket if a socket is wanted.
socket = req.socket
assert isinstance(socket, zmq.Socket)
req = salt.payload.SREQ(
"tcp://[2001:db8:85a3:8d3:1319:8a2e:370:7348]:3434",
id_=b"id",
serial="msgpack",
linger=1,
opts=None,
)
# socket() is a property that auto creates a socket if a socket is wanted.
socket = req.socket
assert isinstance(socket, zmq.Socket)
req = salt.payload.SREQ(
"tcp://salt:3434", id_=None, serial="msgpack", linger=1, opts=None
)
# socket() is a property that auto creates a socket if a socket is wanted.
socket = req.socket
assert isinstance(socket, zmq.Socket)
def test_SREQ_set_tcp_keepalive():
opts = {"tcp_keepalive": True}
req = salt.payload.SREQ(
"tcp://salt:3434", id_=b"id", serial="msgpack", linger=1, opts=opts
)
socket = req.socket
assert req._socket.getsockopt(zmq.TCP_KEEPALIVE)
opts = {"tcp_keepalive_idle": 100}
req = salt.payload.SREQ(
"tcp://salt:3434", id_=b"id", serial="msgpack", linger=1, opts=opts
)
socket = req.socket
assert req._socket.getsockopt(zmq.TCP_KEEPALIVE_IDLE) == 100
opts = {"tcp_keepalive_cnt": 100}
req = salt.payload.SREQ(
"tcp://salt:3434", id_=b"id", serial="msgpack", linger=1, opts=opts
)
socket = req.socket
assert req._socket.getsockopt(zmq.TCP_KEEPALIVE_CNT) == 100
opts = {"tcp_keepalive_intvl": 100}
req = salt.payload.SREQ(
"tcp://salt:3434", id_=b"id", serial="msgpack", linger=1, opts=opts
)
socket = req.socket
assert req._socket.getsockopt(zmq.TCP_KEEPALIVE_INTVL) == 100

8
tests/unit/modules/test_heat.py
View file

@ -4,7 +4,6 @@ import salt.modules.file as file_
import salt.modules.heat as heat
import salt.modules.win_file as win_file
import salt.utils.platform
import salt.utils.win_dacl as dacl
from tests.support.mixins import LoaderModuleMockMixin
from tests.support.mock import MagicMock, patch
from tests.support.runtests import RUNTIME_VARS
@ -78,10 +77,7 @@ class HeatTestCase(TestCase, LoaderModuleMockMixin):
"config.backup_mode": MagicMock(return_value=False),
},
},
win_file: {
"__utils__": {"dacl.check_perms": salt.utils.win_dacl.check_perms}
},
dacl: {"__opts__": {"test": False}},
win_file: {"__opts__": {"test": False}},
}
def setUp(self):
@ -166,7 +162,7 @@ class HeatTestCase(TestCase, LoaderModuleMockMixin):
)
assert ret == {
"result": False,
"comment": "Can not open environment: {}, ".format(env_file),
"comment": f"Can not open environment: {env_file}, ",
}
def test_heat_update_stack(self):

6
tests/unit/states/test_heat.py
View file

@ -5,7 +5,6 @@ import salt.modules.heat
import salt.modules.win_file as win_file
import salt.states.heat as heat
import salt.utils.platform
import salt.utils.win_dacl as dacl
import tests.unit.modules.test_heat
from tests.support.mixins import LoaderModuleMockMixin
from tests.support.mock import MagicMock, patch
@ -38,10 +37,7 @@ class HeatTestCase(TestCase, LoaderModuleMockMixin):
"config.backup_mode": MagicMock(return_value=False),
},
},
win_file: {
"__utils__": {"dacl.check_perms": salt.utils.win_dacl.check_perms}
},
dacl: {"__opts__": {"test": False}},
win_file: {"__opts__": {"test": False}},
}
def setUp(self):