APIPath to group

Author: lgarber-akamai

linode_api4/__init__.py

@@ -1,6 +1,5 @@
 # isort: skip_file
 from linode_api4.objects import *
-from linode_api4.paths import *
 from linode_api4.errors import ApiError, UnexpectedResponseError
 from linode_api4.linode_client import LinodeClient
 from linode_api4.login_client import LinodeLoginClient, OAuthScopes

linode_api4/groups/__init__.py

@@ -8,6 +8,7 @@
 from .image import *
 from .linode import *
 from .lke import *
+from .lke_tier import *
 from .longview import *
 from .networking import *
 from .nodebalancer import *

linode_api4/groups/lke.py

@@ -2,6 +2,7 @@
 
 from linode_api4.errors import UnexpectedResponseError
 from linode_api4.groups import Group
+from linode_api4.groups.lke_tier import LKETierGroup
 from linode_api4.objects import (
     KubeVersion,
     LKECluster,
@@ -11,7 +12,6 @@
     drop_null_keys,
 )
 from linode_api4.objects.base import _flatten_request_body_recursive
-from linode_api4.paths import LKETier
 
 
 class LKEGroup(Group):
@@ -191,7 +191,7 @@ def types(self, *filters):
             LKEType, *filters, endpoint="/lke/types"
         )
 
-    def tier(self, id: str) -> LKETier:
+    def tier(self, id: str) -> LKETierGroup:
         """
         Returns an object representing the LKE tier API path.
 
@@ -202,4 +202,4 @@ def tier(self, id: str) -> LKETier:
         :rtype: LKETier
         """
 
-        return LKETier(self.client, id)
+        return LKETierGroup(self.client, id)

linode_api4/groups/lke_tier.py

@@ -0,0 +1,40 @@
+from linode_api4.groups import Group
+from linode_api4.objects import TieredKubeVersion
+
+
+class LKETierGroup(Group):
+    """
+    Encapsulates methods related to a specific LKE tier. This
+    should not be instantiated on its own, but should instead be used through
+    an instance of :any:`LinodeClient`::
+
+       client = LinodeClient(token)
+       instances = client.lke.tier("standard") # use the LKETierGroup
+
+    This group contains all features beneath the `/lke/tiers/{tier}` group in the API v4.
+    """
+
+    def __init__(self, client: "LinodeClient", tier: str):
+        super().__init__(client)
+        self.tier = tier
+
+    def versions(self, *filters):
+        """
+        Returns a paginated list of versions for this tier matching the given filters.
+
+        API Documentation: Not Yet Available
+
+        :param filters: Any number of filters to apply to this query.
+                        See :doc:`Filtering Collections</linode_api4/objects/filtering>`
+                        for more details on filtering.
+
+        :returns: A paginated list of kube versions that match the query.
+        :rtype: PaginatedList of TieredKubeVersion
+        """
+
+        return self.client._get_and_filter(
+            TieredKubeVersion,
+            endpoint=f"/lke/tiers/{self.tier}/versions",
+            parent_id=self.tier,
+            *filters,
+        )

linode_api4/paths/__init__.py

@@ -139,7 +139,10 @@ def lke_cluster_with_apl(test_linode_client):
 
 @pytest.fixture(scope="session")
 def lke_cluster_enterprise(test_linode_client):
-    version = test_linode_client.lke.tier("enterprise").versions()[0]
+    # We use the oldest version here so we can test upgrades
+    version = sorted(
+        v.id for v in test_linode_client.lke.tier("enterprise").versions()
+    )[0]
 
     region = get_region(
         test_linode_client, {"Kubernetes Enterprise", "Disk Encryption"}
@@ -428,18 +431,25 @@ def test_lke_cluster_with_apl(lke_cluster_with_apl):
     )
 
 
-def test_lke_cluster_enterprise(lke_cluster_enterprise):
+def test_lke_cluster_enterprise(test_linode_client, lke_cluster_enterprise):
     lke_cluster_enterprise.invalidate()
     assert lke_cluster_enterprise.tier == "enterprise"
 
     pool = lke_cluster_enterprise.pools[0]
     assert str(pool.k8s_version) == lke_cluster_enterprise.k8s_version.id
     assert pool.update_strategy == "rolling_update"
 
+    target_version = sorted(
+        v.id for v in test_linode_client.lke.tier("enterprise").versions()
+    )[0]
     pool.update_strategy = "on_recycle"
-    # Should we test updating the pool k8s_version field here?
+    pool.k8s_version = target_version
+
     pool.save()
 
+    pool.invalidate()
+
+    assert pool.k8s_version == target_version
     assert pool.update_strategy == "on_recycle"
 
 

linode_api4/paths/api_path.py

@@ -1,9 +1,9 @@
 from test.unit.base import ClientBaseCase
 
 
-class LKETierPathTest(ClientBaseCase):
+class LKETierGroupTest(ClientBaseCase):
     """
-    Tests methods of the LKETierPath
+    Tests methods under the LKETierGroup class.
     """
 
     def test_list_versions(self):