mirror of
https://github.com/getredash/redash.git
synced 2026-03-23 13:00:10 -04:00
e0e94d79ac
* add some logging to scheduler * schedule jobs only if they are not already scheduled * jobs scheduled with an interval over 24 hours were not repeated * schedule version_check using standard scheduling * clean up old jobs that are not part of the definition anymore * add some tests * add one more test to verify that reschedules are not done when not neccesary * no need to check for func existence - all jobs have a func to run
61 lines
1.9 KiB
Python
61 lines
1.9 KiB
Python
from unittest import TestCase
|
|
from mock import patch, ANY
|
|
|
|
from redash.schedule import rq_scheduler, schedule_periodic_jobs
|
|
|
|
class TestSchedule(TestCase):
|
|
def setUp(self):
|
|
for job in rq_scheduler.get_jobs():
|
|
rq_scheduler.cancel(job)
|
|
job.delete()
|
|
|
|
def test_schedules_a_new_job(self):
|
|
def foo():
|
|
pass
|
|
|
|
schedule_periodic_jobs([{"func": foo, "interval": 60}])
|
|
|
|
jobs = [job for job in rq_scheduler.get_jobs()]
|
|
|
|
self.assertEqual(len(jobs), 1)
|
|
self.assertTrue(jobs[0].func_name.endswith('foo'))
|
|
self.assertEqual(jobs[0].meta['interval'], 60)
|
|
|
|
def test_doesnt_reschedule_an_existing_job(self):
|
|
def foo():
|
|
pass
|
|
|
|
schedule_periodic_jobs([{"func": foo, "interval": 60}])
|
|
with patch('redash.schedule.rq_scheduler.schedule') as schedule:
|
|
schedule_periodic_jobs([{"func": foo, "interval": 60}])
|
|
schedule.assert_not_called()
|
|
|
|
|
|
def test_reschedules_a_modified_job(self):
|
|
def foo():
|
|
pass
|
|
|
|
schedule_periodic_jobs([{"func": foo, "interval": 60}])
|
|
schedule_periodic_jobs([{"func": foo, "interval": 120}])
|
|
|
|
jobs = [job for job in rq_scheduler.get_jobs()]
|
|
|
|
self.assertEqual(len(jobs), 1)
|
|
self.assertTrue(jobs[0].func_name.endswith('foo'))
|
|
self.assertEqual(jobs[0].meta['interval'], 120)
|
|
|
|
def test_removes_jobs_that_are_no_longer_defined(self):
|
|
def foo():
|
|
pass
|
|
|
|
def bar():
|
|
pass
|
|
|
|
schedule_periodic_jobs([{"func": foo, "interval": 60}, {"func": bar, "interval": 90}])
|
|
schedule_periodic_jobs([{"func": foo, "interval": 60}])
|
|
|
|
jobs = [job for job in rq_scheduler.get_jobs()]
|
|
|
|
self.assertEqual(len(jobs), 1)
|
|
self.assertTrue(jobs[0].func_name.endswith('foo'))
|
|
self.assertEqual(jobs[0].meta['interval'], 60) |