Add lke tests (#641)

Author: vshanthe

linodecli/api_request.py

@@ -374,9 +374,9 @@ def _attempt_warn_old_version(ctx, result):
                 "with --suppress-warnings",
                 file=sys.stderr,
             )
-        suppress_version_warning = ctx.config.get_bool("suppress-version-warning") or os.getenv(
-            "LINODE_CLI_SUPPRESS_VERSION_WARNING"
-        )
+        suppress_version_warning = ctx.config.get_bool(
+            "suppress-version-warning"
+        ) or os.getenv("LINODE_CLI_SUPPRESS_VERSION_WARNING")
         if new_version_exists and not suppress_version_warning:
             print(
                 f"The API responded with version {spec_version}, which is newer than "

tests/integration/lke/test_clusters.py

@@ -55,13 +55,30 @@ def test_deploy_an_lke_cluster():
             "--no-defaults",
         ]
     ).stdout.decode()
-
     assert label + ",us-ord," + lke_version in result
 
-    # Sleep needed here for proper deletion of linodes that are related to lke cluster
-    time.sleep(15)
 
-    remove_lke_clusters()
+@pytest.fixture
+def get_cluster_id():
+    cluster_id = (
+        exec_test_command(
+            BASE_CMD
+            + [
+                "clusters-list",
+                "--text",
+                "--no-headers",
+                "--delimiter",
+                ",",
+                "--format",
+                "id",
+            ]
+        )
+        .stdout.decode()
+        .rstrip()
+        .splitlines()
+    )
+    first_id = cluster_id[0]
+    yield first_id
 
 
 def test_lke_cluster_list():
@@ -78,6 +95,175 @@ def test_lke_cluster_list():
     assert_headers_in_lines(headers, lines)
 
 
+def test_view_lke_cluster(get_cluster_id):
+    cluster_id = get_cluster_id
+
+    res = (
+        exec_test_command(
+            BASE_CMD + ["cluster-view", cluster_id, "--text", "--delimiter=,"]
+        )
+        .stdout.decode()
+        .rstrip()
+    )
+    lines = res.splitlines()
+    headers = ["label", "k8s_version"]
+    assert_headers_in_lines(headers, lines)
+
+
+def test_update_kubernetes_cluster(get_cluster_id):
+    cluster_id = get_cluster_id
+    new_label = "cluster_test" + str(time.time_ns())
+    updated_label = (
+        exec_test_command(
+            BASE_CMD
+            + [
+                "cluster-update",
+                cluster_id,
+                "--label",
+                new_label,
+                "--text",
+                "--no-headers",
+                "--format=label",
+            ]
+        )
+        .stdout.decode()
+        .rstrip()
+    )
+    assert new_label == updated_label
+
+
+def test_list_kubernetes_endpoint(get_cluster_id):
+    cluster_id = get_cluster_id
+    res = (
+        exec_test_command(
+            BASE_CMD
+            + ["api-endpoints-list", cluster_id, "--text", "--delimiter=,"]
+        )
+        .stdout.decode()
+        .rstrip()
+    )
+    lines = res.splitlines()
+
+    headers = ["endpoint"]
+    assert_headers_in_lines(headers, lines)
+
+
+def test_cluster_dashboard_url(get_cluster_id):
+    cluster_id = get_cluster_id
+    res = (
+        exec_test_command(
+            BASE_CMD
+            + ["cluster-dashboard-url", cluster_id, "--text", "--delimiter=,"]
+        )
+        .stdout.decode()
+        .rstrip()
+    )
+    lines = res.splitlines()
+
+    headers = ["url"]
+    assert_headers_in_lines(headers, lines)
+
+
+@pytest.fixture
+def get_node_pool_id(get_cluster_id):
+    cluster_id = get_cluster_id
+    nodepool_id = (
+        exec_test_command(
+            BASE_CMD
+            + [
+                "pools-list",
+                cluster_id,
+                "--text",
+                "--no-headers",
+                "--delimiter",
+                ",",
+                "--format",
+                "id",
+            ]
+        )
+        .stdout.decode()
+        .rstrip()
+        .splitlines()
+    )
+    first_id = nodepool_id[0]
+    yield first_id
+
+
+def test_node_pool_list(get_cluster_id):
+    cluster_id = get_cluster_id
+    res = (
+        exec_test_command(
+            BASE_CMD + ["pools-list", cluster_id, "--text", "--delimiter=,"]
+        )
+        .stdout.decode()
+        .rstrip()
+    )
+    lines = res.splitlines()
+
+    headers = ["nodes.id", "nodes.instance_id"]
+    assert_headers_in_lines(headers, lines)
+
+
+def test_view_pool(get_cluster_id, get_node_pool_id):
+    cluster_id = get_cluster_id
+    node_pool_id = get_node_pool_id
+    res = (
+        exec_test_command(
+            BASE_CMD
+            + ["pool-view", cluster_id, node_pool_id, "--text", "--delimiter=,"]
+        )
+        .stdout.decode()
+        .rstrip()
+    )
+    lines = res.splitlines()
+    headers = ["type", "labels.value"]
+    assert_headers_in_lines(headers, lines)
+
+
+@pytest.mark.skip(reason="BUG TPT-TPT-3145")
+def test_update_node_pool(get_cluster_id, get_node_pool_id):
+    cluster_id = get_cluster_id
+    node_pool_id = get_node_pool_id
+    new_label = "cluster_test" + str(time.time_ns())
+    updated_count = (
+        exec_test_command(
+            BASE_CMD
+            + [
+                "pool-update",
+                cluster_id,
+                node_pool_id,
+                "--count",
+                "5",
+                "--label.value",
+                new_label,
+                "--text",
+                "--no-headers",
+                "--format=label",
+            ]
+        )
+        .stdout.decode()
+        .rstrip()
+    )
+    assert new_label == updated_count
+
+
+@pytest.mark.skip(reason="BUG TPT-TPT-3145")
+def test_view_node(get_cluster_id, get_node_pool_id):
+    cluster_id = get_cluster_id
+    node_pool_id = get_node_pool_id
+    res = (
+        exec_test_command(
+            BASE_CMD
+            + ["node-view", cluster_id, node_pool_id, "--text", "--delimiter=,"]
+        )
+        .stdout.decode()
+        .rstrip()
+    )
+    lines = res.splitlines()
+    headers = ["type", "labels.value"]
+    assert_headers_in_lines(headers, lines)
+
+
 @pytest.fixture
 def test_version_id():
     version_id = (
@@ -101,7 +287,7 @@ def test_version_id():
     yield first_id
 
 
-def test_beta_view(test_version_id):
+def test_version_view(test_version_id):
     version_id = test_version_id
     res = (
         exec_test_command(
@@ -114,3 +300,6 @@ def test_beta_view(test_version_id):
 
     headers = ["id"]
     assert_headers_in_lines(headers, lines)
+    # Sleep needed here for proper deletion of linodes that are related to lke cluster
+    time.sleep(5)
+    remove_lke_clusters()