Implement fields related to LKE APL

linode_api4/groups/lke.py

@@ -66,6 +66,7 @@ def cluster_create(
         control_plane: Union[
             LKEClusterControlPlaneOptions, Dict[str, Any]
         ] = None,
+        apl_enabled: bool = False,
         **kwargs,
     ):
         """
@@ -100,8 +101,12 @@ def cluster_create(
                           formatted dicts.
         :param kube_version: The version of Kubernetes to use
         :type kube_version: KubeVersion or str
-        :param control_plane: Dict[str, Any] or LKEClusterControlPlaneRequest
-        :type control_plane: The control plane configuration of this LKE cluster.
+        :param control_plane: The control plane configuration of this LKE cluster.
+        :type control_plane: Dict[str, Any] or LKEClusterControlPlaneRequest
+        :param apl_enabled: Whether this cluster should use APL.
+                            NOTE: This endpoint is in beta and may only
+                            function if base_url is set to `https://api.linode.com/v4beta`.
+        :type apl_enabled: bool
         :param kwargs: Any other arguments to pass along to the API.  See the API
                        docs for possible values.
 
@@ -120,6 +125,10 @@ def cluster_create(
         }
         params.update(kwargs)
 
+        # Prevent errors for users without access to APL
+        if apl_enabled:
+            params["apl_enabled"] = apl_enabled
+
         result = self.client.post(
             "/lke/clusters",
             data=_flatten_request_body_recursive(drop_null_keys(params)),

linode_api4/objects/linode.py

@@ -1920,7 +1920,7 @@ def _serialize(self):
 def _expand_placement_group_assignment(
     pg: Union[
         InstancePlacementGroupAssignment, "PlacementGroup", Dict[str, Any], int
-    ]
+    ],
 ) -> Optional[Dict[str, Any]]:
     """
     Expands the placement group argument into a dict for use in an API request body.

linode_api4/objects/lke.py

@@ -257,6 +257,7 @@ class LKECluster(Base):
         "k8s_version": Property(slug_relationship=KubeVersion, mutable=True),
         "pools": Property(derived_class=LKENodePool),
         "control_plane": Property(mutable=True),
+        "apl_enabled": Property(),
     }
 
     def invalidate(self):
@@ -353,6 +354,36 @@ def control_plane_acl(self) -> LKEClusterControlPlaneACL:
 
         return LKEClusterControlPlaneACL.from_json(self._control_plane_acl)
 
+    @property
+    def apl_console_url(self) -> Optional[str]:
+        """
+        Returns the URL of this cluster's APL installation if this cluster
+        is APL-enabled, else None.
+
+        :returns: The URL of the APL console for this cluster.
+        :rtype: str or None
+        """
+
+        if not self.apl_enabled:
+            return None
+
+        return f"https://console.lke{self.id}.akamai-apl.net"
+
+    @property
+    def apl_health_check_url(self) -> Optional[str]:
+        """
+        Returns the URL of this cluster's APL health check endpoint if this cluster
+        is APL-enabled, else None.
+
+        :returns: The URL of the APL console for this cluster.
+        :rtype: str or None
+        """
+
+        if not self.apl_enabled:
+            return None
+
+        return f"https://auth.lke{self.id}.akamai-apl.net/ready"
+
     def node_pool_create(
         self,
         node_type: Union[Type, str],

test/fixtures/lke_clusters.json

@@ -6,5 +6,6 @@
     "label": "example-cluster",
     "region": "ap-west",
     "k8s_version": "1.19",
-    "tags": []
+    "tags": [],
+    "apl_enabled": true
 }

test/fixtures/lke_clusters_18881.json

@@ -9,5 +9,6 @@
     "tags": [],
     "control_plane": {
         "high_availability": true
-    }
+    },
+    "apl_enabled": true
 }

test/integration/models/image/test_image.py

@@ -32,8 +32,8 @@ def image_upload_url(test_linode_client):
 @pytest.fixture(scope="session")
 def test_uploaded_image(test_linode_client):
     test_image_content = (
-        b"\x1F\x8B\x08\x08\xBD\x5C\x91\x60\x00\x03\x74\x65\x73\x74\x2E\x69"
-        b"\x6D\x67\x00\x03\x00\x00\x00\x00\x00\x00\x00\x00\x00"
+        b"\x1f\x8b\x08\x08\xbd\x5c\x91\x60\x00\x03\x74\x65\x73\x74\x2e\x69"
+        b"\x6d\x67\x00\x03\x00\x00\x00\x00\x00\x00\x00\x00\x00"
     )
 
     label = get_test_label() + "_image"

test/integration/models/lke/test_lke.py

@@ -110,6 +110,32 @@ def lke_cluster_with_labels_and_taints(test_linode_client):
     cluster.delete()
 
 
+@pytest.fixture(scope="session")
+def lke_cluster_with_apl(test_linode_client):
+    version = test_linode_client.lke.versions()[0]
+
+    region = get_region(test_linode_client, {"Kubernetes", "Disk Encryption"})
+
+    # NOTE: g6-dedicated-4 is the minimum APL-compatible Linode type
+    node_pools = test_linode_client.lke.node_pool("g6-dedicated-4", 3)
+    label = get_test_label() + "_cluster"
+
+    cluster = test_linode_client.lke.cluster_create(
+        region,
+        label,
+        node_pools,
+        version,
+        control_plane=LKEClusterControlPlaneOptions(
+            high_availability=True,
+        ),
+        apl_enabled=True,
+    )
+
+    yield cluster
+
+    cluster.delete()
+
+
 def get_cluster_status(cluster: LKECluster, status: str):
     return cluster._raw_json["status"] == status
 
@@ -328,6 +354,19 @@ def test_lke_cluster_labels_and_taints(lke_cluster_with_labels_and_taints):
     assert LKENodePoolTaint.from_json(updated_taints[1]) in pool.taints
 
 
+@pytest.mark.flaky(reruns=3, reruns_delay=2)
+def test_lke_cluster_with_apl(lke_cluster_with_apl):
+    assert lke_cluster_with_apl.apl_enabled == True
+    assert (
+        lke_cluster_with_apl.apl_console_url
+        == f"https://console.lke{lke_cluster_with_apl.id}.akamai-apl.net"
+    )
+    assert (
+        lke_cluster_with_apl.apl_health_check_url
+        == f"https://auth.lke{lke_cluster_with_apl.id}.akamai-apl.net/ready"
+    )
+
+
 def test_lke_types(test_linode_client):
     types = test_linode_client.lke.types()
 

test/unit/objects/image_test.py

@@ -8,8 +8,8 @@
 
 # A minimal gzipped image that will be accepted by the API
 TEST_IMAGE_CONTENT = (
-    b"\x1F\x8B\x08\x08\xBD\x5C\x91\x60\x00\x03\x74\x65\x73\x74\x2E\x69"
-    b"\x6D\x67\x00\x03\x00\x00\x00\x00\x00\x00\x00\x00\x00"
+    b"\x1f\x8b\x08\x08\xbd\x5c\x91\x60\x00\x03\x74\x65\x73\x74\x2e\x69"
+    b"\x6d\x67\x00\x03\x00\x00\x00\x00\x00\x00\x00\x00\x00"
 )
 
 

test/unit/objects/lke_test.py

@@ -39,6 +39,7 @@ def test_get_cluster(self):
         self.assertEqual(cluster.region.id, "ap-west")
         self.assertEqual(cluster.k8s_version.id, "1.19")
         self.assertTrue(cluster.control_plane.high_availability)
+        self.assertTrue(cluster.apl_enabled)
 
     def test_get_pool(self):
         """
@@ -352,6 +353,40 @@ def test_cluster_create_with_labels_and_taints(self):
                 ],
             }
 
+    def test_cluster_create_with_apl(self):
+        """
+        Tests that an LKE cluster can be created with APL enabled.
+        """
+
+        with self.mock_post("lke/clusters") as m:
+            cluster = self.client.lke.cluster_create(
+                "us-mia",
+                "test-aapl-cluster",
+                [
+                    self.client.lke.node_pool(
+                        "g6-dedicated-4",
+                        3,
+                    )
+                ],
+                "1.29",
+                apl_enabled=True,
+                control_plane=LKEClusterControlPlaneOptions(
+                    high_availability=True,
+                ),
+            )
+
+            assert m.call_data["apl_enabled"] == True
+            assert m.call_data["control_plane"]["high_availability"] == True
+
+        assert (
+            cluster.apl_console_url == "https://console.lke18881.akamai-apl.net"
+        )
+
+        assert (
+            cluster.apl_health_check_url
+            == "https://auth.lke18881.akamai-apl.net/ready"
+        )
+
     def test_populate_with_taints(self):
         """
         Tests that LKENodePool correctly handles a list of LKENodePoolTaint and Dict objects.