valitydev/salt
14
0
Fork 0
mirror of https://github.com/valitydev/salt.git synced 2024-11-07 00:55:19 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Merge pull request #45128 from garethgreenaway/scheduler_cron_tests

Browse Source 
[oxygen] Adding tests for cron functionality.
This commit is contained in:
Nicole Thomas 2018-01-02 10:47:36 -05:00 committed by GitHub
commit 9ee4e242e6
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 58 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

58
tests/integration/scheduler/test_eval.py
View File

@ -16,6 +16,7 @@ from tests.support.mixins import SaltReturnAssertsMixin
# Import Salt Testing Libs
from tests.support.mock import MagicMock, patch
from tests.support.unit import skipIf
import tests.integration as integration
# Import Salt libs
@ -23,6 +24,12 @@ import salt.utils.schedule
from salt.modules.test import ping as ping
try:
import croniter # pylint: disable=W0611
HAS_CRONITER = True
except ImportError:
HAS_CRONITER = False
log = logging.getLogger(__name__)
ROOT_DIR = os.path.join(integration.TMP, 'schedule-unit-tests')
SOCK_DIR = os.path.join(ROOT_DIR, 'test-socks')
@ -214,3 +221,54 @@ class SchedulerEvalTest(ModuleCase, SaltReturnAssertsMixin):
self.schedule.eval(now=run_time)
ret = self.schedule.job_status('job1')
self.assertEqual(ret['_last_run'], run_time)
@skipIf(not HAS_CRONITER, 'Cannot find croniter python module')
def test_eval_cron(self):
'''
verify that scheduled job runs
'''
job = {
'schedule': {
'job1': {
'function': 'test.ping',
'cron': '0 16 29 11 *'
}
}
}
# Add the job to the scheduler
self.schedule.opts.update(job)
run_time = int(time.mktime(dateutil_parser.parse('11/29/2017 4:00pm').timetuple()))
with patch('croniter.croniter.get_next', MagicMock(return_value=run_time)):
self.schedule.eval(now=run_time)
ret = self.schedule.job_status('job1')
self.assertEqual(ret['_last_run'], run_time)
@skipIf(not HAS_CRONITER, 'Cannot find croniter python module')
def test_eval_cron_loop_interval(self):
'''
verify that scheduled job runs
'''
job = {
'schedule': {
'job1': {
'function': 'test.ping',
'cron': '0 16 29 11 *'
}
}
}
# Randomn second loop interval
LOOP_INTERVAL = random.randint(0, 59)
self.schedule.opts['loop_interval'] = LOOP_INTERVAL
# Add the job to the scheduler
self.schedule.opts.update(job)
run_time = int(time.mktime(dateutil_parser.parse('11/29/2017 4:00pm').timetuple()))
with patch('croniter.croniter.get_next', MagicMock(return_value=run_time)):
self.schedule.eval(now=run_time)
ret = self.schedule.job_status('job1')
self.assertEqual(ret['_last_run'], run_time)