test

Author: pawelpaszki

tests/e2e/mnist_raycluster_sdk_kind_test.py

@@ -1,4 +1,5 @@
 import requests
+import subprocess
 
 from time import sleep
 
@@ -16,8 +17,16 @@
 class TestRayClusterSDKKind:
     def setup_method(self):
         initialize_kubernetes_client(self)
+        self.port_forward_process = None
+
+    def cleanup_port_forward(self):
+        if self.port_forward_process:
+            self.port_forward_process.terminate()
+            self.port_forward_process.wait(timeout=10)
+            self.port_forward_process = None
 
     def teardown_method(self):
+        self.cleanup_port_forward()
         delete_namespace(self)
         delete_kueue_resources(self)
 
@@ -67,46 +76,74 @@ def run_mnist_raycluster_sdk_kind(
 
         self.assert_jobsubmit_withoutlogin_kind(cluster, accelerator, number_of_gpus)
 
-        assert_get_cluster_and_jobsubmit(
-            self, "mnist", accelerator="gpu", number_of_gpus=1
-        )
+        # Note: assert_get_cluster_and_jobsubmit uses cluster.job_client which requires
+        # the dashboard URL (HTTPRoute/Route). Since this is not available on KinD,
+        # we skip this call. Job submission is already tested above with port-forwarding.
+        # This function is tested on OpenShift in mnist_raycluster_sdk_test.py
+
+        cluster.down()
 
     # Assertions
 
     def assert_jobsubmit_withoutlogin_kind(self, cluster, accelerator, number_of_gpus):
-        ray_dashboard = cluster.cluster_dashboard_uri()
+        # Use port-forwarding to access the dashboard since HTTPRoute/Route is not available on KinD
+        local_port = "8265"
+        dashboard_port = "8265"
+        cluster_name = cluster.config.name
+
+        port_forward_cmd = [
+            "kubectl",
+            "port-forward",
+            "-n",
+            self.namespace,
+            f"svc/{cluster_name}-head-svc",
+            f"{local_port}:{dashboard_port}",
+        ]
+        self.port_forward_process = subprocess.Popen(
+            port_forward_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL
+        )
+
+        # Wait for port-forward to be ready
+        sleep(5)
+
+        ray_dashboard = f"http://localhost:{local_port}"
         client = RayJobClient(address=ray_dashboard, verify=False)
 
-        submission_id = client.submit_job(
-            entrypoint="python mnist.py",
-            runtime_env={
-                "working_dir": "./tests/e2e/",
-                "pip": "./tests/e2e/mnist_pip_requirements.txt",
-                "env_vars": get_setup_env_variables(ACCELERATOR=accelerator),
-            },
-            entrypoint_num_gpus=number_of_gpus,
-        )
-        print(f"Submitted job with ID: {submission_id}")
-        done = False
-        time = 0
-        timeout = 900
-        while not done:
-            status = client.get_job_status(submission_id)
-            if status.is_terminal():
-                break
-            if not done:
-                print(status)
-                if timeout and time >= timeout:
-                    raise TimeoutError(f"job has timed out after waiting {timeout}s")
-                sleep(5)
-                time += 5
-
-        logs = client.get_job_logs(submission_id)
-        print(logs)
-
-        self.assert_job_completion(status)
-
-        client.delete_job(submission_id)
+        try:
+            submission_id = client.submit_job(
+                entrypoint="python mnist.py",
+                runtime_env={
+                    "working_dir": "./tests/e2e/",
+                    "pip": "./tests/e2e/mnist_pip_requirements.txt",
+                    "env_vars": get_setup_env_variables(ACCELERATOR=accelerator),
+                },
+                entrypoint_num_gpus=number_of_gpus,
+            )
+            print(f"Submitted job with ID: {submission_id}")
+            done = False
+            time = 0
+            timeout = 900
+            while not done:
+                status = client.get_job_status(submission_id)
+                if status.is_terminal():
+                    break
+                if not done:
+                    print(status)
+                    if timeout and time >= timeout:
+                        raise TimeoutError(
+                            f"job has timed out after waiting {timeout}s"
+                        )
+                    sleep(5)
+                    time += 5
+
+            logs = client.get_job_logs(submission_id)
+            print(logs)
+
+            self.assert_job_completion(status)
+
+            client.delete_job(submission_id)
+        finally:
+            self.cleanup_port_forward()
 
     def assert_job_completion(self, status):
         if status == "SUCCEEDED":