NeoAnd/netbox
1
0
Fork 0
mirror of https://github.com/netbox-community/netbox.git synced 2025-08-08 08:38:16 -06:00
Code Issues Actions 15 Packages Projects Releases Wiki Activity

Test enqueue in combination with enqueue_once

Browse Source
This commit is contained in:
Alexander Haase 2024-07-25 18:01:02 +02:00
parent e65e87c7b7
commit 3fc3d37919
1 changed files with 17 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
netbox/utilities/tests/test_jobs.py
View File

@ -111,6 +111,23 @@ class EnqueueTest(BackgroundJobTestCase):
self.assertRaises(Job.DoesNotExist, job1.refresh_from_db) self.assertRaises(Job.DoesNotExist, job1.refresh_from_db)
self.assertEqual(TestBackgroundJob.get_jobs(instance).count(), 1) self.assertEqual(TestBackgroundJob.get_jobs(instance).count(), 1)
def test_enqueue_once_with_enqueue(self):
instance = Job()
job1 = TestBackgroundJob.enqueue_once(instance, schedule_at=self.get_schedule_at(2))
job2 = TestBackgroundJob.enqueue(instance, schedule_at=self.get_schedule_at())
self.assertNotEqual(job1, job2)
self.assertEqual(TestBackgroundJob.get_jobs(instance).count(), 2)
def test_enqueue_once_after_enqueue(self):
instance = Job()
job1 = TestBackgroundJob.enqueue(instance, schedule_at=self.get_schedule_at())
job2 = TestBackgroundJob.enqueue_once(instance, schedule_at=self.get_schedule_at(2))
self.assertNotEqual(job1, job2)
self.assertRaises(Job.DoesNotExist, job1.refresh_from_db)
self.assertEqual(TestBackgroundJob.get_jobs(instance).count(), 1)
def test_enqueue_system(self): def test_enqueue_system(self):
job = TestBackgroundJob.enqueue_once(schedule_at=self.get_schedule_at()) job = TestBackgroundJob.enqueue_once(schedule_at=self.get_schedule_at())