Execute a saved integration (connect) with a dummy payload to verify configuration.
(self, request, pk=None)
| 137 | |
| 138 | @action(detail=True, methods=["post"], url_path="test", permission_classes=[IsAdmin]) |
| 139 | def test(self, request, pk=None): |
| 140 | """ |
| 141 | Execute a saved integration (connect) with a dummy payload to verify configuration. |
| 142 | """ |
| 143 | try: |
| 144 | integration = Integration.objects.get(pk=pk) |
| 145 | except Integration.DoesNotExist: |
| 146 | return Response({"detail": "Integration not found"}, status=status.HTTP_404_NOT_FOUND) |
| 147 | |
| 148 | # Build a dummy payload similar to system events |
| 149 | now = timezone.now().isoformat() |
| 150 | dummy_payload = { |
| 151 | "event": "test", |
| 152 | "timestamp": now, |
| 153 | "channel_name": "Test Channel", |
| 154 | "stream_name": "Test Stream", |
| 155 | "stream_url": "http://example.com/stream.m3u8", |
| 156 | "channel_url": "http://example.com/stream.m3u8", |
| 157 | "provider_name": "Test Provider", |
| 158 | "profile_used": "Default", |
| 159 | "test": True, |
| 160 | } |
| 161 | |
| 162 | # Choose handler based on saved type |
| 163 | if integration.type == "webhook": |
| 164 | handler = WebhookHandler(integration, None, dummy_payload) |
| 165 | elif integration.type == "script": |
| 166 | handler = ScriptHandler(integration, None, dummy_payload) |
| 167 | else: |
| 168 | return Response( |
| 169 | {"success": False, "error": f"Unsupported integration type: {integration.type}"}, |
| 170 | status=status.HTTP_400_BAD_REQUEST, |
| 171 | ) |
| 172 | |
| 173 | try: |
| 174 | result = handler.execute() |
| 175 | return Response( |
| 176 | { |
| 177 | "success": bool(result.get("success")), |
| 178 | "type": integration.type, |
| 179 | "request_payload": dummy_payload, |
| 180 | "result": result, |
| 181 | }, |
| 182 | status=status.HTTP_200_OK, |
| 183 | ) |
| 184 | except Exception as e: |
| 185 | return Response( |
| 186 | { |
| 187 | "success": False, |
| 188 | "type": integration.type, |
| 189 | "request_payload": dummy_payload, |
| 190 | "error": str(e), |
| 191 | }, |
| 192 | status=status.HTTP_502_BAD_GATEWAY, |
| 193 | ) |
| 194 | |
| 195 | |
| 196 | class EventSubscriptionViewSet(viewsets.ModelViewSet): |
no test coverage detected