Integration tests (#814)

Author: vshanthe

tests/integration/account/test_account.py

@@ -196,6 +196,7 @@ def test_account_setting_view():
         "longview_subscription",
         "network_helper",
         "interfaces_for_new_linodes",
+        "maintenance_policy",
     ]
 
     settings_text = exec_test_command(
@@ -343,7 +344,15 @@ def test_maintenance_list():
     )
     lines = res.splitlines()
 
-    headers = ["entity.type", "entity.label"]
+    headers = [
+        "complete_time",
+        "entity.type",
+        "entity.label",
+        "maintenance_policy_set",
+        "not_before",
+        "source",
+        "start_time",
+    ]
     assert_headers_in_lines(headers, lines)
 
 

tests/integration/helpers.py

@@ -16,9 +16,9 @@
 # TypeVars for generic type hints below
 T = TypeVar("T")
 
-
 MODULES = [
     "account",
+    "alerts",
     "domains",
     "linodes",
     "nodebalancers",
@@ -33,7 +33,9 @@
     "linodes",
     "lke",
     "longview",
+    "maintenance",
     "managed",
+    "monitor",
     "networking",
     "obj",
     "object-storage",
@@ -100,8 +102,41 @@ def exec_failing_test_command(
 
 
 # Delete/Remove helper functions (mainly used in clean-ups after tests)
-def delete_target_id(target: str, id: str, delete_command: str = "delete"):
-    command = ["linode-cli", target, delete_command, id]
+def delete_target_id(
+    target: str,
+    id: str,
+    delete_command: str = "delete",
+    service_type: str = None,
+    use_retry: bool = False,
+    retries: int = 3,
+    delay: int = 80,
+):
+    if service_type:
+        command = ["linode-cli", target, delete_command, service_type, id]
+    else:
+        command = ["linode-cli", target, delete_command, id]
+
+    if use_retry:
+        last_exc = None
+        for attempt in range(retries):
+            try:
+                subprocess.run(
+                    command,
+                    check=True,
+                    stdout=subprocess.PIPE,
+                    stderr=subprocess.PIPE,
+                    text=True,
+                )
+                return  # success
+            except Exception as e:
+                last_exc = e
+                if attempt < retries - 1:
+                    time.sleep(delay)
+        # If all retries fail, raise
+        raise RuntimeError(
+            f"Error executing command '{' '.join(command)}' after {retries} retries: {last_exc}"
+        )
+
     try:
         subprocess.run(
             command,

tests/integration/lke/test_clusters.py

@@ -52,6 +52,8 @@ def test_deploy_an_lke_cluster():
             '[{"type":"ext4","size":1024}]',
             "--k8s_version",
             lke_version,
+            "--tier",
+            "standard",
             "--text",
             "--delimiter",
             ",",
@@ -75,7 +77,15 @@ def test_lke_cluster_list():
     )
     lines = res.splitlines()
 
-    headers = ["label", "k8s_version"]
+    headers = [
+        "label",
+        "k8s_version",
+        "tier",
+        "apl_enabled",
+        "vpc_id",
+        "subnet_id",
+        "stack_type",
+    ]
     assert_headers_in_lines(headers, lines)
 
 
@@ -87,7 +97,15 @@ def test_view_lke_cluster(lke_cluster):
         + ["cluster-view", cluster_id, "--text", "--delimiter=,"]
     )
     lines = res.splitlines()
-    headers = ["label", "k8s_version"]
+    headers = [
+        "label",
+        "k8s_version",
+        "tier",
+        "apl_enabled",
+        "vpc_id",
+        "subnet_id",
+        "stack_type",
+    ]
     assert_headers_in_lines(headers, lines)
 
 
@@ -160,7 +178,7 @@ def test_view_pool(lke_cluster):
     )
 
     lines = res.splitlines()
-    headers = ["type", "labels"]
+    headers = ["type", "labels", "k8s_version", "label"]
     assert_headers_in_lines(headers, lines)
 
 

tests/integration/maintenance/test_maintenance_policies.py

@@ -0,0 +1,27 @@
+from tests.integration.helpers import (
+    BASE_CMDS,
+    assert_headers_in_lines,
+    exec_test_command,
+)
+
+
+def test_maintenance_policies_list():
+    res = exec_test_command(
+        BASE_CMDS["maintenance"] + ["policies-list", "--text", "--delimiter=,"]
+    )
+    lines = res.splitlines()
+    headers = [
+        "description",
+        "is_default",
+        "label",
+        "is_default",
+        "slug",
+        "type",
+    ]
+    assert_headers_in_lines(headers, lines)
+    rows = [line.split(",") for line in lines[1:] if line.strip()]
+
+    type_index = headers.index("type")
+    types = [row[type_index].strip() for row in rows]
+
+    assert set(types) == {"migrate", "power_off_on"}

tests/integration/monitor/conftest.py

@@ -0,0 +1,24 @@
+import pytest
+
+from tests.integration.helpers import (
+    BASE_CMDS,
+    exec_test_command,
+)
+
+
+@pytest.fixture
+def get_service_type():
+    service_ids = exec_test_command(
+        BASE_CMDS["monitor"]
+        + [
+            "service-list",
+            "--text",
+            "--no-headers",
+            "--delimiter",
+            ",",
+            "--format",
+            "service_type",
+        ]
+    ).splitlines()
+    first_id = service_ids[0].split(",")[0]
+    yield first_id

tests/integration/monitor/test_alerts.py

@@ -0,0 +1,152 @@
+import pytest
+
+from tests.integration.helpers import (
+    BASE_CMDS,
+    assert_headers_in_lines,
+    delete_target_id,
+    exec_test_command,
+    get_random_text,
+    retry_exec_test_command_with_delay,
+)
+
+
+def test_channels_list():
+    res = exec_test_command(
+        BASE_CMDS["alerts"] + ["channels-list", "--text", "--delimiter=,"]
+    )
+    lines = res.splitlines()
+    headers = [
+        "channel_type",
+        "content.email.email_addresses",
+        "id",
+        "label",
+        "type",
+        "updated",
+    ]
+    assert_headers_in_lines(headers, lines)
+
+
+@pytest.fixture
+def get_channel_id():
+    channel_ids = exec_test_command(
+        BASE_CMDS["alerts"]
+        + [
+            "channels-list",
+            "--text",
+            "--no-headers",
+            "--delimiter",
+            ",",
+            "--format",
+            "id",
+        ]
+    ).splitlines()
+    first_id = channel_ids[0].split(",")[0]
+    yield first_id
+
+
+def test_alerts_definition_create(get_channel_id, get_service_type):
+    label = get_random_text(8) + "_alert"
+    exec_test_command(
+        BASE_CMDS["alerts"]
+        + [
+            "definition-create",
+            "--channel_ids",
+            get_channel_id,
+            "--label",
+            label,
+            "--rule_criteria.rules.metric",
+            "cpu_usage",
+            "--rule_criteria.rules.operator",
+            "eq",
+            "--rule_criteria.rules.threshold",
+            "80",
+            "--rule_criteria.rules.aggregate_function",
+            "avg",
+            "--severity",
+            "1",
+            "--trigger_conditions.criteria_condition",
+            "ALL",
+            "--trigger_conditions.evaluation_period_seconds",
+            "300",
+            "--trigger_conditions.polling_interval_seconds",
+            "300",
+            "--trigger_conditions.trigger_occurrences",
+            "3",
+            get_service_type,
+        ]
+    )
+
+
+def test_alerts_list():
+    res = exec_test_command(
+        BASE_CMDS["alerts"]
+        + ["definitions-list-all", "--text", "--delimiter=,"]
+    )
+    lines = res.splitlines()
+    headers = ["class", "created", "label", "severity", "service_type"]
+    assert_headers_in_lines(headers, lines)
+
+
+@pytest.fixture
+def get_alert_id():
+    alert_id = exec_test_command(
+        BASE_CMDS["alerts"]
+        + [
+            "definitions-list-all",
+            "--text",
+            "--no-headers",
+            "--delimiter",
+            ",",
+            "--format",
+            "id",
+        ]
+    ).splitlines()
+    first_id = alert_id[-1]
+    yield first_id
+
+
+def test_alert_view(get_alert_id, get_service_type):
+    alert_id = get_alert_id
+    service_type = get_service_type
+    res = exec_test_command(
+        BASE_CMDS["alerts"]
+        + [
+            "definition-view",
+            service_type,
+            alert_id,
+            "--text",
+            "--delimiter=,",
+        ]
+    )
+    lines = res.splitlines()
+
+    headers = ["class", "created", "label", "severity", "service_type"]
+    assert_headers_in_lines(headers, lines)
+
+
+def test_alert_update(get_alert_id, get_service_type):
+    alert_id = get_alert_id
+    service_type = get_service_type
+    new_label = get_random_text(8) + "_updated"
+    updated_label = retry_exec_test_command_with_delay(
+        BASE_CMDS["alerts"]
+        + [
+            "definition-update",
+            service_type,
+            alert_id,
+            "--label",
+            new_label,
+            "--text",
+            "--no-headers",
+            "--format=label",
+        ],
+        delay=50,
+    )
+    assert updated_label == new_label
+    delete_target_id(
+        target="alerts",
+        delete_command="definition-delete",
+        service_type=service_type,
+        id=alert_id,
+        use_retry=True,
+    )