wip

Author: cunla

scheduler/tests/test_old_models/__init__.py

@@ -3,9 +3,9 @@
 from scheduler import settings
 from scheduler.models import CronTask
 from scheduler.tools import create_worker
-from .test_old_models import BaseTestCases
-from .testtools import old_task_factory
-from ..queues import get_queue
+from scheduler.tests.test_old_models.test_old_models import BaseTestCases
+from scheduler.tests.testtools import old_task_factory
+from scheduler.queues import get_queue
 
 
 class TestCronTask(BaseTestCases.TestBaseTask):

scheduler/tests/test_cron_task.py renamed to scheduler/tests/test_old_models/test_cron_task.py

@@ -10,16 +10,11 @@
 
 from scheduler import settings
 from scheduler.models import BaseTask, TaskArg, TaskKwarg, ScheduledTask
+from scheduler.queues import get_queue
+from scheduler.tests import jobs
+from scheduler.tests.testtools import (
+    old_task_factory, taskarg_factory, _get_job_from_scheduled_registry, SchedulerBaseCase, _get_executions)
 from scheduler.tools import run_task, create_worker
-from . import jobs
-from .testtools import (
-    old_task_factory,
-    taskarg_factory,
-    _get_job_from_scheduled_registry,
-    SchedulerBaseCase,
-    _get_executions,
-)
-from ..queues import get_queue
 
 
 def assert_response_has_msg(response, message):

scheduler/tests/test_old_models.py renamed to scheduler/tests/test_old_models/test_old_models.py

@@ -0,0 +1,207 @@
+from datetime import timedelta
+
+from django.core.exceptions import ValidationError
+from django.test import override_settings
+from django.utils import timezone
+
+from scheduler import settings
+from scheduler.models import RepeatableTask
+from scheduler.tests.test_old_models.test_old_task_model import BaseTestCases
+
+from scheduler.tests.testtools import old_task_factory, _get_job_from_scheduled_registry
+
+
+class TestRepeatableTask(BaseTestCases.TestSchedulableTask):
+    TaskModelClass = RepeatableTask
+
+    def test_unschedulable_old_job(self):
+        job = old_task_factory(self.TaskModelClass, scheduled_time=timezone.now() - timedelta(hours=1), repeat=0)
+        self.assertFalse(job.is_scheduled())
+
+    def test_schedulable_old_job_repeat_none(self):
+        # If repeat is None, the job should be scheduled
+        job = old_task_factory(self.TaskModelClass, scheduled_time=timezone.now() - timedelta(hours=1), repeat=None)
+        self.assertTrue(job.is_scheduled())
+
+    def test_clean(self):
+        job = old_task_factory(self.TaskModelClass)
+        job.queue = list(settings.QUEUES)[0]
+        job.callable = "scheduler.tests.jobs.test_job"
+        job.interval = 1
+        job.result_ttl = -1
+        self.assertIsNone(job.clean())
+
+    def test_clean_seconds(self):
+        job = old_task_factory(self.TaskModelClass)
+        job.queue = list(settings.QUEUES)[0]
+        job.callable = "scheduler.tests.jobs.test_job"
+        job.interval = 60
+        job.result_ttl = -1
+        job.interval_unit = "seconds"
+        self.assertIsNone(job.clean())
+
+    @override_settings(
+        SCHEDULER_CONFIG={
+            "SCHEDULER_INTERVAL": 10,
+        }
+    )
+    def test_clean_too_frequent(self):
+        job = old_task_factory(self.TaskModelClass)
+        job.queue = list(settings.QUEUES)[0]
+        job.callable = "scheduler.tests.jobs.test_job"
+        job.interval = 2  # Smaller than 10
+        job.result_ttl = -1
+        job.interval_unit = "seconds"
+        with self.assertRaises(ValidationError):
+            job.clean_interval_unit()
+
+    def test_clean_not_multiple(self):
+        job = old_task_factory(self.TaskModelClass)
+        job.queue = list(settings.QUEUES)[0]
+        job.callable = "scheduler.tests.jobs.test_job"
+        job.interval = 121
+        job.interval_unit = "seconds"
+        with self.assertRaises(ValidationError):
+            job.clean_interval_unit()
+
+    def test_clean_short_result_ttl(self):
+        job = old_task_factory(self.TaskModelClass)
+        job.queue = list(settings.QUEUES)[0]
+        job.callable = "scheduler.tests.jobs.test_job"
+        job.interval = 1
+        job.repeat = 1
+        job.result_ttl = 3599
+        job.interval_unit = "hours"
+        job.repeat = 42
+        with self.assertRaises(ValidationError):
+            job.clean_result_ttl()
+
+    def test_clean_indefinite_result_ttl(self):
+        job = old_task_factory(self.TaskModelClass)
+        job.queue = list(settings.QUEUES)[0]
+        job.callable = "scheduler.tests.jobs.test_job"
+        job.interval = 1
+        job.result_ttl = -1
+        job.interval_unit = "hours"
+        job.clean_result_ttl()
+
+    def test_clean_undefined_result_ttl(self):
+        job = old_task_factory(self.TaskModelClass)
+        job.queue = list(settings.QUEUES)[0]
+        job.callable = "scheduler.tests.jobs.test_job"
+        job.interval = 1
+        job.interval_unit = "hours"
+        job.clean_result_ttl()
+
+    def test_interval_seconds_weeks(self):
+        job = old_task_factory(self.TaskModelClass, interval=2, interval_unit="weeks")
+        self.assertEqual(1209600.0, job.interval_seconds())
+
+    def test_interval_seconds_days(self):
+        job = old_task_factory(self.TaskModelClass, interval=2, interval_unit="days")
+        self.assertEqual(172800.0, job.interval_seconds())
+
+    def test_interval_seconds_hours(self):
+        job = old_task_factory(self.TaskModelClass, interval=2, interval_unit="hours")
+        self.assertEqual(7200.0, job.interval_seconds())
+
+    def test_interval_seconds_minutes(self):
+        job = old_task_factory(self.TaskModelClass, interval=15, interval_unit="minutes")
+        self.assertEqual(900.0, job.interval_seconds())
+
+    def test_interval_seconds_seconds(self):
+        job = RepeatableTask(interval=15, interval_unit="seconds")
+        self.assertEqual(15.0, job.interval_seconds())
+
+    def test_interval_display(self):
+        job = old_task_factory(self.TaskModelClass, interval=15, interval_unit="minutes")
+        self.assertEqual(job.interval_display(), "15 minutes")
+
+    def test_result_interval(self):
+        job = old_task_factory(
+            self.TaskModelClass,
+        )
+        entry = _get_job_from_scheduled_registry(job)
+        self.assertEqual(entry.meta["interval"], 3600)
+
+    def test_repeat(self):
+        job = old_task_factory(self.TaskModelClass, repeat=10)
+        entry = _get_job_from_scheduled_registry(job)
+        self.assertEqual(entry.meta["repeat"], 10)
+
+    def test_repeat_old_job_exhausted(self):
+        base_time = timezone.now()
+        job = old_task_factory(self.TaskModelClass, scheduled_time=base_time - timedelta(hours=10), repeat=10)
+        self.assertEqual(job.is_scheduled(), False)
+
+    def test_repeat_old_job_last_iter(self):
+        base_time = timezone.now()
+        job = old_task_factory(self.TaskModelClass, scheduled_time=base_time - timedelta(hours=9, minutes=30), repeat=10)
+        self.assertEqual(job.repeat, 0)
+        self.assertEqual(job.is_scheduled(), True)
+
+    def test_repeat_old_job_remaining(self):
+        base_time = timezone.now()
+        job = old_task_factory(self.TaskModelClass, scheduled_time=base_time - timedelta(minutes=30), repeat=5)
+        self.assertEqual(job.repeat, 4)
+        self.assertEqual(job.scheduled_time, base_time + timedelta(minutes=30))
+        self.assertEqual(job.is_scheduled(), True)
+
+    def test_repeat_none_interval_2_min(self):
+        base_time = timezone.now()
+        job = old_task_factory(self.TaskModelClass, scheduled_time=base_time - timedelta(minutes=29), repeat=None)
+        job.interval = 120
+        job.interval_unit = "seconds"
+        job.schedule()
+        self.assertTrue(job.scheduled_time > base_time)
+        self.assertTrue(job.is_scheduled())
+
+    def test_check_rescheduled_after_execution(self):
+        task = old_task_factory(self.TaskModelClass, scheduled_time=timezone.now() + timedelta(seconds=1), repeat=10)
+        queue = task.rqueue
+        first_run_id = task.job_id
+        entry = queue.fetch_job(first_run_id)
+        queue.run_sync(entry)
+        task.refresh_from_db()
+        self.assertEqual(task.failed_runs, 0)
+        self.assertIsNone(task.last_failed_run)
+        self.assertEqual(task.successful_runs, 1)
+        self.assertIsNotNone(task.last_successful_run)
+        self.assertTrue(task.is_scheduled())
+        self.assertNotEqual(task.job_id, first_run_id)
+
+    def test_check_rescheduled_after_execution_failed_job(self):
+        task = old_task_factory(
+            self.TaskModelClass,
+            callable_name="scheduler.tests.jobs.failing_job",
+            scheduled_time=timezone.now() + timedelta(seconds=1),
+            repeat=10,
+        )
+        queue = task.rqueue
+        first_run_id = task.job_id
+        entry = queue.fetch_job(first_run_id)
+        queue.run_sync(entry)
+        task.refresh_from_db()
+        self.assertEqual(task.failed_runs, 1)
+        self.assertIsNotNone(task.last_failed_run)
+        self.assertEqual(task.successful_runs, 0)
+        self.assertIsNone(task.last_successful_run)
+        self.assertTrue(task.is_scheduled())
+        self.assertNotEqual(task.job_id, first_run_id)
+
+    def test_check_not_rescheduled_after_last_repeat(self):
+        task = old_task_factory(
+            self.TaskModelClass,
+            scheduled_time=timezone.now() + timedelta(seconds=1),
+            repeat=1,
+        )
+        queue = task.rqueue
+        first_run_id = task.job_id
+        entry = queue.fetch_job(first_run_id)
+        queue.run_sync(entry)
+        task.refresh_from_db()
+        self.assertEqual(task.failed_runs, 0)
+        self.assertIsNone(task.last_failed_run)
+        self.assertEqual(task.successful_runs, 1)
+        self.assertIsNotNone(task.last_successful_run)
+        self.assertNotEqual(task.job_id, first_run_id)

scheduler/tests/test_old_models/test_old_repeatable_task.py

@@ -11,15 +11,10 @@
 from scheduler import settings
 from scheduler.models import BaseTask, TaskArg, TaskKwarg, ScheduledTask
 from scheduler.tools import run_task, create_worker
-from . import jobs
-from .testtools import (
-    old_task_factory,
-    taskarg_factory,
-    _get_job_from_scheduled_registry,
-    SchedulerBaseCase,
-    _get_executions,
-)
-from ..queues import get_queue
+from scheduler.tests import jobs
+from scheduler.tests.testtools import old_task_factory, taskarg_factory, _get_job_from_scheduled_registry, \
+    SchedulerBaseCase, _get_executions
+from scheduler.queues import get_queue
 
 
 def assert_response_has_msg(response, message):
@@ -512,7 +507,7 @@ def test_admin_delete_selected(self):
             scheduled_jobs = queue.scheduled_job_registry.get_job_ids()
             self.assertNotIn(job_id, scheduled_jobs)
 
-    class TestSchedulableJob(TestBaseTask):
+    class TestSchedulableTask(TestBaseTask):
         # Currently ScheduledJob and RepeatableJob
         TaskModelClass = ScheduledTask
 
@@ -537,7 +532,7 @@ def test_result_ttl_passthrough(self):
             self.assertEqual(entry.result_ttl, 500)
 
 
-class TestScheduledJob(BaseTestCases.TestSchedulableJob):
+class TestScheduledJob(BaseTestCases.TestSchedulableTask):
     TaskModelClass = ScheduledTask
 
     def test_clean(self):

scheduler/tests/test_old_task_model.py renamed to scheduler/tests/test_old_models/test_old_task_model.py

@@ -0,0 +1,78 @@
+from django.core.exceptions import ValidationError
+
+from scheduler import settings
+from scheduler.queues import get_queue
+from scheduler.tests.test_task_types.test_task_model import BaseTestCases
+from scheduler.tests.testtools import task_factory
+from scheduler.tools import create_worker, TaskType
+
+
+class TestCronTask(BaseTestCases.TestBaseTask):
+    task_type = TaskType.CRON
+
+    def test_clean(self):
+        task = task_factory(self.task_type)
+        task.cron_string = "* * * * *"
+        task.queue = list(settings.QUEUES)[0]
+        task.callable = "scheduler.tests.jobs.test_job"
+        self.assertIsNone(task.clean())
+
+    def test_clean_cron_string_invalid(self):
+        task = task_factory(self.task_type)
+        task.cron_string = "not-a-cron-string"
+        task.queue = list(settings.QUEUES)[0]
+        task.callable = "scheduler.tests.jobs.test_job"
+        with self.assertRaises(ValidationError):
+            task.clean_cron_string()
+
+    def test_check_rescheduled_after_execution(self):
+        task = task_factory(self.task_type)
+        queue = task.rqueue
+        first_run_id = task.job_id
+        entry = queue.fetch_job(first_run_id)
+        queue.run_sync(entry)
+        task.refresh_from_db()
+        self.assertEqual(task.failed_runs, 0)
+        self.assertIsNone(task.last_failed_run)
+        self.assertEqual(task.successful_runs, 1)
+        self.assertIsNotNone(task.last_successful_run)
+        self.assertTrue(task.is_scheduled())
+        self.assertNotEqual(task.job_id, first_run_id)
+
+    def test_check_rescheduled_after_failed_execution(self):
+        task = task_factory(
+            self.task_type,
+            callable_name="scheduler.tests.jobs.scheduler.tests.jobs.test_job",
+        )
+        queue = task.rqueue
+        first_run_id = task.job_id
+        entry = queue.fetch_job(first_run_id)
+        queue.run_sync(entry)
+        task.refresh_from_db()
+        self.assertEqual(task.failed_runs, 1)
+        self.assertIsNotNone(task.last_failed_run)
+        self.assertEqual(task.successful_runs, 0)
+        self.assertIsNone(task.last_successful_run)
+        self.assertTrue(task.is_scheduled())
+        self.assertNotEqual(task.job_id, first_run_id)
+
+    def test_cron_task_enqueuing_jobs(self):
+        queue = get_queue()
+        prev_queued = len(queue.scheduled_job_registry)
+        prev_finished = len(queue.finished_job_registry)
+        task = task_factory(self.task_type, callable_name="scheduler.tests.jobs.enqueue_jobs")
+        self.assertEqual(prev_queued + 1, len(queue.scheduled_job_registry))
+        first_run_id = task.job_id
+        entry = queue.fetch_job(first_run_id)
+        queue.run_sync(entry)
+        self.assertEqual(20, len(queue))
+        self.assertEqual(prev_finished + 1, len(queue.finished_job_registry))
+        worker = create_worker(
+            "default",
+            fork_job_execution=False,
+        )
+        worker.work(burst=True)
+        self.assertEqual(prev_finished + 21, len(queue.finished_job_registry))
+        worker.refresh()
+        self.assertEqual(20, worker.successful_job_count)
+        self.assertEqual(0, worker.failed_job_count)