fix: 修改代码层级问题 --story=120737215

env.py

@@ -151,9 +151,9 @@
 PERIODIC_TASK_REMINDER_NOTIFY_TYPE = json.loads(os.getenv("PERIODIC_TASK_REMINDER_NOTIFY_TYPE", '["email"]'))
 
 # 周期任务最短时间间隔，以分钟为单位，例如：30
-PERIODIC_TASK_SHORTEST_TIME = os.getenv("PERIODIC_TASK_SHORTEST_TIME", "")
+PERIODIC_TASK_SHORTEST_TIME = int(os.getenv("PERIODIC_TASK_SHORTEST_TIME", 0))
 # 周期任务迭代次数
-PERIODIC_TASK_ITERATION = os.getenv("PERIODIC_TASK_ITERATION", 10)
+PERIODIC_TASK_ITERATION = int(os.getenv("PERIODIC_TASK_ITERATION", 10))
 
 # bk_audit
 BK_AUDIT_ENDPOINT = os.getenv("BK_AUDIT_ENDPOINT", None)

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

@@ -22,8 +22,8 @@
 from rest_framework.validators import ValidationError
 
 import env
+from gcloud.utils.strings import inspect_time
 from gcloud.conf import settings
-from gcloud.core.apis.drf.exceptions import ValidationException
 from gcloud.constants import PROJECT
 from gcloud.core.apis.drf.serilaziers.project import ProjectSerializer
 from gcloud.core.models import Project, ProjectConfig
@@ -154,9 +154,23 @@ def check_cron_params(cron, project):
         raise ValidationError("周期任务时间格式过长")
 
 
-class CreatePeriodicTaskSerializer(serializers.ModelSerializer):
+class CronFieldSerializer(serializers.Serializer):
+    cron = serializers.DictField(write_only=True, help_text="周期", required=False)
+
+    def validate_inspect_cron(self, value):
+        minute = value.get("minute", "*")
+        hour = value.get("hour", "*")
+        day_of_month = value.get("day_of_month", "*")
+        month = value.get("month", "*")
+        day_of_week = value.get("day_of_week", "*")
+
+        cron_expression = f"{minute} {hour} {day_of_month} {month} {day_of_week}"
+
+        return cron_expression
+
+
+class CreatePeriodicTaskSerializer(CronFieldSerializer, serializers.ModelSerializer):
     project = serializers.IntegerField(write_only=True)
-    cron = serializers.DictField(write_only=True)
     template_source = serializers.CharField(required=False, default=PROJECT)
     pipeline_tree = ReadWriteSerializerMethodField()
     template_scheme_ids = ReadWriteSerializerMethodField()
@@ -195,51 +209,36 @@ def validate_project(self, value):
 
     def validate(self, attrs):
         check_cron_params(attrs.get("cron"), attrs.get("project"))
-        self.inspect_time(self.context["request"], attrs.get("cron"))
-        return attrs
-
-    def inspect_time(self, request, cron):
-        if settings.PERIODIC_TASK_SHORTEST_TIME:
-            result = PeriodicTask().inspect_time(
-                is_superuser=request.user.is_superuser,
-                cron=cron,
-                shortest_time=int(settings.PERIODIC_TASK_SHORTEST_TIME),
-                item_num=int(settings.PERIODIC_TASK_ITERATION),
-            )
+        if settings.PERIODIC_TASK_SHORTEST_TIME and not self.context["request"].user.is_superuser:
+            cron_str = super().validate_inspect_cron(attrs.get("cron"))
+            result = inspect_time(cron_str, settings.PERIODIC_TASK_SHORTEST_TIME, settings.PERIODIC_TASK_ITERATION)
             if not result:
-                raise ValidationException(
+                raise serializers.ValidationError(
                     "The interval between tasks should be at least {} minutes".format(
                         settings.PERIODIC_TASK_SHORTEST_TIME
                     )
                 )
+        return attrs
 
     class Meta:
         model = PeriodicTask
         fields = ["project", "cron", "name", "template_id", "pipeline_tree", "template_source", "template_scheme_ids"]
 
 
-class PatchUpdatePeriodicTaskSerializer(serializers.Serializer):
-    cron = serializers.DictField(help_text="周期", required=False)
+class PatchUpdatePeriodicTaskSerializer(CronFieldSerializer, serializers.Serializer):
     project = serializers.IntegerField(help_text="项目ID", required=False)
     constants = serializers.DictField(help_text="执行参数", required=False)
     name = serializers.CharField(help_text="任务名", required=False)
 
     def validate(self, attrs):
         check_cron_params(attrs.get("cron"), attrs.get("project"))
-        self.inspect_time(self.context["request"], attrs.get("cron"))
-        return attrs
-
-    def inspect_time(self, request, cron):
-        if settings.PERIODIC_TASK_SHORTEST_TIME:
-            result = PeriodicTask().inspect_time(
-                is_superuser=request.user.is_superuser,
-                cron=cron,
-                shortest_time=int(settings.PERIODIC_TASK_SHORTEST_TIME),
-                item_num=int(settings.PERIODIC_TASK_ITERATION),
-            )
+        if settings.PERIODIC_TASK_SHORTEST_TIME and not self.context["request"].user.is_superuser:
+            cron_str = super().validate_inspect_cron(attrs.get("cron"))
+            result = inspect_time(cron_str, settings.PERIODIC_TASK_SHORTEST_TIME, settings.PERIODIC_TASK_ITERATION)
             if not result:
-                raise ValidationException(
+                raise serializers.ValidationError(
                     "The interval between tasks should be at least {} minutes".format(
                         settings.PERIODIC_TASK_SHORTEST_TIME
                     )
                 )
+        return attrs

gcloud/periodictask/models.py

@@ -14,8 +14,6 @@
 import logging
 
 import ujson as json
-from croniter import croniter
-from datetime import datetime, timedelta
 from django.conf import settings
 from django.db import models, transaction
 from django.utils.translation import ugettext_lazy as _
@@ -256,38 +254,6 @@ def delete(self, using=None):
         super(PeriodicTask, self).delete(using)
         PeriodicTaskHistory.objects.filter(task=self).delete()
 
-    def inspect_time(self, is_superuser, cron, shortest_time, item_num):
-        """检查cron时间间隔是否符合要求
-        :param is_superuser: 是否是管理员
-        :type is_superuser bool
-        :param cron: 定时任务配置
-        :type cron dict
-        :param shortest_time: 最短时间间隔
-        :type shortest_time int
-        :param item_num: 迭代次数
-        :type item_num int
-        """
-        if is_superuser:
-            return True
-
-        minute = cron.get("minute")
-        hour = cron.get("hour")
-        day_of_month = cron.get("day_of_month")
-        month_of_year = cron.get("month_of_year")
-        day_of_week = cron.get("day_of_week")
-        # 拼接cron表达式
-        cron_expression = f"{minute} {hour} {day_of_month} {month_of_year} {day_of_week}"
-
-        schedule_iter = croniter(cron_expression)
-        # 计算指定次数内的最短时间间隔
-        next_times = [schedule_iter.get_next(datetime) for _ in range(item_num)]
-        min_interval = min((next_times[i] - next_times[i - 1] for i in range(1, len(next_times))))
-
-        if min_interval < timedelta(minutes=shortest_time):
-            return False
-
-        return True
-
     def modify_cron(self, cron, timezone):
         if self.task.enabled is False:
             self.task.modify_cron(cron, timezone)

gcloud/tests/periodictask/models/test_periodic_task.py

@@ -237,25 +237,6 @@ def test_delete(self):
             PipelinePeriodicTask.DoesNotExist, PipelinePeriodicTask.objects.get, id=pipeline_periodic_task_id
         )
 
-    def test_inspect_time(self):
-        shortest_time = 30
-        item_num = 10
-        cron_1 = {"day_of_month": "*", "day_of_week": "*", "hour": "*", "minute": "*", "month_of_year": "*"}
-        self.periodic_task_1 = self.task.inspect_time(
-            is_superuser=False, cron=cron_1, shortest_time=shortest_time, item_num=item_num
-        )
-        self.assertFalse(self.periodic_task_1)
-        cron_2 = {"day_of_month": "*", "day_of_week": "*", "hour": "*", "minute": "*/20", "month_of_year": "*"}
-        self.periodic_task_2 = self.task.inspect_time(
-            is_superuser=False, cron=cron_2, shortest_time=shortest_time, item_num=item_num
-        )
-        self.assertFalse(self.periodic_task_2)
-        cron_3 = {"day_of_month": "*", "day_of_week": "*", "hour": "*/2", "minute": "*/40", "month_of_year": "*"}
-        self.periodic_task_3 = self.task.inspect_time(
-            is_superuser=False, cron=cron_3, shortest_time=shortest_time, item_num=item_num
-        )
-        self.assertTrue(self.periodic_task_3)
-
     def test_modify_constants(self):
         expect_constants = copy.deepcopy(self.task.task.execution_data["constants"])
         expect_constants["key_1"]["value"] = "val_3"

gcloud/tests/utils/strings/test_inspect_time.py

@@ -0,0 +1,41 @@
+# -*- coding: utf-8 -*-
+"""
+Tencent is pleased to support the open source community by making 蓝鲸智云PaaS平台社区版 (BlueKing PaaS Community
+Edition) available.
+Copyright (C) 2017-2020 THL A29 Limited, a Tencent company. All rights reserved.
+Licensed under the MIT License (the "License"); you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+http://opensource.org/licenses/MIT
+Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+specific language governing permissions and limitations under the License.
+"""
+from django.test import TestCase
+
+from gcloud.utils.strings import inspect_time
+
+
+class InspectTimeTestCase(TestCase):
+    def test_inspect_time(self):
+        cron = "* * * * *"
+        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 * * * *"
+        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 * * *"
+        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 * * * *"
+        shortest_time = 30
+        iter_count = 100
+        self.assertFalse(inspect_time(cron=cron, shortest_time=shortest_time, iter_count=iter_count))

gcloud/utils/strings.py

@@ -13,7 +13,8 @@
 
 import re
 import logging
-
+from croniter import croniter
+from datetime import datetime, timedelta
 from gcloud.constants import TEMPLATE_NODE_NAME_MAX_LENGTH, AE
 
 logger = logging.getLogger("root")
@@ -116,3 +117,24 @@ def django_celery_beat_cron_time_format_fit(cron_str):
     cron_config = {time_format: cron_time for time_format, cron_time in zip(time_formats, cron_times)}
     result_cron_list = [cron_config[unit] for unit in unit_order] + ["({})".format("/".join(unit_order))] + time_zone
     return " ".join(result_cron_list).strip()
+
+
+def inspect_time(cron, shortest_time, iter_count):
+    """检查定时任务时间间隔是否符合要求
+    :param cron: 定时任务配置
+    :type cron str
+    :param shortest_time: 最短时间间隔，以分钟为单位，例如 30
+    :type shortest_time int
+    :param iter_count: 迭代次数
+    :type iter_count int
+    """
+
+    schedule_iter = croniter(cron)
+    # 计算指定次数内的最短时间间隔
+    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))))
+
+    if min_interval < timedelta(minutes=shortest_time):
+        return False
+
+    return True