Integration tests for support Linode integration with alerts

test/integration/conftest.py

@@ -728,3 +728,64 @@ def test_monitor_client(get_monitor_token_for_db_entities):
     )
 
     return client, entity_ids
+
+
+@pytest.fixture(scope="session")
+def create_alert_service_definition(test_linode_client):
+    rule_criteria = {
+        "rules": [
+            {
+                "aggregate_function": "min",
+                "dimension_filters": [
+                    {
+                        "dimension_label": "node_type",
+                        "label": "Node Type",
+                        "operator": "eq",
+                        "value": "primary",
+                    }
+                ],
+                "label": "Memory Usage",
+                "metric": "memory_usage",
+                "operator": "eq",
+                "threshold": 95,
+                "unit": "percent",
+            }
+        ]
+    }
+    trigger_conditions = {
+        "criteria_condition": "ALL",
+        "evaluation_period_seconds": 300,
+        "polling_interval_seconds": 900,
+        "trigger_occurrences": 3,
+    }
+    channels = list(test_linode_client.monitor.alert_channels())
+    if len(channels) == 0:
+        raise Exception(
+            "No alert channels available for testing. Please create an alert channel and try again."
+        )
+    alert = test_linode_client.monitor.create_alert_definition(
+        service_type="dbaas",
+        label=get_test_label() + "-service-definition",
+        severity=1,
+        description="description",
+        channel_ids=[channels[0].id],
+        rule_criteria=rule_criteria,
+        trigger_conditions=trigger_conditions,
+    )
+
+    yield alert
+
+    alert.delete()
+
+
+def get_system_alerts(client: LinodeClient):
+    alerts = client.monitor.alert_definitions()
+    system_alerts = []
+    for alert in alerts.lists[0]:
+        if alert.type == "system":
+            system_alerts.append(alert)
+    if len(system_alerts) == 0:
+        raise Exception(
+            "No system alert definitions found. Cannot run tests dependent on system alert definitions."
+        )
+    return system_alerts

test/integration/models/linode/test_linode.py

@@ -1,6 +1,6 @@
 import ipaddress
 import time
-from test.integration.conftest import get_region
+from test.integration.conftest import get_region, get_system_alerts
 from test.integration.helpers import (
     get_test_label,
     retry_sending_request,
@@ -235,6 +235,16 @@ def instance_type_condition(linode: Instance, type: str):
     return type in str(linode.type)
 
 
+def test_get_linodes_verify_alerts(test_linode_client, create_linode):
+    linodes_list = test_linode_client.linode.instances().lists[0]
+    assert len(linodes_list) > 0
+    assert linodes_list[0].alerts.cpu >= 0
+    assert linodes_list[0].alerts.io >= 0
+    assert linodes_list[0].alerts.network_in >= 0
+    assert linodes_list[0].alerts.network_out >= 0
+    assert linodes_list[0].alerts.transfer_quota >= 0
+
+
 def test_get_linode(test_linode_client, linode_with_volume_firewall):
     linode = test_linode_client.load(Instance, linode_with_volume_firewall.id)
 
@@ -283,6 +293,8 @@ def test_linode_rebuild(test_linode_client):
 
     assert linode.status == "rebuilding"
     assert linode.image.id == "linode/debian12"
+    assert linode.alerts.cpu >= 0
+    assert linode.alerts.io >= 0
 
     assert linode.disk_encryption == InstanceDiskEncryptionType.disabled
 
@@ -346,6 +358,70 @@ def test_linode_reboot(create_linode):
     assert linode.status == "running"
 
 
+def test_linode_alerts_workflow(test_linode_client, create_linode):
+    linode = create_linode
+    parent_linode_id = create_linode.id
+    assert linode.alerts.cpu == 90
+    assert linode.alerts.io == 10000
+    assert linode.alerts.network_in == 10
+    assert linode.alerts.network_out == 10
+    assert linode.alerts.transfer_quota == 80
+    assert isinstance(linode.alerts.system_alerts, list)
+    assert isinstance(linode.alerts.user_alerts, list)
+
+    linode = test_linode_client.load(Instance, parent_linode_id)
+    assert linode.alerts.cpu == 90
+    assert linode.alerts.io == 10000
+    assert linode.alerts.network_in == 10
+    assert linode.alerts.network_out == 10
+    assert linode.alerts.transfer_quota == 80
+    assert isinstance(linode.alerts.system_alerts, list)
+    assert isinstance(linode.alerts.user_alerts, list)
+
+    linode.alerts = {
+        "cpu": 50,
+        "io": 6000,
+        "network_in": 20,
+        "network_out": 20,
+        "transfer_quota": 40,
+    }
+    linode.save()
+
+    wait_for_condition(10, 100, get_status, linode, "running")
+    new_linode = retry_sending_request(
+        5,
+        linode.clone,
+        region=linode.region.id,
+        instance_type=linode.type.id,
+        label=get_test_label(),
+    )
+    assert new_linode.alerts.cpu == 90
+    assert new_linode.alerts.io == 10000
+    assert new_linode.alerts.network_in == 10
+    assert new_linode.alerts.network_out == 10
+    assert new_linode.alerts.transfer_quota == 80
+    assert isinstance(new_linode.alerts.system_alerts, list)
+    assert isinstance(new_linode.alerts.user_alerts, list)
+
+    if new_linode is not None:
+        new_linode.delete()
+
+
+def test_update_linode_aclp_alerts(
+    test_linode_client, create_linode, create_alert_service_definition
+):
+    linode = create_linode
+    sample_system_alert = get_system_alerts(test_linode_client)[0].id
+
+    linode.alerts = {
+        "user_alerts": [create_alert_service_definition.id],
+        "system_alerts": [sample_system_alert],
+    }
+    linode.save()
+    assert linode.alerts["user_alerts"] == [create_alert_service_definition.id]
+    assert linode.alerts["system_alerts"] == [sample_system_alert]
+
+
 def test_linode_shutdown(create_linode):
     linode = create_linode