diff --git a/test/integration/conftest.py b/test/integration/conftest.py index a5c832f4f..7eda63c47 100644 --- a/test/integration/conftest.py +++ b/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 diff --git a/test/integration/models/linode/test_linode.py b/test/integration/models/linode/test_linode.py index 574d5d9d2..bb53cd090 100644 --- a/test/integration/models/linode/test_linode.py +++ b/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, @@ -10,7 +10,7 @@ import pytest -from linode_api4.errors import ApiError +from linode_api4.errors import ApiError, UnexpectedResponseError from linode_api4.objects import ( Config, ConfigInterface, @@ -231,10 +231,37 @@ def get_status(linode: Instance, status: str): return linode.status == status +def wait_for_clone_complete_and_delete_linode( + interval: int, timeout: int, linode: Instance +) -> object: + end_time = time.time() + timeout + while time.time() < end_time: + try: + linode.delete() + return True + except ApiError as err: + if "[400] Linode is the target of an ongoing clone" not in str(err): + raise UnexpectedResponseError(f"Unexpected delete linode error") + time.sleep(interval) + raise TimeoutError( + f"Timeout Error: not possible to delete just cloned linode in {timeout} seconds" + ) + + 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 +310,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 +375,75 @@ 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_status = linode.save() + assert linode_save_status == True + assert linode.alerts["cpu"] == 50 + assert linode.alerts["io"] == 6000 + assert linode.alerts["network_in"] == 20 + assert linode.alerts["network_out"] == 20 + assert linode.alerts["transfer_quota"] == 40 + + 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) + + wait_for_clone_complete_and_delete_linode(10, 100, new_linode) + + +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