Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
66 changes: 66 additions & 0 deletions linode_api4/groups/monitor.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
MonitorService,
MonitorServiceToken,
)
from linode_api4.objects.monitor import ChannelDetails

__all__ = [
"MonitorGroup",
Expand Down Expand Up @@ -332,3 +333,68 @@ 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

.. 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,
"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)
15 changes: 7 additions & 8 deletions linode_api4/objects/monitor.py
Original file line number Diff line number Diff line change
Expand Up @@ -492,23 +492,22 @@ 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}"

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),
Expand Down
79 changes: 78 additions & 1 deletion test/integration/models/monitor/test_monitor.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@

from linode_api4 import LinodeClient, PaginatedList
from linode_api4.objects import (
AlertChannel,
AlertDefinition,
AlertDefinitionEntity,
ApiError,
Expand All @@ -17,7 +18,11 @@
MonitorService,
MonitorServiceToken,
)
from linode_api4.objects.monitor import AlertStatus
from linode_api4.objects.monitor import (
AlertStatus,
ChannelDetails,
EmailDetails,
)


# List all dashboards
Expand Down Expand Up @@ -311,3 +316,75 @@ def test_alert_definition_entities(test_linode_client):
assert entity.label
assert entity.url
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 = 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"

# 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}"
)
71 changes: 71 additions & 0 deletions test/unit/groups/monitor_api_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down Expand Up @@ -180,3 +182,72 @@ 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_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_user"],
"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_user",
"updated_by": "test_user",
}

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_user"],
)
),
)

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
57 changes: 57 additions & 0 deletions test/unit/objects/monitor_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -169,3 +169,60 @@ def test_alert_channels(self):
"/monitor/alert-channels/123/alerts",
)
self.assertEqual(channels[0].alerts.alert_count, 0)

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": channel_id,
"label": "CRUD Test Channel",
"type": "user",
"channel_type": "email",
"details": {
"email": {
"usernames": ["crud_user"],
"recipient_type": "user",
}
},
"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": "crud_user",
"updated_by": "crud_user",
}

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")

# 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.assertEqual(m_delete.call_url, url)
self.assertTrue(result)