Updated get_cluster to account for the new kueue parameter

Author: Bobbins228

src/codeflare_sdk/cluster/cluster.py

@@ -461,6 +461,7 @@ def job_logs(self, job_id: str) -> str:
     def from_k8_cluster_object(
         rc,
         appwrapper=True,
+        kueue=True,
         write_to_file=False,
         verify_tls=True,
     ):
@@ -470,6 +471,11 @@ def from_k8_cluster_object(
             if "orderedinstance" in rc["metadata"]["labels"]
             else []
         )
+        local_queue = (
+            rc["metadata"].get("labels", dict()).get("kueue.x-k8s.io/queue-name", None)
+            if kueue
+            else None
+        )
 
         cluster_config = ClusterConfiguration(
             name=rc["metadata"]["name"],
@@ -499,9 +505,8 @@ def from_k8_cluster_object(
             appwrapper=appwrapper,
             write_to_file=write_to_file,
             verify_tls=verify_tls,
-            local_queue=rc["metadata"]
-            .get("labels", dict())
-            .get("kueue.x-k8s.io/queue-name", None),
+            kueue=kueue,
+            local_queue=local_queue,
         )
         return Cluster(cluster_config)
 
@@ -634,9 +639,11 @@ def get_cluster(
     for rc in rcs["items"]:
         if rc["metadata"]["name"] == cluster_name:
             appwrapper = _check_aw_exists(cluster_name, namespace)
+            kueue = _check_kueue_enabled(rc)
             return Cluster.from_k8_cluster_object(
                 rc,
                 appwrapper=appwrapper,
+                kueue=kueue,
                 write_to_file=write_to_file,
                 verify_tls=verify_tls,
             )
@@ -691,6 +698,14 @@ def _check_aw_exists(name: str, namespace: str) -> bool:
     return False
 
 
+def _check_kueue_enabled(rc) -> bool:
+    # This function will check if a Ray Cluster is being managed by Kueue by checking for the local queue label
+    for label in rc["metadata"]["labels"]:
+        if label == "kueue.x-k8s.io/queue-name":
+            return True
+    return False
+
+
 # Cant test this until get_current_namespace is fixed and placed in this function over using `self`
 def _get_ingress_domain(self):  # pragma: no cover
     config_check()