wip

Author: cunla

scheduler/management/commands/export.py

@@ -4,7 +4,6 @@
 from django.apps import apps
 from django.core.management.base import BaseCommand
 
-from scheduler.models import Task
 from scheduler.tools import MODEL_NAMES
 
 

scheduler/tests/test_old_models/test_old_repeatable_task.py

@@ -7,7 +7,6 @@
 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_task_job_execution_from_registry
 
 
@@ -136,7 +135,9 @@ def test_repeat_old_job_exhausted(self):
 
     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)
+        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)
 

scheduler/tests/test_task_types/test_task_model.py

@@ -478,7 +478,7 @@ def test_admin_delete_selected(self):
             res = self.client.post(url, data=data, follow=True)
             # assert
             self.assertEqual(200, res.status_code)
-            assert_response_has_msg(res, f"Successfully deleted 1 task.")
+            assert_response_has_msg(res, "Successfully deleted 1 task.")
             self.assertIsNone(Task.objects.filter(task_type=self.task_type).filter(id=task.id).first())
             scheduled_jobs = queue.scheduled_job_registry.get_job_ids()
             self.assertNotIn(job_id, scheduled_jobs)
@@ -506,4 +506,3 @@ def test_result_ttl_passthrough(self):
             job = task_factory(self.task_type, result_ttl=500)
             entry = _get_task_job_execution_from_registry(job)
             self.assertEqual(entry.result_ttl, 500)
-