add tests

Signed-off-by: oliver könig <okoenig@nvidia.com>

Author: ko3n1g

test/run/ray/test_kuberay.py

@@ -217,7 +217,6 @@ def test_create_basic_cluster(self, cluster_with_basic_executor, mock_k8s_client
         assert body["metadata"]["name"] == "testuser-test-cluster-raycluster"
         assert body["metadata"]["namespace"] == "test-namespace"
         assert body["spec"]["rayVersion"] == "2.43.0"
-
     def test_create_advanced_cluster(self, cluster_with_advanced_executor, mock_k8s_clients):
         """Test creating an advanced Ray cluster with volumes and custom settings."""
         mock_api, _ = mock_k8s_clients
@@ -748,11 +747,9 @@ class TestKubeRayExecutorUtilityFunctions:
     @pytest.fixture
     def basic_cluster_dict(self):
         """Create a basic cluster dictionary for testing."""
-        from nemo_run.core.execution.kuberay import (
-            populate_meta,
-            populate_ray_head,
-            populate_worker_group,
-        )
+        from nemo_run.core.execution.kuberay import (populate_meta,
+                                                     populate_ray_head,
+                                                     populate_worker_group)
 
         cluster = {}
         cluster = populate_meta(cluster, "test-cluster", "default", {}, "2.43.0")
@@ -866,7 +863,8 @@ def test_populate_ray_head_missing_spec(self):
 
     def test_populate_ray_head_without_dashboard_host(self):
         """Test populate_ray_head automatically adds dashboard-host."""
-        from nemo_run.core.execution.kuberay import populate_meta, populate_ray_head
+        from nemo_run.core.execution.kuberay import (populate_meta,
+                                                     populate_ray_head)
 
         cluster = populate_meta({}, "test-cluster", "default", {}, "2.43.0")
         ray_start_params = {}
@@ -893,7 +891,8 @@ def test_populate_ray_head_without_dashboard_host(self):
 
     def test_update_worker_group_replicas_success(self, basic_cluster_dict):
         """Test successfully updating worker group replicas."""
-        from nemo_run.core.execution.kuberay import update_worker_group_replicas
+        from nemo_run.core.execution.kuberay import \
+            update_worker_group_replicas
 
         cluster, success = update_worker_group_replicas(
             basic_cluster_dict, "workers", max_replicas=5, min_replicas=1, replicas=3
@@ -907,7 +906,8 @@ def test_update_worker_group_replicas_success(self, basic_cluster_dict):
 
     def test_update_worker_group_replicas_not_found(self, basic_cluster_dict):
         """Test updating non-existent worker group replicas."""
-        from nemo_run.core.execution.kuberay import update_worker_group_replicas
+        from nemo_run.core.execution.kuberay import \
+            update_worker_group_replicas
 
         cluster, success = update_worker_group_replicas(
             basic_cluster_dict, "nonexistent", max_replicas=5, min_replicas=1, replicas=3
@@ -917,7 +917,8 @@ def test_update_worker_group_replicas_not_found(self, basic_cluster_dict):
 
     def test_update_worker_group_resources_success(self, basic_cluster_dict):
         """Test successfully updating worker group resources."""
-        from nemo_run.core.execution.kuberay import update_worker_group_resources
+        from nemo_run.core.execution.kuberay import \
+            update_worker_group_resources
 
         cluster, success = update_worker_group_resources(
             basic_cluster_dict,
@@ -937,7 +938,8 @@ def test_update_worker_group_resources_success(self, basic_cluster_dict):
 
     def test_update_worker_group_resources_all_containers(self, basic_cluster_dict):
         """Test updating resources for all containers in a worker group."""
-        from nemo_run.core.execution.kuberay import update_worker_group_resources
+        from nemo_run.core.execution.kuberay import \
+            update_worker_group_resources
 
         # Add another container to test "all_containers" functionality
         basic_cluster_dict["spec"]["workerGroupSpecs"][0]["template"]["spec"]["containers"].append(
@@ -966,7 +968,8 @@ def test_update_worker_group_resources_all_containers(self, basic_cluster_dict):
 
     def test_update_worker_group_resources_specific_container(self, basic_cluster_dict):
         """Test updating resources for a specific container."""
-        from nemo_run.core.execution.kuberay import update_worker_group_resources
+        from nemo_run.core.execution.kuberay import \
+            update_worker_group_resources
 
         # Add another container
         basic_cluster_dict["spec"]["workerGroupSpecs"][0]["template"]["spec"]["containers"].append(
@@ -993,7 +996,8 @@ def test_update_worker_group_resources_specific_container(self, basic_cluster_di
 
     def test_update_worker_group_resources_no_containers(self):
         """Test updating resources when worker group has no containers."""
-        from nemo_run.core.execution.kuberay import update_worker_group_resources
+        from nemo_run.core.execution.kuberay import \
+            update_worker_group_resources
 
         cluster = {
             "spec": {
@@ -1063,7 +1067,8 @@ def test_delete_worker_group_not_found(self, basic_cluster_dict):
 
     def test_worker_group_with_labels_and_annotations(self):
         """Test creating worker group with labels and annotations."""
-        from nemo_run.core.execution.kuberay import populate_meta, populate_worker_group
+        from nemo_run.core.execution.kuberay import (populate_meta,
+                                                     populate_worker_group)
 
         cluster = populate_meta({}, "test-cluster", "default", {}, "2.43.0")
         cluster["spec"]["workerGroupSpecs"] = []
@@ -2074,3 +2079,170 @@ def test_cluster_create_without_lifecycle_kwargs(self, mock_k8s_clients):
             # Should create lifecycle_kwargs and succeed
             assert hasattr(executor, "lifecycle_kwargs")
             assert mock_api.create_namespaced_custom_object.called
+
+
+class TestKubeConfigLoading:
+    """Test kube config loading fallback scenarios."""
+
+    def test_kuberay_cluster_load_kube_config_success(self):
+        """Test successful loading of kube config for KubeRayCluster."""
+        with patch("nemo_run.run.ray.kuberay.config.load_kube_config") as mock_load_kube:
+            with patch("nemo_run.run.ray.kuberay.client.CustomObjectsApi"):
+                with patch("nemo_run.run.ray.kuberay.client.CoreV1Api"):
+                    with patch("nemo_run.run.ray.kuberay.get_user", return_value="testuser"):
+                        executor = KubeRayExecutor(namespace="test-namespace")
+                        cluster = KubeRayCluster(name="test-cluster", executor=executor)
+
+                        # Verify load_kube_config was called and succeeded
+                        mock_load_kube.assert_called_once()
+
+    def test_kuberay_cluster_fallback_to_incluster_config(self):
+        """Test fallback to incluster config when kube config fails for KubeRayCluster."""
+        with patch(
+            "nemo_run.run.ray.kuberay.config.load_kube_config",
+            side_effect=Exception("Kube config not found"),
+        ) as mock_load_kube:
+            with patch(
+                "nemo_run.run.ray.kuberay.config.load_incluster_config"
+            ) as mock_load_incluster:
+                with patch("nemo_run.run.ray.kuberay.client.CustomObjectsApi"):
+                    with patch("nemo_run.run.ray.kuberay.client.CoreV1Api"):
+                        with patch("nemo_run.run.ray.kuberay.get_user", return_value="testuser"):
+                            executor = KubeRayExecutor(namespace="test-namespace")
+                            cluster = KubeRayCluster(name="test-cluster", executor=executor)
+
+                            # Verify fallback to incluster config
+                            mock_load_kube.assert_called_once()
+                            mock_load_incluster.assert_called_once()
+
+    def test_kuberay_cluster_both_configs_fail(self):
+        """Test when both kube config and incluster config fail for KubeRayCluster."""
+        kube_config_error = Exception("Kube config not found")
+        incluster_config_error = Exception("Not running in cluster")
+
+        with patch(
+            "nemo_run.run.ray.kuberay.config.load_kube_config", side_effect=kube_config_error
+        ):
+            with patch(
+                "nemo_run.run.ray.kuberay.config.load_incluster_config",
+                side_effect=incluster_config_error,
+            ):
+                with pytest.raises(Exception) as exc_info:
+                    with patch("nemo_run.run.ray.kuberay.get_user", return_value="testuser"):
+                        executor = KubeRayExecutor(namespace="test-namespace")
+                        KubeRayCluster(name="test-cluster", executor=executor)
+
+                # Should raise the original kube config error
+                assert exc_info.value == kube_config_error
+
+    def test_kuberay_job_load_kube_config_success(self):
+        """Test successful loading of kube config for KubeRayJob."""
+        with patch("nemo_run.run.ray.kuberay.config.load_kube_config") as mock_load_kube:
+            with patch("nemo_run.run.ray.kuberay.client.CustomObjectsApi"):
+                with patch("nemo_run.run.ray.kuberay.client.CoreV1Api"):
+                    with patch("nemo_run.run.ray.kuberay.get_user", return_value="testuser"):
+                        executor = KubeRayExecutor(namespace="test-namespace")
+                        job = KubeRayJob(name="test-job", executor=executor)
+
+                        # Verify load_kube_config was called and succeeded
+                        mock_load_kube.assert_called()
+
+    def test_kuberay_job_fallback_to_incluster_config(self):
+        """Test fallback to incluster config when kube config fails for KubeRayJob."""
+        with patch(
+            "nemo_run.run.ray.kuberay.config.load_kube_config",
+            side_effect=Exception("Kube config not found"),
+        ) as mock_load_kube:
+            with patch(
+                "nemo_run.run.ray.kuberay.config.load_incluster_config"
+            ) as mock_load_incluster:
+                with patch("nemo_run.run.ray.kuberay.client.CustomObjectsApi"):
+                    with patch("nemo_run.run.ray.kuberay.client.CoreV1Api"):
+                        with patch("nemo_run.run.ray.kuberay.get_user", return_value="testuser"):
+                            executor = KubeRayExecutor(namespace="test-namespace")
+                            job = KubeRayJob(name="test-job", executor=executor)
+
+                            # Verify fallback to incluster config
+                            # Called twice: once for executor, once for job
+                            assert mock_load_kube.call_count >= 1
+                            assert mock_load_incluster.call_count >= 1
+
+    def test_kuberay_job_both_configs_fail(self):
+        """Test when both kube config and incluster config fail for KubeRayJob."""
+        kube_config_error = Exception("Kube config not found")
+        incluster_config_error = Exception("Not running in cluster")
+
+        with patch(
+            "nemo_run.run.ray.kuberay.config.load_kube_config", side_effect=kube_config_error
+        ):
+            with patch(
+                "nemo_run.run.ray.kuberay.config.load_incluster_config",
+                side_effect=incluster_config_error,
+            ):
+                with pytest.raises(Exception) as exc_info:
+                    with patch("nemo_run.run.ray.kuberay.get_user", return_value="testuser"):
+                        executor = KubeRayExecutor(namespace="test-namespace")
+                        KubeRayJob(name="test-job", executor=executor)
+
+                # Should raise the original kube config error
+                assert exc_info.value == kube_config_error
+
+    def test_executor_load_kube_config_success(self):
+        """Test successful loading of kube config for KubeRayExecutor."""
+        with patch(
+            "nemo_run.core.execution.kuberay.config.load_kube_config"
+        ) as mock_load_kube:
+            with patch("nemo_run.core.execution.kuberay.client.CustomObjectsApi"):
+                with patch("nemo_run.core.execution.kuberay.client.CoreV1Api"):
+                    executor = KubeRayExecutor(namespace="test-namespace")
+
+                    # Verify load_kube_config was called and succeeded
+                    mock_load_kube.assert_called_once()
+
+    def test_executor_fallback_to_incluster_config(self):
+        """Test fallback to incluster config when kube config fails for KubeRayExecutor."""
+        with patch(
+            "nemo_run.core.execution.kuberay.config.load_kube_config",
+            side_effect=Exception("Kube config not found"),
+        ) as mock_load_kube:
+            with patch(
+                "nemo_run.core.execution.kuberay.config.load_incluster_config"
+            ) as mock_load_incluster:
+                with patch("nemo_run.core.execution.kuberay.client.CustomObjectsApi"):
+                    with patch("nemo_run.core.execution.kuberay.client.CoreV1Api"):
+                        executor = KubeRayExecutor(namespace="test-namespace")
+
+                        # Verify fallback to incluster config
+                        mock_load_kube.assert_called_once()
+                        mock_load_incluster.assert_called_once()
+
+    def test_executor_both_configs_fail(self):
+        """Test when both kube config and incluster config fail for KubeRayExecutor."""
+        kube_config_error = Exception("Kube config not found")
+        incluster_config_error = Exception("Not running in cluster")
+
+        with patch(
+            "nemo_run.core.execution.kuberay.config.load_kube_config",
+            side_effect=kube_config_error,
+        ):
+            with patch(
+                "nemo_run.core.execution.kuberay.config.load_incluster_config",
+                side_effect=incluster_config_error,
+            ):
+                with pytest.raises(Exception) as exc_info:
+                    KubeRayExecutor(namespace="test-namespace")
+
+                # Should raise the original kube config error
+                assert exc_info.value == kube_config_error
+            ):
+                with pytest.raises(Exception) as exc_info:
+                    KubeRayExecutor(namespace="test-namespace")
+
+                # Should raise the original kube config error
+                assert exc_info.value == kube_config_error
+            ):
+                with pytest.raises(Exception) as exc_info:
+                    KubeRayExecutor(namespace="test-namespace")
+
+                # Should raise the original kube config error
+                assert exc_info.value == kube_config_error