NeoAnd/netbox
1
0
Fork 0
mirror of https://github.com/netbox-community/netbox.git synced 2025-07-31 21:06:25 -06:00
Code Issues Actions 12 Packages Projects Releases Wiki Activity

Fix check for existing jobs

Browse Source 
If a job is to be enqueued once and no specific scheduled time is
specified, any scheduled time of existing jobs will be valid. Only if a
specific scheduled time is specified for 'enqueue_once()' can it be
evaluated.
This commit is contained in:
Alexander Haase 2024-10-07 00:40:30 +02:00 committed by Jeremy Stretch
parent 727de0fb59
commit 892c54f6b5
2 changed files with 10 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
netbox/netbox/jobs.py
View File

@ -127,7 +127,7 @@ class JobRunner(ABC):
if job: if job:
# If the job parameters haven't changed, don't schedule a new job and keep the current schedule. Otherwise, # If the job parameters haven't changed, don't schedule a new job and keep the current schedule. Otherwise,
# delete the existing job and schedule a new job instead. # delete the existing job and schedule a new job instead.
if (schedule_at and job.scheduled == schedule_at) and (job.interval == interval): if (not schedule_at or job.scheduled == schedule_at) and (job.interval == interval):
return job return job
job.delete() job.delete()

9
netbox/netbox/tests/test_jobs.py
View File

@ -90,6 +90,15 @@ class EnqueueTest(JobRunnerTestCase):
self.assertEqual(job1, job2) self.assertEqual(job1, job2)
self.assertEqual(TestJobRunner.get_jobs(instance).count(), 1) self.assertEqual(TestJobRunner.get_jobs(instance).count(), 1)
def test_enqueue_once_twice_same_no_schedule_at(self):
instance = Job()
schedule_at = self.get_schedule_at()
job1 = TestJobRunner.enqueue_once(instance, schedule_at=schedule_at)
job2 = TestJobRunner.enqueue_once(instance)
self.assertEqual(job1, job2)
self.assertEqual(TestJobRunner.get_jobs(instance).count(), 1)
def test_enqueue_once_twice_different_schedule_at(self): def test_enqueue_once_twice_different_schedule_at(self):
instance = Job() instance = Job()
job1 = TestJobRunner.enqueue_once(instance, schedule_at=self.get_schedule_at()) job1 = TestJobRunner.enqueue_once(instance, schedule_at=self.get_schedule_at())