fix: 周期任务接口添加验证 --story=121224121

gcloud/apigw/views/modify_cron_for_periodic_task.py

@@ -18,6 +18,8 @@
 
 from blueapps.account.decorators import login_exempt
 from gcloud import err_code
+from gcloud.utils.strings import inspect_time
+from gcloud.conf import settings
 from gcloud.apigw.decorators import mark_request_whether_is_trust, return_json_response
 from gcloud.apigw.decorators import project_inject
 from gcloud.periodictask.models import PeriodicTask
@@ -44,6 +46,12 @@ def modify_cron_for_periodic_task(request, task_id, project_id):
     cron = params.get("cron", {})
     tz = project.time_zone
 
+    if not inspect_time(cron, settings.PERIODIC_TASK_SHORTEST_TIME, settings.PERIODIC_TASK_ITERATION):
+        return {
+            "result": False,
+            "message": "The interval must be at least {} minutes".format(settings.PERIODIC_TASK_SHORTEST_TIME),
+            "code": err_code.REQUEST_PARAM_INVALID.code,
+        }
     try:
         task = PeriodicTask.objects.get(id=task_id, project_id=project.id)
     except PeriodicTask.DoesNotExist:

gcloud/contrib/template_market/serializers.py

@@ -48,6 +48,7 @@ class FileUploadAddrSerializer(serializers.Serializer):
 
 class TemplateSharedRecordSerializer(serializers.Serializer):
     project_code = serializers.CharField(required=True, max_length=32, help_text="项目id")
+    project_name = serializers.CharField(required=True, help_text="项目名称")
     templates = serializers.ListField(required=True, help_text="关联的模板列表")
     name = serializers.CharField(required=True, help_text="共享名称")
     category = serializers.CharField(required=True, help_text="共享分类")

gcloud/core/apis/drf/serilaziers/periodic_task.py

@@ -158,22 +158,15 @@ class CronFieldSerializer(serializers.Serializer):
     cron = serializers.DictField(write_only=True)
 
     def inspect_cron(self, cron):
-        minute = cron.get("minute", "*")
-        hour = cron.get("hour", "*")
-        day_of_month = cron.get("day_of_month", "*")
-        month = cron.get("month", "*")
-        day_of_week = cron.get("day_of_week", "*")
-
-        cron_expression = f"{minute} {hour} {day_of_month} {month} {day_of_week}"
-
-        result = inspect_time(cron_expression, settings.PERIODIC_TASK_SHORTEST_TIME, settings.PERIODIC_TASK_ITERATION)
+        result = inspect_time(cron, settings.PERIODIC_TASK_SHORTEST_TIME, settings.PERIODIC_TASK_ITERATION)
         if not result:
             raise serializers.ValidationError(
                 "The interval between tasks should be at least {} minutes".format(settings.PERIODIC_TASK_SHORTEST_TIME)
             )
 
 
 class CreatePeriodicTaskSerializer(CronFieldSerializer, serializers.ModelSerializer):
+    id = serializers.IntegerField(required=False)
     project = serializers.IntegerField(write_only=True)
     template_source = serializers.CharField(required=False, default=PROJECT)
     pipeline_tree = ReadWriteSerializerMethodField()
@@ -220,6 +213,7 @@ def validate(self, attrs):
     class Meta:
         model = PeriodicTask
         fields = [
+            "id",
             "project",
             "cron",
             "name",

gcloud/tests/utils/strings/test_inspect_time.py

@@ -17,25 +17,25 @@
 
 class InspectTimeTestCase(TestCase):
     def test_inspect_time(self):
-        cron = "* * * * *"
+        cron = {"minute": "*", "hour": "*", "day_of_month": "*", "month": "*", "day_of_week": "*"}
         shortest_time = 30
         iter_count = 10
         self.assertFalse(inspect_time(cron=cron, shortest_time=shortest_time, iter_count=iter_count))
 
     def test_fail_inspect_time(self):
-        cron = "*/15 * * * *"
+        cron = {"minute": "*/15", "hour": "*", "day_of_month": "*", "month": "*", "day_of_week": "*"}
         shortest_time = 30
         iter_count = 10
         self.assertFalse(inspect_time(cron=cron, shortest_time=shortest_time, iter_count=iter_count))
 
     def test_success_inspect_time(self):
-        cron = "15 2 * * *"
+        cron = {"minute": "15", "hour": "2", "day_of_month": "*", "month": "*", "day_of_week": "*"}
         shortest_time = 30
         iter_count = 10
         self.assertTrue(inspect_time(cron=cron, shortest_time=shortest_time, iter_count=iter_count))
 
     def test_iter_count_inspect_time(self):
-        cron = "*/15 * * * *"
+        cron = {"minute": "*/15", "hour": "*", "day_of_month": "*", "month": "*", "day_of_week": "*"}
         shortest_time = 30
         iter_count = 100
         self.assertFalse(inspect_time(cron=cron, shortest_time=shortest_time, iter_count=iter_count))

gcloud/utils/strings.py

@@ -122,14 +122,21 @@ def django_celery_beat_cron_time_format_fit(cron_str):
 def inspect_time(cron, shortest_time, iter_count):
     """检查定时任务时间间隔是否符合要求
     :param cron: 定时任务配置
-    :type cron str
+    :type cron dict
     :param shortest_time: 最短时间间隔，以分钟为单位，例如 30
     :type shortest_time int
     :param iter_count: 迭代次数
     :type iter_count int
     """
+    minute = cron.get("minute", "*")
+    hour = cron.get("hour", "*")
+    day_of_month = cron.get("day_of_month", "*")
+    month = cron.get("month", "*")
+    day_of_week = cron.get("day_of_week", "*")
 
-    schedule_iter = croniter(cron)
+    cron_expression = f"{minute} {hour} {day_of_month} {month} {day_of_week}"
+
+    schedule_iter = croniter(cron_expression)
     # 计算指定次数内的最短时间间隔
     next_times = [schedule_iter.get_next(datetime) for _ in range(iter_count)]
     min_interval = min((next_times[i] - next_times[i - 1] for i in range(1, len(next_times))))