Add endpoint for alert group escalation snapshot (#3615)

# What this PR does
Adds endpoint for alert group escalation snapshot

## Which issue(s) this PR fixes
#3277

## Checklist

- [x] Unit, integration, and e2e (if applicable) tests updated
- [x] Documentation added (or `pr:no public docs` PR label added if not
required)
- [x] `CHANGELOG.md` updated (or `pr:no changelog` PR label added if not
required)

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+### Added
+
+- Add endpoint for alert group escalation snapshot by @Ferril ([#3615](https://github.com/grafana/oncall/pull/3615))
+
 ### Changed
 
 - Do not retry `firebase.messaging.UnregisteredError` exceptions for FCM relay tasks by @joeyorlando ([#3637](https://github.com/grafana/oncall/pull/3637))

engine/apps/api/serializers/alert_group_escalation_snapshot.py

@@ -0,0 +1,55 @@
+from rest_framework import serializers
+
+from apps.api.serializers.custom_button import CustomButtonFastSerializer
+from apps.api.serializers.escalation_policy import EscalationPolicySerializer
+from apps.api.serializers.schedule_base import ScheduleFastSerializer
+from apps.api.serializers.user import FastUserSerializer
+from apps.api.serializers.user_group import UserGroupSerializer
+from apps.api.serializers.webhook import WebhookFastSerializer
+
+
+class EscalationPolicySnapshotAPISerializer(EscalationPolicySerializer):
+    """Serializes AlertGroup escalation policies snapshots for API endpoint"""
+
+    notify_to_users_queue = FastUserSerializer(many=True, read_only=True)
+    notify_schedule = ScheduleFastSerializer(read_only=True)
+    notify_to_group = UserGroupSerializer(read_only=True)
+    custom_button_trigger = CustomButtonFastSerializer(read_only=True)
+    custom_webhook = WebhookFastSerializer(read_only=True)
+
+    class Meta(EscalationPolicySerializer.Meta):
+        fields = [
+            "step",
+            "wait_delay",
+            "notify_to_users_queue",
+            "from_time",
+            "to_time",
+            "num_alerts_in_window",
+            "num_minutes_in_window",
+            "slack_integration_required",
+            "custom_button_trigger",
+            "custom_webhook",
+            "notify_schedule",
+            "notify_to_group",
+            "important",
+        ]
+        read_only_fields = fields
+
+
+class AlertGroupEscalationSnapshotAPISerializer(serializers.Serializer):
+    """Serializes AlertGroup escalation snapshot for API endpoint"""
+
+    escalation_chain = serializers.SerializerMethodField()
+    channel_filter = serializers.SerializerMethodField()
+    escalation_policies = EscalationPolicySnapshotAPISerializer(
+        source="escalation_policies_snapshots", many=True, read_only=True
+    )
+
+    class Meta:
+        fields = ["escalation_chain", "channel_filter", "escalation_policies"]
+
+    def get_escalation_chain(self, obj):
+        return {"name": obj.escalation_chain_snapshot.name}
+
+    def get_channel_filter(self, obj):
+        return {"name": obj.channel_filter_snapshot.str_for_clients}

engine/apps/api/serializers/custom_button.py

@@ -69,3 +69,11 @@ def validate_forward_whole_payload(self, data):
         if data is None:
             return False
         return data
+
+
+class CustomButtonFastSerializer(serializers.ModelSerializer):
+    id = serializers.CharField(read_only=True, source="public_primary_key")
+
+    class Meta:
+        model = CustomButton
+        fields = ["id", "name"]

engine/apps/api/serializers/webhook.py

@@ -220,3 +220,11 @@ def is_field_controlled(self, field_name):
             if field_name not in preset.metadata.controlled_fields:
                 return False
         return True
+
+
+class WebhookFastSerializer(serializers.ModelSerializer):
+    id = serializers.CharField(read_only=True, source="public_primary_key")
+
+    class Meta:
+        model = Webhook
+        fields = ["id", "name"]

engine/apps/api/tests/test_alert_group.py

@@ -1423,6 +1423,41 @@ def test_alert_group_detail_permissions(
     assert response.status_code == expected_status
 
 
+@pytest.mark.django_db
+@pytest.mark.parametrize(
+    "role,expected_status",
+    [
+        (LegacyAccessControlRole.ADMIN, status.HTTP_200_OK),
+        (LegacyAccessControlRole.EDITOR, status.HTTP_200_OK),
+        (LegacyAccessControlRole.VIEWER, status.HTTP_200_OK),
+        (LegacyAccessControlRole.NONE, status.HTTP_403_FORBIDDEN),
+    ],
+)
+def test_alert_group_escalation_snapshot_permissions(
+    alert_group_internal_api_setup,
+    make_user_for_organization,
+    make_user_auth_headers,
+    role,
+    expected_status,
+):
+    _, token, alert_groups = alert_group_internal_api_setup
+    _, _, new_alert_group, _ = alert_groups
+    organization = new_alert_group.channel.organization
+    user = make_user_for_organization(organization, role)
+
+    client = APIClient()
+    url = reverse("api-internal:alertgroup-escalation-snapshot", kwargs={"pk": new_alert_group.public_primary_key})
+
+    with patch(
+        "apps.api.views.alert_group.AlertGroupView.escalation_snapshot",
+        return_value=Response(
+            status=status.HTTP_200_OK,
+        ),
+    ):
+        response = client.get(url, format="json", **make_user_auth_headers(user, token))
+    assert response.status_code == expected_status
+
+
 @pytest.mark.django_db
 def test_silence(alert_group_internal_api_setup, make_user_auth_headers):
     client = APIClient()

engine/apps/api/tests/test_alert_group_escalation_snapshot.py

@@ -0,0 +1,124 @@
+import pytest
+from django.urls import reverse
+from rest_framework import status
+from rest_framework.test import APIClient
+
+from apps.alerts.models import EscalationPolicy
+
+
+@pytest.mark.django_db
+def test_alert_group_escalation_snapshot_with_important(
+    make_organization_and_user_with_plugin_token,
+    make_alert_receive_channel,
+    make_escalation_chain,
+    make_escalation_policy,
+    make_channel_filter,
+    make_alert_group,
+    make_user_auth_headers,
+):
+    organization, user, token = make_organization_and_user_with_plugin_token()
+    alert_receive_channel = make_alert_receive_channel(organization)
+    escalation_chain = make_escalation_chain(organization)
+    notify_to_multiple_users_step = make_escalation_policy(
+        escalation_chain=escalation_chain,
+        escalation_policy_step=EscalationPolicy.STEP_NOTIFY_MULTIPLE_USERS,
+    )
+    notify_to_multiple_users_step.notify_to_users_queue.set([user])
+    notify_to_multiple_users_step_important = make_escalation_policy(
+        escalation_chain=escalation_chain,
+        escalation_policy_step=EscalationPolicy.STEP_NOTIFY_MULTIPLE_USERS_IMPORTANT,
+    )
+    notify_to_multiple_users_step_important.notify_to_users_queue.set([user])
+    channel_filter = make_channel_filter(alert_receive_channel, is_default=True, escalation_chain=escalation_chain)
+    alert_group = make_alert_group(alert_receive_channel, channel_filter=channel_filter)
+    alert_group.raw_escalation_snapshot = alert_group.build_raw_escalation_snapshot()
+    alert_group.save()
+
+    client = APIClient()
+    url = reverse("api-internal:alertgroup-escalation-snapshot", kwargs={"pk": alert_group.public_primary_key})
+    expected_result = {
+        "escalation_chain": {"name": escalation_chain.name},
+        "channel_filter": {"name": "default"},
+        "escalation_policies": [
+            {
+                "step": 13,
+                "wait_delay": None,
+                "notify_to_users_queue": [{"pk": user.public_primary_key, "username": user.username}],
+                "from_time": None,
+                "to_time": None,
+                "num_alerts_in_window": None,
+                "num_minutes_in_window": None,
+                "custom_button_trigger": None,
+                "custom_webhook": None,
+                "notify_schedule": None,
+                "notify_to_group": None,
+                "important": False,
+            },
+            {
+                "step": 13,
+                "wait_delay": None,
+                "notify_to_users_queue": [{"pk": user.public_primary_key, "username": user.username}],
+                "from_time": None,
+                "to_time": None,
+                "num_alerts_in_window": None,
+                "num_minutes_in_window": None,
+                "custom_button_trigger": None,
+                "custom_webhook": None,
+                "notify_schedule": None,
+                "notify_to_group": None,
+                "important": True,
+            },
+        ],
+    }
+    response = client.get(url, **make_user_auth_headers(user, token))
+    assert response.status_code == status.HTTP_200_OK
+    assert response.json() == expected_result
+
+
+@pytest.mark.django_db
+def test_alert_group_no_escalation_snapshot(
+    make_organization_and_user_with_plugin_token,
+    make_alert_receive_channel,
+    make_channel_filter,
+    make_alert_group,
+    make_user_auth_headers,
+):
+    organization, user, token = make_organization_and_user_with_plugin_token()
+    alert_receive_channel = make_alert_receive_channel(organization)
+    channel_filter = make_channel_filter(alert_receive_channel, is_default=True)
+    alert_group = make_alert_group(alert_receive_channel, channel_filter=channel_filter)
+    client = APIClient()
+    url = reverse("api-internal:alertgroup-escalation-snapshot", kwargs={"pk": alert_group.public_primary_key})
+    expected_result = {}
+    response = client.get(url, **make_user_auth_headers(user, token))
+    assert response.status_code == status.HTTP_200_OK
+    assert response.json() == expected_result
+
+
+@pytest.mark.django_db
+def test_alert_group_escalation_snapshot_no_policies(
+    make_organization_and_user_with_plugin_token,
+    make_alert_receive_channel,
+    make_escalation_chain,
+    make_channel_filter,
+    make_alert_group,
+    make_user_auth_headers,
+):
+    organization, user, token = make_organization_and_user_with_plugin_token()
+    alert_receive_channel = make_alert_receive_channel(organization)
+    escalation_chain = make_escalation_chain(organization)
+    channel_filter = make_channel_filter(alert_receive_channel, is_default=True, escalation_chain=escalation_chain)
+    alert_group = make_alert_group(alert_receive_channel, channel_filter=channel_filter)
+    alert_group.raw_escalation_snapshot = alert_group.build_raw_escalation_snapshot()
+    alert_group.save()
+
+    client = APIClient()
+    url = reverse("api-internal:alertgroup-escalation-snapshot", kwargs={"pk": alert_group.public_primary_key})
+    expected_result = {
+        "escalation_chain": {"name": escalation_chain.name},
+        "channel_filter": {"name": "default"},
+        "escalation_policies": [],
+    }
+    response = client.get(url, **make_user_auth_headers(user, token))
+    assert response.status_code == status.HTTP_200_OK
+    assert response.json() == expected_result

engine/apps/api/views/alert_group.py

@@ -21,6 +21,7 @@
 from apps.api.label_filtering import parse_label_query
 from apps.api.permissions import RBACPermission
 from apps.api.serializers.alert_group import AlertGroupListSerializer, AlertGroupSerializer
+from apps.api.serializers.alert_group_escalation_snapshot import AlertGroupEscalationSnapshotAPISerializer
 from apps.api.serializers.team import TeamSerializer
 from apps.auth_token.auth import PluginAuthentication
 from apps.base.models.user_notification_policy_log_record import UserNotificationPolicyLogRecord
@@ -303,6 +304,7 @@ class AlertGroupView(
         "unpage_user": [RBACPermission.Permissions.ALERT_GROUPS_WRITE],
         "bulk_action": [RBACPermission.Permissions.ALERT_GROUPS_WRITE],
         "preview_template": [RBACPermission.Permissions.INTEGRATIONS_TEST],
+        "escalation_snapshot": [RBACPermission.Permissions.ALERT_GROUPS_READ],
     }
 
     http_method_names = ["get", "post", "delete"]
@@ -857,3 +859,10 @@ def bulk_action_options(self, request):
     # This method is required for PreviewTemplateMixin
     def get_alert_to_template(self, payload=None):
         return self.get_object().alerts.first()
+
+    @action(methods=["get"], detail=True)
+    def escalation_snapshot(self, request, pk=None):
+        alert_group = self.get_object()
+        escalation_snapshot = alert_group.escalation_snapshot
+        result = AlertGroupEscalationSnapshotAPISerializer(escalation_snapshot).data if escalation_snapshot else {}
+        return Response(result)