Merge pull request #349 from Jerome-Celle/add_test_cron_manager

Update cron manager logic and add test

cron_manager/models.py

@@ -42,38 +42,35 @@ def __str__(self):
 
     @property
     def last_execution(self):
-        return self.executions.order_by('-executed_at').first()
+        return self.executions.filter(success=True)\
+            .order_by('-executed_at').first()
 
-    @property
-    def can_be_execute(self):
-
-        if self.active:
+    def next_execution_datetime(self):
+        if self.last_execution:
             if self.execution_interval:
-                now_less_intervals = timezone.now() - timezone.timedelta(
-                    milliseconds=self.execution_interval)
+                next_execution_datetime = False
             else:
-                now_less_intervals = timezone.now()
+                next_execution_datetime = \
+                    self.last_execution.executed_at + self.execution_interval
+        else:
+            next_execution_datetime = self.execution_datetime
+        return next_execution_datetime
 
-            last_execution = self.last_execution
+    @property
+    def can_be_execute(self):
 
-            if last_execution:
-                return now_less_intervals > last_execution.executed_at
+        if self.active:
+            next_execution_datetime = self.next_execution_datetime()
+            if next_execution_datetime:
+                return timezone.now() >= self.next_execution_datetime()
             else:
-                return now_less_intervals > self.execution_datetime
+                return False
         else:
             return False
 
     def execute(self):
 
-        last_execution = self.last_execution
-        if last_execution:
-            if self.execution_interval:
-                executed_at = last_execution.executed_at - timezone.timedelta(
-                    milliseconds=self.execution_interval)
-            else:
-                executed_at = last_execution.executed_at
-        else:
-            executed_at = self.execution_datetime
+        executed_at = self.next_execution_datetime()
 
         execution = Execution.objects.create(
             task=self,

cron_manager/tests/test_cron_manager_model.py

@@ -0,0 +1,171 @@
+import json
+from datetime import datetime
+from unittest import mock
+
+import responses
+from django.test import TestCase
+from django.utils import timezone
+
+from cron_manager.models import Task
+from log_management.models import Log
+
+
+class CronManagerTests(TestCase):
+
+    def setUp(self) -> None:
+
+        self.url_test = 'http://local/retreat/wait_queue_places/15/notify'
+
+        self.task = Task.objects.create(
+            url=self.url_test,
+            description='test_description_task',
+            execution_datetime=timezone.now(),
+            execution_interval=86400000
+        )
+
+        self.task_without_interval = Task.objects.create(
+            url=self.url_test,
+            description='test_description_task',
+            execution_datetime=timezone.now(),
+        )
+
+    def test_can_execute_after(self):
+        date_after_execution_datetime = \
+            self.task.execution_datetime + timezone.timedelta(
+                minutes=10)
+        with mock.patch(
+                'django.utils.timezone.now',
+                return_value=date_after_execution_datetime):
+            self.assertTrue(self.task.can_be_execute)
+
+    def test_can_execute_before(self):
+        date_before_execution_datetime = \
+            self.task.execution_datetime - timezone.timedelta(
+                minutes=10
+            )
+        with mock.patch(
+                'django.utils.timezone.now',
+                return_value=date_before_execution_datetime):
+            self.assertFalse(self.task.can_be_execute)
+
+    def test_can_execute_after_without_execution_interval(self):
+        date_after_execution_datetime = \
+            self.task_without_interval.execution_datetime + \
+            timezone.timedelta(
+                minutes=10)
+        with mock.patch(
+                'django.utils.timezone.now',
+                return_value=date_after_execution_datetime):
+
+            self.assertTrue(
+                self.task_without_interval.can_be_execute)
+
+    @responses.activate
+    def test_execution(self):
+        responses.add(
+            responses.GET,
+            self.url_test,
+            json={
+                'stop': False
+            },
+            status=200
+        )
+        date_after_execution_datetime = \
+            self.task.execution_datetime + timezone.timedelta(
+                minutes=10)
+        with mock.patch(
+                'django.utils.timezone.now',
+                return_value=date_after_execution_datetime):
+            self.task.execute()
+
+            self.assertEqual(
+                self.task.executions.count(),
+                1
+            )
+
+            self.assertTrue(self.task.active)
+
+    @responses.activate
+    def test_execution_one_time(self):
+        responses.add(
+            responses.GET,
+            self.url_test,
+            json={
+                'stop': False
+            },
+            status=200
+        )
+        date_after_execution_datetime = \
+            self.task_without_interval.execution_datetime\
+            + timezone.timedelta(
+                minutes=10)
+        with mock.patch(
+                'django.utils.timezone.now',
+                return_value=date_after_execution_datetime):
+            self.task_without_interval.execute()
+
+            self.assertEqual(
+                self.task_without_interval.executions.count(),
+                1
+            )
+
+            self.assertFalse(self.task_without_interval.active)
+            self.assertFalse(
+                self.task_without_interval.can_be_execute)
+
+    @responses.activate
+    def test_execution_one_time_failed(self):
+        responses.add(
+            responses.GET,
+            self.url_test,
+            json={
+                'stop': False
+            },
+            status=300
+        )
+        date_after_execution_datetime = \
+            self.task_without_interval.execution_datetime\
+            + timezone.timedelta(
+                minutes=10)
+        with mock.patch(
+                'django.utils.timezone.now',
+                return_value=date_after_execution_datetime):
+            self.task_without_interval.execute()
+
+            self.assertEqual(
+                self.task_without_interval.executions.count(),
+                1
+            )
+
+            self.assertTrue(self.task_without_interval.active)
+            self.assertTrue(
+                self.task_without_interval.can_be_execute)
+
+            execution = self.task_without_interval.executions.first()
+            self.assertFalse(execution.success)
+
+    @responses.activate
+    def test_execution_fail(self):
+        responses.add(
+            responses.GET,
+            self.url_test,
+            status=300
+        )
+        date_after_execution_datetime = \
+            self.task.execution_datetime\
+            + timezone.timedelta(
+                minutes=10)
+        with mock.patch(
+                'django.utils.timezone.now',
+                return_value=date_after_execution_datetime):
+            self.task.execute()
+
+            self.assertEqual(
+                self.task.executions.count(),
+                1
+            )
+
+            self.assertTrue(self.task.active)
+
+            execution = self.task.executions.first()
+            self.assertFalse(execution.success)