dev to main

engine/apps/api/serializers/escalation_policy.py

@@ -1,4 +1,3 @@
-import time
 from datetime import timedelta
 
 from rest_framework import serializers
@@ -9,6 +8,7 @@
 from apps.user_management.models import Team, User
 from apps.webhooks.models import Webhook
 from common.api_helpers.custom_fields import (
+    DurationSecondsField,
     OrganizationFilteredPrimaryKeyRelatedField,
     UsersFilteredByOrganizationField,
 )
@@ -47,15 +47,17 @@ class EscalationPolicySerializer(EagerLoadingMixin, serializers.ModelSerializer)
         queryset=User.objects,
         required=False,
     )
-    wait_delay = serializers.ChoiceField(
+    wait_delay = DurationSecondsField(
         required=False,
-        choices=EscalationPolicy.WEB_DURATION_CHOICES,
         allow_null=True,
+        min_value=timedelta(minutes=1),
+        max_value=timedelta(hours=24),
     )
-    num_minutes_in_window = serializers.ChoiceField(
+    num_minutes_in_window = serializers.IntegerField(
         required=False,
-        choices=EscalationPolicy.WEB_DURATION_CHOICES_MINUTES,
         allow_null=True,
+        min_value=1,  # 1 minute
+        max_value=24 * 60,  # 24 hours
     )
     notify_schedule = OrganizationFilteredPrimaryKeyRelatedField(
         queryset=OnCallSchedule.objects,
@@ -151,29 +153,12 @@ def validate_step(self, step_type):
             raise serializers.ValidationError("Invalid escalation step type: step is Slack-specific")
         return step_type
 
-    def to_internal_value(self, data):
-        data = self._wait_delay_to_internal_value(data)
-        return super().to_internal_value(data)
-
     def to_representation(self, instance):
         step = instance.step
         result = super().to_representation(instance)
         result = EscalationPolicySerializer._get_important_field(step, result)
         return result
 
-    @staticmethod
-    def _wait_delay_to_internal_value(data):
-        if data.get(WAIT_DELAY, None):
-            try:
-                time.strptime(data[WAIT_DELAY], "%H:%M:%S")
-            except ValueError:
-                try:
-                    data[WAIT_DELAY] = str(timedelta(seconds=float(data[WAIT_DELAY])))
-                except ValueError:
-                    raise serializers.ValidationError("Invalid wait delay format")
-
-        return data
-
     @staticmethod
     def _get_important_field(step, result):
         if step in {*EscalationPolicy.DEFAULT_STEPS_SET, *EscalationPolicy.STEPS_WITH_NO_IMPORTANT_VERSION_SET}:

engine/apps/api/serializers/user_notification_policy.py

@@ -1,12 +1,11 @@
-import time
 from datetime import timedelta
 
 from rest_framework import serializers
 
 from apps.base.models import UserNotificationPolicy
 from apps.base.models.user_notification_policy import NotificationChannelAPIOptions
 from apps.user_management.models import User
-from common.api_helpers.custom_fields import OrganizationFilteredPrimaryKeyRelatedField
+from common.api_helpers.custom_fields import DurationSecondsField, OrganizationFilteredPrimaryKeyRelatedField
 from common.api_helpers.exceptions import Forbidden
 from common.api_helpers.mixins import EagerLoadingMixin
 
@@ -26,6 +25,12 @@ class UserNotificationPolicyBaseSerializer(EagerLoadingMixin, serializers.ModelS
         default=UserNotificationPolicy.Step.NOTIFY,
         choices=UserNotificationPolicy.Step.choices,
     )
+    wait_delay = DurationSecondsField(
+        required=False,
+        allow_null=True,
+        min_value=timedelta(minutes=1),
+        max_value=timedelta(hours=24),
+    )
 
     SELECT_RELATED = [
         "user",
@@ -41,14 +46,6 @@ class Meta:
         read_only_fields = ["order"]
 
     def to_internal_value(self, data):
-        if data.get("wait_delay", None):
-            try:
-                time.strptime(data["wait_delay"], "%H:%M:%S")
-            except ValueError:
-                try:
-                    data["wait_delay"] = str(timedelta(seconds=float(data["wait_delay"])))
-                except ValueError:
-                    raise serializers.ValidationError("Invalid wait delay format")
         data = self._notify_by_to_internal_value(data)
         return super().to_internal_value(data)
 

engine/apps/api/tests/test_escalation_policy.py

@@ -55,6 +55,28 @@ def test_create_escalation_policy(escalation_policy_internal_api_setup, make_use
     assert EscalationPolicy.objects.get(public_primary_key=response.data["id"]).order == max_order + 1
 
 
+@pytest.mark.django_db
+@pytest.mark.parametrize("wait_delay", (timedelta(seconds=59), timedelta(hours=24, seconds=1)))
+def test_create_escalation_policy_wait_delay_invalid(
+    escalation_policy_internal_api_setup, make_user_auth_headers, wait_delay
+):
+    token, escalation_chain, _, user, _ = escalation_policy_internal_api_setup
+    client = APIClient()
+    url = reverse("api-internal:escalation_policy-list")
+
+    data = {
+        "step": EscalationPolicy.STEP_WAIT,
+        "wait_delay": str(wait_delay.total_seconds()),
+        "escalation_chain": escalation_chain.public_primary_key,
+        "notify_to_users_queue": [],
+        "from_time": None,
+        "to_time": None,
+    }
+
+    response = client.post(url, data, format="json", **make_user_auth_headers(user, token))
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
+
+
 @pytest.mark.django_db
 def test_create_escalation_policy_webhook(
     escalation_policy_internal_api_setup, make_custom_webhook, make_user_auth_headers

engine/apps/api/tests/test_user_notification_policy.py

@@ -1,4 +1,5 @@
 import json
+from datetime import timedelta
 from unittest.mock import patch
 
 import pytest
@@ -67,6 +68,26 @@ def test_create_notification_policy(user_notification_policy_internal_api_setup,
     assert response.status_code == status.HTTP_201_CREATED
 
 
+@pytest.mark.django_db
+@pytest.mark.parametrize("wait_delay", (timedelta(seconds=59), timedelta(hours=24, seconds=1)))
+def test_create_notification_policy_wait_delay_invalid(
+    user_notification_policy_internal_api_setup, make_user_auth_headers, wait_delay
+):
+    token, _, users = user_notification_policy_internal_api_setup
+    admin, _ = users
+    client = APIClient()
+    url = reverse("api-internal:notification_policy-list")
+
+    data = {
+        "step": UserNotificationPolicy.Step.WAIT,
+        "wait_delay": str(wait_delay.total_seconds()),
+        "important": False,
+        "user": admin.public_primary_key,
+    }
+    response = client.post(url, data, format="json", **make_user_auth_headers(admin, token))
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
+
+
 @pytest.mark.django_db
 def test_admin_can_create_notification_policy_for_user(
     user_notification_policy_internal_api_setup, make_user_auth_headers

engine/apps/api/views/alert_group.py

@@ -643,6 +643,8 @@ def silence(self, request, pk=None):
     )
     @action(methods=["get"], detail=False)
     def silence_options(self, request):
+        # TODO: DEPRECATED, REMOVE IN A FUTURE RELEASE
+
         """
         Retrieve a list of valid silence options
         """

engine/apps/api/views/escalation_policy.py

@@ -138,13 +138,15 @@ def escalation_options(self, request):
 
     @action(detail=False, methods=["get"])
     def delay_options(self, request):
+        # TODO: DEPRECATED, REMOVE IN A FUTURE RELEASE
         choices = []
         for item in EscalationPolicy.WEB_DURATION_CHOICES:
             choices.append({"value": str(item[0]), "sec_value": item[0], "display_name": item[1]})
         return Response(choices)
 
     @action(detail=False, methods=["get"])
     def num_minutes_in_window_options(self, request):
+        # TODO: DEPRECATED, REMOVE IN A FUTURE RELEASE
         choices = [
             {"value": choice[0], "display_name": choice[1]} for choice in EscalationPolicy.WEB_DURATION_CHOICES_MINUTES
         ]

engine/common/api_helpers/custom_fields.py

@@ -1,3 +1,5 @@
+from datetime import timedelta
+
 from django.core.exceptions import ObjectDoesNotExist
 from drf_spectacular.utils import extend_schema_field
 from rest_framework import fields, serializers
@@ -205,3 +207,11 @@ def __init__(self, **kwargs):
             ],
             **kwargs,
         )
+
+
+class DurationSecondsField(serializers.FloatField):
+    def to_internal_value(self, data):
+        return timedelta(seconds=int(super().to_internal_value(data)))
+
+    def to_representation(self, value):
+        return str(value.total_seconds())