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

Add a state for managing scheduled tasks on Windows

Browse source 
This is a port over from a PR that was merged onto the develop branch.
https://github.com/saltstack/salt/pull/53035
This commit is contained in:
Shane Lee 2024-02-21 09:45:13 -07:00 • committed by Daniel Wozniak
parent de710aa39e
commit d68826c61e
4 changed files with 983 additions and 775 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
changelog/59037.added.md Normal file
View file

@ -0,0 +1 @@
Added a state (win_task) for managing scheduled tasks on Windows

569
salt/states/win_task.py
View file

@ -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
HAS_PYWIN32 = False
ACTION_PARTS = {'Execute': ['cmd'],
'Email': ['from', 'to', 'cc', 'server'],
'Message': ['title', 'message']}
ACTION_PARTS = {
"Execute": ["cmd"],
"Email": ["from", "to", "cc", "server"],
"Message": ["title", "message"],
}
OPTIONAL_ACTION_PARTS = {'start_in': '',
'arguments': ''}
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']}
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_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_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}
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}}
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.
def _get_task_state_data(name, location):
r"""
Get the current state of a task in the task scheduler
:param str name: name of task
Args:
name (str): Task name
location (str): Task location
:param str location: location of task.
: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
:param dict state:
:param dict new_task:
:return task state:
:rtype dict:
'''
Args:
state (dict): A dictionary containing the current task configuration
new_task (dict) A dictionary containing the new task configuration
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
.. 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
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 <YAML file>
'''
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
# 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)
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
# 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: ''
- location: ""
"""
.. code-block:: bash
salt 'minion-id' state.apply <YAML file>
'''
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

681
tests/pytests/unit/states/test_win_task.py Normal file
View file

@ -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"]

507
tests/unit/states/test_win_task.py
View file

@ -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)