diff --git a/linode_api4/groups/monitor.py b/linode_api4/groups/monitor.py index 46f37561f..816794b3f 100644 --- a/linode_api4/groups/monitor.py +++ b/linode_api4/groups/monitor.py @@ -20,8 +20,14 @@ ) from linode_api4.objects.monitor import ( AkamaiObjectStorageLogsDestinationDetails, + BasicAuthenticationDetails, + ChannelDetails, + CustomHeader, CustomHTTPSLogsDestinationDetails, + DestinationAuthentication, + EmailDetails, LogsStreamDetails, + WebhookDetails, ) __all__ = [ @@ -213,6 +219,193 @@ def alert_channels(self, *filters) -> PaginatedList: """ return self.client._get_and_filter(AlertChannel, *filters) + def channel_create( + self, + label: str, + channel_type: str, + details: "ChannelDetails", + ) -> AlertChannel: + """ + Create a new alert channel. + + Alert channels define destinations for alert notifications. Supported + channel types include email, webhook, PagerDuty, and Slack. + + **Webhook Channel Constraints:** + - If ``channel_type`` is "webhook", the following are required: + - ``details.webhook.endpoint_url`` must be provided + - ``details.webhook.authentication.type`` must be specified ("basic" or "none") + - If ``authentication.type`` is "basic", both ``basic_authentication_user`` and + ``basic_authentication_password`` must be provided in ``details.webhook.authentication.details`` + - Client Certificate Configuration (Optional but must be complete): + - If ``details.webhook.client_certificate_details`` is provided, all three certificates + must be included: ``client_ca_certificate``, ``client_certificate``, and ``client_private_key`` + - ``tls_hostname`` is optional + - Custom Headers: + - ``Content-Type`` header must NOT be set by the user; it will be managed by the API + + API Documentation: https://techdocs.akamai.com/linode-api/reference/post-notification-channel + + :param label: A human-readable name for the alert channel. + :type label: str + :param channel_type: The channel type (e.g., ``"email"``, ``"webhook"``). + :type channel_type: str + :param details: Configuration details specific to the channel type. + :type details: ChannelDetails + + :returns: The newly created alert channel. + :rtype: AlertChannel + + :raises ValueError: If webhook channel configuration is invalid or missing required fields. + """ + pass + + # Validate webhook channel requirements + if channel_type == "webhook": + self._validate_webhook_details(details) + + params = { + "label": label, + "channel_type": channel_type, + "details": ( + details._serialize() + if hasattr(details, "_serialize") + else details + ), + } + + result = self.client.post("/monitor/alert-channels", data=params) + + if "id" not in result: + raise UnexpectedResponseError( + "Unexpected response when creating alert channel!", + json=result, + ) + + return AlertChannel(self.client, result["id"], result) + + def verify_webhook( + self, + webhook: "WebhookDetails", + ) -> bool: + """ + Verify a webhook configuration by testing connectivity to the endpoint. + + This endpoint validates that the webhook endpoint is reachable and + accepts the request format. It's recommended to verify webhook + configurations before creating a webhook channel. + + **Webhook Configuration Requirements:** + - ``endpoint_url`` must be provided + - ``authentication.type`` must be specified ("basic" or "none") + - If ``authentication.type`` is "basic", both ``basic_authentication_user`` and + ``basic_authentication_password`` must be provided + - Client certificates (if used) must include all three: ``client_ca_certificate``, + ``client_certificate``, and ``client_private_key`` + + API Documentation: https://techdocs.akamai.com/linode-api/reference/post-verify-webhook + + :param webhook: The webhook configuration to verify. + :type webhook: WebhookDetails + + :returns: True if verification succeeds. + :rtype: bool + + :raises ValueError: If webhook configuration is invalid. + :raises ApiError: If the webhook verification fails. + """ + from linode_api4.objects.monitor import ChannelDetails + + # Validate webhook configuration + self._validate_webhook_details(ChannelDetails(webhook=webhook)) + + data = { + "webhook": ( + webhook._serialize() + if hasattr(webhook, "_serialize") + else webhook + ), + } + + result = self.client.post("/monitor/alert-channels/verify", data=data) + + return result.get("success", True) + + def _validate_webhook_details(self, details: "ChannelDetails") -> None: + """ + Validate webhook channel details against API requirements. + + :param details: The channel details to validate. + :type details: ChannelDetails + + :raises ValueError: If validation fails. + """ + if not details or not details.webhook: + raise ValueError( + "Webhook details are required for webhook channel type" + ) + + webhook = details.webhook + + # Validate required endpoint_url + if not webhook.endpoint_url: + raise ValueError( + "Webhook channel requires 'endpoint_url' to be specified" + ) + + # Validate required authentication.type + if not webhook.authentication or not webhook.authentication.type: + raise ValueError( + "Webhook channel requires 'authentication.type' to be specified " + "(e.g., 'basic' or 'none')" + ) + + auth_type = webhook.authentication.type + if auth_type == "basic": + # For basic auth, both username and password are required + if not webhook.authentication.details: + raise ValueError( + "Basic authentication requires 'authentication.details' to be specified" + ) + + auth_details = webhook.authentication.details + if not auth_details.basic_authentication_user: + raise ValueError( + "Basic authentication requires 'basic_authentication_user' to be specified" + ) + + if not auth_details.basic_authentication_password: + raise ValueError( + "Basic authentication requires 'basic_authentication_password' to be specified" + ) + + # Validate client certificate configuration (all three must be present together) + if webhook.client_certificate_details: + cert_details = webhook.client_certificate_details + + # Check if any certificate field is present + has_ca_cert = bool(cert_details.client_ca_certificate) + has_client_cert = bool(cert_details.client_certificate) + has_private_key = bool(cert_details.client_private_key) + + # If any certificate field is present, all must be present + if has_ca_cert or has_client_cert or has_private_key: + if not (has_ca_cert and has_client_cert and has_private_key): + raise ValueError( + "Client certificate configuration requires all three to be specified: " + "'client_ca_certificate', 'client_certificate', and 'client_private_key'. " + "'tls_hostname' is optional." + ) + + # Validate custom headers don't include Content-Type + if webhook.custom_headers: + for header in webhook.custom_headers: + if header.name and header.name.lower() == "content-type": + raise ValueError( + "Custom headers must NOT include 'Content-Type'; " + "it will be managed by the API" + ) + def create_alert_definition( self, service_type: str, diff --git a/linode_api4/objects/monitor.py b/linode_api4/objects/monitor.py index c23e4cead..3ff21fba0 100644 --- a/linode_api4/objects/monitor.py +++ b/linode_api4/objects/monitor.py @@ -14,12 +14,15 @@ "AlertEntities", "AlertScope", "AlertType", + "ChannelDetails", + "EmailDetails", "MonitorDashboard", "MonitorMetricsDefinition", "MonitorService", "MonitorServiceToken", "RuleCriteria", "TriggerConditions", + "WebhookDetails", "AkamaiObjectStorageLogsDestinationDetails", "AuthenticationType", "BasicAuthenticationDetails", @@ -503,6 +506,48 @@ class AlertDefinition(DerivedBase): } +@dataclass +class BasicAuthenticationDetails(JSONObject): + """ + Includes additional parameters necessary to define basic authentication. + """ + + basic_authentication_user: Optional[str] = None + basic_authentication_password: Optional[str] = None + + +@dataclass +class DestinationAuthentication(JSONObject): + """ + Authentication details required to access the endpoint_url. + """ + + type: Optional[AuthenticationType] = None + details: Optional[BasicAuthenticationDetails] = None + + +@dataclass +class CustomHeader(JSONObject): + """ + Pairs of parameters used to optionally include custom headers in the request. + """ + + name: str = "" + value: str = "" + + +@dataclass +class ClientCertificateDetails(JSONObject): + """ + Contains TLS client certificate information to additionally secure the connection. + """ + + client_ca_certificate: Optional[str] = None + client_certificate: Optional[str] = None + client_private_key: Optional[str] = None + tls_hostname: Optional[str] = None + + @dataclass class EmailDetails(JSONObject): """ @@ -513,13 +558,35 @@ class EmailDetails(JSONObject): recipient_type: Optional[str] = None +@dataclass +class WebhookDetails(JSONObject): + """ + Represents webhook-specific details for an alert channel. + + Fields: + - endpoint_url: The URL where webhook events are sent. + - authentication: Authentication configuration for the webhook endpoint. + - data_compression: Compression method for webhook payloads ("gzip" or "none"). + - client_certificate_details: TLS client certificate configuration. + - custom_headers: List of custom HTTP headers to include in webhook requests. + """ + + endpoint_url: Optional[str] = None + authentication: Optional[DestinationAuthentication] = None + data_compression: Optional[str] = None + client_certificate_details: Optional[ClientCertificateDetails] = None + custom_headers: Optional[List[CustomHeader]] = None + + @dataclass class ChannelDetails(JSONObject): """ Represents the details block for an AlertChannel, which varies by channel type. + Supports email and webhook channel details. """ email: Optional[EmailDetails] = None + webhook: Optional[WebhookDetails] = None @dataclass @@ -546,20 +613,58 @@ class AlertChannel(Base): API Documentation: https://techdocs.akamai.com/linode-api/reference/get-notification-channels This class maps to the Monitor API's `/monitor/alert-channels` resource - and is used by the SDK to list, load, and inspect channels. + and supports full CRUD operations (create, read, update, delete). + + Examples: + # List channels + channels = client.monitor.alert_channels() + + # Create email channel + channel = client.monitor.channel_create( + label="Support Email", + channel_type="email", + details=ChannelDetails( + email=EmailDetails( + recipient_type="user", + usernames=["user@example.com"] + ) + ) + ) + + # Create webhook channel + channel = client.monitor.channel_create( + label="Webhook Receiver", + channel_type="webhook", + details=ChannelDetails( + webhook=WebhookDetails( + endpoint_url="https://example.com/webhook", + authentication=DestinationAuthentication( + type="basic", + details=BasicAuthenticationDetails( + basic_authentication_user="user", + basic_authentication_password="pass" + ) + ) + ) + ) + ) + + # Update channel + channel.label = "Updated Label" + channel.save() - NOTE: Only read operations are supported for AlertChannel at this time. - Create, update, and delete (CRUD) operations are not allowed. + # Delete channel + channel.delete() """ api_endpoint = "/monitor/alert-channels/{id}" properties = { "id": Property(identifier=True), - "label": Property(), + "label": Property(mutable=True), "type": Property(), "channel_type": Property(), - "details": Property(mutable=False, json_object=ChannelDetails), + "details": Property(mutable=True, json_object=ChannelDetails), "alerts": Property(mutable=False, json_object=AlertInfo), "created": Property(is_datetime=True), "updated": Property(is_datetime=True), @@ -568,48 +673,6 @@ class AlertChannel(Base): } -@dataclass -class BasicAuthenticationDetails(JSONObject): - """ - Includes additional parameters necessary to define basic authentication. - """ - - basic_authentication_user: Optional[str] = None - basic_authentication_password: Optional[str] = None - - -@dataclass -class DestinationAuthentication(JSONObject): - """ - Authentication details required to access the endpoint_url. - """ - - type: Optional[AuthenticationType] = None - details: Optional[BasicAuthenticationDetails] = None - - -@dataclass -class CustomHeader(JSONObject): - """ - Pairs of parameters used to optionally include custom headers in the request. - """ - - name: str = "" - value: str = "" - - -@dataclass -class ClientCertificateDetails(JSONObject): - """ - Contains TLS client certificate information to additionally secure the connection. - """ - - client_ca_certificate: Optional[str] = None - client_certificate: Optional[str] = None - client_private_key: Optional[str] = None - tls_hostname: Optional[str] = None - - @dataclass class LogsDestinationDetailsBase(JSONObject): """ diff --git a/test/fixtures/monitor_alert-channels_123.json b/test/fixtures/monitor_alert-channels_123.json new file mode 100644 index 000000000..3a677da7a --- /dev/null +++ b/test/fixtures/monitor_alert-channels_123.json @@ -0,0 +1,24 @@ +{ + "id": 123, + "label": "alert notification channel", + "type": "user", + "channel_type": "email", + "details": { + "email": { + "usernames": [ + "admin-user1", + "admin-user2" + ], + "recipient_type": "user" + } + }, + "alerts": { + "url": "/monitor/alert-channels/123/alerts", + "type": "alerts-definitions", + "alert_count": 2 + }, + "created": "2024-01-01T00:00:00", + "updated": "2024-01-01T00:00:00", + "created_by": "tester", + "updated_by": "tester" +} diff --git a/test/fixtures/monitor_alert-channels_123_alerts.json b/test/fixtures/monitor_alert-channels_123_alerts.json new file mode 100644 index 000000000..d6cc9f89f --- /dev/null +++ b/test/fixtures/monitor_alert-channels_123_alerts.json @@ -0,0 +1,21 @@ +{ + "data": [ + { + "id": 12345, + "label": "DBAAS Alert 1", + "service_type": "dbaas", + "type": "alerts-definitions", + "url": "/monitor/services/dbaas/alerts-definitions/12345" + }, + { + "id": 12346, + "label": "DBAAS Alert 2", + "service_type": "dbaas", + "type": "alerts-definitions", + "url": "/monitor/services/dbaas/alerts-definitions/12346" + } + ], + "page": 1, + "pages": 1, + "results": 2 +} diff --git a/test/fixtures/monitor_services_dbaas_alert-definitions.json b/test/fixtures/monitor_services_dbaas_alert-definitions.json index c7b725524..704b3ec56 100644 --- a/test/fixtures/monitor_services_dbaas_alert-definitions.json +++ b/test/fixtures/monitor_services_dbaas_alert-definitions.json @@ -41,7 +41,7 @@ "metric": "cpu_usage", "operator": "gt", "threshold": 90, - "unit": "percent" + "unit": "%" } ] }, diff --git a/test/fixtures/monitor_services_dbaas_alert-definitions_12345.json b/test/fixtures/monitor_services_dbaas_alert-definitions_12345.json index f88dd7503..36bd66ddd 100644 --- a/test/fixtures/monitor_services_dbaas_alert-definitions_12345.json +++ b/test/fixtures/monitor_services_dbaas_alert-definitions_12345.json @@ -39,7 +39,7 @@ "metric": "cpu_usage", "operator": "gt", "threshold": 90, - "unit": "percent" + "unit": "%" } ] }, diff --git a/test/fixtures/monitor_services_dbaas_metric-definitions.json b/test/fixtures/monitor_services_dbaas_metric-definitions.json index c493b23a3..545013562 100644 --- a/test/fixtures/monitor_services_dbaas_metric-definitions.json +++ b/test/fixtures/monitor_services_dbaas_metric-definitions.json @@ -22,7 +22,7 @@ "metric": "cpu_usage", "metric_type": "gauge", "scrape_interval": "60s", - "unit": "percent" + "unit": "%" }, { "available_aggregate_functions": [ diff --git a/test/integration/models/monitor/test_monitor.py b/test/integration/models/monitor/test_monitor.py index 996f5f728..4b1acd0ce 100644 --- a/test/integration/models/monitor/test_monitor.py +++ b/test/integration/models/monitor/test_monitor.py @@ -17,7 +17,15 @@ MonitorService, MonitorServiceToken, ) -from linode_api4.objects.monitor import AlertStatus +from linode_api4.objects.monitor import ( + AlertChannel, + AlertStatus, + BasicAuthenticationDetails, + ChannelDetails, + CustomHeader, + DestinationAuthentication, + WebhookDetails, +) def wait_for_alert_ready( @@ -429,3 +437,66 @@ def test_integration_clone_alert_definition(test_linode_client): AlertDefinition, created.id, service_type ) delete_source_alert.delete() + + +# Webhook Channel Operations +def test_webhook_channel_crud(test_linode_client): + """ + Test webhook channel create, verify, and delete operations. + + Creates a webhook channel with basic auth, verifies the configuration, + and then deletes the channel. + """ + client = test_linode_client + + # Create webhook channel with basic authentication + webhook = client.monitor.channel_create( + label=f"webhook-test-{get_test_label()}", + channel_type="webhook", + details=ChannelDetails( + webhook=WebhookDetails( + endpoint_url="https://example.com/webhook", + authentication=DestinationAuthentication( + type="basic", + details=BasicAuthenticationDetails( + basic_authentication_user="testuser", + basic_authentication_password="testpass", + ), + ), + data_compression="gzip", + custom_headers=[ + CustomHeader(name="X-API-Key", value="secret123"), + ], + ) + ), + ) + + assert isinstance(webhook, AlertChannel) + assert webhook.channel_type == "webhook" + assert webhook.details.webhook.endpoint_url == "https://example.com/webhook" + assert webhook.details.webhook.authentication.type == "basic" + assert webhook.details.webhook.data_compression == "gzip" + assert len(webhook.details.webhook.custom_headers) == 1 + + # Verify webhook configuration + webhook_config = WebhookDetails( + endpoint_url="https://example.com/webhook", + authentication=DestinationAuthentication( + type="basic", + details=BasicAuthenticationDetails( + basic_authentication_user="user", + basic_authentication_password="pass", + ), + ), + ) + + is_valid = client.monitor.verify_webhook(webhook_config) + assert is_valid is True + + # Delete webhook channel + webhook_id = webhook.id + webhook.delete() + + # Verify deletion + with pytest.raises(ApiError): + client.load(AlertChannel, webhook_id) diff --git a/test/unit/groups/monitor_api_test.py b/test/unit/groups/monitor_api_test.py index 8b2af9fe5..48c89b914 100644 --- a/test/unit/groups/monitor_api_test.py +++ b/test/unit/groups/monitor_api_test.py @@ -3,11 +3,20 @@ from linode_api4 import PaginatedList from linode_api4.objects import ( AggregateFunction, + AlertChannel, AlertDefinition, AlertDefinitionChannel, AlertDefinitionEntity, EntityMetricOptions, ) +from linode_api4.objects.monitor import ( + BasicAuthenticationDetails, + ChannelDetails, + CustomHeader, + DestinationAuthentication, + EmailDetails, + WebhookDetails, +) class MonitorAPITest(MonitorClientBaseCase): @@ -256,3 +265,151 @@ def test_clone_alert_definition_with_optional_fields(self): } assert mock_post.call_data["channel_ids"] == [1, 2] assert mock_post.call_data["group_by"] == ["entity_id"] + + def test_create_email_channel(self): + """ + Test creating an email alert channel. + Verifies that channel_create() properly handles email channel details. + """ + create_url = "/monitor/alert-channels" + channel_id = 789 + channel_url = f"{create_url}/{channel_id}" + + create_response = { + "id": channel_id, + "label": "Email Test Channel", + "type": "user", + "channel_type": "email", + "details": { + "email": { + "usernames": ["test_user1", "test_user2"], + "recipient_type": "user", + } + }, + "alerts": { + "url": f"{channel_url}/alerts", + "type": "alerts-definitions", + "alert_count": 0, + }, + "created": "2024-01-01T00:00:00", + "updated": "2024-01-01T00:00:00", + "created_by": "test_user1", + "updated_by": "test_user1", + } + + with self.mock_post(create_response) as mock_post: + channel = self.client.monitor.channel_create( + label="Email Test Channel", + channel_type="email", + details=ChannelDetails( + email=EmailDetails( + recipient_type="user", + usernames=["test_user1", "test_user2"], + ) + ), + ) + assert mock_post.call_url == create_url + assert isinstance(channel, AlertChannel) + assert channel.id == channel_id + assert channel.label == "Email Test Channel" + assert channel.channel_type == "email" + + def test_create_webhook_channel(self): + """ + Test creating a webhook alert channel. + Verifies that channel_create() properly handles webhook channel details + with authentication, compression, and custom headers. + """ + create_url = "/monitor/alert-channels" + channel_id = 888 + channel_url = f"{create_url}/{channel_id}" + + create_response = { + "id": channel_id, + "label": "Webhook Test Channel", + "type": "user", + "channel_type": "webhook", + "details": { + "webhook": { + "endpoint_url": "https://example.com/webhook", + "authentication": { + "type": "basic", + "details": { + "basic_authentication_user": "testuser", + "basic_authentication_password": "testpass", + }, + }, + "data_compression": "gzip", + "custom_headers": [ + {"name": "X-API-Key", "value": "secret123"} + ], + } + }, + "alerts": { + "url": f"{channel_url}/alerts", + "type": "alerts-definitions", + "alert_count": 0, + }, + "created": "2024-01-01T00:00:00", + "updated": "2024-01-01T00:00:00", + "created_by": "webhook_user", + "updated_by": "webhook_user", + } + + with self.mock_post(create_response) as mock_post: + webhook_channel = self.client.monitor.channel_create( + label="Webhook Test Channel", + channel_type="webhook", + details=ChannelDetails( + webhook=WebhookDetails( + endpoint_url="https://example.com/webhook", + authentication=DestinationAuthentication( + type="basic", + details=BasicAuthenticationDetails( + basic_authentication_user="testuser", + basic_authentication_password="testpass", + ), + ), + data_compression="gzip", + custom_headers=[ + CustomHeader(name="X-API-Key", value="secret123") + ], + ) + ), + ) + assert mock_post.call_url == create_url + assert isinstance(webhook_channel, AlertChannel) + assert webhook_channel.id == channel_id + assert webhook_channel.label == "Webhook Test Channel" + assert webhook_channel.channel_type == "webhook" + assert ( + webhook_channel.details.webhook.endpoint_url + == "https://example.com/webhook" + ) + assert ( + webhook_channel.details.webhook.authentication.type == "basic" + ) + + def test_verify_webhook_channel(self): + """ + Test verifying a webhook channel configuration. + Verifies that verify_webhook() returns success for valid webhook config. + """ + verify_url = "/monitor/alert-channels/verify" + verify_response = {"success": True} + + with self.mock_post(verify_response) as mock_verify: + is_valid = self.client.monitor.verify_webhook( + WebhookDetails( + endpoint_url="https://example.com/webhook", + authentication=DestinationAuthentication( + type="basic", + details=BasicAuthenticationDetails( + basic_authentication_user="testuser", + basic_authentication_password="testpass", + ), + ), + ) + ) + assert mock_verify.call_url == verify_url + assert is_valid is True diff --git a/test/unit/objects/monitor_test.py b/test/unit/objects/monitor_test.py index c0999e485..6e6c80450 100644 --- a/test/unit/objects/monitor_test.py +++ b/test/unit/objects/monitor_test.py @@ -12,11 +12,16 @@ ) from linode_api4.objects.monitor import ( AkamaiObjectStorageLogsDestinationDetails, + BasicAuthenticationDetails, + ChannelDetails, + ClientCertificateDetails, + CustomHeader, CustomHTTPSLogsDestinationDetails, DestinationAuthentication, LogsDestinationDetailsBase, LogsStreamDetails, LogsStreamType, + WebhookDetails, ) @@ -190,6 +195,201 @@ def test_alert_channels(self): ) self.assertEqual(channels[0].alerts.alert_count, 0) + def test_webhook_channel_validation(self): + """ + Test webhook channel validation constraints for create and verify operations. + + Validates all constraint checks: endpoint_url, authentication.type, + basic auth credentials, client certificates, and custom headers. + """ + # Test 1: Missing endpoint_url + webhook_details = ChannelDetails( + webhook=WebhookDetails( + endpoint_url=None, + authentication=DestinationAuthentication(type="none"), + ) + ) + with self.assertRaises(ValueError) as cm: + self.client.monitor._validate_webhook_details(webhook_details) + self.assertIn("endpoint_url", str(cm.exception)) + + # Test 2: Missing authentication.type + webhook_details = ChannelDetails( + webhook=WebhookDetails( + endpoint_url="https://example.com/webhook", + authentication=DestinationAuthentication(type=None), + ) + ) + with self.assertRaises(ValueError) as cm: + self.client.monitor._validate_webhook_details(webhook_details) + self.assertIn("authentication.type", str(cm.exception)) + + # Test 3: Basic auth missing username + webhook_details = ChannelDetails( + webhook=WebhookDetails( + endpoint_url="https://example.com/webhook", + authentication=DestinationAuthentication( + type="basic", + details=BasicAuthenticationDetails( + basic_authentication_user=None, + basic_authentication_password="password", + ), + ), + ) + ) + with self.assertRaises(ValueError) as cm: + self.client.monitor._validate_webhook_details(webhook_details) + self.assertIn("basic_authentication_user", str(cm.exception)) + + # Test 4: Basic auth missing password + webhook_details = ChannelDetails( + webhook=WebhookDetails( + endpoint_url="https://example.com/webhook", + authentication=DestinationAuthentication( + type="basic", + details=BasicAuthenticationDetails( + basic_authentication_user="user", + basic_authentication_password=None, + ), + ), + ) + ) + with self.assertRaises(ValueError) as cm: + self.client.monitor._validate_webhook_details(webhook_details) + self.assertIn("basic_authentication_password", str(cm.exception)) + + # Test 5: Partial client certificates + webhook_details = ChannelDetails( + webhook=WebhookDetails( + endpoint_url="https://example.com/webhook", + authentication=DestinationAuthentication(type="none"), + client_certificate_details=ClientCertificateDetails( + client_ca_certificate="-----BEGIN CERTIFICATE-----", + client_certificate=None, + client_private_key=None, + ), + ) + ) + with self.assertRaises(ValueError) as cm: + self.client.monitor._validate_webhook_details(webhook_details) + self.assertIn("client_ca_certificate", str(cm.exception)) + + # Test 6: Content-Type header validation + webhook_details = ChannelDetails( + webhook=WebhookDetails( + endpoint_url="https://example.com/webhook", + authentication=DestinationAuthentication(type="none"), + custom_headers=[ + CustomHeader(name="Content-Type", value="application/json") + ], + ) + ) + with self.assertRaises(ValueError) as cm: + self.client.monitor._validate_webhook_details(webhook_details) + self.assertIn("Content-Type", str(cm.exception)) + + def test_create_verify_delete_webhook_channel(self): + """ + Test webhook channel create, verify, and delete operations. + Verifies the full lifecycle of a webhook alert channel object. + """ + create_url = "/monitor/alert-channels" + channel_id = 888 + channel_url = f"{create_url}/{channel_id}" + + # CREATE: Create the webhook channel via channel_create() + create_response = { + "id": channel_id, + "label": "Webhook Test Channel", + "type": "user", + "channel_type": "webhook", + "details": { + "webhook": { + "endpoint_url": "https://example.com/webhook", + "authentication": { + "type": "basic", + "details": { + "basic_authentication_user": "testuser", + "basic_authentication_password": "testpass", + }, + }, + "data_compression": "gzip", + "custom_headers": [ + {"name": "X-API-Key", "value": "secret123"} + ], + } + }, + "alerts": { + "url": f"{channel_url}/alerts", + "type": "alerts-definitions", + "alert_count": 0, + }, + "created": "2024-01-01T00:00:00", + "updated": "2024-01-01T00:00:00", + "created_by": "webhook_user", + "updated_by": "webhook_user", + } + + with self.mock_post(create_response) as m_post: + webhook_channel = self.client.monitor.channel_create( + label="Webhook Test Channel", + channel_type="webhook", + details=ChannelDetails( + webhook=WebhookDetails( + endpoint_url="https://example.com/webhook", + authentication=DestinationAuthentication( + type="basic", + details=BasicAuthenticationDetails( + basic_authentication_user="testuser", + basic_authentication_password="testpass", + ), + ), + data_compression="gzip", + custom_headers=[ + CustomHeader(name="X-API-Key", value="secret123") + ], + ) + ), + ) + self.assertEqual(m_post.call_url, create_url) + self.assertIsInstance(webhook_channel, AlertChannel) + self.assertEqual(webhook_channel.id, channel_id) + self.assertEqual(webhook_channel.label, "Webhook Test Channel") + self.assertEqual( + webhook_channel.details.webhook.endpoint_url, + "https://example.com/webhook", + ) + self.assertEqual( + webhook_channel.details.webhook.authentication.type, "basic" + ) + + # VERIFY: Verify the webhook configuration + verify_url = "/monitor/alert-channels/verify" + verify_response = {"valid": True} + + with self.mock_post(verify_response) as m_verify: + is_valid = self.client.monitor.verify_webhook( + WebhookDetails( + endpoint_url="https://example.com/webhook", + authentication=DestinationAuthentication( + type="basic", + details=BasicAuthenticationDetails( + basic_authentication_user="testuser", + basic_authentication_password="testpass", + ), + ), + ) + ) + self.assertEqual(m_verify.call_url, verify_url) + self.assertTrue(is_valid) + + # DELETE: Delete the webhook channel + with self.mock_delete() as m_delete: + result = webhook_channel.delete() + + self.assertEqual(m_delete.call_url, channel_url) + self.assertTrue(result) + class LogsDestinationTest(ClientBaseCase): """