[e2e] Enhance RayCluster Auth E2E (#4231)

* [e2e] Enhance RayCluster Auth E2E

Signed-off-by: seanlaii <qazwsx0939059006@gmail.com>

* fix test

* fix test

* fix test

* fix test

* fix test

---------

Signed-off-by: seanlaii <qazwsx0939059006@gmail.com>

Author: seanlaii

ray-operator/test/e2e/raycluster_auth_test.go

@@ -1,9 +1,13 @@
 package e2e
 
 import (
+	"fmt"
 	"testing"
+	"time"
 
 	. "github.com/onsi/gomega"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	rayv1 "github.com/ray-project/kuberay/ray-operator/apis/ray/v1"
 	"github.com/ray-project/kuberay/ray-operator/controllers/ray/utils"
@@ -52,6 +56,80 @@ func TestRayClusterAuthOptions(t *testing.T) {
 			VerifyContainerAuthTokenEnvVars(test, rayCluster, &workerPod.Spec.Containers[utils.RayContainerIndex])
 		}
 
-		// TODO(andrewsykim): add job submission test with and without token once a Ray version with token support is released.
+		// Get auth token for job submission tests
+		authToken := getAuthTokenFromPod(test, rayCluster, headPod)
+		g.Expect(authToken).NotTo(BeEmpty(), "Auth token should be present")
+
+		// Test job submission with auth token using Ray Job CLI
+		test.T().Run("Submit job with auth token should succeed", func(_ *testing.T) {
+			LogWithTimestamp(test.T(), "Testing job submission WITH auth token")
+
+			submissionId := fmt.Sprintf("test-job-with-auth-%d", time.Now().Unix())
+
+			// Submit job via Ray Job CLI with auth token
+			// Set RAY_AUTH_TOKEN environment variable for authentication
+			submitCmd := []string{
+				"bash", "-c",
+				fmt.Sprintf("RAY_AUTH_TOKEN=%s ray job submit --address http://127.0.0.1:8265 --submission-id %s --no-wait -- python -c 'import ray; ray.init(); print(\"Job with auth succeeded\")'",
+					authToken, submissionId),
+			}
+
+			stdout, _ := ExecPodCmd(test, headPod, headPod.Spec.Containers[utils.RayContainerIndex].Name, submitCmd)
+
+			// Verify job was submitted successfully
+			g.Expect(stdout.String()).To(ContainSubstring(submissionId), "Job submission should succeed with valid auth token")
+
+			// Verify job status is queryable with auth token (confirms auth works)
+			g.Eventually(func(g Gomega) {
+				statusCmd := []string{
+					"bash", "-c",
+					fmt.Sprintf("RAY_AUTH_TOKEN=%s ray job status --address http://127.0.0.1:8265 %s", authToken, submissionId),
+				}
+				stdout, _ := ExecPodCmd(test, headPod, headPod.Spec.Containers[utils.RayContainerIndex].Name, statusCmd)
+				g.Expect(stdout.String()).To(ContainSubstring("succeeded"))
+			}, TestTimeoutShort).Should(Succeed())
+
+			LogWithTimestamp(test.T(), "Successfully submitted and verified job with auth token")
+		})
+
+		test.T().Run("Submit job with incorrect auth token should fail", func(_ *testing.T) {
+			LogWithTimestamp(test.T(), "Testing job submission WITH incorrect auth token (should fail)")
+
+			submissionId := fmt.Sprintf("test-job-bad-auth-%d", time.Now().Unix())
+
+			// Submit job via Ray Job CLI with INCORRECT auth token
+			incorrectToken := "incorrect-token-12345"
+			submitCmd := []string{
+				"bash", "-c",
+				fmt.Sprintf("RAY_AUTH_TOKEN=%s ray job submit --address http://127.0.0.1:8265 --submission-id %s --no-wait -- python -c 'print(\"Should not run\")'",
+					incorrectToken, submissionId),
+			}
+
+			_, stderr := ExecPodCmd(test, headPod, headPod.Spec.Containers[utils.RayContainerIndex].Name, submitCmd, true)
+
+			// Verify response indicates authentication failure
+			g.Expect(stderr.String()).To(ContainSubstring("Unauthenticated"), "Job submission should fail with Unauthorized when auth token is incorrect")
+
+			LogWithTimestamp(test.T(), "Job submission correctly rejected with incorrect auth token")
+		})
 	})
 }
+
+// getAuthTokenFromPod extracts the auth token from the pod's environment variables.
+// It reads the token from the secret referenced by the RAY_AUTH_TOKEN environment variable.
+func getAuthTokenFromPod(test Test, rayCluster *rayv1.RayCluster, pod *corev1.Pod) string {
+	test.T().Helper()
+	g := NewWithT(test.T())
+
+	for _, envVar := range pod.Spec.Containers[utils.RayContainerIndex].Env {
+		if envVar.Name == utils.RAY_AUTH_TOKEN_ENV_VAR {
+			if envVar.ValueFrom != nil && envVar.ValueFrom.SecretKeyRef != nil {
+				secret, err := test.Client().Core().CoreV1().Secrets(rayCluster.Namespace).
+					Get(test.Ctx(), envVar.ValueFrom.SecretKeyRef.Name, metav1.GetOptions{})
+				g.Expect(err).NotTo(HaveOccurred())
+				return string(secret.Data[envVar.ValueFrom.SecretKeyRef.Key])
+			}
+		}
+	}
+	return ""
+}

ray-operator/test/support/core.go

@@ -68,7 +68,9 @@ func storeContainerLog(t Test, namespace *corev1.Namespace, podName, containerNa
 	WriteToOutputDir(t, containerLogFileName, Log, bytes)
 }
 
-func ExecPodCmd(t Test, pod *corev1.Pod, containerName string, cmd []string) (bytes.Buffer, bytes.Buffer) {
+func ExecPodCmd(t Test, pod *corev1.Pod, containerName string, cmd []string, allowError ...bool) (bytes.Buffer, bytes.Buffer) {
+	shouldAllowError := len(allowError) > 0 && allowError[0]
+
 	req := t.Client().Core().CoreV1().RESTClient().
 		Post().
 		Resource("pods").
@@ -99,7 +101,11 @@ func ExecPodCmd(t Test, pod *corev1.Pod, containerName string, cmd []string) (by
 	})
 	LogWithTimestamp(t.T(), "Command stdout: %s", stdout.String())
 	LogWithTimestamp(t.T(), "Command stderr: %s", stderr.String())
-	require.NoError(t.T(), err)
+
+	if !shouldAllowError {
+		require.NoError(t.T(), err, "Command failed unexpectedly")
+	}
+
 	return stdout, stderr
 }