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 2

schedule tests to ensure pillar jobs are not modified

Browse source
This commit is contained in:
Vyacheslav Spiridonov 2017-03-15 18:36:15 +03:00
parent 27385ff49c
commit d3a2489c9c
1 changed files with 45 additions and 29 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

74
tests/unit/utils/schedule_test.py
View file

@ -55,12 +55,12 @@ class ScheduleTestCase(TestCase):
def test_delete_job_in_pillar(self):
'''
Tests deleting job in pillar
Tests ignoring deletion job from pillar
'''
self.schedule.opts.update({'pillar': {'schedule': {'foo': 'bar'}}, 'schedule': ''})
self.assertIn('foo', self.schedule.opts['pillar']['schedule'])
self.schedule.delete_job('foo', where='pillar')
self.assertNotIn('foo', self.schedule.opts['pillar']['schedule'])
self.schedule.delete_job('foo')
self.assertIn('foo', self.schedule.opts['pillar']['schedule'])
def test_delete_job_intervals(self):
'''
@ -85,14 +85,12 @@ class ScheduleTestCase(TestCase):
def test_delete_job_prefix_in_pillar(self):
'''
Tests deleting jobs by prefix in pillar
Tests ignoring deletion jobs by prefix from pillar
'''
self.schedule.opts.update({'pillar': {'schedule': {'foobar': 'bar', 'foobaz': 'baz', 'fooboo': 'boo'}},
'schedule': ''})
ret = copy.deepcopy(self.schedule.opts)
del ret['pillar']['schedule']['foobar']
del ret['pillar']['schedule']['foobaz']
self.schedule.delete_job_prefix('fooba', where='pillar')
self.schedule.delete_job_prefix('fooba')
self.assertEqual(self.schedule.opts, ret)
# add_job tests
@ -118,8 +116,10 @@ class ScheduleTestCase(TestCase):
data = {'foo': {'bar': 'baz'}}
ret = copy.deepcopy(self.schedule.opts)
ret.update({'schedule': {'foo': {'bar': 'baz', 'enabled': True},
'hello': {'world': 'peace', 'enabled': True}}})
self.schedule.opts.update({'schedule': {'hello': {'world': 'peace', 'enabled': True}}})
'hello': {'world': 'peace', 'enabled': True}},
'pillar': ''})
self.schedule.opts.update({'schedule': {'hello': {'world': 'peace', 'enabled': True}},
'pillar': ''})
Schedule.add_job(self.schedule, data)
self.assertEqual(self.schedule.opts, ret)
@ -135,11 +135,11 @@ class ScheduleTestCase(TestCase):
def test_enable_job_pillar(self):
'''
Tests enabling a job in pillar
Tests ignoring enable a job from pillar
'''
self.schedule.opts.update({'pillar': {'schedule': {'name': {'enabled': 'foo'}}}})
Schedule.enable_job(self.schedule, 'name', persist=False, where='pillar')
self.assertTrue(self.schedule.opts['pillar']['schedule']['name']['enabled'])
self.schedule.opts.update({'pillar': {'schedule': {'name': {'enabled': False}}}})
Schedule.enable_job(self.schedule, 'name', persist=False)
self.assertFalse(self.schedule.opts['pillar']['schedule']['name']['enabled'])
# disable_job tests
@ -147,17 +147,17 @@ class ScheduleTestCase(TestCase):
'''
Tests disabling a job
'''
self.schedule.opts.update({'schedule': {'name': {'enabled': 'foo'}}})
self.schedule.opts.update({'schedule': {'name': {'enabled': 'foo'}}, 'pillar': {}})
Schedule.disable_job(self.schedule, 'name')
self.assertFalse(self.schedule.opts['schedule']['name']['enabled'])
def test_disable_job_pillar(self):
'''
Tests disabling a job in pillar
Tests ignoring disable a job in pillar
'''
self.schedule.opts.update({'pillar': {'schedule': {'name': {'enabled': 'foo'}}}})
Schedule.disable_job(self.schedule, 'name', persist=False, where='pillar')
self.assertFalse(self.schedule.opts['pillar']['schedule']['name']['enabled'])
self.schedule.opts.update({'pillar': {'schedule': {'name': {'enabled': True}}}, 'schedule': {}})
Schedule.disable_job(self.schedule, 'name', persist=False)
self.assertTrue(self.schedule.opts['pillar']['schedule']['name']['enabled'])
# modify_job tests
@ -165,22 +165,32 @@ class ScheduleTestCase(TestCase):
'''
Tests modifying a job in the scheduler
'''
schedule = {'schedule': {'foo': 'bar'}}
schedule = {'foo': 'bar'}
self.schedule.opts.update({'schedule': {'name': 'baz'}, 'pillar': {}})
ret = copy.deepcopy(self.schedule.opts)
ret.update({'schedule': {'foo': 'bar', 'name': {'schedule': {'foo': 'bar'}}}})
self.schedule.opts.update({'schedule': {'foo': 'bar'}})
ret.update({'schedule': {'name': {'foo': 'bar'}}})
Schedule.modify_job(self.schedule, 'name', schedule)
self.assertEqual(self.schedule.opts, ret)
def test_modify_job_not_exists(self):
'''
Tests modifying a job in the scheduler if jobs not exists
'''
schedule = {'foo': 'bar'}
self.schedule.opts.update({'schedule': {}, 'pillar': {}})
ret = copy.deepcopy(self.schedule.opts)
ret.update({'schedule': {'name': {'foo': 'bar'}}})
Schedule.modify_job(self.schedule, 'name', schedule)
self.assertEqual(self.schedule.opts, ret)
def test_modify_job_pillar(self):
'''
Tests modifying a job in the scheduler in pillar
Tests ignoring modification of job from pillar
'''
schedule = {'foo': 'bar'}
self.schedule.opts.update({'schedule': {}, 'pillar': {'schedule': {'name': 'baz'}}})
ret = copy.deepcopy(self.schedule.opts)
ret.update({'pillar': {'schedule': {'name': {'foo': 'bar'}}}})
self.schedule.opts.update({'pillar': {'schedule': {'name': {'foo': 'bar'}}}})
Schedule.modify_job(self.schedule, 'name', schedule, persist=False, where='pillar')
Schedule.modify_job(self.schedule, 'name', schedule, persist=False)
self.assertEqual(self.schedule.opts, ret)
maxDiff = None
@ -191,7 +201,7 @@ class ScheduleTestCase(TestCase):
'''
Tests enabling the scheduler
'''
self.schedule.opts.update({'schedule': {'enabled': 'foo'}})
self.schedule.opts.update({'schedule': {'enabled': 'foo'}, 'pillar': {}})
Schedule.enable_schedule(self.schedule)
self.assertTrue(self.schedule.opts['schedule']['enabled'])
@ -201,7 +211,7 @@ class ScheduleTestCase(TestCase):
'''
Tests disabling the scheduler
'''
self.schedule.opts.update({'schedule': {'enabled': 'foo'}})
self.schedule.opts.update({'schedule': {'enabled': 'foo'}, 'pillar': {}})
Schedule.disable_schedule(self.schedule)
self.assertFalse(self.schedule.opts['schedule']['enabled'])
@ -259,11 +269,17 @@ class ScheduleTestCase(TestCase):
def test_eval_schedule_is_not_dict(self):
'''
Tests if the schedule is a dictionary
Tests eval if the schedule is not a dictionary
'''
self.schedule.opts.update({'schedule': ''})
self.schedule.opts.update({'schedule': '', 'pillar': {'schedule': {}}})
self.assertRaises(ValueError, Schedule.eval, self.schedule)
def test_eval_schedule_is_not_dict_in_pillar(self):
'''
Tests eval if the schedule from pillar is not a dictionary
'''
self.schedule.opts.update({'schedule': {}, 'pillar': {'schedule': ''}})
self.assertRaises(ValueError, Schedule.eval, self.schedule)
if __name__ == '__main__':
from integration import run_tests