Method test_webhook

workers/shared/clients/webhook_client.py:158–218

Test webhook configuration and connectivity. Args: url: Webhook endpoint URL to test payload: Test payload data authorization_type: Type of authorization (none, bearer, basic, custom) authorization_key: Authorization key/token auth

(
        self,
        url: str,
        payload: dict[str, Any],
        authorization_type: str | AuthorizationType = AuthorizationType.NONE,
        authorization_key: str | None = None,
        authorization_header: str | None = None,
        timeout: int = 10,
        organization_id: str | None = None,
    )

            raise

    def test_webhook(
        self,
        url: str,
        payload: dict[str, Any],
        authorization_type: str | AuthorizationType = AuthorizationType.NONE,
        authorization_key: str | None = None,
        authorization_header: str | None = None,
        timeout: int = 10,
        organization_id: str | None = None,
    ) -> APIResponse:
        """Test webhook configuration and connectivity.

        Args:
            url: Webhook endpoint URL to test
            payload: Test payload data
            authorization_type: Type of authorization (none, bearer, basic, custom)
            authorization_key: Authorization key/token
            authorization_header: Custom authorization header
            timeout: Request timeout in seconds
            organization_id: Optional organization ID override

        Returns:
            Test result with success status and response details
        """
        # Convert authorization_type to string if it's an enum
        auth_type_str = (
            authorization_type.value
            if isinstance(authorization_type, AuthorizationType)
            else authorization_type
        )

        data = {
            "url": url,
            "payload": payload,
            "authorization_type": auth_type_str,
            "timeout": timeout,
        }

        # Add optional authorization
        if authorization_key:
            data["authorization_key"] = authorization_key
        if authorization_header:
            data["authorization_header"] = authorization_header

        logger.info(f"Testing webhook configuration for {url}")
        logger.debug(f"Test webhook authorization_type: {auth_type_str}")

        try:
            response = self.post(
                self._build_url("webhook", "test/"), data, organization_id=organization_id
            )

            success = response.get("success", False)
            logger.info(f"Webhook test for {url} {'succeeded' if success else 'failed'}")
            return APIResponse(
                success=success, data=response, status_code=response.get("status_code")
            )