test: Add LKE ACL cases (#684)

tests/integration/helpers.py

@@ -194,3 +194,24 @@ def get_random_region_with_caps(
     matching_region_ids = [region["id"] for region in matching_regions]
 
     return random.choice(matching_region_ids) if matching_region_ids else None
+
+
+def get_cluster_id(label: str):
+    cluster_id = (
+        exec_test_command(
+            [
+                "linode-cli",
+                "lke",
+                "clusters-list",
+                "--text",
+                "--format=id",
+                "--no-headers",
+                "--label",
+                label,
+            ]
+        )
+        .stdout.decode()
+        .rstrip()
+    )
+
+    return cluster_id

tests/integration/lke/test_clusters.py

@@ -4,6 +4,7 @@
     assert_headers_in_lines,
     delete_target_id,
     exec_test_command,
+    get_cluster_id,
     get_random_region_with_caps,
     get_random_text,
     retry_exec_test_command_with_delay,
@@ -84,26 +85,6 @@ def get_pool_nodesid(cluster_id):
     return first_id
 
 
-def get_cluster_id(label: str):
-    cluster_id = (
-        exec_test_command(
-            BASE_CMD
-            + [
-                "clusters-list",
-                "--text",
-                "--format=id",
-                "--no-headers",
-                "--label",
-                label,
-            ]
-        )
-        .stdout.decode()
-        .rstrip()
-    )
-
-    return cluster_id
-
-
 @pytest.fixture
 def test_lke_cluster():
     label = get_random_text(8) + "_cluster"
@@ -512,3 +493,78 @@ def test_pool_view(test_lke_cluster, test_node_pool):
     ]
 
     assert_headers_in_lines(headers, lines)
+
+
+def test_update_autoscaler(test_lke_cluster, test_node_pool):
+    cluster_id = test_lke_cluster
+    node_pool_id = test_node_pool
+
+    result = (
+        exec_test_command(
+            BASE_CMD
+            + [
+                "pool-update",
+                cluster_id,
+                node_pool_id,
+                "--autoscaler.enabled",
+                "true",
+                "--autoscaler.min",
+                "1",
+                "--autoscaler.max",
+                "3",
+                "--text",
+            ]
+        )
+        .stdout.decode()
+        .rstrip()
+    )
+
+    headers = [
+        "autoscaler.enabled",
+        "autoscaler.max",
+        "autoscaler.min",
+        "count",
+        "disk_encryption",
+        "id",
+        "labels.key",
+        "labels.value",
+        "tags",
+        "taints",
+        "type",
+    ]
+
+    assert_headers_in_lines(headers, result.splitlines())
+
+    assert "3" in result
+    assert "1" in result
+
+
+def test_kubeconfig_view(test_lke_cluster):
+    cluster_id = test_lke_cluster
+
+    kubeconfig = (
+        retry_exec_test_command_with_delay(
+            BASE_CMD
+            + [
+                "kubeconfig-view",
+                cluster_id,
+                "--text",
+            ],
+            retries=5,
+            delay=60,
+        )
+        .stdout.decode()
+        .strip()
+    )
+
+    header = ["kubeconfig"]
+
+    assert_headers_in_lines(header, kubeconfig.splitlines())
+
+    assert kubeconfig
+
+
+def test_cluster_nodes_recycle(test_lke_cluster):
+    cluster_id = test_lke_cluster
+
+    exec_test_command(BASE_CMD + ["cluster-nodes-recycle", cluster_id])

tests/integration/lke/test_lke_acl.py

@@ -0,0 +1,172 @@
+# │ cluster-acl-delete    │ Delete the control plane access control list. │
+# │ cluster-acl-update    │ Update the control plane access control list. │
+# │ cluster-acl-view      │ Get the control plane access control list.    │
+import pytest
+
+from tests.integration.helpers import (
+    assert_headers_in_lines,
+    delete_target_id,
+    exec_test_command,
+    get_cluster_id,
+    get_random_region_with_caps,
+    get_random_text,
+    retry_exec_test_command_with_delay,
+)
+
+BASE_CMD = ["linode-cli", "lke"]
+
+
+@pytest.fixture
+def test_lke_cluster_acl():
+    label = get_random_text(8) + "_cluster"
+
+    test_region = get_random_region_with_caps(
+        required_capabilities=["Linodes", "Kubernetes"]
+    )
+    lke_version = (
+        exec_test_command(
+            BASE_CMD
+            + [
+                "versions-list",
+                "--text",
+                "--no-headers",
+            ]
+        )
+        .stdout.decode()
+        .rstrip()
+        .splitlines()[0]
+    )
+
+    cluster_label = (
+        exec_test_command(
+            BASE_CMD
+            + [
+                "cluster-create",
+                "--region",
+                test_region,
+                "--label",
+                label,
+                "--node_pools.type",
+                "g6-standard-1",
+                "--node_pools.count",
+                "1",
+                "--node_pools.disks",
+                '[{"type":"ext4","size":1024}]',
+                "--k8s_version",
+                lke_version,
+                "--control_plane.high_availability",
+                "true",
+                "--control_plane.acl.enabled",
+                "true",
+                "--text",
+                "--delimiter",
+                ",",
+                "--no-headers",
+                "--format",
+                "label",
+                "--no-defaults",
+            ]
+        )
+        .stdout.decode()
+        .rstrip()
+    )
+
+    cluster_id = get_cluster_id(label=cluster_label)
+
+    yield cluster_id
+
+    delete_target_id(
+        target="lke", id=cluster_id, delete_command="cluster-delete"
+    )
+
+
+def test_cluster_acl_view(test_lke_cluster_acl):
+    cluster_id = test_lke_cluster_acl
+
+    acl = (
+        exec_test_command(
+            BASE_CMD
+            + [
+                "cluster-acl-view",
+                cluster_id,
+                "--text",
+            ]
+        )
+        .stdout.decode()
+        .strip()
+    )
+
+    headers = [
+        "acl.enabled",
+        "acl.addresses.ipv4",
+        "acl.addresses.ipv6",
+        "acl.revision-id",
+    ]
+
+    assert_headers_in_lines(headers, acl.splitlines())
+
+    assert "True" in acl
+
+
+def test_cluster_acl_update(test_lke_cluster_acl):
+    cluster_id = test_lke_cluster_acl
+
+    print("RUNNING TEST")
+
+    # Verify the update
+    acl = (
+        exec_test_command(
+            BASE_CMD
+            + [
+                "cluster-acl-update",
+                cluster_id,
+                "--acl.addresses.ipv4",
+                "203.0.113.1",
+                "--acl.addresses.ipv6",
+                "2001:db8:1234:abcd::/64",
+                "--acl.enabled",
+                "true",
+                "--text",
+            ]
+        )
+        .stdout.decode()
+        .strip()
+    )
+
+    headers = [
+        "acl.enabled",
+        "acl.addresses.ipv4",
+        "acl.addresses.ipv6",
+        "acl.revision-id",
+    ]
+
+    assert_headers_in_lines(headers, acl.splitlines())
+
+    assert "203.0.113.1" in acl
+    assert "2001:db8:1234:abcd::/64" in acl
+
+
+def test_cluster_acl_delete(test_lke_cluster_acl):
+    cluster_id = test_lke_cluster_acl
+
+    retry_exec_test_command_with_delay(
+        BASE_CMD + ["cluster-acl-delete", cluster_id]
+    )
+
+    # Verify the deletion
+    acl = (
+        exec_test_command(
+            BASE_CMD
+            + [
+                "cluster-acl-view",
+                cluster_id,
+                "--text",
+                "--format=acl.enabled",
+                "--text",
+            ]
+        )
+        .stdout.decode()
+        .strip()
+    )
+
+    assert "False" in acl