test: add deployment multikueue e2e tests

test/e2e/multikueue/e2e_test.go

@@ -28,6 +28,7 @@ import (
 	"github.com/onsi/ginkgo/v2"
 	"github.com/onsi/gomega"
 	rayv1 "github.com/ray-project/kuberay/ray-operator/apis/ray/v1"
+	appsv1 "k8s.io/api/apps/v1"
 	batchv1 "k8s.io/api/batch/v1"
 	corev1 "k8s.io/api/core/v1"
 	apimeta "k8s.io/apimachinery/pkg/api/meta"
@@ -48,6 +49,7 @@ import (
 	workloadrayjob "sigs.k8s.io/kueue/pkg/controller/jobs/rayjob"
 	utilpod "sigs.k8s.io/kueue/pkg/util/pod"
 	utiltesting "sigs.k8s.io/kueue/pkg/util/testing"
+	testingdeployment "sigs.k8s.io/kueue/pkg/util/testingjobs/deployment"
 	testingjob "sigs.k8s.io/kueue/pkg/util/testingjobs/job"
 	testingjobset "sigs.k8s.io/kueue/pkg/util/testingjobs/jobset"
 	testingmpijob "sigs.k8s.io/kueue/pkg/util/testingjobs/mpijob"
@@ -270,6 +272,75 @@ var _ = ginkgo.Describe("MultiKueue", func() {
 			})
 		})
 
+		ginkgo.It("Should create pods on worker if the deployment is admitted", func() {
+			deployment := testingdeployment.MakeDeployment("deployment", managerNs.Name).
+				Image(util.E2eTestSleepImage, []string{"10m"}).
+				Replicas(3).
+				Queue(managerLq.Name).
+				Obj()
+
+			ginkgo.By("Creating the deployment", func() {
+				gomega.Expect(k8sManagerClient.Create(ctx, deployment)).Should(gomega.Succeed())
+				gomega.Eventually(func(g gomega.Gomega) {
+					createdDeployment := &appsv1.Deployment{}
+					g.Expect(k8sManagerClient.Get(ctx, client.ObjectKeyFromObject(deployment), createdDeployment)).To(gomega.Succeed())
+				}, util.Timeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			ginkgo.By("Wait for replicas ready", func() {
+				createdDeployment := &appsv1.Deployment{}
+				gomega.Eventually(func(g gomega.Gomega) {
+					g.Expect(k8sManagerClient.Get(ctx, client.ObjectKeyFromObject(deployment), createdDeployment)).To(gomega.Succeed())
+					g.Expect(createdDeployment.Status.ReadyReplicas).To(gomega.Equal(int32(3)))
+				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+			})
+
+			pods := &corev1.PodList{}
+			gomega.Expect(k8sManagerClient.List(ctx, pods, client.InNamespace(managerNs.Namespace),
+				client.MatchingLabels(deployment.Spec.Selector.MatchLabels))).To(gomega.Succeed())
+
+			ginkgo.By("Ensure Pod Workloads are created and local Pods receive status update", func() {
+				for _, pod := range pods.Items { // We want to test that all deployment pods have workloads.
+					createdLeaderWorkload := &kueue.Workload{}
+					wlLookupKey := types.NamespacedName{Name: workloadpod.GetWorkloadNameForPod(pod.Name, pod.UID), Namespace: managerNs.Name}
+					gomega.Eventually(func(g gomega.Gomega) {
+						g.Expect(k8sManagerClient.Get(ctx, wlLookupKey, createdLeaderWorkload)).To(gomega.Succeed())
+					}, util.Timeout, util.Interval).Should(gomega.Succeed())
+
+					// Local pods should be in "Running" phase
+					gomega.Eventually(func(g gomega.Gomega) {
+						createdPod := &corev1.Pod{}
+						g.Expect(k8sManagerClient.Get(ctx, client.ObjectKey{Namespace: pod.Namespace, Name: pod.Name}, createdPod)).To(gomega.Succeed())
+						g.Expect(utilpod.HasGate(createdPod, "kueue.x-k8s.io/admission")).To(gomega.BeTrue())
+						g.Expect(createdPod.Status.Phase).Should(gomega.Equal(corev1.PodRunning))
+					}, util.Timeout, util.Interval).Should(gomega.Succeed())
+				}
+			})
+			deploymentConditions := []appsv1.DeploymentCondition{
+				{
+					Type:    appsv1.DeploymentAvailable,
+					Status:  corev1.ConditionTrue,
+					Reason:  "MinimumReplicasAvailable",
+					Message: "Deployment has minimum availability.",
+				},
+				{
+					Type:   appsv1.DeploymentProgressing,
+					Status: corev1.ConditionTrue,
+					Reason: "NewReplicaSetAvailable",
+				},
+			}
+
+			ginkgo.By("Waiting for the deployment to get status updates", func() {
+				gomega.Eventually(func(g gomega.Gomega) {
+					createdDeployment := appsv1.Deployment{}
+					g.Expect(k8sManagerClient.Get(ctx, client.ObjectKeyFromObject(deployment), &createdDeployment)).To(gomega.Succeed())
+					g.Expect(createdDeployment.Status.ReadyReplicas).To(gomega.Equal(int32(3)))
+					g.Expect(createdDeployment.Status.Replicas).To(gomega.Equal(int32(3)))
+					g.Expect(createdDeployment.Status.UpdatedReplicas).To(gomega.Equal(int32(3)))
+					g.Expect(createdDeployment.Status.Conditions).To(gomega.BeComparableTo(deploymentConditions, cmpopts.IgnoreFields(appsv1.DeploymentCondition{}, "LastTransitionTime", "LastUpdateTime", "Message"))) // Ignoring message as it is required to gather the pod replica set
+				}, util.LongTimeout, util.Interval).Should(gomega.Succeed())
+			})
+		})
 		ginkgo.It("Should run a job on worker if admitted", func() {
 			if managerK8SVersion.LessThan(versionutil.MustParseSemantic("1.30.0")) {
 				ginkgo.Skip("the managers kubernetes version is less then 1.30")