From f4ab4af61327376d062be90f8ad799e74d7e7bb5 Mon Sep 17 00:00:00 2001 From: mawasthy Date: Thu, 18 Jun 2026 15:48:21 +0530 Subject: [PATCH 1/9] adding create channel api changes --- linode_api4/groups/monitor.py | 62 +++++++++++++++++++ linode_api4/objects/monitor.py | 11 ++-- .../models/monitor/test_monitor.py | 61 +++++++++++++++++- test/unit/groups/monitor_api_test.py | 54 ++++++++++++++++ test/unit/objects/monitor_test.py | 57 +++++++++++++++++ 5 files changed, 238 insertions(+), 7 deletions(-) diff --git a/linode_api4/groups/monitor.py b/linode_api4/groups/monitor.py index 0d7f19ce8..45b12cbe9 100644 --- a/linode_api4/groups/monitor.py +++ b/linode_api4/groups/monitor.py @@ -13,6 +13,7 @@ MonitorService, MonitorServiceToken, ) +from linode_api4.objects.monitor import ChannelDetails __all__ = [ "MonitorGroup", @@ -332,3 +333,64 @@ def alert_definition_entities( *filters, endpoint=endpoint, ) + + def channel_create( + self, + label: str, + channel_type: str, + details: ChannelDetails, + ) -> AlertChannel: + """ + Creates a new alert channel for the authenticated account. + + An alert channel defines a notification destination (for example: an + email list) that can be associated with one or more alert definitions. + Currently only ``email`` is supported as a ``channel_type``. + + Example usage:: + + from linode_api4.objects.monitor import ChannelDetails, EmailDetails + + client = LinodeClient(TOKEN) + + new_channel = client.monitor.channel_create( + label="Email channel for api change", + channel_type="email", + details=ChannelDetails( + email=EmailDetails( + recipient_type="user", + usernames=["username-test"], + ) + ), + ) + + API Documentation: https://techdocs.akamai.com/linode-api/reference/post-alert-channel + + :param label: Human-readable name for the new alert channel. + :type label: str + :param channel_type: The type of notification channel (e.g. ``"email"``). + :type channel_type: str + :param details: Notification-type-specific configuration. Use + :class:`~linode_api4.objects.monitor.ChannelDetails` with + a nested :class:`~linode_api4.objects.monitor.EmailDetails` + for email channels. + :type details: ChannelDetails + + :returns: The newly created :class:`AlertChannel`. + :rtype: AlertChannel + """ + params = { + "label": label, + "channel_type": channel_type, + "details": details.dict, + } + + 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) diff --git a/linode_api4/objects/monitor.py b/linode_api4/objects/monitor.py index 7e0f4ae4d..6efd23354 100644 --- a/linode_api4/objects/monitor.py +++ b/linode_api4/objects/monitor.py @@ -492,13 +492,12 @@ class AlertChannel(Base): fire. Alert channels define a destination and configuration for notifications (for example: email lists, webhooks, PagerDuty, Slack, etc.). - API Documentation: https://techdocs.akamai.com/linode-api/reference/get-notification-channels + API Documentation: + List/Get: https://techdocs.akamai.com/linode-api/reference/get-alert-channels + Create: https://techdocs.akamai.com/linode-api/reference/post-alert-channel - This class maps to the Monitor API's `/monitor/alert-channels` resource - and is used by the SDK to list, load, and inspect channels. - - NOTE: Only read operations are supported for AlertChannel at this time. - Create, update, and delete (CRUD) operations are not allowed. + This class maps to the Monitor API's ``/monitor/alert-channels`` resource + and is used by the SDK to list, load, create, and inspect channels. """ api_endpoint = "/monitor/alert-channels/{id}" diff --git a/test/integration/models/monitor/test_monitor.py b/test/integration/models/monitor/test_monitor.py index ceb9fdc3a..765bc8b21 100644 --- a/test/integration/models/monitor/test_monitor.py +++ b/test/integration/models/monitor/test_monitor.py @@ -9,6 +9,7 @@ from linode_api4 import LinodeClient, PaginatedList from linode_api4.objects import ( + AlertChannel, AlertDefinition, AlertDefinitionEntity, ApiError, @@ -17,7 +18,7 @@ MonitorService, MonitorServiceToken, ) -from linode_api4.objects.monitor import AlertStatus +from linode_api4.objects.monitor import AlertStatus, ChannelDetails, EmailDetails # List all dashboards @@ -311,3 +312,61 @@ def test_alert_definition_entities(test_linode_client): assert entity.label assert entity.url assert entity._type == service_type + + +def test_integration_create_get_delete_alert_channel(test_linode_client): + """E2E: create an alert channel, fetch it, then delete it. + + This test creates an alert channel with email details, retrieves it, + and then deletes it. It ensures the feature is working end-to-end + against the actual API. + """ + client = test_linode_client + label = get_test_label() + "-e2e-channel" + label = f"{label}-{int(time.time())}" + + created_channel = None + + try: + # Create an alert channel with email details + created_channel = client.monitor.channel_create( + label=label, + channel_type="email", + details=ChannelDetails( + email=EmailDetails( + recipient_type="user", + usernames=["mawasthy_tenant02_admin"], + ) + ), + ) + + # Assert the created channel has expected properties + assert isinstance(created_channel, AlertChannel) + assert created_channel.id is not None + assert created_channel.label == label + assert created_channel.channel_type == "email" + assert created_channel.details is not None + + # Fetch the channel to verify it exists + channels = list(client.monitor.alert_channels()) + assert len(channels) > 0, "No channels found after creation" + + # Find the created channel in the list + found_channel = None + for ch in channels: + if ch.id == created_channel.id: + found_channel = ch + break + + assert found_channel is not None, "Created channel not found in list" + assert found_channel.label == label + assert found_channel.channel_type == "email" + + finally: + if created_channel: + # Clean up: delete the created channel + try: + created_channel.delete() + except Exception as e: + # Log but don't fail if cleanup fails + print(f"Warning: Failed to delete channel {created_channel.id}: {e}") diff --git a/test/unit/groups/monitor_api_test.py b/test/unit/groups/monitor_api_test.py index fdc93060c..fd7e1c784 100644 --- a/test/unit/groups/monitor_api_test.py +++ b/test/unit/groups/monitor_api_test.py @@ -3,11 +3,13 @@ from linode_api4 import PaginatedList from linode_api4.objects import ( AggregateFunction, + AlertChannel, AlertDefinition, AlertDefinitionChannel, AlertDefinitionEntity, EntityMetricOptions, ) +from linode_api4.objects.monitor import ChannelDetails, EmailDetails class MonitorAPITest(MonitorClientBaseCase): @@ -180,3 +182,55 @@ def test_alert_definition_entities(self): assert entities[2].label == "mydatabase-3" assert entities[2].url == "/v4/databases/mysql/instances/3" assert entities[2]._type == "dbaas" + + def test_create_channel(self): + url = "/monitor/alert-channels" + result = { + "id": 123, + "label": "email channel for api change", + "type": "user", + "channel_type": "email", + "details": { + "email": { + "usernames": ["mawasthy_tenant02_admin"], + "recipient_type": "user", + } + }, + "alerts": { + "url": "/monitor/alert-channels/123/alerts", + "type": "alerts-definitions", + "alert_count": 0, + }, + "created": "2024-01-01T00:00:00", + "updated": "2024-01-01T00:00:00", + "created_by": "mawasthy_tenant02_admin", + "updated_by": "mawasthy_tenant02_admin", + } + + with self.mock_post(result) as mock_post: + channel = self.client.monitor.channel_create( + label="email channel for api change", + channel_type="email", + details=ChannelDetails( + email=EmailDetails( + recipient_type="user", + usernames=["mawasthy_tenant02_admin"], + ) + ), + ) + + assert mock_post.call_url == url + # payload should include the provided fields + assert mock_post.call_data["label"] == "email channel for api change" + assert mock_post.call_data["channel_type"] == "email" + assert "details" in mock_post.call_data + + assert isinstance(channel, AlertChannel) + assert channel.id == 123 + assert channel.label == "email channel for api change" + assert channel.channel_type == "email" + + # fetch the same response from the client and assert + resp = self.client.post(url, data={}) + assert resp["label"] == "email channel for api change" + assert resp["channel_type"] == "email" diff --git a/test/unit/objects/monitor_test.py b/test/unit/objects/monitor_test.py index 5913b3b28..2daa1486c 100644 --- a/test/unit/objects/monitor_test.py +++ b/test/unit/objects/monitor_test.py @@ -2,6 +2,7 @@ from test.unit.base import ClientBaseCase from linode_api4.objects import AlertChannel, MonitorDashboard, MonitorService +from linode_api4.objects.monitor import ChannelDetails, EmailDetails class MonitorTest(ClientBaseCase): @@ -169,3 +170,59 @@ def test_alert_channels(self): "/monitor/alert-channels/123/alerts", ) self.assertEqual(channels[0].alerts.alert_count, 0) + + def test_create_channel(self): + + create_response = { + "id": 456, + "label": "Email channel for api change", + "type": "user", + "channel_type": "email", + "details": { + "email": { + "recipient_type": "user", + "usernames": ["mawasthy_tenant02_admin"], + } + }, + "alerts": { + "url": "/monitor/alert-channels/456/alerts", + "type": "alerts-definitions", + "alert_count": 0, + }, + "created": "2024-01-01T00:00:00", + "updated": "2024-01-01T00:00:00", + "created_by": "mawasthy_tenant02_admin", + "updated_by": "mawasthy_tenant02_admin", + } + + with self.mock_post(create_response) as m: + result = self.client.monitor.channel_create( + label="Email channel for api change", + channel_type="email", + details=ChannelDetails( + email=EmailDetails( + recipient_type="user", + usernames=["mawasthy_tenant02_admin"], + ) + ), + ) + + self.assertEqual(m.call_url, "/monitor/alert-channels") + self.assertEqual(m.call_data["label"], "Email channel for api change") + self.assertEqual(m.call_data["channel_type"], "email") + self.assertEqual( + m.call_data["details"]["email"]["recipient_type"], "user" + ) + self.assertEqual( + m.call_data["details"]["email"]["usernames"], ["mawasthy_tenant02_admin"] + ) + + self.assertIsInstance(result, AlertChannel) + self.assertEqual(result.id, 456) + self.assertEqual(result.label, "Email channel for api change") + self.assertEqual(result.type, "user") + self.assertEqual(result.channel_type, "email") + self.assertIsNotNone(result.details) + self.assertIsNotNone(result.details.email) + self.assertEqual(result.details.email.recipient_type, "user") + self.assertEqual(result.details.email.usernames, ["mawasthy_tenant02_admin"]) From d33a7140adf3457579fbb075bee028dfd7517e35 Mon Sep 17 00:00:00 2001 From: mawasthy Date: Wed, 1 Jul 2026 17:08:40 +0530 Subject: [PATCH 2/9] adding delete and update channel changes --- linode_api4/groups/monitor.py | 4 + linode_api4/objects/monitor.py | 4 +- .../models/monitor/test_monitor.py | 30 ++++++-- test/unit/groups/monitor_api_test.py | 67 ++++++++++------ test/unit/objects/monitor_test.py | 76 +++++++++---------- 5 files changed, 110 insertions(+), 71 deletions(-) diff --git a/linode_api4/groups/monitor.py b/linode_api4/groups/monitor.py index 45b12cbe9..6dafc1a36 100644 --- a/linode_api4/groups/monitor.py +++ b/linode_api4/groups/monitor.py @@ -378,6 +378,10 @@ def channel_create( :returns: The newly created :class:`AlertChannel`. :rtype: AlertChannel + + .. note:: + For updating an alert channel, use the ``save()`` method on the :class:`AlertChannel` object. + For deleting an alert channel, use the ``delete()`` method directly on the :class:`AlertChannel` object. """ params = { "label": label, diff --git a/linode_api4/objects/monitor.py b/linode_api4/objects/monitor.py index 6efd23354..00fb12cc2 100644 --- a/linode_api4/objects/monitor.py +++ b/linode_api4/objects/monitor.py @@ -504,10 +504,10 @@ class AlertChannel(Base): 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), diff --git a/test/integration/models/monitor/test_monitor.py b/test/integration/models/monitor/test_monitor.py index 765bc8b21..b247bacd1 100644 --- a/test/integration/models/monitor/test_monitor.py +++ b/test/integration/models/monitor/test_monitor.py @@ -18,7 +18,11 @@ MonitorService, MonitorServiceToken, ) -from linode_api4.objects.monitor import AlertStatus, ChannelDetails, EmailDetails +from linode_api4.objects.monitor import ( + AlertStatus, + ChannelDetails, + EmailDetails, +) # List all dashboards @@ -314,12 +318,12 @@ def test_alert_definition_entities(test_linode_client): assert entity._type == service_type -def test_integration_create_get_delete_alert_channel(test_linode_client): - """E2E: create an alert channel, fetch it, then delete it. +def test_integration_create_get_update_delete_alert_channel(test_linode_client): + """E2E: create an alert channel, fetch it, update it, then delete it. This test creates an alert channel with email details, retrieves it, - and then deletes it. It ensures the feature is working end-to-end - against the actual API. + updates it, and then deletes it. It ensures the full CRUD feature is + working end-to-end against the actual API. """ client = test_linode_client label = get_test_label() + "-e2e-channel" @@ -362,6 +366,18 @@ def test_integration_create_get_delete_alert_channel(test_linode_client): assert found_channel.label == label assert found_channel.channel_type == "email" + # Update the channel label + updated_label = f"{label}-updated" + created_channel.label = updated_label + result = created_channel.save() + assert result is True, "Failed to update channel" + + # Fetch the updated channel to verify the change + reloaded_channel = client.load(AlertChannel, created_channel.id) + assert ( + reloaded_channel.label == updated_label + ), "Channel label was not updated" + finally: if created_channel: # Clean up: delete the created channel @@ -369,4 +385,6 @@ def test_integration_create_get_delete_alert_channel(test_linode_client): created_channel.delete() except Exception as e: # Log but don't fail if cleanup fails - print(f"Warning: Failed to delete channel {created_channel.id}: {e}") + print( + f"Warning: Failed to delete channel {created_channel.id}: {e}" + ) diff --git a/test/unit/groups/monitor_api_test.py b/test/unit/groups/monitor_api_test.py index fd7e1c784..48b8bd57d 100644 --- a/test/unit/groups/monitor_api_test.py +++ b/test/unit/groups/monitor_api_test.py @@ -183,54 +183,71 @@ def test_alert_definition_entities(self): assert entities[2].url == "/v4/databases/mysql/instances/3" assert entities[2]._type == "dbaas" - def test_create_channel(self): - url = "/monitor/alert-channels" - result = { - "id": 123, - "label": "email channel for api change", + def test_create_update_delete_alert_channel(self): + """ + E2E test for alert channel CRUD: create, update, and delete. + Verifies the full lifecycle of an alert channel. + """ + create_url = "/monitor/alert-channels" + channel_id = 789 + channel_url = f"{create_url}/{channel_id}" + + # Create channel + create_response = { + "id": channel_id, + "label": "Test Channel", "type": "user", "channel_type": "email", "details": { "email": { - "usernames": ["mawasthy_tenant02_admin"], + "usernames": ["test_user"], "recipient_type": "user", } }, "alerts": { - "url": "/monitor/alert-channels/123/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": "mawasthy_tenant02_admin", - "updated_by": "mawasthy_tenant02_admin", + "created_by": "test_user", + "updated_by": "test_user", } - with self.mock_post(result) as mock_post: + with self.mock_post(create_response) as mock_post: channel = self.client.monitor.channel_create( - label="email channel for api change", + label="Test Channel", channel_type="email", details=ChannelDetails( email=EmailDetails( recipient_type="user", - usernames=["mawasthy_tenant02_admin"], + usernames=["test_user"], ) ), ) - assert mock_post.call_url == url - # payload should include the provided fields - assert mock_post.call_data["label"] == "email channel for api change" - assert mock_post.call_data["channel_type"] == "email" - assert "details" in mock_post.call_data - + assert mock_post.call_url == create_url assert isinstance(channel, AlertChannel) - assert channel.id == 123 - assert channel.label == "email channel for api change" - assert channel.channel_type == "email" + assert channel.id == channel_id + assert channel.label == "Test Channel" - # fetch the same response from the client and assert - resp = self.client.post(url, data={}) - assert resp["label"] == "email channel for api change" - assert resp["channel_type"] == "email" + # Update channel + updated_response = create_response.copy() + updated_response["label"] = "Test Channel Updated" + updated_response["updated"] = "2024-01-02T00:00:00" + + with self.mock_put(updated_response) as mock_put: + channel.label = "Test Channel Updated" + result = channel.save() + + assert mock_put.call_url == channel_url + assert result is True + assert channel.label == "Test Channel Updated" + + # Delete channel + with self.mock_delete() as mock_delete: + result = channel.delete() + + assert mock_delete.call_url == channel_url + assert result is True diff --git a/test/unit/objects/monitor_test.py b/test/unit/objects/monitor_test.py index 2daa1486c..52f2b629d 100644 --- a/test/unit/objects/monitor_test.py +++ b/test/unit/objects/monitor_test.py @@ -2,7 +2,6 @@ from test.unit.base import ClientBaseCase from linode_api4.objects import AlertChannel, MonitorDashboard, MonitorService -from linode_api4.objects.monitor import ChannelDetails, EmailDetails class MonitorTest(ClientBaseCase): @@ -171,58 +170,59 @@ def test_alert_channels(self): ) self.assertEqual(channels[0].alerts.alert_count, 0) - def test_create_channel(self): - + def test_create_update_delete_channel(self): + """ + Test CRUD operations for AlertChannel: create, update, and delete. + Verifies the full lifecycle of an alert channel object. + """ + channel_id = 999 + url = f"/monitor/alert-channels/{channel_id}" + + # Create the channel create_response = { - "id": 456, - "label": "Email channel for api change", + "id": channel_id, + "label": "CRUD Test Channel", "type": "user", "channel_type": "email", "details": { "email": { + "usernames": ["crud_user"], "recipient_type": "user", - "usernames": ["mawasthy_tenant02_admin"], } }, "alerts": { - "url": "/monitor/alert-channels/456/alerts", + "url": f"{url}/alerts", "type": "alerts-definitions", "alert_count": 0, }, "created": "2024-01-01T00:00:00", "updated": "2024-01-01T00:00:00", - "created_by": "mawasthy_tenant02_admin", - "updated_by": "mawasthy_tenant02_admin", + "created_by": "crud_user", + "updated_by": "crud_user", } - with self.mock_post(create_response) as m: - result = self.client.monitor.channel_create( - label="Email channel for api change", - channel_type="email", - details=ChannelDetails( - email=EmailDetails( - recipient_type="user", - usernames=["mawasthy_tenant02_admin"], - ) - ), - ) + with self.mock_get(create_response) as m_get: + channel = self.client.load(AlertChannel, channel_id) + self.assertIsInstance(channel, AlertChannel) + self.assertEqual(channel.id, channel_id) + self.assertEqual(channel.label, "CRUD Test Channel") - self.assertEqual(m.call_url, "/monitor/alert-channels") - self.assertEqual(m.call_data["label"], "Email channel for api change") - self.assertEqual(m.call_data["channel_type"], "email") - self.assertEqual( - m.call_data["details"]["email"]["recipient_type"], "user" - ) - self.assertEqual( - m.call_data["details"]["email"]["usernames"], ["mawasthy_tenant02_admin"] - ) + # Update the channel + updated_response = create_response.copy() + updated_response["label"] = "CRUD Test Channel Updated" + updated_response["updated"] = "2024-01-02T00:00:00" + + with self.mock_put(updated_response) as m_put: + channel.label = "CRUD Test Channel Updated" + result = channel.save() + + self.assertEqual(m_put.call_url, url) + self.assertTrue(result) + self.assertEqual(channel.label, "CRUD Test Channel Updated") + + # Delete the channel + with self.mock_delete() as m_delete: + result = channel.delete() - self.assertIsInstance(result, AlertChannel) - self.assertEqual(result.id, 456) - self.assertEqual(result.label, "Email channel for api change") - self.assertEqual(result.type, "user") - self.assertEqual(result.channel_type, "email") - self.assertIsNotNone(result.details) - self.assertIsNotNone(result.details.email) - self.assertEqual(result.details.email.recipient_type, "user") - self.assertEqual(result.details.email.usernames, ["mawasthy_tenant02_admin"]) + self.assertEqual(m_delete.call_url, url) + self.assertTrue(result) From 54d4efc753695d57f650425983fc66b13c348e11 Mon Sep 17 00:00:00 2001 From: mawasthy Date: Tue, 7 Jul 2026 21:13:55 +0530 Subject: [PATCH 3/9] fixing test case to not hardcode usernames --- .../models/monitor/test_monitor.py | 12 ++++++- test/unit/groups/monitor_api_test.py | 8 ++--- test/unit/objects/monitor_test.py | 36 ++++++++++++------- 3 files changed, 39 insertions(+), 17 deletions(-) diff --git a/test/integration/models/monitor/test_monitor.py b/test/integration/models/monitor/test_monitor.py index b247bacd1..144b2e93a 100644 --- a/test/integration/models/monitor/test_monitor.py +++ b/test/integration/models/monitor/test_monitor.py @@ -332,6 +332,16 @@ def test_integration_create_get_update_delete_alert_channel(test_linode_client): created_channel = None try: + # Get valid users to use for the email alert channel + users = list(client.account.users()) + if len(users) == 0: + pytest.skip("No account users available for creating alert channels") + + # Use the first user, or first two if available + usernames = [users[0].username] + if len(users) > 1: + usernames.append(users[1].username) + # Create an alert channel with email details created_channel = client.monitor.channel_create( label=label, @@ -339,7 +349,7 @@ def test_integration_create_get_update_delete_alert_channel(test_linode_client): details=ChannelDetails( email=EmailDetails( recipient_type="user", - usernames=["mawasthy_tenant02_admin"], + usernames=usernames, ) ), ) diff --git a/test/unit/groups/monitor_api_test.py b/test/unit/groups/monitor_api_test.py index 48b8bd57d..09537ffa0 100644 --- a/test/unit/groups/monitor_api_test.py +++ b/test/unit/groups/monitor_api_test.py @@ -200,7 +200,7 @@ def test_create_update_delete_alert_channel(self): "channel_type": "email", "details": { "email": { - "usernames": ["test_user"], + "usernames": ["test_user1", "test_user2"], "recipient_type": "user", } }, @@ -211,8 +211,8 @@ def test_create_update_delete_alert_channel(self): }, "created": "2024-01-01T00:00:00", "updated": "2024-01-01T00:00:00", - "created_by": "test_user", - "updated_by": "test_user", + "created_by": "test_user1", + "updated_by": "test_user1", } with self.mock_post(create_response) as mock_post: @@ -222,7 +222,7 @@ def test_create_update_delete_alert_channel(self): details=ChannelDetails( email=EmailDetails( recipient_type="user", - usernames=["test_user"], + usernames=["test_user1", "test_user2"], ) ), ) diff --git a/test/unit/objects/monitor_test.py b/test/unit/objects/monitor_test.py index 52f2b629d..6cff23f87 100644 --- a/test/unit/objects/monitor_test.py +++ b/test/unit/objects/monitor_test.py @@ -2,6 +2,7 @@ from test.unit.base import ClientBaseCase from linode_api4.objects import AlertChannel, MonitorDashboard, MonitorService +from linode_api4.objects.monitor import ChannelDetails, EmailDetails class MonitorTest(ClientBaseCase): @@ -175,10 +176,11 @@ def test_create_update_delete_channel(self): Test CRUD operations for AlertChannel: create, update, and delete. Verifies the full lifecycle of an alert channel object. """ + create_url = "/monitor/alert-channels" channel_id = 999 - url = f"/monitor/alert-channels/{channel_id}" + channel_url = f"{create_url}/{channel_id}" - # Create the channel + # CREATE: Create the channel via channel_create() create_response = { "id": channel_id, "label": "CRUD Test Channel", @@ -186,28 +188,38 @@ def test_create_update_delete_channel(self): "channel_type": "email", "details": { "email": { - "usernames": ["crud_user"], + "usernames": ["crud_user1", "crud_user2"], "recipient_type": "user", } }, "alerts": { - "url": f"{url}/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": "crud_user", - "updated_by": "crud_user", + "created_by": "crud_user1", + "updated_by": "crud_user1", } - with self.mock_get(create_response) as m_get: - channel = self.client.load(AlertChannel, channel_id) + with self.mock_post(create_response) as m_post: + channel = self.client.monitor.channel_create( + label="CRUD Test Channel", + channel_type="email", + details=ChannelDetails( + email=EmailDetails( + recipient_type="user", + usernames=["crud_user1", "crud_user2"], + ) + ), + ) + self.assertEqual(m_post.call_url, create_url) self.assertIsInstance(channel, AlertChannel) self.assertEqual(channel.id, channel_id) self.assertEqual(channel.label, "CRUD Test Channel") - # Update the channel + # UPDATE: Update the channel label updated_response = create_response.copy() updated_response["label"] = "CRUD Test Channel Updated" updated_response["updated"] = "2024-01-02T00:00:00" @@ -216,13 +228,13 @@ def test_create_update_delete_channel(self): channel.label = "CRUD Test Channel Updated" result = channel.save() - self.assertEqual(m_put.call_url, url) + self.assertEqual(m_put.call_url, channel_url) self.assertTrue(result) self.assertEqual(channel.label, "CRUD Test Channel Updated") - # Delete the channel + # DELETE: Delete the channel with self.mock_delete() as m_delete: result = channel.delete() - self.assertEqual(m_delete.call_url, url) + self.assertEqual(m_delete.call_url, channel_url) self.assertTrue(result) From 21ef5c19bfeb10b71a1b876fdfb3e01172dd1c6c Mon Sep 17 00:00:00 2001 From: mawasthy Date: Wed, 8 Jul 2026 12:15:31 +0530 Subject: [PATCH 4/9] updating review comment fixes --- linode_api4/groups/monitor.py | 24 ++----------------- linode_api4/objects/monitor.py | 18 ++++++++++---- .../models/monitor/test_monitor.py | 6 +++-- 3 files changed, 20 insertions(+), 28 deletions(-) diff --git a/linode_api4/groups/monitor.py b/linode_api4/groups/monitor.py index 6dafc1a36..bb52ba671 100644 --- a/linode_api4/groups/monitor.py +++ b/linode_api4/groups/monitor.py @@ -347,33 +347,13 @@ def channel_create( email list) that can be associated with one or more alert definitions. Currently only ``email`` is supported as a ``channel_type``. - Example usage:: - - from linode_api4.objects.monitor import ChannelDetails, EmailDetails - - client = LinodeClient(TOKEN) - - new_channel = client.monitor.channel_create( - label="Email channel for api change", - channel_type="email", - details=ChannelDetails( - email=EmailDetails( - recipient_type="user", - usernames=["username-test"], - ) - ), - ) - - API Documentation: https://techdocs.akamai.com/linode-api/reference/post-alert-channel + API Documentation: https://techdocs.akamai.com/linode-api/reference/post-notification-channel :param label: Human-readable name for the new alert channel. :type label: str :param channel_type: The type of notification channel (e.g. ``"email"``). :type channel_type: str - :param details: Notification-type-specific configuration. Use - :class:`~linode_api4.objects.monitor.ChannelDetails` with - a nested :class:`~linode_api4.objects.monitor.EmailDetails` - for email channels. + :param details: Notification-type-specific configuration. :type details: ChannelDetails :returns: The newly created :class:`AlertChannel`. diff --git a/linode_api4/objects/monitor.py b/linode_api4/objects/monitor.py index 00fb12cc2..7b3fd3f37 100644 --- a/linode_api4/objects/monitor.py +++ b/linode_api4/objects/monitor.py @@ -8,6 +8,7 @@ __all__ = [ "AggregateFunction", "AlertChannel", + "AlertChannelType", "AlertDefinition", "AlertDefinitionChannel", "AlertDefinitionEntity", @@ -387,6 +388,15 @@ class AlertScope(StrEnum): account = "account" +class AlertChannelType(StrEnum): + """ + Type values for alert channels. + """ + + system = "system" + user = "user" + + @dataclass class AlertEntities(JSONObject): """ @@ -490,11 +500,11 @@ class AlertChannel(Base): """ Represents an alert channel used to deliver notifications when alerts fire. Alert channels define a destination and configuration for - notifications (for example: email lists, webhooks, PagerDuty, Slack, etc.). + notifications (for example: email lists, webhooks, Slack, etc.). API Documentation: - List/Get: https://techdocs.akamai.com/linode-api/reference/get-alert-channels - Create: https://techdocs.akamai.com/linode-api/reference/post-alert-channel + List/Get: https://techdocs.akamai.com/linode-api/reference/get-notification-channel + Create: https://techdocs.akamai.com/linode-api/reference/post-notification-channel This class maps to the Monitor API's ``/monitor/alert-channels`` resource and is used by the SDK to list, load, create, and inspect channels. @@ -505,7 +515,7 @@ class AlertChannel(Base): properties = { "id": Property(identifier=True), "label": Property(mutable=True), - "type": Property(), + "type": Property(AlertChannelType), "channel_type": Property(), "details": Property(mutable=True, json_object=ChannelDetails), "alerts": Property(mutable=False, json_object=AlertInfo), diff --git a/test/integration/models/monitor/test_monitor.py b/test/integration/models/monitor/test_monitor.py index 144b2e93a..ae7dc3041 100644 --- a/test/integration/models/monitor/test_monitor.py +++ b/test/integration/models/monitor/test_monitor.py @@ -335,8 +335,10 @@ def test_integration_create_get_update_delete_alert_channel(test_linode_client): # Get valid users to use for the email alert channel users = list(client.account.users()) if len(users) == 0: - pytest.skip("No account users available for creating alert channels") - + pytest.skip( + "No account users available for creating alert channels" + ) + # Use the first user, or first two if available usernames = [users[0].username] if len(users) > 1: From b7d03677043d3d5959bcc8bb2e52163d7d49ccba Mon Sep 17 00:00:00 2001 From: mawasthy Date: Mon, 20 Jul 2026 16:05:04 +0530 Subject: [PATCH 5/9] adding sdk support for get channel endpoints --- linode_api4/groups/monitor.py | 84 ++++++++++++++++ test/fixtures/monitor_alert-channels_123.json | 24 +++++ .../monitor_alert-channels_123_alerts.json | 21 ++++ ...itor_services_dbaas_alert-definitions.json | 2 +- ...ervices_dbaas_alert-definitions_12345.json | 2 +- ...tor_services_dbaas_metric-definitions.json | 2 +- .../models/monitor/test_monitor.py | 54 +++++++++++ test/unit/groups/monitor_api_test.py | 96 +++++++++++++++++++ test/unit/objects/monitor_test.py | 2 +- 9 files changed, 283 insertions(+), 4 deletions(-) create mode 100644 test/fixtures/monitor_alert-channels_123.json create mode 100644 test/fixtures/monitor_alert-channels_123_alerts.json diff --git a/linode_api4/groups/monitor.py b/linode_api4/groups/monitor.py index bb52ba671..a9aa48091 100644 --- a/linode_api4/groups/monitor.py +++ b/linode_api4/groups/monitor.py @@ -13,6 +13,7 @@ MonitorService, MonitorServiceToken, ) +from linode_api4.objects.filtering import and_ from linode_api4.objects.monitor import ChannelDetails __all__ = [ @@ -378,3 +379,86 @@ def channel_create( ) return AlertChannel(self.client, result["id"], result) + + def alert_channel(self, channel_id: int) -> AlertChannel: + """ + Retrieve a specific notification channel definition details by its channel ID. + + Returns an :class:`AlertChannel` object for the specified channel ID. + The channel object contains all configuration details for the notification + destination (e.g., email lists, webhooks, etc.). + + .. note:: This endpoint is in beta and requires using the v4beta base URL. + + API Documentation: https://techdocs.akamai.com/linode-api/reference/get-notification-channel + + :param channel_id: The ID of the alert channel to retrieve. + :type channel_id: int + + :returns: The requested :class:`AlertChannel` object. + :rtype: AlertChannel + :raises ApiError: if the requested channel could not be loaded. + """ + return self.client.load(AlertChannel, channel_id) + + def alert_channel_alerts(self, channel_id: int, *filters) -> PaginatedList: + """ + Retrieve all alerts associated with a specific alert channel. + + Returns a paginated collection of alert definitions associated with the + specified alert channel. This allows you to see which alert definitions + are configured to notify this specific channel. + + .. note:: This endpoint is in beta and requires using the v4beta base URL. + + API Documentation: https://techdocs.akamai.com/linode-api/reference/get-notification-channel-alerts + + :param channel_id: The ID of the alert channel to retrieve alerts for. + :type channel_id: int + :param filters: Optional filter expressions to apply to the collection. + See :doc:`Filtering Collections` for details. + + :returns: A paginated list of alert definitions associated with this channel. + :rtype: PaginatedList[AlertDefinition] + """ + endpoint = f"/monitor/alert-channels/{channel_id}/alerts" + parsed_filters = None + if filters: + if len(filters) > 1: + parsed_filters = and_( + *filters + ).dct # pylint: disable=no-value-for-parameter + else: + parsed_filters = filters[0].dct + + response_json = self.client.get(endpoint, filters=parsed_filters) + + if "data" not in response_json: + raise UnexpectedResponseError( + "Unexpected response when retrieving alert channel alerts!", + json=response_json, + ) + + # For each alert definition in the response, extract the service_type + # and use it as the parent_id when creating AlertDefinition objects + result = [] + for obj in response_json.get("data", []): + if "id" in obj and "service_type" in obj: + alert = AlertDefinition.make_instance( + obj["id"], + self.client, + parent_id=obj["service_type"], + json=obj, + ) + result.append(alert) + + # Return paginated list with pagination metadata from response + return PaginatedList( + self.client, + endpoint[1:], + page=result, + max_pages=response_json.get("pages", 1), + total_items=response_json.get("results", len(result)), + parent_id=None, + filters=parsed_filters, + ) 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 67ea9d2ab..1c70368a3 100644 --- a/test/fixtures/monitor_services_dbaas_alert-definitions.json +++ b/test/fixtures/monitor_services_dbaas_alert-definitions.json @@ -38,7 +38,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 4d70f66b1..affb48228 100644 --- a/test/fixtures/monitor_services_dbaas_alert-definitions_12345.json +++ b/test/fixtures/monitor_services_dbaas_alert-definitions_12345.json @@ -36,7 +36,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 ae7dc3041..8c2cd01fe 100644 --- a/test/integration/models/monitor/test_monitor.py +++ b/test/integration/models/monitor/test_monitor.py @@ -400,3 +400,57 @@ def test_integration_create_get_update_delete_alert_channel(test_linode_client): print( f"Warning: Failed to delete channel {created_channel.id}: {e}" ) + + +def test_integration_alert_channel(test_linode_client): + """Test retrieving a single alert channel by ID. + + This test fetches an existing alert channel and verifies that all + expected properties are populated correctly. + """ + client = test_linode_client + + # Get an existing alert channel to test with + channels = list(client.monitor.alert_channels()) + if len(channels) == 0: + pytest.skip("No alert channels available on account for testing") + + channel_id = channels[0].id + + # Test the alert_channel() method + fetched_channel = client.monitor.alert_channel(channel_id) + + assert isinstance(fetched_channel, AlertChannel) + assert fetched_channel.id == channel_id + assert fetched_channel.label is not None + assert fetched_channel.channel_type is not None + assert fetched_channel.details is not None + + +def test_integration_alert_channel_alerts(test_linode_client): + """Test retrieving alerts associated with a specific alert channel. + + This test fetches alerts for an existing alert channel and verifies + the paginated list of alert definitions is returned correctly. + """ + client = test_linode_client + + # Get an existing alert channel to test with + channels = list(client.monitor.alert_channels()) + if len(channels) == 0: + pytest.skip("No alert channels available on account for testing") + + channel_id = channels[0].id + + # Test the alert_channel_alerts() method + alerts = client.monitor.alert_channel_alerts(channel_id) + + assert isinstance(alerts, PaginatedList) + + # If there are alerts, verify their structure + if len(alerts) > 0: + alert = alerts[0] + assert isinstance(alert, AlertDefinition) + assert alert.id is not None + assert alert.label is not None + assert alert.service_type is not None diff --git a/test/unit/groups/monitor_api_test.py b/test/unit/groups/monitor_api_test.py index 09537ffa0..3f7a8caa1 100644 --- a/test/unit/groups/monitor_api_test.py +++ b/test/unit/groups/monitor_api_test.py @@ -251,3 +251,99 @@ def test_create_update_delete_alert_channel(self): assert mock_delete.call_url == channel_url assert result is True + + def test_alert_channel(self): + """ + Test retrieval of a specific alert channel by ID. + Verifies the alert_channel method returns a single AlertChannel object. + """ + channel_id = 123 + channel_url = f"/monitor/alert-channels/{channel_id}" + + channel_response = { + "id": channel_id, + "label": "alert notification channel", + "type": "user", + "channel_type": "email", + "details": { + "email": { + "usernames": ["admin-user1", "admin-user2"], + "recipient_type": "user", + } + }, + "alerts": { + "url": f"{channel_url}/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", + } + + with self.mock_get(channel_response) as mock_get: + channel = self.client.monitor.alert_channel(channel_id=channel_id) + + assert mock_get.call_url == channel_url + assert isinstance(channel, AlertChannel) + assert channel.id == channel_id + assert channel.label == "alert notification channel" + assert channel.channel_type == "email" + assert channel.details.email.usernames == [ + "admin-user1", + "admin-user2", + ] + assert channel.alerts.alert_count == 2 + + def test_alert_channel_alerts(self): + """ + Test retrieval of alerts associated with a specific alert channel. + Verifies the alert_channel_alerts method returns a paginated list + of AlertDefinition objects associated with the channel. + """ + channel_id = 123 + alerts_url = f"/monitor/alert-channels/{channel_id}/alerts" + + alerts_response = { + "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, + } + + with self.mock_get(alerts_response) as mock_get: + alerts = self.client.monitor.alert_channel_alerts( + channel_id=channel_id + ) + + assert mock_get.call_url == alerts_url + assert isinstance(alerts, PaginatedList) + assert len(alerts) == 2 + + # Verify first alert + assert isinstance(alerts[0], AlertDefinition) + assert alerts[0].id == 12345 + assert alerts[0].label == "DBAAS Alert 1" + assert alerts[0].service_type == "dbaas" + + # Verify second alert + assert isinstance(alerts[1], AlertDefinition) + assert alerts[1].id == 12346 + assert alerts[1].label == "DBAAS Alert 2" + assert alerts[1].service_type == "dbaas" diff --git a/test/unit/objects/monitor_test.py b/test/unit/objects/monitor_test.py index 6cff23f87..0df086af0 100644 --- a/test/unit/objects/monitor_test.py +++ b/test/unit/objects/monitor_test.py @@ -127,7 +127,7 @@ def test_metric_definitions(self): self.assertEqual(metrics[0].metric, "cpu_usage") self.assertEqual(metrics[0].metric_type, "gauge") self.assertEqual(metrics[0].scrape_interval, "60s") - self.assertEqual(metrics[0].unit, "percent") + self.assertEqual(metrics[0].unit, "%") self.assertEqual(metrics[0].dimensions[0].dimension_label, "node_type") self.assertEqual(metrics[0].dimensions[0].label, "Node Type") self.assertEqual( From 98ba8684446403f6dd7aec25d44c3e70f788da34 Mon Sep 17 00:00:00 2001 From: mawasthy Date: Wed, 22 Jul 2026 16:27:20 +0530 Subject: [PATCH 6/9] resolving review comments --- linode_api4/groups/monitor.py | 62 ++++++----------- .../models/monitor/test_monitor.py | 66 +++++++++---------- test/unit/groups/monitor_api_test.py | 44 ------------- 3 files changed, 49 insertions(+), 123 deletions(-) diff --git a/linode_api4/groups/monitor.py b/linode_api4/groups/monitor.py index 957258741..a74f30983 100644 --- a/linode_api4/groups/monitor.py +++ b/linode_api4/groups/monitor.py @@ -451,6 +451,8 @@ def channel_create( :rtype: AlertChannel .. note:: + If you need to obtain a single :class:`AlertChannel`, use :meth:`LinodeClient.load`. + Example: ``client.load(AlertChannel, channel_id)``. For updating an alert channel, use the ``save()`` method on the :class:`AlertChannel` object. For deleting an alert channel, use the ``delete()`` method directly on the :class:`AlertChannel` object. """ @@ -470,27 +472,6 @@ def channel_create( return AlertChannel(self.client, result["id"], result) - def alert_channel(self, channel_id: int) -> AlertChannel: - """ - Retrieve a specific notification channel definition details by its channel ID. - - Returns an :class:`AlertChannel` object for the specified channel ID. - The channel object contains all configuration details for the notification - destination (e.g., email lists, webhooks, etc.). - - .. note:: This endpoint is in beta and requires using the v4beta base URL. - - API Documentation: https://techdocs.akamai.com/linode-api/reference/get-notification-channel - - :param channel_id: The ID of the alert channel to retrieve. - :type channel_id: int - - :returns: The requested :class:`AlertChannel` object. - :rtype: AlertChannel - :raises ApiError: if the requested channel could not be loaded. - """ - return self.client.load(AlertChannel, channel_id) - def alert_channel_alerts(self, channel_id: int, *filters) -> PaginatedList: """ Retrieve all alerts associated with a specific alert channel. @@ -499,8 +480,6 @@ def alert_channel_alerts(self, channel_id: int, *filters) -> PaginatedList: specified alert channel. This allows you to see which alert definitions are configured to notify this specific channel. - .. note:: This endpoint is in beta and requires using the v4beta base URL. - API Documentation: https://techdocs.akamai.com/linode-api/reference/get-notification-channel-alerts :param channel_id: The ID of the alert channel to retrieve alerts for. @@ -512,14 +491,13 @@ def alert_channel_alerts(self, channel_id: int, *filters) -> PaginatedList: :rtype: PaginatedList[AlertDefinition] """ endpoint = f"/monitor/alert-channels/{channel_id}/alerts" + + # Build filter dict if filters provided parsed_filters = None if filters: - if len(filters) > 1: - parsed_filters = and_( - *filters - ).dct # pylint: disable=no-value-for-parameter - else: - parsed_filters = filters[0].dct + parsed_filters = ( + and_(*filters).dct if len(filters) > 1 else filters[0].dct + ) response_json = self.client.get(endpoint, filters=parsed_filters) @@ -529,20 +507,18 @@ def alert_channel_alerts(self, channel_id: int, *filters) -> PaginatedList: json=response_json, ) - # For each alert definition in the response, extract the service_type - # and use it as the parent_id when creating AlertDefinition objects - result = [] - for obj in response_json.get("data", []): - if "id" in obj and "service_type" in obj: - alert = AlertDefinition.make_instance( - obj["id"], - self.client, - parent_id=obj["service_type"], - json=obj, - ) - result.append(alert) - - # Return paginated list with pagination metadata from response + # Create AlertDefinition objects with proper parent_id (service_type) + result = [ + AlertDefinition.make_instance( + obj["id"], + self.client, + parent_id=obj["service_type"], + json=obj, + ) + for obj in response_json.get("data", []) + if "id" in obj and "service_type" in obj + ] + return PaginatedList( self.client, endpoint[1:], diff --git a/test/integration/models/monitor/test_monitor.py b/test/integration/models/monitor/test_monitor.py index 33cdc979a..f3d7000c4 100644 --- a/test/integration/models/monitor/test_monitor.py +++ b/test/integration/models/monitor/test_monitor.py @@ -245,25 +245,36 @@ def test_integration_create_get_update_delete_alert_definition( label = f"{label}-{int(time.time())}" description = "E2E alert created by SDK integration test" - # Pick an existing alert channel to attach to the definition; skip if none - channels = list( - client.monitor.alert_channels() - ) # TODO: create channel instead of relying on pre-existing one - if not channels: - pytest.skip( - "No alert channels available on account for creating alert definitions" - ) + # Get valid users to create an alert channel for the alert definition + users = list(client.account.users()) + if len(users) == 0: + pytest.skip("No account users available for creating alert channels") + + # Use the first user for the alert channel + usernames = [users[0].username] created = None + created_channel = None try: + # Create a new alert channel for this test + created_channel = client.monitor.channel_create( + label=f"{get_test_label()}-channel-{int(time.time())}", + channel_type="email", + details=ChannelDetails( + email=EmailDetails( + recipient_type="user", + usernames=usernames, + ) + ), + ) # Create the alert definition using API-compliant top-level fields created = client.monitor.create_alert_definition( service_type=service_type, label=label, severity=1, description=description, - channel_ids=[channels[0].id], + channel_ids=[created_channel.id], rule_criteria=rule_criteria, trigger_conditions=trigger_conditions, ) @@ -293,6 +304,15 @@ def test_integration_create_get_update_delete_alert_definition( AlertDefinition, created.id, service_type ) delete_alert.delete() + if created_channel: + # Clean up the created channel + try: + created_channel.delete() + except Exception as e: + # Log but don't fail if cleanup fails + print( + f"Warning: Failed to delete channel {created_channel.id}: {e}" + ) def test_alert_definition_entities(test_linode_client): @@ -337,8 +357,7 @@ def test_integration_create_get_update_delete_alert_channel(test_linode_client): working end-to-end against the actual API. """ client = test_linode_client - label = get_test_label() + "-e2e-channel" - label = f"{label}-{int(time.time())}" + label = "pythonsdk-alert-channel-test" created_channel = None @@ -413,31 +432,6 @@ def test_integration_create_get_update_delete_alert_channel(test_linode_client): ) -def test_integration_alert_channel(test_linode_client): - """Test retrieving a single alert channel by ID. - - This test fetches an existing alert channel and verifies that all - expected properties are populated correctly. - """ - client = test_linode_client - - # Get an existing alert channel to test with - channels = list(client.monitor.alert_channels()) - if len(channels) == 0: - pytest.skip("No alert channels available on account for testing") - - channel_id = channels[0].id - - # Test the alert_channel() method - fetched_channel = client.monitor.alert_channel(channel_id) - - assert isinstance(fetched_channel, AlertChannel) - assert fetched_channel.id == channel_id - assert fetched_channel.label is not None - assert fetched_channel.channel_type is not None - assert fetched_channel.details is not None - - def test_integration_alert_channel_alerts(test_linode_client): """Test retrieving alerts associated with a specific alert channel. diff --git a/test/unit/groups/monitor_api_test.py b/test/unit/groups/monitor_api_test.py index 516910f51..1777013d8 100644 --- a/test/unit/groups/monitor_api_test.py +++ b/test/unit/groups/monitor_api_test.py @@ -258,50 +258,6 @@ def test_create_update_delete_alert_channel(self): assert mock_delete.call_url == channel_url assert result is True - def test_alert_channel(self): - """ - Test retrieval of a specific alert channel by ID. - Verifies the alert_channel method returns a single AlertChannel object. - """ - channel_id = 123 - channel_url = f"/monitor/alert-channels/{channel_id}" - - channel_response = { - "id": channel_id, - "label": "alert notification channel", - "type": "user", - "channel_type": "email", - "details": { - "email": { - "usernames": ["admin-user1", "admin-user2"], - "recipient_type": "user", - } - }, - "alerts": { - "url": f"{channel_url}/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", - } - - with self.mock_get(channel_response) as mock_get: - channel = self.client.monitor.alert_channel(channel_id=channel_id) - - assert mock_get.call_url == channel_url - assert isinstance(channel, AlertChannel) - assert channel.id == channel_id - assert channel.label == "alert notification channel" - assert channel.channel_type == "email" - assert channel.details.email.usernames == [ - "admin-user1", - "admin-user2", - ] - assert channel.alerts.alert_count == 2 - def test_alert_channel_alerts(self): """ Test retrieval of alerts associated with a specific alert channel. From c22db1abb8c561783001b64a36c4496fbcd0134f Mon Sep 17 00:00:00 2001 From: mawasthy Date: Tue, 28 Jul 2026 11:40:26 +0530 Subject: [PATCH 7/9] updating clone alert integration test --- .../models/monitor/test_monitor.py | 38 +++++++++++++++---- 1 file changed, 31 insertions(+), 7 deletions(-) diff --git a/test/integration/models/monitor/test_monitor.py b/test/integration/models/monitor/test_monitor.py index f3d7000c4..276a3e801 100644 --- a/test/integration/models/monitor/test_monitor.py +++ b/test/integration/models/monitor/test_monitor.py @@ -495,18 +495,32 @@ def test_integration_clone_alert_definition(test_linode_client): "trigger_occurrences": 1, } - channels = list( - client.monitor.alert_channels() - ) # TODO: create channel instead of relying on pre-existing one - if not channels: - pytest.skip( - "No alert channels available on account for creating/cloning alert definitions" - ) + # Get valid users to create an alert channel + users = list(client.account.users()) + if len(users) == 0: + pytest.skip("No account users available for creating alert channels") + + # Use the first user for the alert channel + usernames = [users[0].username] created = None cloned_alert = None + created_channel = None try: + # Create a new alert channel for this test + created_channel = client.monitor.channel_create( + label=f"{get_test_label()}-channel-{int(time.time())}", + channel_type="email", + details=ChannelDetails( + email=EmailDetails( + recipient_type="user", + usernames=usernames, + ) + ), + ) + channels = [created_channel] + # Create the source alert definition created = client.monitor.create_alert_definition( service_type=service_type, @@ -566,3 +580,13 @@ def test_integration_clone_alert_definition(test_linode_client): AlertDefinition, created.id, service_type ) delete_source_alert.delete() + + if created_channel: + # Clean up the created channel + try: + created_channel.delete() + except Exception as e: + # Log but don't fail if cleanup fails + print( + f"Warning: Failed to delete channel {created_channel.id}: {e}" + ) From 006bf41b1b08e11932677d79c12574afe270de2a Mon Sep 17 00:00:00 2001 From: mawasthy Date: Tue, 11 Aug 2026 00:46:05 +0530 Subject: [PATCH 8/9] updating code as per review comments --- linode_api4/objects/monitor.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/linode_api4/objects/monitor.py b/linode_api4/objects/monitor.py index 38cadc538..ed72e22de 100644 --- a/linode_api4/objects/monitor.py +++ b/linode_api4/objects/monitor.py @@ -505,7 +505,7 @@ class AlertDefinition(DerivedBase): "entity_ids": Property(mutable=True), "description": Property(mutable=True), "service_class": Property(alias_of="class"), - "scope": Property(AlertScope), + "scope": Property(), "regions": Property(mutable=True), "entities": Property(json_object=AlertEntities), "channel_ids": Property(mutable=True), @@ -566,7 +566,7 @@ class AlertChannel(Base): properties = { "id": Property(identifier=True), "label": Property(mutable=True), - "type": Property(AlertChannelType), + "type": Property(), "channel_type": Property(), "details": Property(mutable=True, json_object=ChannelDetails), "alerts": Property(mutable=False, json_object=AlertInfo), From 5eea1dce5ea5599eb233ad8b7feb96655c4d39cd Mon Sep 17 00:00:00 2001 From: mawasthy Date: Fri, 14 Aug 2026 13:38:56 +0530 Subject: [PATCH 9/9] feat: add webhook channel support for alerts --- linode_api4/groups/monitor.py | 297 +++++++++++------- linode_api4/objects/monitor.py | 172 ++++++---- .../models/monitor/test_monitor.py | 256 +++++---------- test/unit/groups/monitor_api_test.py | 278 +++++++++------- test/unit/objects/monitor_test.py | 204 +++++++++--- 5 files changed, 711 insertions(+), 496 deletions(-) diff --git a/linode_api4/groups/monitor.py b/linode_api4/groups/monitor.py index a74f30983..816794b3f 100644 --- a/linode_api4/groups/monitor.py +++ b/linode_api4/groups/monitor.py @@ -18,12 +18,16 @@ MonitorService, MonitorServiceToken, ) -from linode_api4.objects.filtering import and_ from linode_api4.objects.monitor import ( AkamaiObjectStorageLogsDestinationDetails, + BasicAuthenticationDetails, ChannelDetails, + CustomHeader, CustomHTTPSLogsDestinationDetails, + DestinationAuthentication, + EmailDetails, LogsStreamDetails, + WebhookDetails, ) __all__ = [ @@ -215,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, @@ -425,110 +616,6 @@ def alert_definition_entities( endpoint=endpoint, ) - def channel_create( - self, - label: str, - channel_type: str, - details: ChannelDetails, - ) -> AlertChannel: - """ - Creates a new alert channel for the authenticated account. - - An alert channel defines a notification destination (for example: an - email list) that can be associated with one or more alert definitions. - Currently only ``email`` is supported as a ``channel_type``. - - API Documentation: https://techdocs.akamai.com/linode-api/reference/post-notification-channel - - :param label: Human-readable name for the new alert channel. - :type label: str - :param channel_type: The type of notification channel (e.g. ``"email"``). - :type channel_type: str - :param details: Notification-type-specific configuration. - :type details: ChannelDetails - - :returns: The newly created :class:`AlertChannel`. - :rtype: AlertChannel - - .. note:: - If you need to obtain a single :class:`AlertChannel`, use :meth:`LinodeClient.load`. - Example: ``client.load(AlertChannel, channel_id)``. - For updating an alert channel, use the ``save()`` method on the :class:`AlertChannel` object. - For deleting an alert channel, use the ``delete()`` method directly on the :class:`AlertChannel` object. - """ - params = { - "label": label, - "channel_type": channel_type, - "details": details.dict, - } - - 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 alert_channel_alerts(self, channel_id: int, *filters) -> PaginatedList: - """ - Retrieve all alerts associated with a specific alert channel. - - Returns a paginated collection of alert definitions associated with the - specified alert channel. This allows you to see which alert definitions - are configured to notify this specific channel. - - API Documentation: https://techdocs.akamai.com/linode-api/reference/get-notification-channel-alerts - - :param channel_id: The ID of the alert channel to retrieve alerts for. - :type channel_id: int - :param filters: Optional filter expressions to apply to the collection. - See :doc:`Filtering Collections` for details. - - :returns: A paginated list of alert definitions associated with this channel. - :rtype: PaginatedList[AlertDefinition] - """ - endpoint = f"/monitor/alert-channels/{channel_id}/alerts" - - # Build filter dict if filters provided - parsed_filters = None - if filters: - parsed_filters = ( - and_(*filters).dct if len(filters) > 1 else filters[0].dct - ) - - response_json = self.client.get(endpoint, filters=parsed_filters) - - if "data" not in response_json: - raise UnexpectedResponseError( - "Unexpected response when retrieving alert channel alerts!", - json=response_json, - ) - - # Create AlertDefinition objects with proper parent_id (service_type) - result = [ - AlertDefinition.make_instance( - obj["id"], - self.client, - parent_id=obj["service_type"], - json=obj, - ) - for obj in response_json.get("data", []) - if "id" in obj and "service_type" in obj - ] - - return PaginatedList( - self.client, - endpoint[1:], - page=result, - max_pages=response_json.get("pages", 1), - total_items=response_json.get("results", len(result)), - parent_id=None, - filters=parsed_filters, - ) - def destinations(self, *filters) -> PaginatedList: """ List available logs destinations. diff --git a/linode_api4/objects/monitor.py b/linode_api4/objects/monitor.py index ed72e22de..3ff21fba0 100644 --- a/linode_api4/objects/monitor.py +++ b/linode_api4/objects/monitor.py @@ -8,19 +8,21 @@ __all__ = [ "AggregateFunction", "AlertChannel", - "AlertChannelType", "AlertDefinition", "AlertDefinitionChannel", "AlertDefinitionEntity", "AlertEntities", "AlertScope", "AlertType", + "ChannelDetails", + "EmailDetails", "MonitorDashboard", "MonitorMetricsDefinition", "MonitorService", "MonitorServiceToken", "RuleCriteria", "TriggerConditions", + "WebhookDetails", "AkamaiObjectStorageLogsDestinationDetails", "AuthenticationType", "BasicAuthenticationDetails", @@ -438,15 +440,6 @@ class AlertScope(StrEnum): account = "account" -class AlertChannelType(StrEnum): - """ - Type values for alert channels. - """ - - system = "system" - user = "user" - - @dataclass class AlertEntities(JSONObject): """ @@ -505,7 +498,7 @@ class AlertDefinition(DerivedBase): "entity_ids": Property(mutable=True), "description": Property(mutable=True), "service_class": Property(alias_of="class"), - "scope": Property(), + "scope": Property(AlertScope), "regions": Property(mutable=True), "entities": Property(json_object=AlertEntities), "channel_ids": Property(mutable=True), @@ -513,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): """ @@ -523,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 @@ -551,14 +608,53 @@ class AlertChannel(Base): """ Represents an alert channel used to deliver notifications when alerts fire. Alert channels define a destination and configuration for - notifications (for example: email lists, webhooks, Slack, etc.). + notifications (for example: email lists, webhooks, PagerDuty, Slack, etc.). + + 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 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" + ) + ) + ) + ) + ) - API Documentation: - List/Get: https://techdocs.akamai.com/linode-api/reference/get-notification-channel - Create: https://techdocs.akamai.com/linode-api/reference/post-notification-channel + # Update channel + channel.label = "Updated Label" + channel.save() - This class maps to the Monitor API's ``/monitor/alert-channels`` resource - and is used by the SDK to list, load, create, and inspect channels. + # Delete channel + channel.delete() """ api_endpoint = "/monitor/alert-channels/{id}" @@ -577,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/integration/models/monitor/test_monitor.py b/test/integration/models/monitor/test_monitor.py index 276a3e801..4b1acd0ce 100644 --- a/test/integration/models/monitor/test_monitor.py +++ b/test/integration/models/monitor/test_monitor.py @@ -9,7 +9,6 @@ from linode_api4 import LinodeClient, PaginatedList from linode_api4.objects import ( - AlertChannel, AlertDefinition, AlertDefinitionEntity, ApiError, @@ -19,9 +18,13 @@ MonitorServiceToken, ) from linode_api4.objects.monitor import ( + AlertChannel, AlertStatus, + BasicAuthenticationDetails, ChannelDetails, - EmailDetails, + CustomHeader, + DestinationAuthentication, + WebhookDetails, ) @@ -245,36 +248,25 @@ def test_integration_create_get_update_delete_alert_definition( label = f"{label}-{int(time.time())}" description = "E2E alert created by SDK integration test" - # Get valid users to create an alert channel for the alert definition - users = list(client.account.users()) - if len(users) == 0: - pytest.skip("No account users available for creating alert channels") - - # Use the first user for the alert channel - usernames = [users[0].username] + # Pick an existing alert channel to attach to the definition; skip if none + channels = list( + client.monitor.alert_channels() + ) # TODO: create channel instead of relying on pre-existing one + if not channels: + pytest.skip( + "No alert channels available on account for creating alert definitions" + ) created = None - created_channel = None try: - # Create a new alert channel for this test - created_channel = client.monitor.channel_create( - label=f"{get_test_label()}-channel-{int(time.time())}", - channel_type="email", - details=ChannelDetails( - email=EmailDetails( - recipient_type="user", - usernames=usernames, - ) - ), - ) # Create the alert definition using API-compliant top-level fields created = client.monitor.create_alert_definition( service_type=service_type, label=label, severity=1, description=description, - channel_ids=[created_channel.id], + channel_ids=[channels[0].id], rule_criteria=rule_criteria, trigger_conditions=trigger_conditions, ) @@ -304,15 +296,6 @@ def test_integration_create_get_update_delete_alert_definition( AlertDefinition, created.id, service_type ) delete_alert.delete() - if created_channel: - # Clean up the created channel - try: - created_channel.delete() - except Exception as e: - # Log but don't fail if cleanup fails - print( - f"Warning: Failed to delete channel {created_channel.id}: {e}" - ) def test_alert_definition_entities(test_linode_client): @@ -349,118 +332,6 @@ def test_alert_definition_entities(test_linode_client): assert entity._type == service_type -def test_integration_create_get_update_delete_alert_channel(test_linode_client): - """E2E: create an alert channel, fetch it, update it, then delete it. - - This test creates an alert channel with email details, retrieves it, - updates it, and then deletes it. It ensures the full CRUD feature is - working end-to-end against the actual API. - """ - client = test_linode_client - label = "pythonsdk-alert-channel-test" - - created_channel = None - - try: - # Get valid users to use for the email alert channel - users = list(client.account.users()) - if len(users) == 0: - pytest.skip( - "No account users available for creating alert channels" - ) - - # Use the first user, or first two if available - usernames = [users[0].username] - if len(users) > 1: - usernames.append(users[1].username) - - # Create an alert channel with email details - created_channel = client.monitor.channel_create( - label=label, - channel_type="email", - details=ChannelDetails( - email=EmailDetails( - recipient_type="user", - usernames=usernames, - ) - ), - ) - - # Assert the created channel has expected properties - assert isinstance(created_channel, AlertChannel) - assert created_channel.id is not None - assert created_channel.label == label - assert created_channel.channel_type == "email" - assert created_channel.details is not None - - # Fetch the channel to verify it exists - channels = list(client.monitor.alert_channels()) - assert len(channels) > 0, "No channels found after creation" - - # Find the created channel in the list - found_channel = None - for ch in channels: - if ch.id == created_channel.id: - found_channel = ch - break - - assert found_channel is not None, "Created channel not found in list" - assert found_channel.label == label - assert found_channel.channel_type == "email" - - # Update the channel label - updated_label = f"{label}-updated" - created_channel.label = updated_label - result = created_channel.save() - assert result is True, "Failed to update channel" - - # Fetch the updated channel to verify the change - reloaded_channel = client.load(AlertChannel, created_channel.id) - assert ( - reloaded_channel.label == updated_label - ), "Channel label was not updated" - - finally: - if created_channel: - # Clean up: delete the created channel - try: - created_channel.delete() - except Exception as e: - # Log but don't fail if cleanup fails - print( - f"Warning: Failed to delete channel {created_channel.id}: {e}" - ) - - -def test_integration_alert_channel_alerts(test_linode_client): - """Test retrieving alerts associated with a specific alert channel. - - This test fetches alerts for an existing alert channel and verifies - the paginated list of alert definitions is returned correctly. - """ - client = test_linode_client - - # Get an existing alert channel to test with - channels = list(client.monitor.alert_channels()) - if len(channels) == 0: - pytest.skip("No alert channels available on account for testing") - - channel_id = channels[0].id - - # Test the alert_channel_alerts() method - alerts = client.monitor.alert_channel_alerts(channel_id) - - assert isinstance(alerts, PaginatedList) - - # If there are alerts, verify their structure - if len(alerts) > 0: - alert = alerts[0] - assert isinstance(alert, AlertDefinition) - assert alert.id is not None - assert alert.label is not None - assert alert.service_type is not None - - def test_integration_clone_alert_definition(test_linode_client): """E2E: create a source alert definition, clone it, then delete both.""" client = test_linode_client @@ -495,32 +366,18 @@ def test_integration_clone_alert_definition(test_linode_client): "trigger_occurrences": 1, } - # Get valid users to create an alert channel - users = list(client.account.users()) - if len(users) == 0: - pytest.skip("No account users available for creating alert channels") - - # Use the first user for the alert channel - usernames = [users[0].username] + channels = list( + client.monitor.alert_channels() + ) # TODO: create channel instead of relying on pre-existing one + if not channels: + pytest.skip( + "No alert channels available on account for creating/cloning alert definitions" + ) created = None cloned_alert = None - created_channel = None try: - # Create a new alert channel for this test - created_channel = client.monitor.channel_create( - label=f"{get_test_label()}-channel-{int(time.time())}", - channel_type="email", - details=ChannelDetails( - email=EmailDetails( - recipient_type="user", - usernames=usernames, - ) - ), - ) - channels = [created_channel] - # Create the source alert definition created = client.monitor.create_alert_definition( service_type=service_type, @@ -581,12 +438,65 @@ def test_integration_clone_alert_definition(test_linode_client): ) delete_source_alert.delete() - if created_channel: - # Clean up the created channel - try: - created_channel.delete() - except Exception as e: - # Log but don't fail if cleanup fails - print( - f"Warning: Failed to delete channel {created_channel.id}: {e}" - ) + +# 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 1777013d8..48c89b914 100644 --- a/test/unit/groups/monitor_api_test.py +++ b/test/unit/groups/monitor_api_test.py @@ -9,7 +9,14 @@ AlertDefinitionEntity, EntityMetricOptions, ) -from linode_api4.objects.monitor import ChannelDetails, EmailDetails +from linode_api4.objects.monitor import ( + BasicAuthenticationDetails, + ChannelDetails, + CustomHeader, + DestinationAuthentication, + EmailDetails, + WebhookDetails, +) class MonitorAPITest(MonitorClientBaseCase): @@ -189,127 +196,6 @@ def test_alert_definition_entities(self): assert entities[2].url == "/v4/databases/mysql/instances/3" assert entities[2]._type == "dbaas" - def test_create_update_delete_alert_channel(self): - """ - E2E test for alert channel CRUD: create, update, and delete. - Verifies the full lifecycle of an alert channel. - """ - create_url = "/monitor/alert-channels" - channel_id = 789 - channel_url = f"{create_url}/{channel_id}" - - # Create channel - create_response = { - "id": channel_id, - "label": "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="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 == "Test Channel" - - # Update channel - updated_response = create_response.copy() - updated_response["label"] = "Test Channel Updated" - updated_response["updated"] = "2024-01-02T00:00:00" - - with self.mock_put(updated_response) as mock_put: - channel.label = "Test Channel Updated" - result = channel.save() - - assert mock_put.call_url == channel_url - assert result is True - assert channel.label == "Test Channel Updated" - - # Delete channel - with self.mock_delete() as mock_delete: - result = channel.delete() - - assert mock_delete.call_url == channel_url - assert result is True - - def test_alert_channel_alerts(self): - """ - Test retrieval of alerts associated with a specific alert channel. - Verifies the alert_channel_alerts method returns a paginated list - of AlertDefinition objects associated with the channel. - """ - channel_id = 123 - alerts_url = f"/monitor/alert-channels/{channel_id}/alerts" - - alerts_response = { - "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, - } - - with self.mock_get(alerts_response) as mock_get: - alerts = self.client.monitor.alert_channel_alerts( - channel_id=channel_id - ) - - assert mock_get.call_url == alerts_url - assert isinstance(alerts, PaginatedList) - assert len(alerts) == 2 - - # Verify first alert - assert isinstance(alerts[0], AlertDefinition) - assert alerts[0].id == 12345 - assert alerts[0].label == "DBAAS Alert 1" - assert alerts[0].service_type == "dbaas" - - # Verify second alert - assert isinstance(alerts[1], AlertDefinition) - assert alerts[1].id == 12346 - assert alerts[1].label == "DBAAS Alert 2" - assert alerts[1].service_type == "dbaas" - def test_clone_alert_definition(self): service_type = "dbaas" source_id = 12345 @@ -379,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 b6534fe72..6e6c80450 100644 --- a/test/unit/objects/monitor_test.py +++ b/test/unit/objects/monitor_test.py @@ -12,13 +12,16 @@ ) from linode_api4.objects.monitor import ( AkamaiObjectStorageLogsDestinationDetails, + BasicAuthenticationDetails, ChannelDetails, + ClientCertificateDetails, + CustomHeader, CustomHTTPSLogsDestinationDetails, DestinationAuthentication, - EmailDetails, LogsDestinationDetailsBase, LogsStreamDetails, LogsStreamType, + WebhookDetails, ) @@ -148,7 +151,7 @@ def test_metric_definitions(self): self.assertEqual(metrics[0].metric, "cpu_usage") self.assertEqual(metrics[0].metric_type, "gauge") self.assertEqual(metrics[0].scrape_interval, "60s") - self.assertEqual(metrics[0].unit, "%") + self.assertEqual(metrics[0].unit, "percent") self.assertEqual(metrics[0].dimensions[0].dimension_label, "node_type") self.assertEqual(metrics[0].dimensions[0].label, "Node Type") self.assertEqual( @@ -192,25 +195,128 @@ def test_alert_channels(self): ) self.assertEqual(channels[0].alerts.alert_count, 0) - def test_create_update_delete_channel(self): + def test_webhook_channel_validation(self): """ - Test CRUD operations for AlertChannel: create, update, and delete. - Verifies the full lifecycle of an alert channel object. + 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 = 999 + channel_id = 888 channel_url = f"{create_url}/{channel_id}" - # CREATE: Create the channel via channel_create() + # CREATE: Create the webhook channel via channel_create() create_response = { "id": channel_id, - "label": "CRUD Test Channel", + "label": "Webhook Test Channel", "type": "user", - "channel_type": "email", + "channel_type": "webhook", "details": { - "email": { - "usernames": ["crud_user1", "crud_user2"], - "recipient_type": "user", + "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": { @@ -220,42 +326,66 @@ def test_create_update_delete_channel(self): }, "created": "2024-01-01T00:00:00", "updated": "2024-01-01T00:00:00", - "created_by": "crud_user1", - "updated_by": "crud_user1", + "created_by": "webhook_user", + "updated_by": "webhook_user", } with self.mock_post(create_response) as m_post: - channel = self.client.monitor.channel_create( - label="CRUD Test Channel", - channel_type="email", + webhook_channel = self.client.monitor.channel_create( + label="Webhook Test Channel", + channel_type="webhook", details=ChannelDetails( - email=EmailDetails( - recipient_type="user", - usernames=["crud_user1", "crud_user2"], + 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(channel, AlertChannel) - self.assertEqual(channel.id, channel_id) - self.assertEqual(channel.label, "CRUD Test Channel") - - # UPDATE: Update the channel label - updated_response = create_response.copy() - updated_response["label"] = "CRUD Test Channel Updated" - updated_response["updated"] = "2024-01-02T00:00:00" - - with self.mock_put(updated_response) as m_put: - channel.label = "CRUD Test Channel Updated" - result = channel.save() + 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" + ) - self.assertEqual(m_put.call_url, channel_url) - self.assertTrue(result) - self.assertEqual(channel.label, "CRUD Test Channel Updated") + # 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 channel + # DELETE: Delete the webhook channel with self.mock_delete() as m_delete: - result = channel.delete() + result = webhook_channel.delete() self.assertEqual(m_delete.call_url, channel_url) self.assertTrue(result)