Improve APIs for creating/updating direct paging integrations (#2603)

# What this PR does

- Disallow creating direct paging integrations for a team that already
has one + make sure this constraint is also enforced on update
- Refactor some direct paging API parts to be more reusable & simple
- Add tests

## 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

@@ -16,6 +16,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Changed
 
 - Update direct paging docs by @vadimkerr ([#2600](https://github.com/grafana/oncall/pull/2600))
+- Improve APIs for creating/updating direct paging integrations by @vadimkerr ([#2603](https://github.com/grafana/oncall/pull/2603))
 
 ### Fixed
 

engine/apps/alerts/models/alert_receive_channel.py

@@ -251,6 +251,25 @@ def delete(self):
     def hard_delete(self):
         super(AlertReceiveChannel, self).delete()
 
+    class DuplicateDirectPagingError(Exception):
+        """Only one Direct Paging integration is allowed per team."""
+
+        DETAIL = "Direct paging integration already exists for this team"  # Returned in BadRequest responses
+
+    def save(self, *args, **kwargs):
+        # Don't allow multiple Direct Paging integrations per team
+        if (
+            self.integration == AlertReceiveChannel.INTEGRATION_DIRECT_PAGING
+            and AlertReceiveChannel.objects.filter(
+                organization=self.organization, team=self.team, integration=self.integration
+            )
+            .exclude(pk=self.pk)
+            .exists()
+        ):
+            raise self.DuplicateDirectPagingError
+
+        return super().save(*args, **kwargs)
+
     def change_team(self, team_id, user):
         if team_id == self.team_id:
             raise TeamCanNotBeChangedError("Integration is already in this team")

engine/apps/api/serializers/alert_receive_channel.py

@@ -121,15 +121,24 @@ def create(self, validated_data):
             if _integration.slug == integration:
                 is_able_to_autoresolve = _integration.is_able_to_autoresolve
 
-        instance = AlertReceiveChannel.create(
-            **validated_data,
-            organization=organization,
-            author=self.context["request"].user,
-            allow_source_based_resolving=is_able_to_autoresolve,
-        )
+        try:
+            instance = AlertReceiveChannel.create(
+                **validated_data,
+                organization=organization,
+                author=self.context["request"].user,
+                allow_source_based_resolving=is_able_to_autoresolve,
+            )
+        except AlertReceiveChannel.DuplicateDirectPagingError:
+            raise BadRequest(detail=AlertReceiveChannel.DuplicateDirectPagingError.DETAIL)
 
         return instance
 
+    def update(self, *args, **kwargs):
+        try:
+            return super().update(*args, **kwargs)
+        except AlertReceiveChannel.DuplicateDirectPagingError:
+            raise BadRequest(detail=AlertReceiveChannel.DuplicateDirectPagingError.DETAIL)
+
     def get_instructions(self, obj):
         if obj.integration in [AlertReceiveChannel.INTEGRATION_MAINTENANCE]:
             return ""
@@ -146,6 +155,12 @@ def get_default_channel_filter(self, obj):
             if filter.is_default:
                 return filter.public_primary_key
 
+    @staticmethod
+    def validate_integration(integration):
+        if integration is None or integration not in AlertReceiveChannel.WEB_INTEGRATION_CHOICES:
+            raise BadRequest(detail="invalid integration")
+        return integration
+
     def validate_verbal_name(self, verbal_name):
         organization = self.context["request"].auth.organization
         if verbal_name is None or (self.instance and verbal_name == self.instance.verbal_name):

engine/apps/api/tests/test_alert_receive_channel.py

@@ -102,6 +102,27 @@ def test_create_invalid_alert_receive_channel(alert_receive_channel_internal_api
     assert response.status_code == status.HTTP_400_BAD_REQUEST
 
 
+@pytest.mark.django_db
+def test_create_invalid_alert_receive_channel_type(alert_receive_channel_internal_api_setup, make_user_auth_headers):
+    user, token, _ = alert_receive_channel_internal_api_setup
+
+    client = APIClient()
+    url = reverse("api-internal:alert_receive_channel-list")
+
+    response_1 = client.post(
+        url,
+        data={"integration": "random123", "verbal_name": "Random 123"},
+        format="json",
+        **make_user_auth_headers(user, token),
+    )
+    response_2 = client.post(
+        url, data={"verbal_name": "Random 123"}, format="json", **make_user_auth_headers(user, token)
+    )
+
+    assert response_1.status_code == status.HTTP_400_BAD_REQUEST
+    assert response_2.status_code == status.HTTP_400_BAD_REQUEST
+
+
 @pytest.mark.django_db
 def test_update_alert_receive_channel(alert_receive_channel_internal_api_setup, make_user_auth_headers):
     user, token, alert_receive_channel = alert_receive_channel_internal_api_setup
@@ -693,6 +714,74 @@ def test_get_alert_receive_channels_direct_paging_present_for_filters(
     assert response.json()["results"][0]["value"] == alert_receive_channel.public_primary_key
 
 
+@pytest.mark.django_db
+def test_create_alert_receive_channels_direct_paging(
+    make_organization_and_user_with_plugin_token, make_team, make_alert_receive_channel, make_user_auth_headers
+):
+    organization, user, token = make_organization_and_user_with_plugin_token()
+    team = make_team(organization)
+
+    client = APIClient()
+    url = reverse("api-internal:alert_receive_channel-list")
+
+    response_1 = client.post(
+        url, data={"integration": "direct_paging"}, format="json", **make_user_auth_headers(user, token)
+    )
+    response_2 = client.post(
+        url, data={"integration": "direct_paging"}, format="json", **make_user_auth_headers(user, token)
+    )
+
+    response_3 = client.post(
+        url,
+        data={"integration": "direct_paging", "team": team.public_primary_key},
+        format="json",
+        **make_user_auth_headers(user, token),
+    )
+    response_4 = client.post(
+        url,
+        data={"integration": "direct_paging", "team": team.public_primary_key},
+        format="json",
+        **make_user_auth_headers(user, token),
+    )
+
+    # Check direct paging integration for "No team" is created
+    assert response_1.status_code == status.HTTP_201_CREATED
+    # Check direct paging integration is not created, as it already exists for "No team"
+    assert response_2.status_code == status.HTTP_400_BAD_REQUEST
+
+    # Check direct paging integration for team is created
+    assert response_3.status_code == status.HTTP_201_CREATED
+    # Check direct paging integration is not created, as it already exists for team
+    assert response_4.status_code == status.HTTP_400_BAD_REQUEST
+    assert response_4.json()["detail"] == AlertReceiveChannel.DuplicateDirectPagingError.DETAIL
+
+
+@pytest.mark.django_db
+def test_update_alert_receive_channels_direct_paging(
+    make_organization_and_user_with_plugin_token, make_team, make_alert_receive_channel, make_user_auth_headers
+):
+    organization, user, token = make_organization_and_user_with_plugin_token()
+    team = make_team(organization)
+    integration = make_alert_receive_channel(
+        organization, integration=AlertReceiveChannel.INTEGRATION_DIRECT_PAGING, team=None
+    )
+    make_alert_receive_channel(organization, integration=AlertReceiveChannel.INTEGRATION_DIRECT_PAGING, team=team)
+
+    client = APIClient()
+    url = reverse("api-internal:alert_receive_channel-detail", kwargs={"pk": integration.public_primary_key})
+
+    # Move direct paging integration from "No team" to team
+    response = client.put(
+        url,
+        data={"integration": "direct_paging", "team": team.public_primary_key},
+        format="json",
+        **make_user_auth_headers(user, token),
+    )
+
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
+    assert response.json()["detail"] == AlertReceiveChannel.DuplicateDirectPagingError.DETAIL
+
+
 @pytest.mark.django_db
 def test_start_maintenance_integration(
     make_user_auth_headers,

engine/apps/api/views/alert_receive_channel.py

@@ -18,7 +18,6 @@
 )
 from apps.api.throttlers import DemoAlertThrottler
 from apps.auth_token.auth import PluginAuthentication
-from apps.user_management.models.team import Team
 from common.api_helpers.exceptions import BadRequest
 from common.api_helpers.filters import ByTeamModelFieldFilterMixin, TeamModelMultipleChoiceFilter
 from common.api_helpers.mixins import (
@@ -104,42 +103,6 @@ class AlertReceiveChannelView(
         "stop_maintenance": [RBACPermission.Permissions.INTEGRATIONS_WRITE],
     }
 
-    def create(self, request, *args, **kwargs):
-        organization = request.auth.organization
-        user = request.user
-        team_lookup = {}
-        if "team" in request.data:
-            team_public_pk = request.data.get("team", None)
-            if team_public_pk is not None:
-                try:
-                    team = user.available_teams.get(public_primary_key=team_public_pk)
-                    team_lookup = {"team": team}
-                except Team.DoesNotExist:
-                    return Response(data="invalid team", status=status.HTTP_400_BAD_REQUEST)
-            else:
-                team_lookup = {"team__isnull": True}
-
-        if request.data["integration"] is not None:
-            if request.data["integration"] in AlertReceiveChannel.WEB_INTEGRATION_CHOICES:
-                # Don't allow multiple Direct Paging integrations
-                if request.data["integration"] == AlertReceiveChannel.INTEGRATION_DIRECT_PAGING:
-                    try:
-                        AlertReceiveChannel.objects.get(
-                            organization=organization,
-                            integration=AlertReceiveChannel.INTEGRATION_DIRECT_PAGING,
-                            deleted_at=None,
-                            **team_lookup,
-                        )
-                        return Response(
-                            data="Direct paging integration already exists for this team",
-                            status=status.HTTP_400_BAD_REQUEST,
-                        )
-                    except AlertReceiveChannel.DoesNotExist:
-                        pass
-                return super().create(request, *args, **kwargs)
-
-        return Response(data="invalid integration", status=status.HTTP_400_BAD_REQUEST)
-
     def perform_update(self, serializer):
         prev_state = serializer.instance.insight_logs_serialized
         serializer.save()

engine/apps/public_api/serializers/integrations.py

@@ -125,11 +125,14 @@ def create(self, validated_data):
             if connection_error:
                 raise serializers.ValidationError(connection_error)
         with transaction.atomic():
-            instance = AlertReceiveChannel.create(
-                **validated_data,
-                author=self.context["request"].user,
-                organization=organization,
-            )
+            try:
+                instance = AlertReceiveChannel.create(
+                    **validated_data,
+                    author=self.context["request"].user,
+                    organization=organization,
+                )
+            except AlertReceiveChannel.DuplicateDirectPagingError:
+                raise BadRequest(detail=AlertReceiveChannel.DuplicateDirectPagingError.DETAIL)
             if default_route_data:
                 serializer = DefaultChannelFilterSerializer(
                     instance.default_channel_filter, default_route_data, context=self.context
@@ -138,6 +141,12 @@ def create(self, validated_data):
                 serializer.save()
         return instance
 
+    def update(self, *args, **kwargs):
+        try:
+            return super().update(*args, **kwargs)
+        except AlertReceiveChannel.DuplicateDirectPagingError:
+            raise BadRequest(detail=AlertReceiveChannel.DuplicateDirectPagingError.DETAIL)
+
     def validate(self, attrs):
         organization = self.context["request"].auth.organization
         verbal_name = attrs.get("verbal_name", None)

engine/apps/public_api/tests/test_integrations.py

@@ -817,3 +817,57 @@ def test_update_integration_default_route(
 
     assert response.status_code == status.HTTP_200_OK
     assert response.data["default_route"]["escalation_chain_id"] == escalation_chain.public_primary_key
+
+
+@pytest.mark.django_db
+def test_create_integrations_direct_paging(
+    make_organization_and_user_with_token, make_team, make_alert_receive_channel, make_user_auth_headers
+):
+    organization, _, token = make_organization_and_user_with_token()
+    team = make_team(organization)
+
+    client = APIClient()
+    url = reverse("api-public:integrations-list")
+
+    response_1 = client.post(url, data={"type": "direct_paging"}, format="json", HTTP_AUTHORIZATION=token)
+    response_2 = client.post(url, data={"type": "direct_paging"}, format="json", HTTP_AUTHORIZATION=token)
+
+    response_3 = client.post(
+        url, data={"type": "direct_paging", "team_id": team.public_primary_key}, format="json", HTTP_AUTHORIZATION=token
+    )
+    response_4 = client.post(
+        url, data={"type": "direct_paging", "team_id": team.public_primary_key}, format="json", HTTP_AUTHORIZATION=token
+    )
+
+    # Check direct paging integration for "No team" is created
+    assert response_1.status_code == status.HTTP_201_CREATED
+    # Check direct paging integration is not created, as it already exists for "No team"
+    assert response_2.status_code == status.HTTP_400_BAD_REQUEST
+
+    # Check direct paging integration for team is created
+    assert response_3.status_code == status.HTTP_201_CREATED
+    # Check direct paging integration is not created, as it already exists for team
+    assert response_4.status_code == status.HTTP_400_BAD_REQUEST
+    assert response_4.data["detail"] == AlertReceiveChannel.DuplicateDirectPagingError.DETAIL
+
+
+@pytest.mark.django_db
+def test_update_integrations_direct_paging(
+    make_organization_and_user_with_token, make_team, make_alert_receive_channel, make_user_auth_headers
+):
+    organization, _, token = make_organization_and_user_with_token()
+    team = make_team(organization)
+
+    integration = make_alert_receive_channel(
+        organization, integration=AlertReceiveChannel.INTEGRATION_DIRECT_PAGING, team=None
+    )
+    make_alert_receive_channel(organization, integration=AlertReceiveChannel.INTEGRATION_DIRECT_PAGING, team=team)
+
+    client = APIClient()
+    url = reverse("api-public:integrations-detail", args=[integration.public_primary_key])
+
+    # Move direct paging integration from "No team" to team
+    response = client.put(url, data={"team_id": team.public_primary_key}, format="json", HTTP_AUTHORIZATION=token)
+
+    assert response.status_code == status.HTTP_400_BAD_REQUEST
+    assert response.data["detail"] == AlertReceiveChannel.DuplicateDirectPagingError.DETAIL