feat: Test webhook from backend (#5354)

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Co-authored-by: Kim Gustyr <kim.gustyr@flagsmith.com>

Author: 3 people

api/api/urls/v1.py

@@ -80,6 +80,8 @@
         schema_view.with_ui("swagger", cache_timeout=0),
         name="schema-swagger-ui",
     ),
+    # Test webhook url
+    re_path(r"^webhooks/", include("webhooks.urls", namespace="webhooks")),
 ]
 
 if settings.SPLIT_TESTING_INSTALLED:

api/environments/views.py

@@ -1,10 +1,11 @@
 import logging
+from typing import Generic, Type, TypeVar
 
 from common.environments.permissions import (
     TAG_SUPPORTED_PERMISSIONS,
     VIEW_ENVIRONMENT,
 )
-from django.db.models import Count, Q
+from django.db.models import Count, Q, QuerySet
 from django.utils.decorators import method_decorator
 from drf_yasg import openapi  # type: ignore[import-untyped]
 from drf_yasg.utils import no_body, swagger_auto_schema  # type: ignore[import-untyped]
@@ -15,7 +16,9 @@
 from rest_framework.permissions import IsAuthenticated
 from rest_framework.request import Request
 from rest_framework.response import Response
+from rest_framework.serializers import BaseSerializer
 
+from core.models import AbstractBaseExportableModel
 from environments.permissions.permissions import (
     EnvironmentAdminPermission,
     EnvironmentPermissions,
@@ -35,7 +38,6 @@
 )
 from projects.models import Project
 from users.models import FFAdminUser
-from webhooks.mixins import TriggerSampleWebhookMixin
 from webhooks.webhooks import WebhookType
 
 from .identities.traits.models import Trait
@@ -57,6 +59,8 @@
     WebhookSerializer,
 )
 
+T = TypeVar("T", bound=AbstractBaseExportableModel)
+
 logger = logging.getLogger(__name__)
 
 
@@ -307,43 +311,44 @@ def disable_v2_versioning(self, request: Request, api_key: str) -> Response:
         return Response(status=status.HTTP_202_ACCEPTED)
 
 
-class NestedEnvironmentViewSet(viewsets.GenericViewSet):  # type: ignore[type-arg]
-    model_class = None
+class NestedEnvironmentViewSet(Generic[T], viewsets.GenericViewSet[T]):
+    model_class: Type[T]
     webhook_type = WebhookType.ENVIRONMENT
 
-    def get_queryset(self):  # type: ignore[no-untyped-def]
-        return self.model_class.objects.filter(  # type: ignore[attr-defined]
+    def get_queryset(self) -> QuerySet[T]:
+        return self.model_class.objects.filter(
             environment__api_key=self.kwargs.get("environment_api_key")
         )
 
-    def perform_create(self, serializer):  # type: ignore[no-untyped-def]
-        serializer.save(environment=self._get_environment())  # type: ignore[no-untyped-call]
+    def perform_create(self, serializer: BaseSerializer[T]) -> None:
+        serializer.save(environment=self._get_environment())
 
-    def perform_update(self, serializer):  # type: ignore[no-untyped-def]
-        serializer.save(environment=self._get_environment())  # type: ignore[no-untyped-call]
+    def perform_update(self, serializer: BaseSerializer[T]) -> None:
+        serializer.save(environment=self._get_environment())
 
-    def _get_environment(self):  # type: ignore[no-untyped-def]
-        return Environment.objects.get(api_key=self.kwargs.get("environment_api_key"))
+    def _get_environment(self) -> Environment:
+        environment: Environment = Environment.objects.get(
+            api_key=self.kwargs.get("environment_api_key")
+        )
+        return environment
 
 
 class WebhookViewSet(
-    NestedEnvironmentViewSet,
+    NestedEnvironmentViewSet[Webhook],
     mixins.ListModelMixin,
     mixins.CreateModelMixin,
     mixins.UpdateModelMixin,
     mixins.DestroyModelMixin,
-    TriggerSampleWebhookMixin,
 ):
     serializer_class = WebhookSerializer
     pagination_class = None
     permission_classes = [IsAuthenticated, NestedEnvironmentPermissions]
-    model_class = Webhook  # type: ignore[assignment]
-
-    webhook_type = WebhookType.ENVIRONMENT  # type: ignore[assignment]
+    model_class: Type[Webhook] = Webhook
+    webhook_type: WebhookType = WebhookType.ENVIRONMENT
 
 
 class EnvironmentAPIKeyViewSet(
-    NestedEnvironmentViewSet,
+    NestedEnvironmentViewSet[EnvironmentAPIKey],
     mixins.ListModelMixin,
     mixins.CreateModelMixin,
     mixins.UpdateModelMixin,
@@ -352,4 +357,4 @@ class EnvironmentAPIKeyViewSet(
     serializer_class = EnvironmentAPIKeySerializer
     pagination_class = None
     permission_classes = [IsAuthenticated, EnvironmentAdminPermission]
-    model_class = EnvironmentAPIKey  # type: ignore[assignment]
+    model_class: Type[EnvironmentAPIKey] = EnvironmentAPIKey

api/organisations/views.py

@@ -59,7 +59,6 @@
 from projects.serializers import ProjectListSerializer
 from users.models import FFAdminUser
 from users.serializers import UserIdSerializer
-from webhooks.mixins import TriggerSampleWebhookMixin
 from webhooks.webhooks import WebhookType
 
 from .serializers import OrganisationAPIUsageNotificationSerializer
@@ -335,11 +334,11 @@ def chargebee_webhook(request: Request) -> Response:
     return webhook_handlers.process_subscription(request)
 
 
-class OrganisationWebhookViewSet(viewsets.ModelViewSet, TriggerSampleWebhookMixin):  # type: ignore[type-arg]
+class OrganisationWebhookViewSet(viewsets.ModelViewSet):  # type: ignore[type-arg]
     serializer_class = OrganisationWebhookSerializer
     permission_classes = [IsAuthenticated, NestedOrganisationEntityPermission]
 
-    webhook_type = WebhookType.ORGANISATION  # type: ignore[assignment]
+    webhook_type = WebhookType.ORGANISATION
 
     def get_queryset(self):  # type: ignore[no-untyped-def]
         if getattr(self, "swagger_fake_view", False):

api/tests/unit/environments/test_unit_environments_views.py

@@ -1,5 +1,4 @@
 import json
-from unittest import mock
 
 import pytest
 from common.environments.permissions import (
@@ -512,32 +511,6 @@ def test_cannot_delete_webhooks_for_environment_user_does_not_belong_to(
     assert Webhook.objects.filter(id=webhook.id).exists()
 
 
-@mock.patch("webhooks.mixins.trigger_sample_webhook")
-def test_trigger_sample_webhook_calls_trigger_sample_webhook_method_with_correct_arguments(
-    trigger_sample_webhook_mock: mock.MagicMock,
-    environment: Environment,
-    admin_client: APIClient,
-) -> None:
-    # Given
-    valid_webhook_url = "http://my.webhook.com/webhooks"
-    mocked_response = mock.MagicMock(status_code=200)
-    trigger_sample_webhook_mock.return_value = mocked_response
-    url = reverse(
-        "api-v1:environments:environment-webhooks-trigger-sample-webhook",
-        args=[environment.api_key],
-    )
-    data = {"url": valid_webhook_url}
-
-    # When
-    response = admin_client.post(url, data)
-
-    # Then
-    assert response.json()["message"] == "Request returned 200"
-    assert response.status_code == status.HTTP_200_OK
-    args, _ = trigger_sample_webhook_mock.call_args
-    assert args[0].url == valid_webhook_url
-
-
 def test_list_api_keys(
     environment: Environment,
     admin_client_new: APIClient,

api/tests/unit/organisations/test_unit_organisations_views.py

@@ -1006,33 +1006,6 @@ def test_can_update_secret_for_webhook(
     assert webhook.secret == data["secret"]
 
 
-@mock.patch("webhooks.mixins.trigger_sample_webhook")
-def test_trigger_sample_webhook_calls_trigger_sample_webhook_method_with_correct_arguments(
-    trigger_sample_webhook: MagicMock,
-    organisation: Organisation,
-    admin_client: APIClient,
-) -> None:
-    # Given
-    mocked_response = mock.MagicMock(status_code=200)
-    trigger_sample_webhook.return_value = mocked_response
-
-    url = reverse(
-        "api-v1:organisations:organisation-webhooks-trigger-sample-webhook",
-        args=[organisation.id],
-    )
-    valid_webhook_url = "http://my.webhook.com/webhooks"
-    data = {"url": valid_webhook_url}
-
-    # When
-    response = admin_client.post(url, data)
-
-    # Then
-    assert response.json()["message"] == "Request returned 200"
-    assert response.status_code == status.HTTP_200_OK
-    args, _ = trigger_sample_webhook.call_args
-    assert args[0].url == valid_webhook_url
-
-
 def test_get_subscription_metadata_when_subscription_information_cache_exist(
     organisation: Organisation,
     admin_client: APIClient,