From 2736865ba9071bde8bffb32264fdbbf5f1795a19 Mon Sep 17 00:00:00 2001 From: Pawel Snoch Date: Tue, 10 Mar 2026 15:27:06 +0100 Subject: [PATCH 1/4] Create tests for Linode integration with alerts --- test/integration/models/linode/test_linode.py | 72 +++++++++++++++++++ 1 file changed, 72 insertions(+) diff --git a/test/integration/models/linode/test_linode.py b/test/integration/models/linode/test_linode.py index 574d5d9d2..cf7561259 100644 --- a/test/integration/models/linode/test_linode.py +++ b/test/integration/models/linode/test_linode.py @@ -234,6 +234,17 @@ def get_status(linode: Instance, status: str): def instance_type_condition(linode: Instance, type: str): return type in str(linode.type) +def test_get_linodes_verify_alerts(test_linode_client): + 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 + assert isinstance(linodes_list[0].alerts.system_alerts, list) + assert isinstance(linodes_list[0].alerts.user_alerts, list) + def test_get_linode(test_linode_client, linode_with_volume_firewall): linode = test_linode_client.load(Instance, linode_with_volume_firewall.id) @@ -283,6 +294,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 +359,65 @@ 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": 80, + } + linode.save() + + wait_for_condition(10, 100, get_status, linode, "running") + new_linode = retry_sending_request(5, linode.clone, parent_linode_id) + assert new_linode.alerts.cpu == 50 + assert new_linode.alerts.io == 6000 + assert new_linode.alerts.network_in == 20 + assert new_linode.alerts.network_out == 20 + assert new_linode.alerts.transfer_quota == 80 + assert isinstance(new_linode.alerts.system_alerts, list) + assert isinstance(new_linode.alerts.user_alerts, list) + + +def test_try_to_update_linode_alerts_legacy_and_aclp_at_the_same_time(create_linode): + linode = create_linode + + linode.alerts={ + "cpu": 50, + "io": 6000, + "network_in": 20, + "network_out": 20, + "transfer": 50, + "system_alerts": [1,436], + "user_alerts": [555], + } + + with pytest.raises(RuntimeError) as err: + linode.save() + assert "Cannot set both legacy and ACLP alerts simultaneously" in str(err.value) + assert "[400] alerts" in str(err.value) + + def test_linode_shutdown(create_linode): linode = create_linode From e256458ebf00e08dcad8e7893f2c816f78c6c35f Mon Sep 17 00:00:00 2001 From: Pawel Snoch Date: Mon, 30 Mar 2026 19:13:05 +0200 Subject: [PATCH 2/4] Code review fixes --- test/integration/models/linode/test_linode.py | 44 ++++++++++++------- 1 file changed, 28 insertions(+), 16 deletions(-) diff --git a/test/integration/models/linode/test_linode.py b/test/integration/models/linode/test_linode.py index cf7561259..a1bd3e5a0 100644 --- a/test/integration/models/linode/test_linode.py +++ b/test/integration/models/linode/test_linode.py @@ -234,7 +234,8 @@ def get_status(linode: Instance, status: str): def instance_type_condition(linode: Instance, type: str): return type in str(linode.type) -def test_get_linodes_verify_alerts(test_linode_client): + +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 @@ -242,8 +243,6 @@ def test_get_linodes_verify_alerts(test_linode_client): assert linodes_list[0].alerts.network_in >= 0 assert linodes_list[0].alerts.network_out >= 0 assert linodes_list[0].alerts.transfer_quota >= 0 - assert isinstance(linodes_list[0].alerts.system_alerts, list) - assert isinstance(linodes_list[0].alerts.user_alerts, list) def test_get_linode(test_linode_client, linode_with_volume_firewall): @@ -379,42 +378,55 @@ def test_linode_alerts_workflow(test_linode_client, create_linode): assert isinstance(linode.alerts.system_alerts, list) assert isinstance(linode.alerts.user_alerts, list) - linode.alerts={ + linode.alerts = { "cpu": 50, "io": 6000, "network_in": 20, "network_out": 20, - "transfer": 80, + "transfer_quota": 40, } linode.save() wait_for_condition(10, 100, get_status, linode, "running") - new_linode = retry_sending_request(5, linode.clone, parent_linode_id) - assert new_linode.alerts.cpu == 50 - assert new_linode.alerts.io == 6000 - assert new_linode.alerts.network_in == 20 - assert new_linode.alerts.network_out == 20 + 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_try_to_update_linode_alerts_legacy_and_aclp_at_the_same_time(create_linode): +def test_try_to_update_linode_alerts_legacy_and_aclp_at_the_same_time( + create_linode, +): linode = create_linode - linode.alerts={ + linode.alerts = { "cpu": 50, "io": 6000, "network_in": 20, "network_out": 20, - "transfer": 50, - "system_alerts": [1,436], + "transfer_quota": 50, + "system_alerts": [1, 436], "user_alerts": [555], } - with pytest.raises(RuntimeError) as err: + with pytest.raises(ApiError) as err: linode.save() - assert "Cannot set both legacy and ACLP alerts simultaneously" in str(err.value) + assert "Cannot set both legacy and ACLP alerts simultaneously" in str( + err.value + ) assert "[400] alerts" in str(err.value) From 2ca53ccfbb8c4a7eb5292d19fcea1f02e06c98cd Mon Sep 17 00:00:00 2001 From: Pawel Snoch Date: Tue, 31 Mar 2026 18:31:29 +0200 Subject: [PATCH 3/4] Remove test test_try_to_update_linode_alerts_legacy_and_aclp_at_the_same_time and add test test_update_linode_aclp_alerts --- test/integration/conftest.py | 61 +++++++++++++++++++ test/integration/models/linode/test_linode.py | 26 +++----- 2 files changed, 70 insertions(+), 17 deletions(-) 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 a1bd3e5a0..23073ae3d 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, @@ -407,27 +407,19 @@ def test_linode_alerts_workflow(test_linode_client, create_linode): new_linode.delete() -def test_try_to_update_linode_alerts_legacy_and_aclp_at_the_same_time( - create_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 = { - "cpu": 50, - "io": 6000, - "network_in": 20, - "network_out": 20, - "transfer_quota": 50, - "system_alerts": [1, 436], - "user_alerts": [555], + "user_alerts": [create_alert_service_definition.id], + "system_alerts": [sample_system_alert], } - - with pytest.raises(ApiError) as err: - linode.save() - assert "Cannot set both legacy and ACLP alerts simultaneously" in str( - err.value - ) - assert "[400] alerts" in str(err.value) + 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): From 9d9834d680d6bcce7c43a92225fd13243c8052f3 Mon Sep 17 00:00:00 2001 From: Pawel Snoch Date: Wed, 1 Apr 2026 18:00:45 +0200 Subject: [PATCH 4/4] Apply code review suggestions --- test/integration/models/linode/test_linode.py | 30 ++++++++++++++++--- 1 file changed, 26 insertions(+), 4 deletions(-) diff --git a/test/integration/models/linode/test_linode.py b/test/integration/models/linode/test_linode.py index 23073ae3d..bb53cd090 100644 --- a/test/integration/models/linode/test_linode.py +++ b/test/integration/models/linode/test_linode.py @@ -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,6 +231,23 @@ 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) @@ -385,7 +402,13 @@ def test_linode_alerts_workflow(test_linode_client, create_linode): "network_out": 20, "transfer_quota": 40, } - linode.save() + 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( @@ -403,8 +426,7 @@ def test_linode_alerts_workflow(test_linode_client, create_linode): 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() + wait_for_clone_complete_and_delete_linode(10, 100, new_linode) def test_update_linode_aclp_alerts(