adding sdk support for get channel endpoints

Author: mawasthy-lgtm

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</linode_api4/objects/filtering>` 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,
+        )

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"
+}

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
+}

test/fixtures/monitor_services_dbaas_alert-definitions.json

@@ -38,7 +38,7 @@
             "metric": "cpu_usage",
             "operator": "gt",
             "threshold": 90,
-            "unit": "percent"
+            "unit": "%"
           }
         ]
       },

test/fixtures/monitor_services_dbaas_alert-definitions_12345.json

@@ -36,7 +36,7 @@
         "metric": "cpu_usage",
         "operator": "gt",
         "threshold": 90,
-        "unit": "percent"
+        "unit": "%"
       }
     ]
   },

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": [

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

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"

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(