Fixes #17685 : Fix job field validation

netbox/core/models/jobs.py

@@ -130,7 +130,7 @@ def clean(self):
         super().clean()
 
         # Validate the assigned object type
-        if self.object_type not in ObjectType.objects.with_feature('jobs'):
+        if self.object_type and self.object_type not in ObjectType.objects.with_feature('jobs'):
             raise ValidationError(
                 _("Jobs cannot be assigned to this object type ({type}).").format(type=self.object_type)
             )
@@ -223,7 +223,7 @@ def enqueue(cls, func, instance=None, name='', user=None, schedule_at=None, inte
         rq_queue_name = get_queue_for_model(object_type.model if object_type else None)
         queue = django_rq.get_queue(rq_queue_name)
         status = JobStatusChoices.STATUS_SCHEDULED if schedule_at else JobStatusChoices.STATUS_PENDING
-        job = Job.objects.create(
+        job = Job(
             object_type=object_type,
             object_id=object_id,
             name=name,
@@ -233,6 +233,8 @@ def enqueue(cls, func, instance=None, name='', user=None, schedule_at=None, inte
             user=user,
             job_id=uuid.uuid4()
         )
+        job.full_clean()
+        job.save()
 
         # Run the job immediately, rather than enqueuing it as a background task. Note that this is a synchronous
         # (blocking) operation, and execution will pause until the job completes.

netbox/netbox/tests/test_jobs.py

@@ -5,7 +5,7 @@
 from django_rq import get_queue
 
 from ..jobs import *
-from core.models import Job
+from core.models import DataSource, Job
 from core.choices import JobStatusChoices
 
 
@@ -68,21 +68,21 @@ class EnqueueTest(JobRunnerTestCase):
     """
 
     def test_enqueue(self):
-        instance = Job()
+        instance = DataSource()
         for i in range(1, 3):
             job = TestJobRunner.enqueue(instance, schedule_at=self.get_schedule_at())
 
             self.assertIsInstance(job, Job)
             self.assertEqual(TestJobRunner.get_jobs(instance).count(), i)
 
     def test_enqueue_once(self):
-        job = TestJobRunner.enqueue_once(instance=Job(), schedule_at=self.get_schedule_at())
+        job = TestJobRunner.enqueue_once(instance=DataSource(), schedule_at=self.get_schedule_at())
 
         self.assertIsInstance(job, Job)
         self.assertEqual(job.name, TestJobRunner.__name__)
 
     def test_enqueue_once_twice_same(self):
-        instance = Job()
+        instance = DataSource()
         schedule_at = self.get_schedule_at()
         job1 = TestJobRunner.enqueue_once(instance, schedule_at=schedule_at)
         job2 = TestJobRunner.enqueue_once(instance, schedule_at=schedule_at)
@@ -91,7 +91,7 @@ def test_enqueue_once_twice_same(self):
         self.assertEqual(TestJobRunner.get_jobs(instance).count(), 1)
 
     def test_enqueue_once_twice_different_schedule_at(self):
-        instance = Job()
+        instance = DataSource()
         job1 = TestJobRunner.enqueue_once(instance, schedule_at=self.get_schedule_at())
         job2 = TestJobRunner.enqueue_once(instance, schedule_at=self.get_schedule_at(2))
 
@@ -100,7 +100,7 @@ def test_enqueue_once_twice_different_schedule_at(self):
         self.assertEqual(TestJobRunner.get_jobs(instance).count(), 1)
 
     def test_enqueue_once_twice_different_interval(self):
-        instance = Job()
+        instance = DataSource()
         schedule_at = self.get_schedule_at()
         job1 = TestJobRunner.enqueue_once(instance, schedule_at=schedule_at)
         job2 = TestJobRunner.enqueue_once(instance, schedule_at=schedule_at, interval=60)
@@ -112,15 +112,15 @@ def test_enqueue_once_twice_different_interval(self):
         self.assertEqual(TestJobRunner.get_jobs(instance).count(), 1)
 
     def test_enqueue_once_with_enqueue(self):
-        instance = Job()
+        instance = DataSource()
         job1 = TestJobRunner.enqueue_once(instance, schedule_at=self.get_schedule_at(2))
         job2 = TestJobRunner.enqueue(instance, schedule_at=self.get_schedule_at())
 
         self.assertNotEqual(job1, job2)
         self.assertEqual(TestJobRunner.get_jobs(instance).count(), 2)
 
     def test_enqueue_once_after_enqueue(self):
-        instance = Job()
+        instance = DataSource()
         job1 = TestJobRunner.enqueue(instance, schedule_at=self.get_schedule_at())
         job2 = TestJobRunner.enqueue_once(instance, schedule_at=self.get_schedule_at(2))