diff --git a/changelog/59037.added.md b/changelog/59037.added.md new file mode 100644 index 000000000000..6d74b4ba63c2 --- /dev/null +++ b/changelog/59037.added.md @@ -0,0 +1 @@ +Added a state (win_task) for managing scheduled tasks on Windows diff --git a/salt/states/win_task.py b/salt/states/win_task.py index 47be689f6ba9..6e99950280c9 100644 --- a/salt/states/win_task.py +++ b/salt/states/win_task.py @@ -1,174 +1,168 @@ -# -*- coding: utf-8 -*- # https://msdn.microsoft.com/en-us/library/windows/desktop/aa383608(v=vs.85).aspx -from __future__ import absolute_import, print_function, unicode_literals +""" +State module for adding and removing scheduled tasks using the Windows Task +Scheduler. +""" - -# State Module for managing task scheduler on Windows. -# You can add or remove tasks. - -# Import Python libs import copy import logging import time -# Import salt libs import salt.utils.data -import salt.utils.platform import salt.utils.dateutils +import salt.utils.platform -# Import 3rd-party libs try: import pywintypes + + HAS_PYWIN32 = True except ImportError: - pass - - -ACTION_PARTS = {'Execute': ['cmd'], - 'Email': ['from', 'to', 'cc', 'server'], - 'Message': ['title', 'message']} - -OPTIONAL_ACTION_PARTS = {'start_in': '', - 'arguments': ''} - -TRIGGER_PARTS = {'Event': ['subscription'], - 'Once': [], - 'Daily': ['days_interval'], - 'Weekly': ['days_of_week', 'weeks_interval'], - 'Monthly': ['months_of_year', 'days_of_month', 'last_day_of_month'], - 'MonthlyDay': ['months_of_year', 'weeks_of_month', 'days_of_week'], - 'OnIdle': [], - 'OnTaskCreation': [], - 'OnBoot': [], - 'OnLogon': [], - 'OnSessionChange': ['state_change']} - -OPTIONAL_TRIGGER_PARTS = {'trigger_enabled': True, - 'start_date': time.strftime('%Y-%m-%d'), - 'start_time': time.strftime('%H:%M:%S'), - 'end_date': None, - 'end_time': "00:00:00", - 'random_delay': False, - 'repeat_interval': None, - 'repeat_duration': None, - 'repeat_stop_at_duration_end': False, - 'execution_time_limit': '3 days', - 'delay': False} - -OPTIONAL_CONDITIONS_PARTS = {'ac_only': True, - 'run_if_idle': False, - 'run_if_network': False, - 'start_when_available': False} - -OPTIONAL_SETTINGS_PARTS = {'allow_demand_start': True, - 'delete_after': False, - 'execution_time_limit': '3 days', - 'force_stop': True, - 'multiple_instances': 'No New Instance', - 'restart_interval': False, - 'stop_if_on_batteries': True, - 'wake_to_run': False} - -TASK_PARTS = {'actions': {'parts': ACTION_PARTS, 'optional': OPTIONAL_ACTION_PARTS}, - 'triggers': {'parts': TRIGGER_PARTS, 'optional': OPTIONAL_TRIGGER_PARTS}, - 'conditions': {'parts': {}, 'optional': OPTIONAL_CONDITIONS_PARTS}, - 'settings': {'parts': {}, 'optional': OPTIONAL_SETTINGS_PARTS}} + HAS_PYWIN32 = False + + +ACTION_PARTS = { + "Execute": ["cmd"], + "Email": ["from", "to", "cc", "server"], + "Message": ["title", "message"], +} + +OPTIONAL_ACTION_PARTS = {"start_in": "", "arguments": ""} + +TRIGGER_PARTS = { + "Event": ["subscription"], + "Once": [], + "Daily": ["days_interval"], + "Weekly": ["days_of_week", "weeks_interval"], + "Monthly": ["months_of_year", "days_of_month", "last_day_of_month"], + "MonthlyDay": ["months_of_year", "weeks_of_month", "days_of_week"], + "OnIdle": [], + "OnTaskCreation": [], + "OnBoot": [], + "OnLogon": [], + "OnSessionChange": ["state_change"], +} + +OPTIONAL_TRIGGER_PARTS = { + "trigger_enabled": True, + "start_date": time.strftime("%Y-%m-%d"), + "start_time": time.strftime("%H:%M:%S"), + "end_date": None, + "end_time": "00:00:00", + "random_delay": False, + "repeat_interval": None, + "repeat_duration": None, + "repeat_stop_at_duration_end": False, + "execution_time_limit": "3 days", + "delay": False, +} + +OPTIONAL_CONDITIONS_PARTS = { + "ac_only": True, + "run_if_idle": False, + "run_if_network": False, + "start_when_available": False, +} + +OPTIONAL_SETTINGS_PARTS = { + "allow_demand_start": True, + "delete_after": False, + "execution_time_limit": "3 days", + "force_stop": True, + "multiple_instances": "No New Instance", + "restart_interval": False, + "stop_if_on_batteries": True, + "wake_to_run": False, +} + +TASK_PARTS = { + "actions": {"parts": ACTION_PARTS, "optional": OPTIONAL_ACTION_PARTS}, + "triggers": {"parts": TRIGGER_PARTS, "optional": OPTIONAL_TRIGGER_PARTS}, + "conditions": {"parts": {}, "optional": OPTIONAL_CONDITIONS_PARTS}, + "settings": {"parts": {}, "optional": OPTIONAL_SETTINGS_PARTS}, +} log = logging.getLogger(__name__) -__virtualname__ = 'task' +__virtualname__ = "task" def __virtual__(): - ''' + """ Load only on minions running on Windows and with task Module loaded. - ''' - if not salt.utils.platform.is_windows() or 'task.list_tasks' not in __salt__ or\ - 'pywintypes' not in globals(): - return False, 'State win_task: state only works on Window systems with task loaded' + """ + if not salt.utils.platform.is_windows(): + return False, "State task: Not a Windows System" + if not HAS_PYWIN32: + return False, "State task: Missing PyWin32 library" return __virtualname__ -def _get_state_data(name): - r''' - will return a new blank state dict. - - :param str name: name of task. - - :return new state data: - :rtype dict: - ''' - return {'name': name, - 'changes': {}, - 'result': True, - 'comment': ''} - - def _valid_location(location): - r''' - will test to see if task location is valid. + r""" + Test to see if the task location is valid. - :param str location: location of task. + Args: + location (str): task location to check - :return True if location is vaild, False if location is not valid: - :rtype bool: - ''' + Returns: + bool: ``True`` if location is valid, otherwise ``False`` + """ try: - __salt__['task.list_tasks'](location) + __salt__["task.list_tasks"](location) except pywintypes.com_error: return False return True -def _get_task_state_data(name, - location): - r''' - will get the state of a task. - - :param str name: name of task +def _get_task_state_data(name, location): + r""" + Get the current state of a task in the task scheduler - :param str location: location of task. + Args: + name (str): Task name + location (str): Task location - :return task state: - :rtype dict: - ''' - task_state = {'location_valid': False, - 'task_found': False, - 'task_info': {}} + Return: + dict: A dictionary containing task configuration information + """ + task_state = {"location_valid": False, "task_found": False, "task_info": {}} # if valid location then try to get more info on task if _valid_location(location): - task_state['location_valid'] = True - task_state['task_found'] = name in __salt__['task.list_tasks'](location) + task_state["location_valid"] = True + task_state["task_found"] = name in __salt__["task.list_tasks"](location) # if task was found then get actions and triggers info - if task_state['task_found']: - task_info = __salt__['task.info'](name, location) - task_state['task_info'] = {key: task_info[key] for key in TASK_PARTS} + if task_state["task_found"]: + task_info = __salt__["task.info"](name, location) + task_state["task_info"] = {key: task_info[key] for key in TASK_PARTS} return task_state -def _get_arguments(arguments_given, - key_arguments, - arguments_need_it, - optional_arguments): +def _get_arguments( + arguments_given, key_arguments, arguments_need_it, optional_arguments +): + """ + Make sure all required arguments are passed + """ block = {} # check if key arguments are present for key in key_arguments: if key not in arguments_given: - return 'Missing key argument {0}'.format(repr(key)) + return f"Missing key argument {repr(key)}" block[key] = arguments_given[key] - # check is key item valid + # check if key item valid if block[key] not in arguments_need_it: - return '{0} item {1} is not in key item list {2}'.format(repr(key), repr(block[key]), list(arguments_need_it)) + return f"{repr(key)} item {repr(block[key])} is not in key item list {list(arguments_need_it)}" - # check is key2 present + # check if key2 present for key2 in arguments_need_it[block[key]]: if key2 not in arguments_given: - return 'Missing {0} argument'.format(key2) + return f"Missing {key2} argument" block[key2] = arguments_given[key2] - # add optional arguments if their not present + # add optional arguments if they're not present for key in optional_arguments: if key in arguments_given: block[key] = arguments_given[key] @@ -179,89 +173,110 @@ def _get_arguments(arguments_given, def _task_state_prediction_bandage(state): - r''' - A bandage to format and add arguments to a task state. - This is so task states can be compared. + r""" + A bandage that standardizes date/time formats and adds additional arguments + to a task state. This is so task states can be compared with existing tasks + one the system. - :param dict state: task state - :return task state: - :rtype dict: - ''' + Args: + state (dict): The dictionary of the current settings for the task - # add 'enabled = True' to all triggers - # this is because triggers will add this argument after their made - if 'triggers' in state['task_info']: - for trigger in state['task_info']['triggers']: - trigger['enabled'] = True + Returns: + dict: A dictionary with parameters with in the expected format + """ + + # Add "enabled = True" to all triggers + # This is because triggers will add this argument after they're made + if "triggers" in state["task_info"]: + for trigger in state["task_info"]["triggers"]: + trigger["enabled"] = True # format dates - for trigger in state['task_info']['triggers']: - for key in ['start_date', 'end_data']: + for trigger in state["task_info"]["triggers"]: + for key in ["start_date", "end_data"]: if key in trigger: - # if except is triggered don't format the date + # if except is triggered don"t format the date try: - div = [d for d in ['-', '/'] if d in trigger[key]][0] + div = [d for d in ["-", "/"] if d in trigger[key]][0] part1, part2, part3 = trigger[key].split(div) if len(part1) == 4: year, month, day = part1, part2, part3 else: month, day, year = part1, part2, part3 if len(year) != 4: - year = time.strftime('%Y')[:2] + year + year = time.strftime("%Y")[:2] + year - trigger[key] = salt.utils.dateutils.strftime("{0}-{1}-{2}".format(year, month, day), '%Y-%m-%d') + trigger[key] = salt.utils.dateutils.strftime( + f"{year}-{month}-{day}", "%Y-%m-%d" + ) except IndexError: pass except ValueError: pass # format times - for trigger in state['task_info']['triggers']: - for key in ['start_time', 'end_time']: + for trigger in state["task_info"]["triggers"]: + for key in ["start_time", "end_time"]: if key in trigger: try: - trigger[key] = salt.utils.dateutils.strftime(trigger[key], '%H:%M:%S') + trigger[key] = salt.utils.dateutils.strftime( + trigger[key], "%H:%M:%S" + ) except ValueError: pass return state -def _get_task_state_prediction(state, - new_task): - r''' - predicts what a the new task will look like +def _get_task_state_prediction(state, new_task): + r""" + Predicts what the new task will look like after it is applied. Used for + test=True + + Args: + state (dict): A dictionary containing the current task configuration + new_task (dict) A dictionary containing the new task configuration - :param dict state: - :param dict new_task: - :return task state: - :rtype dict: - ''' + Returns: + dict: A dictionary containing predicted result + """ new_state = copy.deepcopy(state) - # if location not valid state can't be made - if state['location_valid']: - new_state['task_found'] = True - new_state['task_info'] = {'actions': [new_task['action']], - 'triggers': [new_task['trigger']], - 'conditions': new_task['conditions'], - 'settings': new_task['settings']} + # if location not valid state can"t be made + if state["location_valid"]: + new_state["task_found"] = True + new_state["task_info"] = { + "actions": [new_task["action"]], + "triggers": [new_task["trigger"]], + "conditions": new_task["conditions"], + "settings": new_task["settings"], + } action_keys = set() trigger_keys = set() - if state['task_found']: + if state["task_found"]: # get all the arguments used by actions - for action in state['task_info']['actions']: + for action in state["task_info"]["actions"]: action_keys = action_keys.union(set(action)) # get all the arguments used by triggers - for trigger in state['task_info']['triggers']: + for trigger in state["task_info"]["triggers"]: trigger_keys = trigger_keys.union(set(trigger)) # get setup for the for loop below - arguments_filter = [[new_state['task_info']['actions'], action_keys, TASK_PARTS['actions']['optional']], - [new_state['task_info']['triggers'], trigger_keys, TASK_PARTS['triggers']['optional']]] + arguments_filter = [ + [ + new_state["task_info"]["actions"], + action_keys, + TASK_PARTS["actions"]["optional"], + ], + [ + new_state["task_info"]["triggers"], + trigger_keys, + TASK_PARTS["triggers"]["optional"], + ], + ] # removes any optional arguments that are equal to the default and is not used by the state for argument_list, safe_keys, optional_keys in arguments_filter: @@ -273,8 +288,8 @@ def _get_task_state_prediction(state, # removes add on arguments from triggers # this is because task info does not give this info - argument_add_on = set(sum([TRIGGER_PARTS[key] for key in TRIGGER_PARTS], [])) - for trigger in new_state['task_info']['triggers']: + argument_add_on = set(sum((TRIGGER_PARTS[key] for key in TRIGGER_PARTS), [])) + for trigger in new_state["task_info"]["triggers"]: for key in list(trigger): if key in argument_add_on: del trigger[key] @@ -282,13 +297,10 @@ def _get_task_state_prediction(state, return _task_state_prediction_bandage(new_state) -def present(name, - location='\\', - user_name='System', - password=None, - force=False, - **kwargs): - r''' +def present( + name, location="\\", user_name="System", password=None, force=False, **kwargs +): + r""" Create a new task in the designated location. This function has many keyword arguments that are not listed here. For additional arguments see: @@ -296,95 +308,117 @@ def present(name, - :py:func:`add_action` - :py:func:`add_trigger` - :param str name: The name of the task. This will be displayed in the task - scheduler. + Args: + name (str): The name of the task. This will be displayed in the task + scheduler. - :param str location: A string value representing the location in which to - create the task. Default is '\\' which is the root for the task - scheduler (C:\Windows\System32\tasks). + location (str): A string value representing the location in which to + create the task. Default is "\\" which is the root for the task + scheduler (C:\Windows\System32\tasks). - :param str user_name: The user account under which to run the task. To - specify the 'System' account, use 'System'. The password will be - ignored. + user_name (str): The user account under which to run the task. To + specify the "System" account, use "System". The password will be + ignored. - :param str password: The password to use for authentication. This should set - the task to run whether the user is logged in or not, but is currently - not working. + password (str): The password to use for authentication. This should set + the task to run whether the user is logged in or not, but is + currently not working. - :param bool force: If the task exists, overwrite the existing task. + force (bool): Overwrite the existing task. - :return dict state: - :rtype dict: + Returns: + dict: A dictionary containing the results of the state CLI Example: - .. code-block::YAML - - test_win_task_present: - task.present: - - name: salt - - location: '' - - force: True - - action_type: Execute - - cmd: 'del /Q /S C:\\Temp' - - trigger_type: Once - - start_date: 12-1-16 - - start_time: 01:00 - - .. code-block:: bash - - salt 'minion-id' state.apply - ''' + .. code-block:: YAML - ret = _get_state_data(name) + test_win_task_present: + task.present: + - name: salt + - location: "" + - force: True + - action_type: Execute + - cmd: "del /Q /S C:\\Temp" + - trigger_type: Once + - start_date: 12-1-16 + - start_time: 01:00 + """ + + ret = {"name": name, "changes": {}, "result": True, "comment": ""} before = _get_task_state_data(name, location) # if location not valid the task present will fail - if not before['location_valid']: - ret['result'] = False - ret['comment'] = '{0} is not a valid file location'.format(repr(location)) + if not before["location_valid"]: + ret["result"] = False + ret["comment"] = f"{repr(location)} is not a valid file location" return ret # split up new task into all its parts - new_task = {'action': _get_arguments(kwargs, ['action_type'], - TASK_PARTS['actions']['parts'], TASK_PARTS['actions']['optional']), - 'trigger': _get_arguments(kwargs, ['trigger_type'], - TASK_PARTS['triggers']['parts'], TASK_PARTS['triggers']['optional']), - 'conditions': _get_arguments(kwargs, [], - TASK_PARTS['conditions']['parts'], TASK_PARTS['conditions']['optional']), - 'settings': _get_arguments(kwargs, [], - TASK_PARTS['settings']['parts'], TASK_PARTS['settings']['optional'])} + new_task = { + "action": _get_arguments( + kwargs, + ["action_type"], + TASK_PARTS["actions"]["parts"], + TASK_PARTS["actions"]["optional"], + ), + "trigger": _get_arguments( + kwargs, + ["trigger_type"], + TASK_PARTS["triggers"]["parts"], + TASK_PARTS["triggers"]["optional"], + ), + "conditions": _get_arguments( + kwargs, + [], + TASK_PARTS["conditions"]["parts"], + TASK_PARTS["conditions"]["optional"], + ), + "settings": _get_arguments( + kwargs, + [], + TASK_PARTS["settings"]["parts"], + TASK_PARTS["settings"]["optional"], + ), + } # if win os is higher than 7 then Email and Message action_type is not supported try: - if int(__grains__['osversion'].split('.')[0]) >= 8 and new_task['action']['action_type'] in ['Email', 'Message']: - log.warning('This OS %s does not support Email or Message action_type.', __grains__['osversion']) + if int(__grains__["osversion"].split(".")[0]) >= 8 and new_task["action"][ + "action_type" + ] in ["Email", "Message"]: + log.warning( + "This OS %s does not support Email or Message action_type.", + __grains__["osversion"], + ) except ValueError: pass for key in new_task: # if string is returned then an error happened if isinstance(new_task[key], str): - ret['comment'] = '{0}: {1}'.format(key, new_task[key]) - ret['result'] = False + ret["comment"] = f"{key}: {new_task[key]}" + ret["result"] = None return ret - if __opts__['test']: + if __opts__["test"]: # if force is False and task is found then no changes will take place - if not force and before['task_found']: - ret['comment'] = '\'force=True\' will allow the new task to replace the old one' - ret['result'] = False + if not force and before["task_found"]: + ret[ + "comment" + ] = '"force=True" will allow the new task to replace the old one' + ret["result"] = None log.warning("force=False") return ret after = _get_task_state_prediction(before, new_task) - ret['result'] = None - # the task will not change - if before == after: - ret['result'] = True + + ret["changes"] = salt.utils.data.compare_dicts(before, after) + + if ret["changes"]: + ret["result"] = None return ret - ret['changes'] = salt.utils.data.compare_dicts(before, after) return ret # put all the arguments to kwargs @@ -392,92 +426,91 @@ def present(name, kwargs.update(new_task[key]) # make task - ret['result'] = __salt__['task.create_task'](name=name, - location=location, - user_name=user_name, - password=password, - force=force, - **kwargs) - - # if 'task.crate_task' returns a str then task did not change - if isinstance(ret['result'], str): - ret['comment'] = '\'force=True\' will allow the new task to replace the old one' - ret['result'] = False + result = __salt__["task.create_task"]( + name=name, + location=location, + user_name=user_name, + password=password, + force=force, + **kwargs, + ) + + # if "task.create_task" returns a str then task did not change + if isinstance(result, str): + ret["comment"] = '"force=True" will allow the new task to replace the old one' + ret["result"] = False log.warning("force=False") return ret after = _get_task_state_data(name, location) - if after['task_info']['actions'][0]['action_type'] != kwargs['action_type']: - ret['comment'] = 'failed to make action' - ret['result'] = False - elif after['task_info']['triggers'][0]['trigger_type'] != kwargs['trigger_type']: - ret['comment'] = 'failed to make trigger' - ret['result'] = False + if after["task_info"]["actions"][0]["action_type"] != kwargs["action_type"]: + ret["comment"] = "failed to make action" + ret["result"] = False + elif after["task_info"]["triggers"][0]["trigger_type"] != kwargs["trigger_type"]: + ret["comment"] = "failed to make trigger" + ret["result"] = False - ret['changes'] = salt.utils.data.compare_dicts(before, after) + ret["changes"] = salt.utils.data.compare_dicts(before, after) return ret -def absent(name, - location='\\'): - r''' +def absent(name, location="\\"): + r""" Delete a task from the task scheduler. - :param str name: The name of the task to delete. + Args: + name (str): The name of the task to delete. - :param str location: A string value representing the location of the task. - Default is '\\' which is the root for the task scheduler - (C:\Windows\System32\tasks). + location (str): A string value representing the location of the task. + Default is "\\" which is the root for the task scheduler + (C:\Windows\System32\tasks). - :return True if successful, False if unsuccessful: - :rtype bool: + Returns: + bool: ``True`` if successful, otherwise ``False`` CLI Example: - .. code-block::YAML + .. code-block:: YAML test_win_task_absent: task.absent: - name: salt - - location: '' - - .. code-block:: bash - - salt 'minion-id' state.apply - ''' + - location: "" + """ - ret = _get_state_data(name) + ret = {"name": name, "changes": {}, "result": True, "comment": ""} before = _get_task_state_data(name, location) # if location not valid the task present will fail - if not before['location_valid']: - ret['result'] = False - ret['comment'] = '{0} is not a valid file location'.format(repr(location)) + if not before["location_valid"]: + ret["result"] = False + ret["comment"] = f"{repr(location)} is not a valid file location" return ret - if __opts__['test']: + if __opts__["test"]: # if task was not found then no changes - if not before['task_found']: - ret['result'] = True - ret['changes'] = salt.utils.data.compare_dicts(before, before) + if not before["task_found"]: + ret["result"] = True + ret["changes"] = salt.utils.data.compare_dicts(before, before) else: # if task was found then changes will happen - ret['result'] = None - ret['changes'] = salt.utils.data.compare_dicts(before, {'location_valid': True, - 'task_found': False, - 'task_info': {}}) + ret["result"] = None + ret["changes"] = salt.utils.data.compare_dicts( + before, {"location_valid": True, "task_found": False, "task_info": {}} + ) return ret # if task was found then delete it - if before['task_found']: + if before["task_found"]: # try to delete task - ret['result'] = __salt__['task.delete_task'](name=name, - location=location) + ret["result"] = __salt__["task.delete_task"](name=name, location=location) - # if 'task.delete_task' returns a str then task was not deleted - if isinstance(ret['result'], str): - ret['result'] = False + # if "task.delete_task" returns a str then task was not deleted + if isinstance(ret["result"], str): + ret["result"] = False - ret['changes'] = salt.utils.data.compare_dicts(before, _get_task_state_data(name, location)) + ret["changes"] = salt.utils.data.compare_dicts( + before, _get_task_state_data(name, location) + ) return ret diff --git a/tests/pytests/unit/states/test_win_task.py b/tests/pytests/unit/states/test_win_task.py new file mode 100644 index 000000000000..73daffc6eebf --- /dev/null +++ b/tests/pytests/unit/states/test_win_task.py @@ -0,0 +1,681 @@ +# https://msdn.microsoft.com/en-us/library/windows/desktop/aa383608(v=vs.85).aspx + +import pytest + +import salt.modules.win_task +import salt.states.win_task as win_task +import salt.utils.platform +from tests.support.mock import MagicMock, patch + +pytestmark = [ + pytest.mark.windows_whitelisted, + pytest.mark.skip_unless_on_windows, + pytest.mark.destructive_test, +] + + +@pytest.fixture +def configure_loader_modules(): + return {win_task: {}} + + +def test_present(): + kwargs = { + "action_type": "Execute", + "cmd": "del /Q /S C:\\\\Temp", + "trigger_type": "Once", + "start_data": "2019-05-14", + "start_time": "01:00 pm", + } + + ret = {"result": False} + try: + with patch.dict( + win_task.__salt__, + { + "task.list_tasks": salt.modules.win_task.list_tasks, + "task.info": salt.modules.win_task.info, + "task.create_task": salt.modules.win_task.create_task, + }, + ), patch.dict(win_task.__opts__, {"test": False}), patch.dict( + win_task.__grains__, {"osversion": "7.1"} + ): + + ret = win_task.present(name="salt", location="", force=True, **kwargs) + finally: + try: + salt.modules.win_task.delete_task(name="salt", location="") + finally: + pass + + assert ret["result"] + + +def test_absent(): + with patch.dict( + win_task.__salt__, + { + "task.list_tasks": salt.modules.win_task.list_tasks, + "task.info": salt.modules.win_task.info, + "task.delete_task": salt.modules.win_task.delete_task, + }, + ), patch.dict(win_task.__opts__, {"test": False}): + ret = win_task.absent("salt", "") + + assert ret["result"] + + kwargs = { + "action_type": "Execute", + "cmd": "del /Q /S C:\\\\Temp", + "trigger_type": "Once", + "start_data": "2019-05-14", + "start_time": "01:00 pm", + } + + try: + with patch.dict( + win_task.__salt__, + { + "task.list_tasks": salt.modules.win_task.list_tasks, + "task.info": salt.modules.win_task.info, + "task.create_task": salt.modules.win_task.create_task, + }, + ), patch.dict(win_task.__opts__, {"test": False}), patch.dict( + win_task.__grains__, {"osversion": "7.1"} + ): + + win_task.present(name="salt", location="", force=True, **kwargs) + finally: + try: + with patch.dict( + win_task.__salt__, + { + "task.list_tasks": salt.modules.win_task.list_tasks, + "task.info": salt.modules.win_task.info, + "task.delete_task": salt.modules.win_task.delete_task, + }, + ), patch.dict(win_task.__opts__, {"test": False}): + ret = win_task.absent("salt", "") + finally: + pass + + assert ret["result"] + + +def test__get_arguments(): + kwargs = {"salt": True, "cat": "nice", "idk": 404} + + true_ret = {"salt": True, "cat": "nice", "fat": True, "idk": 404} + + ret = win_task._get_arguments( + kwargs, ["cat"], {"nice": ["idk"], "sad": ["why"]}, {"fat": True, "salt": None} + ) + + assert ret == true_ret + + +def test__get_task_state_prediction(): + state = { + "task_found": True, + "location_valid": True, + "task_info": { + "conditions": { + "ac_only": True, + "run_if_idle": False, + "run_if_network": False, + "start_when_available": False, + }, + "actions": [{"cmd": "del /Q /S C:\\\\Temp", "action_type": "Execute"}], + "triggers": [ + { + "delay": False, + "execution_time_limit": "3 days", + "trigger_type": "OnSessionChange", + "start_date": "2019-05-14", + "enabled": True, + "start_time": "13:00:00", + } + ], + "settings": { + "delete_after": False, + "multiple_instances": "No New Instance", + "execution_time_limit": "3 days", + "allow_demand_start": True, + "restart_interval": False, + "stop_if_on_batteries": True, + "force_stop": True, + "wake_to_run": False, + }, + }, + } + + task_info = { + "conditions": { + "ac_only": True, + "run_if_idle": False, + "run_if_network": False, + "start_when_available": False, + }, + "trigger": { + "end_date": None, + "execution_time_limit": "3 days", + "state_change": "SessionUnlock", + "random_delay": False, + "end_time": "00:00:00", + "start_date": "2019-05-14", + "repeat_duration": None, + "start_time": "01:00 pm", + "repeat_interval": None, + "delay": False, + "trigger_enabled": True, + "trigger_type": "OnSessionChange", + "repeat_stop_at_duration_end": False, + }, + "action": { + "start_in": "", + "cmd": "del /Q /S C:\\\\Temp", + "arguments": "", + "action_type": "Execute", + }, + "settings": { + "delete_after": False, + "multiple_instances": "No New Instance", + "execution_time_limit": "3 days", + "allow_demand_start": True, + "restart_interval": False, + "stop_if_on_batteries": True, + "force_stop": True, + "wake_to_run": False, + }, + } + + prediction = win_task._get_task_state_prediction(state, task_info) + assert state == prediction + + +# The following tests check if the state prediction is correct. A lot of tests +# might look the same but under the hood a lot of checks are happening. The +# Triggers Test does not test Once or Event +def test_daily(): + kwargs = { + "action_type": "Execute", + "cmd": "del /Q /S C:\\\\Temp", + "trigger_type": "Daily", + "start_date": "2019-05-14", + "start_time": "01:00 pm", + "days_interval": 101, + } + + info = { + "triggers": [ + { + "random_delay": False, + "trigger_type": "Daily", + "execution_time_limit": "3 days", + "start_time": "13:00:00", + "enabled": True, + "start_date": "2019-05-14", + } + ], + "actions": [{"cmd": "del /Q /S C:\\\\Temp", "action_type": "Execute"}], + "conditions": { + "start_when_available": False, + "run_if_network": False, + "ac_only": True, + "run_if_idle": False, + }, + "settings": { + "wake_to_run": False, + "allow_demand_start": True, + "multiple_instances": "No New Instance", + "execution_time_limit": "3 days", + "force_stop": True, + "delete_after": False, + "stop_if_on_batteries": True, + "restart_interval": False, + }, + } + + with patch.dict( + win_task.__salt__, + { + "task.list_tasks": MagicMock(side_effect=[["salt"]] * 2), + "task.info": MagicMock(side_effect=[info]), + }, + ), patch.dict(win_task.__opts__, {"test": True}), patch.dict( + win_task.__grains__, {"osversion": "7.1"} + ): + ret = win_task.present(name="salt", location="", force=True, **kwargs) + assert ret["result"] + + +def test_weekly(): + kwargs = { + "action_type": "Execute", + "cmd": "del /Q /S C:\\\\Temp", + "trigger_type": "Weekly", + "start_date": "2019-05-14", + "start_time": "01:00 pm", + "days_of_week": ["Monday", "Wednesday", "Friday"], + "weeks_interval": 1, + } + + info = { + "triggers": [ + { + "start_date": "2019-05-14", + "execution_time_limit": "3 days", + "random_delay": False, + "enabled": True, + "start_time": "13:00:00", + "trigger_type": "Weekly", + } + ], + "actions": [{"cmd": "del /Q /S C:\\\\Temp", "action_type": "Execute"}], + "conditions": { + "start_when_available": False, + "run_if_idle": False, + "run_if_network": False, + "ac_only": True, + }, + "settings": { + "allow_demand_start": True, + "wake_to_run": False, + "execution_time_limit": "3 days", + "force_stop": True, + "multiple_instances": "No New Instance", + "stop_if_on_batteries": True, + "restart_interval": False, + "delete_after": False, + }, + } + + with patch.dict( + win_task.__salt__, + { + "task.list_tasks": MagicMock(side_effect=[["salt"]] * 2), + "task.info": MagicMock(side_effect=[info]), + }, + ), patch.dict(win_task.__opts__, {"test": True}), patch.dict( + win_task.__grains__, {"osversion": "7.1"} + ): + ret = win_task.present(name="salt", location="", force=True, **kwargs) + assert ret["result"] + + +def test_monthly(): + kwargs = { + "action_type": "Execute", + "cmd": "del /Q /S C:\\\\Temp", + "trigger_type": "Monthly", + "start_date": "2019-05-14", + "start_time": "01:00 pm", + "months_of_year": ["January", "July"], + "days_of_month": [6, 16, 26], + "last_day_of_month": True, + } + + info = { + "triggers": [ + { + "start_date": "2019-05-14", + "random_delay": False, + "trigger_type": "Monthly", + "execution_time_limit": "3 days", + "start_time": "13:00:00", + "enabled": True, + } + ], + "actions": [{"cmd": "del /Q /S C:\\\\Temp", "action_type": "Execute"}], + "conditions": { + "run_if_idle": False, + "run_if_network": False, + "start_when_available": False, + "ac_only": True, + }, + "settings": { + "force_stop": True, + "allow_demand_start": True, + "delete_after": False, + "multiple_instances": "No New Instance", + "execution_time_limit": "3 days", + "stop_if_on_batteries": True, + "restart_interval": False, + "wake_to_run": False, + }, + } + + with patch.dict( + win_task.__salt__, + { + "task.list_tasks": MagicMock(side_effect=[["salt"]] * 2), + "task.info": MagicMock(side_effect=[info]), + }, + ), patch.dict(win_task.__opts__, {"test": True}), patch.dict( + win_task.__grains__, {"osversion": "7.1"} + ): + ret = win_task.present(name="salt", location="", force=True, **kwargs) + assert ret["result"] + + +def test_monthly_day(): + kwargs = { + "action_type": "Execute", + "cmd": "del /Q /S C:\\\\Temp", + "trigger_type": "MonthlyDay", + "start_date": "2019-05-14", + "start_time": "01:00 pm", + "months_of_year": ["January", "July"], + "weeks_of_month": ["First", "Third"], + "last_week_of_month": True, + "days_of_week": ["Monday", "Wednesday", "Friday"], + } + + info = { + "triggers": [ + { + "start_date": "2019-05-14", + "random_delay": False, + "trigger_type": "MonthlyDay", + "execution_time_limit": "3 days", + "start_time": "13:00:00", + "enabled": True, + } + ], + "actions": [{"cmd": "del /Q /S C:\\\\Temp", "action_type": "Execute"}], + "conditions": { + "run_if_idle": False, + "run_if_network": False, + "start_when_available": False, + "ac_only": True, + }, + "settings": { + "force_stop": True, + "allow_demand_start": True, + "delete_after": False, + "multiple_instances": "No New Instance", + "execution_time_limit": "3 days", + "stop_if_on_batteries": True, + "restart_interval": False, + "wake_to_run": False, + }, + } + + with patch.dict( + win_task.__salt__, + { + "task.list_tasks": MagicMock(side_effect=[["salt"]] * 2), + "task.info": MagicMock(side_effect=[info]), + }, + ), patch.dict(win_task.__opts__, {"test": True}), patch.dict( + win_task.__grains__, {"osversion": "7.1"} + ): + + ret = win_task.present(name="salt", location="", force=True, **kwargs) + assert ret["result"] + + +def test_on_idle(): + kwargs = { + "action_type": "Execute", + "cmd": "del /Q /S C:\\\\Temp", + "trigger_type": "OnIdle", + "start_date": "2019-05-14", + "start_time": "01:00 pm", + } + + info = { + "triggers": [ + { + "start_date": "2019-05-14", + "random_delay": False, + "trigger_type": "OnIdle", + "execution_time_limit": "3 days", + "start_time": "13:00:00", + "enabled": True, + } + ], + "actions": [{"cmd": "del /Q /S C:\\\\Temp", "action_type": "Execute"}], + "conditions": { + "run_if_idle": False, + "run_if_network": False, + "start_when_available": False, + "ac_only": True, + }, + "settings": { + "force_stop": True, + "allow_demand_start": True, + "delete_after": False, + "multiple_instances": "No New Instance", + "execution_time_limit": "3 days", + "stop_if_on_batteries": True, + "restart_interval": False, + "wake_to_run": False, + }, + } + + with patch.dict( + win_task.__salt__, + { + "task.list_tasks": MagicMock(side_effect=[["salt"]] * 2), + "task.info": MagicMock(side_effect=[info]), + }, + ), patch.dict(win_task.__opts__, {"test": True}), patch.dict( + win_task.__grains__, {"osversion": "7.1"} + ): + ret = win_task.present(name="salt", location="", force=True, **kwargs) + assert ret["result"] + + +def test_on_task_creation(): + kwargs = { + "action_type": "Execute", + "cmd": "del /Q /S C:\\\\Temp", + "trigger_type": "OnTaskCreation", + "start_date": "2019-05-14", + "start_time": "01:00 pm", + } + + info = { + "triggers": [ + { + "start_date": "2019-05-14", + "random_delay": False, + "trigger_type": "OnTaskCreation", + "execution_time_limit": "3 days", + "start_time": "13:00:00", + "enabled": True, + } + ], + "actions": [{"cmd": "del /Q /S C:\\\\Temp", "action_type": "Execute"}], + "conditions": { + "run_if_idle": False, + "run_if_network": False, + "start_when_available": False, + "ac_only": True, + }, + "settings": { + "force_stop": True, + "allow_demand_start": True, + "delete_after": False, + "multiple_instances": "No New Instance", + "execution_time_limit": "3 days", + "stop_if_on_batteries": True, + "restart_interval": False, + "wake_to_run": False, + }, + } + + with patch.dict( + win_task.__salt__, + { + "task.list_tasks": MagicMock(side_effect=[["salt"]] * 2), + "task.info": MagicMock(side_effect=[info]), + }, + ), patch.dict(win_task.__opts__, {"test": True}), patch.dict( + win_task.__grains__, {"osversion": "7.1"} + ): + + ret = win_task.present(name="salt", location="", force=True, **kwargs) + assert ret["result"] + + +def test_on_boot(): + kwargs = { + "action_type": "Execute", + "cmd": "del /Q /S C:\\\\Temp", + "trigger_type": "OnBoot", + "start_date": "2019-05-14", + "start_time": "01:00 pm", + } + + info = { + "triggers": [ + { + "start_date": "2019-05-14", + "random_delay": False, + "trigger_type": "OnBoot", + "execution_time_limit": "3 days", + "start_time": "13:00:00", + "enabled": True, + "delay": False, + } + ], + "actions": [{"cmd": "del /Q /S C:\\\\Temp", "action_type": "Execute"}], + "conditions": { + "run_if_idle": False, + "run_if_network": False, + "start_when_available": False, + "ac_only": True, + }, + "settings": { + "force_stop": True, + "allow_demand_start": True, + "delete_after": False, + "multiple_instances": "No New Instance", + "execution_time_limit": "3 days", + "stop_if_on_batteries": True, + "restart_interval": False, + "wake_to_run": False, + }, + } + + with patch.dict( + win_task.__salt__, + { + "task.list_tasks": MagicMock(side_effect=[["salt"]] * 2), + "task.info": MagicMock(side_effect=[info]), + }, + ), patch.dict(win_task.__opts__, {"test": True}), patch.dict( + win_task.__grains__, {"osversion": "7.1"} + ): + + ret = win_task.present(name="salt", location="", force=True, **kwargs) + assert ret["result"] + + +def test_on_logon(): + kwargs = { + "action_type": "Execute", + "cmd": "del /Q /S C:\\\\Temp", + "trigger_type": "OnLogon", + "start_date": "2019-05-14", + "start_time": "01:00 pm", + } + + info = { + "triggers": [ + { + "start_date": "2019-05-14", + "random_delay": False, + "trigger_type": "OnLogon", + "execution_time_limit": "3 days", + "start_time": "13:00:00", + "enabled": True, + } + ], + "actions": [{"cmd": "del /Q /S C:\\\\Temp", "action_type": "Execute"}], + "conditions": { + "run_if_idle": False, + "run_if_network": False, + "start_when_available": False, + "ac_only": True, + }, + "settings": { + "force_stop": True, + "allow_demand_start": True, + "delete_after": False, + "multiple_instances": "No New Instance", + "execution_time_limit": "3 days", + "stop_if_on_batteries": True, + "restart_interval": False, + "wake_to_run": False, + }, + } + + with patch.dict( + win_task.__salt__, + { + "task.list_tasks": MagicMock(side_effect=[["salt"]] * 2), + "task.info": MagicMock(side_effect=[info]), + }, + ), patch.dict(win_task.__opts__, {"test": True}), patch.dict( + win_task.__grains__, {"osversion": "7.1"} + ): + + ret = win_task.present(name="salt", location="", force=True, **kwargs) + assert ret["result"] + + +def test_on_session_change(): + kwargs = { + "action_type": "Execute", + "cmd": "del /Q /S C:\\\\Temp", + "trigger_type": "OnSessionChange", + "start_date": "2019-05-14", + "start_time": "01:00 pm", + "state_change": "SessionUnlock", + } + + info = { + "actions": [{"cmd": "del /Q /S C:\\\\Temp", "action_type": "Execute"}], + "settings": { + "delete_after": False, + "execution_time_limit": "3 days", + "wake_to_run": False, + "force_stop": True, + "multiple_instances": "No New Instance", + "stop_if_on_batteries": True, + "restart_interval": False, + "allow_demand_start": True, + }, + "triggers": [ + { + "trigger_type": "OnSessionChange", + "execution_time_limit": "3 days", + "delay": False, + "enabled": True, + "start_date": "2019-05-14", + "start_time": "13:00:00", + } + ], + "conditions": { + "run_if_idle": False, + "ac_only": True, + "run_if_network": False, + "start_when_available": False, + }, + } + + with patch.dict( + win_task.__salt__, + { + "task.list_tasks": MagicMock(side_effect=[["salt"]] * 2), + "task.info": MagicMock(side_effect=[info]), + }, + ), patch.dict(win_task.__opts__, {"test": True}), patch.dict( + win_task.__grains__, {"osversion": "7.1"} + ): + ret = win_task.present(name="salt", location="", force=True, **kwargs) + + assert ret["result"] diff --git a/tests/unit/states/test_win_task.py b/tests/unit/states/test_win_task.py deleted file mode 100644 index a00418b929de..000000000000 --- a/tests/unit/states/test_win_task.py +++ /dev/null @@ -1,507 +0,0 @@ -# -*- coding: utf-8 -*- -# https://msdn.microsoft.com/en-us/library/windows/desktop/aa383608(v=vs.85).aspx -from __future__ import absolute_import, print_function, unicode_literals - -# Import Salt Libs -import salt.modules.win_task -import salt.states.win_task as win_task - -# Import Salt Testing Libs -import salt.utils.platform -from tests.support.mixins import LoaderModuleMockMixin -from tests.support.mock import NO_MOCK, NO_MOCK_REASON, MagicMock, patch -from tests.support.unit import skipIf, TestCase -from tests.support.helpers import destructiveTest - - -@destructiveTest -@skipIf(NO_MOCK, NO_MOCK_REASON) -@skipIf(not salt.utils.platform.is_windows(), "Windows is required") -class WinTaskCase(TestCase, LoaderModuleMockMixin): - ''' - Test cases for salt.states.win_task - ''' - def setup_loader_modules(self): - return {win_task: {}} - - def test_present(self): - kwargs = {'action_type': 'Execute', - 'cmd': 'del /Q /S C:\\\\Temp', - 'trigger_type': 'Once', - 'start_data': '2019-05-14', - 'start_time': '01:00 pm'} - - ret = {'result': False} - try: - with patch.dict(win_task.__salt__, {'task.list_tasks': salt.modules.win_task.list_tasks, - 'task.info': salt.modules.win_task.info, - 'task.create_task': salt.modules.win_task.create_task}), \ - patch.dict(win_task.__opts__, {"test": False}), \ - patch.dict(win_task.__grains__, {'osversion': '7.1'}): - - ret = win_task.present(name='salt', location='', force=True, **kwargs) - finally: - try: - salt.modules.win_task.delete_task(name='salt', location='') - finally: - pass - - self.assertEqual(ret['result'], True) - - def test_absent(self): - with patch.dict(win_task.__salt__, {'task.list_tasks': salt.modules.win_task.list_tasks, - 'task.info': salt.modules.win_task.info, - 'task.delete_task': salt.modules.win_task.delete_task}), \ - patch.dict(win_task.__opts__, {"test": False}): - ret = win_task.absent('salt', '') - - self.assertEqual(ret['result'], True) - - kwargs = {'action_type': 'Execute', - 'cmd': 'del /Q /S C:\\\\Temp', - 'trigger_type': 'Once', - 'start_data': '2019-05-14', - 'start_time': '01:00 pm'} - - try: - with patch.dict(win_task.__salt__, {'task.list_tasks': salt.modules.win_task.list_tasks, - 'task.info': salt.modules.win_task.info, - 'task.create_task': salt.modules.win_task.create_task}), \ - patch.dict(win_task.__opts__, {"test": False}), \ - patch.dict(win_task.__grains__, {'osversion': '7.1'}): - - win_task.present(name='salt', location='', force=True, **kwargs) - finally: - try: - with patch.dict(win_task.__salt__, {'task.list_tasks': salt.modules.win_task.list_tasks, - 'task.info': salt.modules.win_task.info, - 'task.delete_task': salt.modules.win_task.delete_task}), \ - patch.dict(win_task.__opts__, {"test": False}): - ret = win_task.absent('salt', '') - finally: - pass - - self.assertEqual(ret['result'], True) - - -@skipIf(NO_MOCK, NO_MOCK_REASON) -@skipIf(not salt.utils.platform.is_windows(), "Windows is required") -class WinTaskPrivateCase(TestCase, LoaderModuleMockMixin): - def setup_loader_modules(self): - return {win_task: {}} - - def test__get_arguments(self): - kwargs = {'salt': True, - 'cat': 'nice', - 'idk': 404} - - true_ret = {'salt': True, - 'cat': 'nice', - 'fat': True, - 'idk': 404} - - ret = win_task._get_arguments(kwargs, - ['cat'], - {'nice': ['idk'], - 'sad': ['why']}, - {'fat': True, - 'salt': None}) - - self.assertEqual(ret, true_ret) - - def test__get_task_state_prediction(self): - state = {'task_found': True, - 'location_valid': True, - 'task_info': {'conditions': {'ac_only': True, - 'run_if_idle': False, - 'run_if_network': False, - 'start_when_available': False}, - 'actions': [{'cmd': 'del /Q /S C:\\\\Temp', - 'action_type': 'Execute'}], - 'triggers': [{'delay': False, - 'execution_time_limit': '3 days', - 'trigger_type': 'OnSessionChange', - 'start_date': '2019-05-14', 'enabled': True, - 'start_time': '13:00:00'}], - 'settings': {'delete_after': False, - 'multiple_instances': 'No New Instance', - 'execution_time_limit': '3 days', - 'allow_demand_start': True, - 'restart_interval': False, - 'stop_if_on_batteries': True, - 'force_stop': True, - 'wake_to_run': False}}} - - task_info = {'conditions': {'ac_only': True, - 'run_if_idle': False, - 'run_if_network': False, - 'start_when_available': False}, - 'trigger': {'end_date': None, - 'execution_time_limit': '3 days', - 'state_change': 'SessionUnlock', 'random_delay': False, - 'end_time': '00:00:00', - 'start_date': '2019-05-14', - 'repeat_duration': None, - 'start_time': '01:00 pm', - 'repeat_interval': None, - 'delay': False, - 'trigger_enabled': True, - 'trigger_type': 'OnSessionChange', - 'repeat_stop_at_duration_end': False}, - 'action': {'start_in': '', - 'cmd': 'del /Q /S C:\\\\Temp', - 'arguments': '', - 'action_type': 'Execute'}, - 'settings': {'delete_after': False, - 'multiple_instances': 'No New Instance', - 'execution_time_limit': '3 days', - 'allow_demand_start': True, - 'restart_interval': False, - 'stop_if_on_batteries': True, - 'force_stop': True, - 'wake_to_run': False}} - - prediction = win_task._get_task_state_prediction(state, task_info) - self.assertEqual(state, prediction) - - -@skipIf(NO_MOCK, NO_MOCK_REASON) -@skipIf(not salt.utils.platform.is_windows(), "Windows is required") -class WinTaskTriggersCase(TestCase, LoaderModuleMockMixin): - ''' - The test below just checks if the state perdition is correct. - A lot of test might look the same but under hud a lot of checks are happening. - Triggers Test does not test Once or Event - ''' - def setup_loader_modules(self): - return {win_task: {}} - - def test_Daily(self): - kwargs = {'action_type': 'Execute', - 'cmd': 'del /Q /S C:\\\\Temp', - 'trigger_type': 'Daily', - 'start_date': '2019-05-14', - 'start_time': '01:00 pm', - 'days_interval': 101} - - info = {'triggers': [{'random_delay': False, - 'trigger_type': 'Daily', - 'execution_time_limit': '3 days', - 'start_time': '13:00:00', - 'enabled': True, - 'start_date': '2019-05-14'}], - 'actions': [{'cmd': 'del /Q /S C:\\\\Temp', - 'action_type': 'Execute'}], - 'conditions': {'start_when_available': False, - 'run_if_network': False, - 'ac_only': True, - 'run_if_idle': False}, - 'settings': {'wake_to_run': False, - 'allow_demand_start': True, - 'multiple_instances': 'No New Instance', - 'execution_time_limit': '3 days', - 'force_stop': True, - 'delete_after': False, - 'stop_if_on_batteries': True, - 'restart_interval': False}} - - with patch.dict(win_task.__salt__, {'task.list_tasks': MagicMock(side_effect=[['salt']] * 2), - 'task.info': MagicMock(side_effect=[info])}), \ - patch.dict(win_task.__opts__, {"test": True}), \ - patch.dict(win_task.__grains__, {'osversion': '7.1'}): - ret = win_task.present(name='salt', location='', force=True, **kwargs) - self.assertEqual(ret['result'], True) - - def test_Weekly(self): - kwargs = {'action_type': 'Execute', - 'cmd': 'del /Q /S C:\\\\Temp', - 'trigger_type': 'Weekly', - 'start_date': '2019-05-14', - 'start_time': '01:00 pm', - 'days_of_week': ['Monday', 'Wednesday', 'Friday'], - 'weeks_interval': 1} - - info = {'triggers': [{'start_date': '2019-05-14', - 'execution_time_limit': '3 days', - 'random_delay': False, - 'enabled': True, - 'start_time': '13:00:00', - 'trigger_type': 'Weekly'}], - 'actions': [{'cmd': 'del /Q /S C:\\\\Temp', - 'action_type': 'Execute'}], - 'conditions': {'start_when_available': False, - 'run_if_idle': False, - 'run_if_network': False, - 'ac_only': True}, - 'settings': {'allow_demand_start': True, - 'wake_to_run': False, - 'execution_time_limit': '3 days', - 'force_stop': True, - 'multiple_instances': 'No New Instance', - 'stop_if_on_batteries': True, - 'restart_interval': False, - 'delete_after': False}} - - with patch.dict(win_task.__salt__, {'task.list_tasks': MagicMock(side_effect=[['salt']] * 2), - 'task.info': MagicMock(side_effect=[info])}), \ - patch.dict(win_task.__opts__, {"test": True}), \ - patch.dict(win_task.__grains__, {'osversion': '7.1'}): - ret = win_task.present(name='salt', location='', force=True, **kwargs) - self.assertEqual(ret['result'], True) - - def test_Monthly(self): - kwargs = {'action_type': 'Execute', - 'cmd': 'del /Q /S C:\\\\Temp', - 'trigger_type': 'Monthly', - 'start_date': '2019-05-14', - 'start_time': '01:00 pm', - 'months_of_year': ['January', 'July'], - 'days_of_month': [6, 16, 26], - 'last_day_of_month': True} - - info = {'triggers': [{'start_date': '2019-05-14', - 'random_delay': False, - 'trigger_type': 'Monthly', - 'execution_time_limit': '3 days', - 'start_time': '13:00:00', - 'enabled': True}], - 'actions': [{'cmd': 'del /Q /S C:\\\\Temp', - 'action_type': 'Execute'}], - 'conditions': {'run_if_idle': False, - 'run_if_network': False, - 'start_when_available': False, - 'ac_only': True}, - 'settings': {'force_stop': True, - 'allow_demand_start': True, - 'delete_after': False, - 'multiple_instances': 'No New Instance', - 'execution_time_limit': '3 days', 'stop_if_on_batteries': True, - 'restart_interval': False, - 'wake_to_run': False}} - - with patch.dict(win_task.__salt__, {'task.list_tasks': MagicMock(side_effect=[['salt']] * 2), - 'task.info': MagicMock(side_effect=[info])}), \ - patch.dict(win_task.__opts__, {"test": True}), \ - patch.dict(win_task.__grains__, {'osversion': '7.1'}): - ret = win_task.present(name='salt', location='', force=True, **kwargs) - self.assertEqual(ret['result'], True) - - def test_MonthlyDay(self): - kwargs = {'action_type': 'Execute', - 'cmd': 'del /Q /S C:\\\\Temp', - 'trigger_type': 'MonthlyDay', - 'start_date': '2019-05-14', - 'start_time': '01:00 pm', - 'months_of_year': ['January', 'July'], - 'weeks_of_month': ['First', 'Third'], - 'last_week_of_month': True, - 'days_of_week': ['Monday', 'Wednesday', 'Friday']} - - info = {'triggers': [{'start_date': '2019-05-14', - 'random_delay': False, - 'trigger_type': 'MonthlyDay', - 'execution_time_limit': '3 days', - 'start_time': '13:00:00', - 'enabled': True}], - 'actions': [{'cmd': 'del /Q /S C:\\\\Temp', - 'action_type': 'Execute'}], - 'conditions': {'run_if_idle': False, - 'run_if_network': False, - 'start_when_available': False, - 'ac_only': True}, - 'settings': {'force_stop': True, - 'allow_demand_start': True, - 'delete_after': False, - 'multiple_instances': 'No New Instance', - 'execution_time_limit': '3 days', 'stop_if_on_batteries': True, - 'restart_interval': False, - 'wake_to_run': False}} - - with patch.dict(win_task.__salt__, {'task.list_tasks': MagicMock(side_effect=[['salt']] * 2), - 'task.info': MagicMock(side_effect=[info])}), \ - patch.dict(win_task.__opts__, {"test": True}), \ - patch.dict(win_task.__grains__, {'osversion': '7.1'}): - - ret = win_task.present(name='salt', location='', force=True, **kwargs) - self.assertEqual(ret['result'], True) - - def test_OnIdle(self): - kwargs = {'action_type': 'Execute', - 'cmd': 'del /Q /S C:\\\\Temp', - 'trigger_type': 'OnIdle', - 'start_date': '2019-05-14', - 'start_time': '01:00 pm'} - - info = {'triggers': [{'start_date': '2019-05-14', - 'random_delay': False, - 'trigger_type': 'OnIdle', - 'execution_time_limit': '3 days', - 'start_time': '13:00:00', - 'enabled': True}], - 'actions': [{'cmd': 'del /Q /S C:\\\\Temp', - 'action_type': 'Execute'}], - 'conditions': {'run_if_idle': False, - 'run_if_network': False, - 'start_when_available': False, - 'ac_only': True}, - 'settings': {'force_stop': True, - 'allow_demand_start': True, - 'delete_after': False, - 'multiple_instances': 'No New Instance', - 'execution_time_limit': '3 days', - 'stop_if_on_batteries': True, - 'restart_interval': False, - 'wake_to_run': False}} - - with patch.dict(win_task.__salt__, {'task.list_tasks': MagicMock(side_effect=[['salt']] * 2), - 'task.info': MagicMock(side_effect=[info])}), \ - patch.dict(win_task.__opts__, {"test": True}), \ - patch.dict(win_task.__grains__, {'osversion': '7.1'}): - ret = win_task.present(name='salt', location='', force=True, **kwargs) - self.assertEqual(ret['result'], True) - - def test_OnTaskCreation(self): - kwargs = {'action_type': 'Execute', - 'cmd': 'del /Q /S C:\\\\Temp', - 'trigger_type': 'OnTaskCreation', - 'start_date': '2019-05-14', - 'start_time': '01:00 pm'} - - info = {'triggers': [{'start_date': '2019-05-14', - 'random_delay': False, - 'trigger_type': 'OnTaskCreation', - 'execution_time_limit': '3 days', - 'start_time': '13:00:00', - 'enabled': True}], - 'actions': [{'cmd': 'del /Q /S C:\\\\Temp', - 'action_type': 'Execute'}], - 'conditions': {'run_if_idle': False, - 'run_if_network': False, - 'start_when_available': False, - 'ac_only': True}, - 'settings': {'force_stop': True, - 'allow_demand_start': True, - 'delete_after': False, - 'multiple_instances': 'No New Instance', - 'execution_time_limit': '3 days', - 'stop_if_on_batteries': True, - 'restart_interval': False, - 'wake_to_run': False}} - - with patch.dict(win_task.__salt__, {'task.list_tasks': MagicMock(side_effect=[['salt']] * 2), - 'task.info': MagicMock(side_effect=[info])}), \ - patch.dict(win_task.__opts__, {"test": True}), \ - patch.dict(win_task.__grains__, {'osversion': '7.1'}): - - ret = win_task.present(name='salt', location='', force=True, **kwargs) - self.assertEqual(ret['result'], True) - - def test_OnBoot(self): - kwargs = {'action_type': 'Execute', - 'cmd': 'del /Q /S C:\\\\Temp', - 'trigger_type': 'OnBoot', - 'start_date': '2019-05-14', - 'start_time': '01:00 pm'} - - info = {'triggers': [{'start_date': '2019-05-14', - 'random_delay': False, - 'trigger_type': 'OnBoot', - 'execution_time_limit': '3 days', - 'start_time': '13:00:00', - 'enabled': True, - 'delay': False}], - 'actions': [{'cmd': 'del /Q /S C:\\\\Temp', - 'action_type': 'Execute'}], - 'conditions': {'run_if_idle': False, - 'run_if_network': False, - 'start_when_available': False, - 'ac_only': True}, - 'settings': {'force_stop': True, - 'allow_demand_start': True, - 'delete_after': False, - 'multiple_instances': 'No New Instance', - 'execution_time_limit': '3 days', - 'stop_if_on_batteries': True, - 'restart_interval': False, - 'wake_to_run': False}} - - with patch.dict(win_task.__salt__, {'task.list_tasks': MagicMock(side_effect=[['salt']] * 2), - 'task.info': MagicMock(side_effect=[info])}), \ - patch.dict(win_task.__opts__, {"test": True}), \ - patch.dict(win_task.__grains__, {'osversion': '7.1'}): - - ret = win_task.present(name='salt', location='', force=True, **kwargs) - self.assertEqual(ret['result'], True) - - def test_OnLogon(self): - kwargs = {'action_type': 'Execute', - 'cmd': 'del /Q /S C:\\\\Temp', - 'trigger_type': 'OnLogon', - 'start_date': '2019-05-14', - 'start_time': '01:00 pm'} - - info = {'triggers': [{'start_date': '2019-05-14', - 'random_delay': False, - 'trigger_type': 'OnLogon', - 'execution_time_limit': '3 days', - 'start_time': '13:00:00', - 'enabled': True}], - 'actions': [{'cmd': 'del /Q /S C:\\\\Temp', - 'action_type': 'Execute'}], - 'conditions': {'run_if_idle': False, - 'run_if_network': False, - 'start_when_available': False, - 'ac_only': True}, - 'settings': {'force_stop': True, - 'allow_demand_start': True, - 'delete_after': False, - 'multiple_instances': 'No New Instance', - 'execution_time_limit': '3 days', - 'stop_if_on_batteries': True, - 'restart_interval': False, - 'wake_to_run': False}} - - with patch.dict(win_task.__salt__, {'task.list_tasks': MagicMock(side_effect=[['salt']] * 2), - 'task.info': MagicMock(side_effect=[info])}), \ - patch.dict(win_task.__opts__, {"test": True}), \ - patch.dict(win_task.__grains__, {'osversion': '7.1'}): - - ret = win_task.present(name='salt', location='', force=True, **kwargs) - self.assertEqual(ret['result'], True) - - def test_OnSessionChange(self): - kwargs = {'action_type': 'Execute', - 'cmd': 'del /Q /S C:\\\\Temp', - 'trigger_type': 'OnSessionChange', - 'start_date': '2019-05-14', - 'start_time': '01:00 pm', - 'state_change': 'SessionUnlock'} - - info = {'actions': [{'cmd': 'del /Q /S C:\\\\Temp', - 'action_type': 'Execute'}], - 'settings': {'delete_after': False, - 'execution_time_limit': '3 days', - 'wake_to_run': False, - 'force_stop': True, - 'multiple_instances': 'No New Instance', - 'stop_if_on_batteries': True, - 'restart_interval': False, - 'allow_demand_start': True}, - 'triggers': [{'trigger_type': 'OnSessionChange', - 'execution_time_limit': '3 days', - 'delay': False, - 'enabled': True, - 'start_date': '2019-05-14', - 'start_time': '13:00:00'}], - 'conditions': {'run_if_idle': False, - 'ac_only': True, - 'run_if_network': False, - 'start_when_available': False}} - - with patch.dict(win_task.__salt__, {'task.list_tasks': MagicMock(side_effect=[['salt']] * 2), - 'task.info': MagicMock(side_effect=[info])}), \ - patch.dict(win_task.__opts__, {"test": True}), \ - patch.dict(win_task.__grains__, {'osversion': '7.1'}): - ret = win_task.present(name='salt', location='', force=True, **kwargs) - - self.assertEqual(ret['result'], True)