(self, request: Request)
| 33 | methods=["POST"], |
| 34 | ) |
| 35 | def test(self, request: Request) -> Response: |
| 36 | serializer = TestWebhookSerializer(data=request.data) |
| 37 | serializer.is_valid(raise_exception=True) |
| 38 | |
| 39 | data = serializer.validated_data |
| 40 | secret = data.get("secret") |
| 41 | webhook_url = data["webhook_url"] |
| 42 | scope_type = data.get("scope", {}).get("type") |
| 43 | try: |
| 44 | webhook_type = ( |
| 45 | WebhookType.ORGANISATION |
| 46 | if scope_type == "organisation" |
| 47 | else WebhookType.ENVIRONMENT |
| 48 | ) |
| 49 | response = send_test_request_to_webhook(webhook_url, secret, webhook_type) |
| 50 | if response.status_code != 200: |
| 51 | return Response( |
| 52 | { |
| 53 | "detail": "Webhook returned invalid status", |
| 54 | "body": "Please check the webhook endpoint to validate it returns a 200 OK.", |
| 55 | "status": response.status_code, |
| 56 | }, |
| 57 | status=status.HTTP_400_BAD_REQUEST, |
| 58 | ) |
| 59 | return Response( |
| 60 | { |
| 61 | "detail": "Webhook test successful", |
| 62 | "status": response.status_code, |
| 63 | }, |
| 64 | status=status.HTTP_200_OK, |
| 65 | ) |
| 66 | except requests.exceptions.RequestException: |
| 67 | return Response( |
| 68 | { |
| 69 | "detail": "Could not connect to webhook URL", |
| 70 | "body": "Please check the URL, and ensure it is valid and accessible from the server.", |
| 71 | }, |
| 72 | status=status.HTTP_400_BAD_REQUEST, |
| 73 | ) |
no test coverage detected