Add support for pending_reboot grain

Adds pending_reboot graint
Adds new win_system salt util
Adds tests for new grain
Adds tests for new salt util
This commit is contained in:
twangboy 2020-03-31 14:23:03 -06:00 committed by Daniel Wozniak
parent a800845776
commit 400322f318
7 changed files with 1119 additions and 152 deletions

2
.gitignore vendored
View file

@ -1,6 +1,8 @@
/build
/src
__pycache__/
*.py[co]
pkg/arch/*.tar.xz
*.sw[a-p]
doc/_build

View file

@ -0,0 +1,31 @@
# -*- coding: utf-8 -*-
'''
Grain that indicates the system is pending a reboot
See functions in salt.utils.win_system to see what conditions would indicate
a reboot is pending
'''
from __future__ import absolute_import, print_function, unicode_literals
# Import python libs
import logging
# Import salt libs
import salt.utils.platform
import salt.utils.win_system
log = logging.getLogger(__name__)
__virtualname__ = 'pending_reboot'
def __virtual__():
if not salt.utils.platform.is_windows():
return False, "'pending_reboot' grain only available on Windows"
return __virtualname__
def pending_reboot():
'''
A grain that indicates that the system is pending a reboot.
'''
return {'pending_reboot': salt.utils.win_system.get_pending_reboot()}

View file

@ -1,6 +1,7 @@
# -*- coding: utf-8 -*-
"""
Module for managing windows systems.
Module for managing Windows systems and getting Windows system information.
Support for reboot, shutdown, join domain, rename
:depends:
- pywintypes
@ -8,8 +9,6 @@ Module for managing windows systems.
- win32con
- win32net
- wmi
Support for reboot, shutdown, etc
"""
from __future__ import absolute_import, print_function, unicode_literals
@ -25,6 +24,7 @@ import salt.utils.functools
import salt.utils.locales
import salt.utils.platform
import salt.utils.winapi
import salt.utils.win_system
from salt.exceptions import CommandExecutionError
from salt.ext import six
@ -440,13 +440,7 @@ def get_pending_computer_name():
salt 'minion-id' system.get_pending_computer_name
"""
current = get_computer_name()
pending = __utils__["reg.read_value"](
"HKLM", r"SYSTEM\CurrentControlSet\Services\Tcpip\Parameters", "NV Hostname"
)["vdata"]
if pending:
return pending if pending != current else None
return False
return salt.utils.win_system.get_pending_computer_name()
def get_computer_name():
@ -462,8 +456,7 @@ def get_computer_name():
salt 'minion-id' system.get_computer_name
"""
name = win32api.GetComputerNameEx(win32con.ComputerNamePhysicalDnsHostname)
return name if name else False
return salt.utils.win_system.get_computer_name()
def set_computer_desc(desc=None):
@ -1331,16 +1324,7 @@ def get_pending_component_servicing():
salt '*' system.get_pending_component_servicing
"""
key = r"SOFTWARE\Microsoft\Windows\CurrentVersion\Component Based Servicing\RebootPending"
# So long as the registry key exists, a reboot is pending.
if __utils__["reg.key_exists"]("HKLM", key):
log.debug("Key exists: %s", key)
return True
else:
log.debug("Key does not exist: %s", key)
return False
return salt.utils.win_system.get_pending_component_servicing()
def get_pending_domain_join():
@ -1360,25 +1344,7 @@ def get_pending_domain_join():
salt '*' system.get_pending_domain_join
"""
base_key = r"SYSTEM\CurrentControlSet\Services\Netlogon"
avoid_key = r"{0}\AvoidSpnSet".format(base_key)
join_key = r"{0}\JoinDomain".format(base_key)
# If either the avoid_key or join_key is present,
# then there is a reboot pending.
if __utils__["reg.key_exists"]("HKLM", avoid_key):
log.debug("Key exists: %s", avoid_key)
return True
else:
log.debug("Key does not exist: %s", avoid_key)
if __utils__["reg.key_exists"]("HKLM", join_key):
log.debug("Key exists: %s", join_key)
return True
else:
log.debug("Key does not exist: %s", join_key)
return False
return salt.utils.win_system.get_pending_domain_join()
def get_pending_file_rename():
@ -1398,23 +1364,7 @@ def get_pending_file_rename():
salt '*' system.get_pending_file_rename
"""
vnames = ("PendingFileRenameOperations", "PendingFileRenameOperations2")
key = r"SYSTEM\CurrentControlSet\Control\Session Manager"
# If any of the value names exist and have value data set,
# then a reboot is pending.
for vname in vnames:
reg_ret = __utils__["reg.read_value"]("HKLM", key, vname)
if reg_ret["success"]:
log.debug("Found key: %s", key)
if reg_ret["vdata"] and (reg_ret["vdata"] != "(value not set)"):
return True
else:
log.debug("Unable to access key: %s", key)
return False
return salt.utils.win_system.get_pending_file_rename()
def get_pending_servermanager():
@ -1434,26 +1384,7 @@ def get_pending_servermanager():
salt '*' system.get_pending_servermanager
"""
vname = "CurrentRebootAttempts"
key = r"SOFTWARE\Microsoft\ServerManager"
# There are situations where it's possible to have '(value not set)' as
# the value data, and since an actual reboot won't be pending in that
# instance, just catch instances where we try unsuccessfully to cast as int.
reg_ret = __utils__["reg.read_value"]("HKLM", key, vname)
if reg_ret["success"]:
log.debug("Found key: %s", key)
try:
if int(reg_ret["vdata"]) > 0:
return True
except ValueError:
pass
else:
log.debug("Unable to access key: %s", key)
return False
return salt.utils.win_system.get_pending_servermanager()
def get_pending_update():
@ -1471,22 +1402,7 @@ def get_pending_update():
salt '*' system.get_pending_update
"""
key = r"SOFTWARE\Microsoft\Windows\CurrentVersion\WindowsUpdate\Auto Update\RebootRequired"
# So long as the registry key exists, a reboot is pending.
if __utils__["reg.key_exists"]("HKLM", key):
log.debug("Key exists: %s", key)
return True
else:
log.debug("Key does not exist: %s", key)
return False
MINION_VOLATILE_KEY = r"SYSTEM\CurrentControlSet\Services\salt-minion\Volatile-Data"
REBOOT_REQUIRED_NAME = "Reboot required"
return salt.utils.win_system.get_pending_update()
def set_reboot_required_witnessed():
@ -1516,14 +1432,7 @@ def set_reboot_required_witnessed():
salt '*' system.set_reboot_required_witnessed
"""
return __utils__["reg.set_value"](
hive="HKLM",
key=MINION_VOLATILE_KEY,
volatile=True,
vname=REBOOT_REQUIRED_NAME,
vdata=1,
vtype="REG_DWORD",
)
return salt.utils.win_system.set_reboot_required_witnessed()
def get_reboot_required_witnessed():
@ -1548,10 +1457,7 @@ def get_reboot_required_witnessed():
salt '*' system.get_reboot_required_witnessed
"""
value_dict = __utils__["reg.read_value"](
hive="HKLM", key=MINION_VOLATILE_KEY, vname=REBOOT_REQUIRED_NAME
)
return value_dict["vdata"] == 1
return salt.utils.win_system.get_reboot_required_witnessed()
def get_pending_reboot():
@ -1569,20 +1475,46 @@ def get_pending_reboot():
salt '*' system.get_pending_reboot
"""
return salt.utils.win_system.get_pending_reboot()
# Order the checks for reboot pending in most to least likely.
checks = (
get_pending_update,
get_pending_file_rename,
get_pending_servermanager,
get_pending_component_servicing,
get_reboot_required_witnessed,
get_pending_computer_name,
get_pending_domain_join,
)
for check in checks:
if check():
return True
def get_pending_reboot_details():
"""
Determine which check is signalling that the system is pending a reboot.
Useful in determining why your system is signalling that it needs a reboot.
return False
.. versionadded:: Sodium
Returns:
dict: A dictionary of the results of each system that would indicate a
pending reboot
CLI Example:
.. code-block:: bash
salt '*' system.get_pending_reboot_details
"""
return salt.utils.win_system.get_pending_reboot_details()
def get_pending_windows_update():
"""
Check the Windows Update system for a pending reboot state.
This leverages the Windows Update System to determine if the system is
pending a reboot.
.. versionadded:: Sodium
Returns:
bool: ``True`` if the Windows Update system reports a pending update,
otherwise ``False``
CLI Example:
.. code-block:: bash
salt '*' system.get_pending_windows_update
"""
return salt.utils.win_system.get_pending_windows_update()

538
salt/utils/win_system.py Normal file
View file

@ -0,0 +1,538 @@
# -*- coding: utf-8 -*-
'''
Win System Utils
Functions shared with salt.modules.win_system and salt.grains.pending_reboot
.. versionadded:: Sodium
'''
# When production windows installer is using Python 3, Python 2 code can be removed
from __future__ import absolute_import, print_function, unicode_literals
# Import python libs
import logging
# Import 3rd-party Libs
try:
import win32api
import win32con
HAS_WIN32_MODS = True
except ImportError:
HAS_WIN32_MODS = False
# Import Salt libs
import salt.utils.win_reg
import salt.utils.win_update
log = logging.getLogger(__name__)
# Define the module's virtual name
__virtualname__ = 'win_system'
MINION_VOLATILE_KEY = r'SYSTEM\CurrentControlSet\Services\salt-minion\Volatile-Data'
REBOOT_REQUIRED_NAME = 'Reboot required'
def __virtual__():
'''
Only works on Windows systems
'''
if not salt.utils.platform.is_windows():
return (False, 'win_system salt util failed to load: '
'The util will only run on Windows systems')
if not HAS_WIN32_MODS:
return (False, 'win_system salt util failed to load: '
'The util will only run on Windows systems')
return __virtualname__
def get_computer_name():
'''
Get the Windows computer name. Uses the win32api to get the current computer
name.
.. versionadded:: Sodium
Returns:
str: Returns the computer name if found. Otherwise returns ``False``.
Example:
.. code-block:: python
import salt.utils.win_system
salt.utils.win_system.get_computer_name()
'''
name = win32api.GetComputerNameEx(win32con.ComputerNamePhysicalDnsHostname)
return name if name else False
def get_pending_computer_name():
'''
Get a pending computer name. If the computer name has been changed, and the
change is pending a system reboot, this function will return the pending
computer name. Otherwise, ``None`` will be returned. If there was an error
retrieving the pending computer name, ``False`` will be returned, and an
error message will be logged to the minion log.
.. versionadded:: Sodium
Returns:
str:
Returns the pending name if pending restart. Returns ``None`` if not
pending restart.
Example:
.. code-block:: python
import salt.utils.win_system
salt.utils.win_system.get_pending_computer_name()
'''
current = get_computer_name()
try:
pending = salt.utils.win_reg.read_value(
hive='HKLM',
key=r'SYSTEM\CurrentControlSet\Services\Tcpip\Parameters',
vname='NV Hostname')['vdata']
except TypeError:
# This should never happen as the above key and vname are system names
# and should always be present
return None
if pending:
return pending if pending != current else None
def get_pending_component_servicing():
r'''
Determine whether there are pending Component Based Servicing tasks that
require a reboot.
If any the following registry keys exist then a reboot is pending:
``HKLM:\\SOFTWARE\Microsoft\Windows\CurrentVersion\Component Based Servicing\RebootPending``
``HKLM:\\SOFTWARE\Microsoft\Windows\CurrentVersion\Component Based Servicing\RebootInProgress``
``HKLM:\\SOFTWARE\Microsoft\Windows\CurrentVersion\Component Based Servicing\PackagesPending``
.. versionadded:: Sodium
Returns:
bool: ``True`` if there are pending Component Based Servicing tasks,
otherwise ``False``
CLI Example:
.. code-block:: bash
salt '*' system.get_pending_component_servicing
'''
# So long as the registry key exists, a reboot is pending
key = r'SOFTWARE\Microsoft\Windows\CurrentVersion\Component Based ' \
r'Servicing\RebootPending'
if salt.utils.win_reg.key_exists(hive='HKLM', key=key):
return True
key = r'SOFTWARE\Microsoft\Windows\CurrentVersion\Component Based ' \
r'Servicing\RebootInProgress'
if salt.utils.win_reg.key_exists(hive='HKLM', key=key):
return True
key = r'SOFTWARE\Microsoft\Windows\CurrentVersion\Component Based ' \
r'Servicing\PackagesPending'
if salt.utils.win_reg.key_exists(hive='HKLM', key=key):
return True
return False
def get_pending_domain_join():
r'''
Determine whether there is a pending domain join action that requires a
reboot.
If any the following registry keys exist then a reboot is pending:
``HKLM:\\SYSTEM\CurrentControlSet\Services\Netlogon\AvoidSpnSet``
``HKLM:\\SYSTEM\CurrentControlSet\Services\Netlogon\JoinDomain``
.. versionadded:: Sodium
Returns:
bool: ``True`` if there is a pending domain join action, otherwise
``False``
Example:
.. code-block:: python
import salt.utils.win_system
salt.utils.win_system.get_pending_domain_join()
'''
base_key = r'SYSTEM\CurrentControlSet\Services\Netlogon'
avoid_key = r'{0}\AvoidSpnSet'.format(base_key)
join_key = r'{0}\JoinDomain'.format(base_key)
# If either the avoid_key or join_key is present,
# then there is a reboot pending.
if salt.utils.win_reg.key_exists(hive='HKLM', key=avoid_key):
return True
if salt.utils.win_reg.key_exists('HKLM', join_key):
return True
return False
def get_pending_file_rename():
'''
Determine whether there are pending file rename operations that require a
reboot.
A reboot is pending if any of the following value names exist and have value
data set:
- ``PendingFileRenameOperations``
- ``PendingFileRenameOperations2``
in the following registry key:
``HKLM:\\SYSTEM\CurrentControlSet\Control\Session Manager``
.. versionadded:: Sodium
Returns:
bool: ``True`` if there are pending file rename operations, otherwise
``False``
Example:
.. code-block:: python
import salt.utils.win_system
salt.utils.win_system.get_pending_file_rename()
'''
vnames = ('PendingFileRenameOperations', 'PendingFileRenameOperations2')
key = r'SYSTEM\CurrentControlSet\Control\Session Manager'
for vname in vnames:
reg_ret = salt.utils.win_reg.read_value(
hive='HKLM',
key=key,
vname=vname
)
if reg_ret['success']:
if reg_ret['vdata'] and (reg_ret['vdata'] != '(value not set)'):
return True
return False
def get_pending_servermanager():
r'''
Determine whether there are pending Server Manager tasks that require a
reboot.
A reboot is pending if the ``CurrentRebootAttempts`` value name exists and
has an integer value. The value name resides in the following registry key:
``HKLM:\\SOFTWARE\Microsoft\ServerManager``
.. versionadded:: Sodium
Returns:
bool: ``True`` if there are pending Server Manager tasks, otherwise
``False``
Example:
.. code-block:: python
import salt.utils.win_system
salt.utils.win_system.get_pending_servermanager()
'''
vname = 'CurrentRebootAttempts'
key = r'SOFTWARE\Microsoft\ServerManager'
# There are situations where it's possible to have '(value not set)' as
# the value data, and since an actual reboot won't be pending in that
# instance, just catch instances where we try unsuccessfully to cast as int.
reg_ret = salt.utils.win_reg.read_value(hive='HKLM', key=key, vname=vname)
if reg_ret['success']:
try:
if int(reg_ret['vdata']) > 0:
return True
except ValueError:
pass
return False
def get_pending_dvd_reboot():
r'''
Determine whether the DVD Reboot flag is set.
The system requires a reboot if the ``DVDRebootSignal`` value name exists
at the following registry location:
``HKLM:\SOFTWARE\Microsoft\Windows\CurrentVersion\RunOnce``
.. versionadded:: Sodium
Returns:
bool: ``True`` if the above condition is met, otherwise ``False``
Example:
.. code-block:: python
import salt.utils.win_system
salt.utils.win_system.get_pending_dvd_reboot()
'''
# So long as the registry key exists, a reboot is pending.
return salt.utils.win_reg.value_exists(
hive='HKLM',
key=r'SOFTWARE\Microsoft\Windows\CurrentVersion\RunOnce',
vname='DVDRebootSignal')
def get_pending_update():
r'''
Determine whether there are pending updates that require a reboot.
If either of the following registry keys exists, a reboot is pending:
``HKLM:\SOFTWARE\Microsoft\Windows\CurrentVersion\WindowsUpdate\Auto Update\RebootRequired``
``HKLM:\SOFTWARE\Microsoft\Windows\CurrentVersion\WindowsUpdate\Auto Update\PostRebootReporting``
If there are any subkeys under the following registry key, a reboot is
pending:
``HKLM:\SOFTWARE\Microsoft\Windows\CurrentVersion\WindowsUpdate\Services\Pending``
.. versionadded:: Sodium
Returns:
bool: ``True`` if any of the above conditions are met, otherwise
``False``
Example:
.. code-block:: python
import salt.utils.win_system
salt.utils.win_system.get_pending_update()
'''
# So long as the registry key exists, a reboot is pending.
key = r'SOFTWARE\Microsoft\Windows\CurrentVersion\WindowsUpdate\Auto ' \
r'Update\RebootRequired'
if salt.utils.win_reg.key_exists(hive='HKLM', key=key):
return True
key = r'SOFTWARE\Microsoft\Windows\CurrentVersion\WindowsUpdate\Auto ' \
r'Update\PostRebootReporting'
if salt.utils.win_reg.key_exists(hive='HKLM', key=key):
return True
# So long as the registry key has subkeys, a reboot is pending.
key = r'SOFTWARE\Microsoft\Windows\CurrentVersion\WindowsUpdate\Services' \
r'\Pending'
list_keys = salt.utils.win_reg.list_keys(hive='HKLM', key=key)
if len(list_keys) > 0:
return True
return False
def get_reboot_required_witnessed():
r'''
Determine if at any time during the current boot session the salt minion
witnessed an event indicating that a reboot is required.
This function will return ``True`` if an install completed with exit
code 3010 during the current boot session and can be extended where
appropriate in the future.
If the ``Reboot required`` value name exists in the following location and
has a value of ``1`` then the system is pending reboot:
``HKLM:\\SYSTEM\CurrentControlSet\Services\salt-minion\Volatile-Data``
.. versionadded:: Sodium
Returns:
bool: ``True`` if the ``Requires reboot`` registry flag is set to ``1``,
otherwise ``False``
Example:
.. code-block:: python
import salt.utils.win_system
salt.utils.win_system.get_reboot_required_witnessed()
'''
value_dict = salt.utils.win_reg.read_value(
hive='HKLM',
key=MINION_VOLATILE_KEY,
vname=REBOOT_REQUIRED_NAME
)
return value_dict['vdata'] == 1
def set_reboot_required_witnessed():
r'''
This function is used to remember that an event indicating that a reboot is
required was witnessed. This function relies on the salt-minion's ability to
create the following volatile registry key in the *HKLM* hive:
*SYSTEM\\CurrentControlSet\\Services\\salt-minion\\Volatile-Data*
Because this registry key is volatile, it will not persist beyond the
current boot session. Also, in the scope of this key, the name *'Reboot
required'* will be assigned the value of *1*.
For the time being, this function is being used whenever an install
completes with exit code 3010 and can be extended where appropriate in the
future.
.. versionadded:: Sodium
Returns:
bool: ``True`` if successful, otherwise ``False``
Example:
.. code-block:: python
import salt.utils.win_system
salt.utils.win_system.set_reboot_required_witnessed()
'''
return salt.utils.win_reg.set_value(
hive='HKLM',
key=MINION_VOLATILE_KEY,
volatile=True,
vname=REBOOT_REQUIRED_NAME,
vdata=1,
vtype='REG_DWORD'
)
def get_pending_update_exe_volatile():
r'''
Determine whether there is a volatile update exe that requires a reboot.
Checks ``HKLM:\Microsoft\Updates``. If the ``UpdateExeVolatile`` value name
is anything other than 0 there is a reboot pending
.. versionadded:: Sodium
Returns:
bool: ``True`` if there is a volatile exe, otherwise ``False``
Example:
.. code-block:: python
import salt.utils.win_system
salt.utils.win_system.get_pending_update_exe_volatile()
'''
key = r'SOFTWARE\Microsoft\Updates'
reg_ret = salt.utils.win_reg.read_value(
hive='HKLM',
key=key,
vname='UpdateExeVolatile'
)
if reg_ret['success']:
try:
if int(reg_ret['vdata']) != 0:
return True
except ValueError:
pass
return False
def get_pending_windows_update():
'''
Check the Windows Update system for a pending reboot state.
This leverages the Windows Update System to determine if the system is
pending a reboot.
.. versionadded:: Sodium
Returns:
bool: ``True`` if the Windows Update system reports a pending update,
otherwise ``False``
Example:
.. code-block:: python
import salt.utils.win_system
salt.utils.win_system.get_pending_windows_update()
'''
return salt.utils.win_update.needs_reboot()
def get_pending_reboot():
'''
Determine whether there is a reboot pending.
.. versionadded:: Sodium
Returns:
bool: ``True`` if the system is pending reboot, otherwise ``False``
Example:
.. code-block:: python
import salt.utils.win_system
salt.utils.win_system.get_pending_reboot()
'''
# Order the checks for reboot pending in most to least likely.
checks = (get_pending_update,
get_pending_windows_update,
get_pending_update_exe_volatile,
get_pending_file_rename,
get_pending_servermanager,
get_pending_component_servicing,
get_pending_dvd_reboot,
get_reboot_required_witnessed,
get_pending_computer_name,
get_pending_domain_join)
for check in checks:
if check():
return True
return False
def get_pending_reboot_details():
'''
Determine which check is signalling that the system is pending a reboot.
Useful in determining why your system is signalling that it needs a reboot.
.. versionadded:: Sodium
Returns:
dict: A dictionary of the results of each function that checks for a
pending reboot
Example:
.. code-block:: python
import salt.utils.win_system
salt.utils.win_system.get_pending_reboot_details()
'''
return{
'Pending Component Servicing': get_pending_component_servicing(),
'Pending Computer Rename': get_pending_computer_name() is not None,
'Pending DVD Reboot': get_pending_dvd_reboot(),
'Pending File Rename': get_pending_file_rename(),
'Pending Join Domain': get_pending_domain_join(),
'Pending ServerManager': get_pending_servermanager(),
'Pending Update': get_pending_update(),
'Pending Windows Update': get_pending_windows_update(),
'Reboot Required Witnessed': get_reboot_required_witnessed(),
'Volatile Update Exe': get_pending_update_exe_volatile(),
}

View file

@ -0,0 +1,30 @@
# -*- coding: utf-8 -*-
'''
:codeauthor: :email:`Shane Lee <slee@saltstack.com>`
'''
# Import Python libs
from __future__ import absolute_import, print_function, unicode_literals
# Import Salt Testing Libs
from tests.support.unit import TestCase
from tests.support.mock import patch
# Import Salt Libs
import salt.grains.pending_reboot as pending_reboot
class PendingRebootGrainTestCase(TestCase):
'''
Test cases for pending_reboot grain
'''
def test_pending_reboot_false(self):
with patch('salt.utils.win_system.get_pending_reboot',
return_value=False):
result = pending_reboot.pending_reboot()
self.assertFalse(result['pending_reboot'])
def test_pending_reboot_true(self):
with patch('salt.utils.win_system.get_pending_reboot',
return_value=True):
result = pending_reboot.pending_reboot()
self.assertTrue(result['pending_reboot'])

View file

@ -216,9 +216,7 @@ class WinSystemTestCase(TestCase, LoaderModuleMockMixin):
"""
Test to reboot the system
"""
with patch(
"salt.modules.win_system.shutdown", MagicMock(return_value=True)
) as shutdown:
with patch("salt.modules.win_system.shutdown", MagicMock(return_value=True)):
self.assertEqual(win_system.reboot(), True)
@skipIf(not win_system.HAS_WIN32NET_MODS, "Missing win32 libraries")
@ -288,19 +286,15 @@ class WinSystemTestCase(TestCase, LoaderModuleMockMixin):
with patch(
"salt.modules.win_system.windll.kernel32.SetComputerNameExW",
MagicMock(return_value=True),
), patch.object(
win_system, "get_computer_name", MagicMock(return_value="salt")
), patch.object(
win_system, "get_pending_computer_name", MagicMock(return_value="salt_new"),
):
with patch.object(
win_system, "get_computer_name", MagicMock(return_value="salt")
):
with patch.object(
win_system,
"get_pending_computer_name",
MagicMock(return_value="salt_new"),
):
self.assertDictEqual(
win_system.set_computer_name("salt_new"),
{"Computer Name": {"Current": "salt", "Pending": "salt_new"}},
)
self.assertDictEqual(
win_system.set_computer_name("salt_new"),
{"Computer Name": {"Current": "salt", "Pending": "salt_new"}},
)
# Test set_computer_name failure
with patch(
"salt.modules.win_system.windll.kernel32.SetComputerNameExW",
@ -470,24 +464,33 @@ class WinSystemTestCase(TestCase, LoaderModuleMockMixin):
"""
Test to join a computer to an Active Directory domain
"""
with patch("salt.modules.win_system._join_domain", MagicMock(return_value=0)):
with patch(
"salt.modules.win_system.get_domain_workgroup",
MagicMock(return_value={"Workgroup": "Workgroup"}),
):
self.assertDictEqual(
win_system.join_domain("saltstack", "salt", "salt@123"),
{"Domain": "saltstack", "Restart": False},
)
with patch(
"salt.modules.win_system._join_domain", MagicMock(return_value=0)
), patch(
"salt.modules.win_system.get_domain_workgroup",
MagicMock(return_value={"Workgroup": "Workgroup"}),
):
self.assertDictEqual(
win_system.join_domain("saltstack", "salt", "salt@123"),
{"Domain": "saltstack", "Restart": False},
)
with patch(
"salt.modules.win_system.get_domain_workgroup",
MagicMock(return_value={"Domain": "saltstack"}),
):
self.assertEqual(
win_system.join_domain("saltstack", "salt", "salt@123"),
"Already joined to saltstack",
)
@skipIf(not win_system.HAS_WIN32NET_MODS, "Missing win32 libraries")
def test_join_domain_already_joined(self):
"""
Test to join a computer to an Active Directory domain when it is
already joined
"""
with patch(
"salt.modules.win_system._join_domain", MagicMock(return_value=0)
), patch(
"salt.modules.win_system.get_domain_workgroup",
MagicMock(return_value={"Domain": "saltstack"}),
):
self.assertEqual(
win_system.join_domain("saltstack", "salt", "salt@123"),
"Already joined to saltstack",
)
def test_get_system_time(self):
"""

View file

@ -0,0 +1,431 @@
# -*- coding: utf-8 -*-
# Import Python Libs
from __future__ import absolute_import, unicode_literals, print_function
import os
# Import Salt Testing Libs
from tests.support.mock import patch
from tests.support.unit import TestCase, skipIf
# Import Salt Libs
import salt.utils.platform
import salt.utils.win_system as win_system
@skipIf(not salt.utils.platform.is_windows(), 'Only test on Windows systems')
class WinSystemTestCase(TestCase):
'''
Test cases for salt.utils.win_system
'''
def test_get_computer_name(self):
'''
Should return the computer name
'''
with patch('win32api.GetComputerNameEx', return_value='FAKENAME'):
self.assertEqual(win_system.get_computer_name(), 'FAKENAME')
def test_get_computer_name_fail(self):
'''
If it fails, it returns False
'''
with patch('win32api.GetComputerNameEx', return_value=None):
self.assertFalse(win_system.get_computer_name())
def test_get_pending_computer_name(self):
'''
Will return the pending computer name if one is pending
'''
expected = 'PendingName'
patch_value = {'vdata': expected}
with patch('salt.utils.win_reg.read_value', return_value=patch_value):
result = win_system.get_pending_computer_name()
self.assertEqual(expected, result)
def test_get_pending_computer_name_none(self):
'''
Will return the None if the pending computer is the current name
'''
patch_value = {'vdata': os.environ.get('COMPUTERNAME')}
with patch('salt.utils.win_reg.read_value', return_value=patch_value):
self.assertIsNone(win_system.get_pending_computer_name())
def test_get_pending_computer_name_false(self):
'''
Will return False if there is no pending computer name
'''
with patch('salt.utils.win_reg.read_value', return_value=False):
self.assertIsNone(win_system.get_pending_computer_name())
def test_get_pending_component_servicing(self):
'''
If none of the keys exist, should return False
'''
with patch('salt.utils.win_reg.key_exists', return_value=False):
self.assertFalse(win_system.get_pending_component_servicing())
def test_get_pending_component_servicing_true_1(self):
'''
If the RebootPending key exists, should return True
'''
with patch('salt.utils.win_reg.key_exists', side_effect=[True]):
self.assertTrue(win_system.get_pending_component_servicing())
def test_get_pending_component_servicing_true_2(self):
'''
If the RebootInProgress key exists, should return True
'''
with patch('salt.utils.win_reg.key_exists', side_effect=[False, True]):
self.assertTrue(win_system.get_pending_component_servicing())
def test_get_pending_component_servicing_true_3(self):
'''
If the PackagesPending key exists, should return True
'''
with patch('salt.utils.win_reg.key_exists',
side_effect=[False, False, True]):
self.assertTrue(win_system.get_pending_component_servicing())
def test_get_pending_domain_join(self):
'''
If none of the keys exist, should return False
'''
with patch('salt.utils.win_reg.key_exists', return_value=False):
self.assertFalse(win_system.get_pending_domain_join())
def test_get_pending_domain_join_true_1(self):
'''
If the AvoidSpnSet key exists, should return True
'''
with patch('salt.utils.win_reg.key_exists', side_effect=[True]):
self.assertTrue(win_system.get_pending_domain_join())
def test_get_pending_domain_join_true_2(self):
'''
If the JoinDomain key exists, should return True
'''
with patch('salt.utils.win_reg.key_exists', side_effect=[False, True]):
self.assertTrue(win_system.get_pending_domain_join())
def test_get_pending_file_rename_false_1(self):
'''
If none of the value names exist, should return False
'''
patched_return = {'success': False}
with patch('salt.utils.win_reg.read_value',
return_value=patched_return):
self.assertFalse(win_system.get_pending_file_rename())
def test_get_pending_file_rename_false_2(self):
'''
If one of the value names exists but is not set, should return False
'''
patched_return = {'success': True,
'vdata': '(value not set)'}
with patch('salt.utils.win_reg.read_value',
return_value=patched_return):
self.assertFalse(win_system.get_pending_file_rename())
def test_get_pending_file_rename_true_1(self):
'''
If one of the value names exists and is set, should return True
'''
patched_return = {'success': True,
'vdata': 'some value'}
with patch('salt.utils.win_reg.read_value',
return_value=patched_return):
self.assertTrue(win_system.get_pending_file_rename())
def test_get_pending_servermanager_false_1(self):
'''
If the CurrentRebootAttempts value name does not exist, should return
False
'''
patched_return = {'success': False}
with patch('salt.utils.win_reg.read_value',
return_value=patched_return):
self.assertFalse(win_system.get_pending_servermanager())
def test_get_pending_servermanager_false_2(self):
'''
If the CurrentRebootAttempts value name exists but is not an integer,
should return False
'''
patched_return = {'success': True,
'vdata': '(value not set)'}
with patch('salt.utils.win_reg.read_value',
return_value=patched_return):
self.assertFalse(win_system.get_pending_file_rename())
def test_get_pending_servermanager_true(self):
'''
If the CurrentRebootAttempts value name exists and is an integer,
should return True
'''
patched_return = {'success': True,
'vdata': 1}
with patch('salt.utils.win_reg.read_value',
return_value=patched_return):
self.assertTrue(win_system.get_pending_file_rename())
def test_get_pending_dvd_reboot(self):
'''
If the DVDRebootSignal value name does not exist, should return False
'''
with patch('salt.utils.win_reg.value_exists', return_value=False):
self.assertFalse(win_system.get_pending_dvd_reboot())
def test_get_pending_dvd_reboot_true(self):
'''
If the DVDRebootSignal value name exists, should return True
'''
with patch('salt.utils.win_reg.value_exists', return_value=True):
self.assertTrue(win_system.get_pending_dvd_reboot())
def test_get_pending_update(self):
'''
If none of the keys exist and there are not subkeys, should return False
'''
with patch('salt.utils.win_reg.key_exists', return_value=False), \
patch('salt.utils.win_reg.list_keys', return_value=[]):
self.assertFalse(win_system.get_pending_update())
def test_get_pending_update_true_1(self):
'''
If the RebootRequired key exists, should return True
'''
with patch('salt.utils.win_reg.key_exists', side_effect=[True]):
self.assertTrue(win_system.get_pending_update())
def test_get_pending_update_true_2(self):
'''
If the PostRebootReporting key exists, should return True
'''
with patch('salt.utils.win_reg.key_exists', side_effect=[False, True]):
self.assertTrue(win_system.get_pending_update())
def test_get_pending_update_true_3(self):
'''
If the Pending key contains subkeys, should return True
'''
with patch('salt.utils.win_reg.key_exists',
side_effect=[False, False]), \
patch('salt.utils.win_reg.list_keys', return_value=['subkey']):
self.assertTrue(win_system.get_pending_update())
def test_get_reboot_required_witnessed_false_1(self):
'''
The ``Reboot Required`` value name does not exist, should return False
'''
patched_data = {'vdata': None}
with patch('salt.utils.win_reg.read_value', return_value=patched_data):
self.assertFalse(win_system.get_reboot_required_witnessed())
def test_get_reboot_required_witnessed_false_2(self):
'''
The ``Reboot required`` value name is set to 0, should return False
'''
patched_data = {'vdata': 0}
with patch('salt.utils.win_reg.read_value', return_value=patched_data):
self.assertFalse(win_system.get_reboot_required_witnessed())
def test_get_reboot_required_witnessed_true(self):
'''
The ``Reboot required`` value name is set to 1, should return True
'''
patched_data = {'vdata': 1}
with patch('salt.utils.win_reg.read_value', return_value=patched_data):
self.assertTrue(win_system.get_reboot_required_witnessed())
def test_set_reboot_required_witnessed(self):
'''
The call to ``set_value`` should return True and should be called with
the specified parameters
'''
with patch('salt.utils.win_reg.set_value', return_value=True) as sv:
self.assertTrue(win_system.set_reboot_required_witnessed())
sv.assert_called_once_with(
hive='HKLM',
key=win_system.MINION_VOLATILE_KEY,
volatile=True,
vname=win_system.REBOOT_REQUIRED_NAME,
vdata=1,
vtype='REG_DWORD'
)
def test_get_pending_update_exe_volatile_false_1(self):
'''
If UpdateExeVolatile value name is 0, should return False
'''
patched_data = {'success': True, 'vdata': 0}
with patch('salt.utils.win_reg.read_value', return_value=patched_data):
self.assertFalse(win_system.get_pending_update_exe_volatile())
def test_get_pending_update_exe_volatile_false_2(self):
'''
If UpdateExeVolatile value name is not present, should return False
'''
patched_data = {'success': False}
with patch('salt.utils.win_reg.read_value', return_value=patched_data):
self.assertFalse(win_system.get_pending_update_exe_volatile())
def test_get_pending_update_exe_volatile_true_1(self):
'''
If UpdateExeVolatile value name is not 0, should return True
'''
patched_data = {'success': True, 'vdata': 1}
with patch('salt.utils.win_reg.read_value', return_value=patched_data):
self.assertTrue(win_system.get_pending_update_exe_volatile())
def test_get_pending_reboot(self):
'''
If all functions return Falsy data, should return False
'''
with patch('salt.utils.win_system.get_pending_update',
return_value=False), \
patch('salt.utils.win_update.needs_reboot',
return_value=False), \
patch('salt.utils.win_system.get_pending_update_exe_volatile',
return_value=False), \
patch('salt.utils.win_system.get_pending_file_rename',
return_value=False), \
patch('salt.utils.win_system.get_pending_servermanager',
return_value=False), \
patch('salt.utils.win_system.get_pending_component_servicing',
return_value=False), \
patch('salt.utils.win_system.get_pending_dvd_reboot',
return_value=False), \
patch('salt.utils.win_system.get_reboot_required_witnessed',
return_value=False), \
patch('salt.utils.win_system.get_pending_computer_name',
return_value=None), \
patch('salt.utils.win_system.get_pending_domain_join',
return_value=False):
self.assertFalse(win_system.get_pending_reboot())
def test_get_pending_reboot_true_1(self):
'''
If any boolean returning functions return True, should return True
'''
with patch('salt.utils.win_system.get_pending_update',
return_value=False), \
patch('salt.utils.win_update.needs_reboot',
return_value=False), \
patch('salt.utils.win_system.get_pending_update_exe_volatile',
return_value=False), \
patch('salt.utils.win_system.get_pending_file_rename',
return_value=False), \
patch('salt.utils.win_system.get_pending_servermanager',
return_value=False), \
patch('salt.utils.win_system.get_pending_component_servicing',
return_value=False), \
patch('salt.utils.win_system.get_pending_dvd_reboot',
return_value=False), \
patch('salt.utils.win_system.get_reboot_required_witnessed',
return_value=False), \
patch('salt.utils.win_system.get_pending_computer_name',
return_value=None), \
patch('salt.utils.win_system.get_pending_domain_join',
return_value=True):
self.assertTrue(win_system.get_pending_reboot())
def test_get_pending_reboot_true_2(self):
'''
If a computer name is returned, should return True
'''
with patch('salt.utils.win_system.get_pending_update',
return_value=False), \
patch('salt.utils.win_update.needs_reboot',
return_value=False), \
patch('salt.utils.win_system.get_pending_update_exe_volatile',
return_value=False), \
patch('salt.utils.win_system.get_pending_file_rename',
return_value=False), \
patch('salt.utils.win_system.get_pending_servermanager',
return_value=False), \
patch('salt.utils.win_system.get_pending_component_servicing',
return_value=False), \
patch('salt.utils.win_system.get_pending_dvd_reboot',
return_value=False), \
patch('salt.utils.win_system.get_reboot_required_witnessed',
return_value=False), \
patch('salt.utils.win_system.get_pending_computer_name',
return_value='pending name'):
self.assertTrue(win_system.get_pending_reboot())
def test_get_pending_reboot_details(self):
'''
All items False should return a dictionary with all items False
'''
with patch('salt.utils.win_system.get_pending_update',
return_value=False), \
patch('salt.utils.win_update.needs_reboot',
return_value=False), \
patch('salt.utils.win_system.get_pending_update_exe_volatile',
return_value=False), \
patch('salt.utils.win_system.get_pending_file_rename',
return_value=False), \
patch('salt.utils.win_system.get_pending_servermanager',
return_value=False), \
patch('salt.utils.win_system.get_pending_component_servicing',
return_value=False), \
patch('salt.utils.win_system.get_pending_dvd_reboot',
return_value=False), \
patch('salt.utils.win_system.get_reboot_required_witnessed',
return_value=False), \
patch('salt.utils.win_system.get_pending_computer_name',
return_value=None), \
patch('salt.utils.win_system.get_pending_domain_join',
return_value=False):
expected = {
'Pending Component Servicing': False,
'Pending Computer Rename': False,
'Pending DVD Reboot': False,
'Pending File Rename': False,
'Pending Join Domain': False,
'Pending ServerManager': False,
'Pending Update': False,
'Pending Windows Update': False,
'Reboot Required Witnessed': False,
'Volatile Update Exe': False,
}
result = win_system.get_pending_reboot_details()
self.assertDictEqual(expected, result)
def test_get_pending_reboot_details_true(self):
'''
All items True should return a dictionary with all items True
'''
with patch('salt.utils.win_system.get_pending_update',
return_value=True), \
patch('salt.utils.win_update.needs_reboot',
return_value=True), \
patch('salt.utils.win_system.get_pending_update_exe_volatile',
return_value=True), \
patch('salt.utils.win_system.get_pending_file_rename',
return_value=True), \
patch('salt.utils.win_system.get_pending_servermanager',
return_value=True), \
patch('salt.utils.win_system.get_pending_component_servicing',
return_value=True), \
patch('salt.utils.win_system.get_pending_dvd_reboot',
return_value=True), \
patch('salt.utils.win_system.get_reboot_required_witnessed',
return_value=True), \
patch('salt.utils.win_system.get_pending_computer_name',
return_value='pending name'), \
patch('salt.utils.win_system.get_pending_domain_join',
return_value=True):
expected = {
'Pending Component Servicing': True,
'Pending Computer Rename': True,
'Pending DVD Reboot': True,
'Pending File Rename': True,
'Pending Join Domain': True,
'Pending ServerManager': True,
'Pending Update': True,
'Pending Windows Update': True,
'Reboot Required Witnessed': True,
'Volatile Update Exe': True,
}
result = win_system.get_pending_reboot_details()
self.assertDictEqual(expected, result)