fix:creating new ONCE task without a scheduled time #270

Author: cunla

docs/changelog.md

@@ -5,6 +5,7 @@
 ### 🐛 Bug Fixes
 
 - Add type hint for `JOB_METHODS_LIST`
+- Fix issue creating new `ONCE` task without a scheduled time #270
 
 ### 🧰 Maintenance
 

scheduler/models/task.py

@@ -441,6 +441,12 @@ def clean(self):
         if self.task_type == TaskType.REPEATABLE:
             self.clean_interval_unit()
             self.clean_result_ttl()
+        if self.task_type == TaskType.ONCE and self.scheduled_time is None:
+            raise ValidationError({"scheduled_time": ValidationError(_("Scheduled time is required"), code="invalid")})
+        if self.task_type == TaskType.ONCE and self.scheduled_time < timezone.now():
+            raise ValidationError(
+                {"scheduled_time": ValidationError(_("Scheduled time must be in the future"), code="invalid")}
+            )
 
 
 def get_next_cron_time(cron_string: Optional[str]) -> Optional[timezone.datetime]:

scheduler/tests/test_task_types/test_once_task.py

@@ -1,5 +1,6 @@
-from datetime import timedelta
+from datetime import timedelta, datetime
 
+from django.core.exceptions import ValidationError
 from django.utils import timezone
 
 from scheduler import settings
@@ -18,6 +19,21 @@ def test_clean(self):
         job.callable = "scheduler.tests.jobs.test_job"
         self.assertIsNone(job.clean())
 
+    def test_create_without_date__fail(self):
+        task = task_factory(self.task_type, scheduled_time=None, instance_only=True)
+        self.assertIsNone(task.scheduled_time)
+        with self.assertRaises(Exception) as cm:
+            task.clean()
+        self.assertTrue(isinstance(cm.exception, ValidationError))
+        self.assertEqual(str(cm.exception), "{'scheduled_time': ['Scheduled time is required']}")
+
+    def test_create_with_date_in_the_past__fail(self):
+        task = task_factory(self.task_type, scheduled_time=datetime.now() - timedelta(days=1), instance_only=True)
+        with self.assertRaises(Exception) as cm:
+            task.clean()
+        self.assertTrue(isinstance(cm.exception, ValidationError))
+        self.assertEqual(str(cm.exception), "{'scheduled_time': ['Scheduled time must be in the future']}")
+
     def test_unschedulable_old_job(self):
         job = task_factory(self.task_type, scheduled_time=timezone.now() - timedelta(hours=1))
         self.assertFalse(job.is_scheduled())