Unit testing for task scheduling

InvenTree/InvenTree/tasks.py

@@ -23,6 +23,7 @@ def schedule_task(taskname, **kwargs):
 
     # If unspecified, repeat indefinitely
     repeats = kwargs.pop('repeats', -1)
+    kwargs['repeats'] = repeats
 
     try:
         from django_q.models import Schedule
@@ -31,15 +32,18 @@ def schedule_task(taskname, **kwargs):
         return
 
     try:
+        # If this task is already scheduled, don't schedule it again
+        # Instead, update the scheduling parameters
         if Schedule.objects.filter(func=taskname).exists():
-            logger.info(f"Scheduled task '{taskname}' already exists. (Skipping)")
+            logger.info(f"Scheduled task '{taskname}' already exists - updating!")
+
+            Schedule.objects.filter(func=taskname).update(**kwargs)
         else:
             logger.info(f"Creating scheduled task '{taskname}'")
 
             Schedule.objects.create(
                 name=taskname,
                 func=taskname,
-                repeats=repeats,
                 **kwargs
             )
     except (OperationalError, ProgrammingError):
@@ -82,8 +86,8 @@ def delete_successful_tasks():
 
     try:
         from django_q.models import Success
-        logger.warning("Could not perform 'delete_successful_tasks' - App registry not ready")
     except AppRegistryNotReady:
+        logger.warning("Could not perform 'delete_successful_tasks' - App registry not ready")
         return
 
     threshold = datetime.now() - timedelta(days=30)

InvenTree/InvenTree/test_tasks.py

@@ -0,0 +1,43 @@
+"""
+Unit tests for task management
+"""
+
+from django.test import TestCase
+from django_q.models import Schedule
+
+import InvenTree.tasks
+
+
+class ScheduledTaskTests(TestCase):
+    """
+    Unit tests for scheduled tasks
+    """
+
+    def get_tasks(self, name):
+
+        return Schedule.objects.filter(func=name)
+
+    def test_add_task(self):
+        """
+        Ensure that duplicate tasks cannot be added.
+        """
+
+        task = 'InvenTree.tasks.heartbeat'
+
+        self.assertEqual(self.get_tasks(task).count(), 0)
+
+        InvenTree.tasks.schedule_task(task, schedule_type=Schedule.MINUTES, minutes=10)
+
+        self.assertEqual(self.get_tasks(task).count(), 1)
+
+        t = Schedule.objects.get(func=task)
+
+        self.assertEqual(t.minutes, 10)
+
+        # Attempt to schedule the same task again
+        InvenTree.tasks.schedule_task(task, schedule_type=Schedule.MINUTES, minutes=5)
+        self.assertEqual(self.get_tasks(task).count(), 1)
+        
+        # But the 'minutes' should have been updated
+        t = Schedule.objects.get(func=task)
+        self.assertEqual(t.minutes, 5)