diff --git a/salt/states/win_task.py b/salt/states/win_task.py new file mode 100644 index 000000000000..47be689f6ba9 --- /dev/null +++ b/salt/states/win_task.py @@ -0,0 +1,483 @@ +# -*- 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 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 3rd-party libs +try: + import pywintypes +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}} + +log = logging.getLogger(__name__) + +__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' + 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. + + :param str location: location of task. + + :return True if location is vaild, False if location is not valid: + :rtype bool: + ''' + try: + __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 + + :param str location: location of task. + + :return task state: + :rtype dict: + ''' + 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) + # 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} + + return task_state + + +def _get_arguments(arguments_given, + key_arguments, + arguments_need_it, + optional_arguments): + 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)) + block[key] = arguments_given[key] + + # check is 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)) + + # check is key2 present + for key2 in arguments_need_it[block[key]]: + if key2 not in arguments_given: + return 'Missing {0} argument'.format(key2) + block[key2] = arguments_given[key2] + + # add optional arguments if their not present + for key in optional_arguments: + if key in arguments_given: + block[key] = arguments_given[key] + else: + block[key] = optional_arguments[key] + + return block + + +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. + + :param dict state: task state + :return task state: + :rtype dict: + ''' + + # 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 + + # format dates + 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 + try: + 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 + + trigger[key] = salt.utils.dateutils.strftime("{0}-{1}-{2}".format(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']: + if key in trigger: + try: + 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 + + :param dict state: + :param dict new_task: + :return task state: + :rtype dict: + ''' + + 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']} + + action_keys = set() + trigger_keys = set() + if state['task_found']: + # get all the arguments used by 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']: + 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']]] + + # 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: + for dic in argument_list: + for key in list(dic): + if key not in safe_keys and key in optional_keys: + if dic[key] == optional_keys[key]: + del dic[key] + + # 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']: + for key in list(trigger): + if key in argument_add_on: + del trigger[key] + + return _task_state_prediction_bandage(new_state) + + +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: + + - :py:func:`edit_task` + - :py:func:`add_action` + - :py:func:`add_trigger` + + :param str name: 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). + + :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. + + :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. + + :param bool force: If the task exists, overwrite the existing task. + + :return dict state: + :rtype dict: + + 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 + ''' + + ret = _get_state_data(name) + 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)) + 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'])} + + # 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']) + 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 + return ret + + 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 + 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 + return ret + + ret['changes'] = salt.utils.data.compare_dicts(before, after) + return ret + + # put all the arguments to kwargs + for key in new_task: + 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 + 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 + + ret['changes'] = salt.utils.data.compare_dicts(before, after) + return ret + + +def absent(name, + location='\\'): + r''' + Delete a task from the task scheduler. + + :param str name: 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). + + :return True if successful, False if unsuccessful: + :rtype bool: + + CLI Example: + + .. code-block::YAML + + test_win_task_absent: + task.absent: + - name: salt + - location: '' + + .. code-block:: bash + + salt 'minion-id' state.apply + ''' + + ret = _get_state_data(name) + 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)) + return ret + + 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) + 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': {}}) + return ret + + # if task was found then delete it + if before['task_found']: + # try to delete task + 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 + + ret['changes'] = salt.utils.data.compare_dicts(before, _get_task_state_data(name, location)) + return ret diff --git a/tests/unit/states/test_win_task.py b/tests/unit/states/test_win_task.py new file mode 100644 index 000000000000..a00418b929de --- /dev/null +++ b/tests/unit/states/test_win_task.py @@ -0,0 +1,507 @@ +# -*- 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)