Add validation to Job schedule to prevent scheduling jobs every second

dkron/__init__.py

@@ -1,4 +1,4 @@
-__version__ = '1.1.2'
+__version__ = '1.2.0'
 
 # set default_app_config when using django earlier than 3.2
 try:

dkron/admin.py

@@ -8,6 +8,7 @@
 from django.urls import reverse
 from django.utils.html import format_html
 from dkron import models, utils
+from dkron.forms import JobForm
 
 
 class JobAdminForm(forms.ModelForm):
@@ -53,6 +54,7 @@ class JobAdmin(admin.ModelAdmin):
     search_fields = ('name', 'schedule', 'command', 'description')
     list_filter = ('name', 'schedule', 'enabled', 'last_run_success', 'notify_on_error')
     actions = ['disable_jobs', 'enable_jobs']
+    form = JobForm
 
     def has_dashboard_permission(self, request):
         return request.user.has_perm('dkron.can_use_dashboard')

dkron/forms.py

@@ -0,0 +1,16 @@
+from django import forms
+from dkron.models import Job
+
+
+class JobForm(forms.ModelForm):
+    def clean_schedule(self):
+        data = self.cleaned_data["schedule"]
+        if data.startswith("*"):
+            raise forms.ValidationError(
+                "Job schedule cannot start with * as this will schedule a job to start every second and can have unintended consequences."
+            )
+        return data
+
+    class Meta:
+        model = Job
+        fields = "__all__"

testapp/testapp/urls.py

@@ -13,6 +13,7 @@
     1. Import the include() function: from django.urls import include, path
     2. Add a URL to urlpatterns:  path('blog/', include('blog.urls'))
 """
+
 from django.contrib import admin
 from django.urls import path, include
 

testapp/tests/test_generic.py

@@ -17,6 +17,7 @@
 from notifications import models as notify_models
 from dkron import admin, models, utils
 from dkron.apps import DkronConfig
+from dkron.forms import JobForm
 
 PROXY_VIEW = 'dkron:proxy'
 JOBS_URL = 'http://dkron/v1/jobs'
@@ -180,6 +181,20 @@ def test_sync_job(self, job_prefix=''):
             self.assertEqual(exc.exception.code, 500)
             self.assertEqual(exc.exception.message, 'Whatever')
 
+    def test_job_form(self, job_prefix=''):
+        form_data = {'name': 'job1', 'schedule': '* 0 1 * * *', 'command': 'echo test', "retries": 0}
+        form = JobForm(data=form_data)
+        self.assertEqual(form.is_valid(), False)
+        self.assertEquals(
+            form.errors['schedule'],
+            [
+                "Job schedule cannot start with * as this will schedule a job to start every second and can have unintended consequences."
+            ],
+        )
+        form_data = {'name': 'job1', 'schedule': '0 0 1 * * *', 'command': 'echo test', "retries": 0}
+        form = JobForm(data=form_data)
+        self.assertEqual(form.is_valid(), True)
+
     def test_delete_job(self, job_prefix=''):
         j = models.Job.objects.create(name='job1')
         with mock.patch('requests.delete') as mp: