test_helpers.py 2.0 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768
  1. # -*- coding: utf-8 -*-
  2. # Import Python libs
  3. from __future__ import absolute_import
  4. import copy
  5. import logging
  6. import os
  7. # Import Salt Testing libs
  8. from tests.support.case import ModuleCase
  9. from tests.support.mixins import SaltReturnAssertsMixin
  10. # Import Salt Testing Libs
  11. from tests.support.mock import MagicMock, patch
  12. import tests.integration as integration
  13. # Import Salt libs
  14. import salt.utils.schedule
  15. import salt.utils.platform
  16. from salt.modules.test import ping as ping
  17. log = logging.getLogger(__name__)
  18. ROOT_DIR = os.path.join(integration.TMP, 'schedule-unit-tests')
  19. SOCK_DIR = os.path.join(ROOT_DIR, 'test-socks')
  20. DEFAULT_CONFIG = salt.config.minion_config(None)
  21. DEFAULT_CONFIG['conf_dir'] = ROOT_DIR
  22. DEFAULT_CONFIG['root_dir'] = ROOT_DIR
  23. DEFAULT_CONFIG['sock_dir'] = SOCK_DIR
  24. DEFAULT_CONFIG['pki_dir'] = os.path.join(ROOT_DIR, 'pki')
  25. DEFAULT_CONFIG['cachedir'] = os.path.join(ROOT_DIR, 'cache')
  26. class SchedulerHelpersTest(ModuleCase, SaltReturnAssertsMixin):
  27. '''
  28. Test scheduler helper functions
  29. '''
  30. def setUp(self):
  31. with patch('salt.utils.schedule.clean_proc_dir', MagicMock(return_value=None)):
  32. functions = {'test.ping': ping}
  33. self.schedule = salt.utils.schedule.Schedule(copy.deepcopy(DEFAULT_CONFIG), functions, returners={})
  34. self.schedule.opts['loop_interval'] = 1
  35. def tearDown(self):
  36. self.schedule.reset()
  37. def test_get_schedule(self):
  38. '''
  39. verify that the _get_schedule function works
  40. when remove_hidden is True and schedule data
  41. contains enabled key
  42. '''
  43. job_name = 'test_get_schedule'
  44. job = {
  45. 'schedule': {
  46. 'enabled': True,
  47. job_name: {
  48. 'function': 'test.ping',
  49. 'seconds': 60
  50. }
  51. }
  52. }
  53. # Add the job to the scheduler
  54. self.schedule.opts.update(job)
  55. ret = self.schedule._get_schedule(remove_hidden=True)
  56. self.assertEqual(job['schedule'], ret)