Enhance end-to-end tests (#62)

* add e2e tests for completion detection and load testing
* add e2e tests for validating webhook

Author: dgrove-oss

.github/workflows/CI-standalone.yaml

@@ -57,4 +57,4 @@ jobs:
         make deploy-aw -e GIT_BRANCH=${{ env.GIT_BRANCH }} TAG=${{ env.GIT_BRANCH }}-${{ env.TAG }}
 
     - name: Run E2E tests
-      run: LABEL_FILTER="Standalone" ./hack/run-tests-on-cluster.sh
+      run: LABEL_FILTER="Standalone,Webhook" ./hack/run-tests-on-cluster.sh

hack/e2e-util.sh

@@ -167,7 +167,7 @@ function configure_cluster {
   echo ""
 
   # Sleep until the kubeflow operator is running
-  echo "Waiting for pods in the kueueflow namespace to become ready"
+  echo "Waiting for pods in the kubeflow namespace to become ready"
   while [[ $(kubectl get pods -n kubeflow -o 'jsonpath={..status.conditions[?(@.type=="Ready")].status}' | tr ' ' '\n' | sort -u) != "True" ]]
   do
       echo -n "." && sleep 1;

hack/run-tests-on-cluster.sh

@@ -19,7 +19,7 @@ export GORACE=1
 export CLEANUP_CLUSTER=${CLEANUP_CLUSTER:-"false"}
 export CLUSTER_STARTED="true"
 export KUTTL_TEST_SUITES=("")
-export LABEL_FILTER=${LABEL_FILTER:-"Kueue"}
+export LABEL_FILTER=${LABEL_FILTER:-"Kueue,Webhook"}
 
 source ${ROOT_DIR}/hack/e2e-util.sh
 

test/e2e/appwrapper_test.go

@@ -17,10 +17,17 @@ limitations under the License.
 package e2e
 
 import (
+	"context"
+	"encoding/json"
+	"fmt"
 	"time"
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/utils/ptr"
 
 	workloadv1beta2 "github.com/project-codeflare/appwrapper/api/v1beta2"
 )
@@ -72,14 +79,137 @@ var _ = Describe("AppWrapper E2E Test", func() {
 			appwrappers = append(appwrappers, aw)
 			Expect(waitAWPodsReady(ctx, aw)).Should(Succeed())
 		})
+	})
+
+	// TODO: KubeRay GVKs (would have to deploy KubeRay operator on e2e test cluster)
+
+	// TODO: JobSets (would have to deploy JobSet controller on e2e test cluster)
+
+	Describe("Webhook Enforces AppWrapper Invariants", Label("Webhook"), func() {
+		Context("Structural Invariants", func() {
+			It("There must be at least one podspec (a)", func() {
+				aw := toAppWrapper()
+				Expect(getClient(ctx).Create(ctx, aw)).ShouldNot(Succeed())
+			})
+
+			It("There must be at least one podspec (b)", func() {
+				aw := toAppWrapper(service())
+				Expect(getClient(ctx).Create(ctx, aw)).ShouldNot(Succeed())
+			})
+
+			It("There must be no more than 8 podspecs", func() {
+				aw := toAppWrapper(pod(100), pod(100), pod(100), pod(100), pod(100), pod(100), pod(100), pod(100), pod(100))
+				Expect(getClient(ctx).Create(ctx, aw)).ShouldNot(Succeed())
+			})
+
+			It("Non-existent PodSpec paths are rejected", func() {
+				comp := deployment(4, 100)
+				comp.PodSets[0].Path = "template.spec.missing"
+				aw := toAppWrapper(comp)
+				Expect(getClient(ctx).Create(ctx, aw)).ShouldNot(Succeed())
+
+				comp.PodSets[0].Path = ""
+				aw = toAppWrapper(comp)
+				Expect(getClient(ctx).Create(ctx, aw)).ShouldNot(Succeed())
+			})
+
+			It("PodSpec paths must refer to a PodSpecTemplate", func() {
+				comp := deployment(4, 100)
+				comp.PodSets[0].Path = "template.spec.template.metadata"
+				aw := toAppWrapper(comp)
+				Expect(getClient(ctx).Create(ctx, aw)).ShouldNot(Succeed())
+			})
+
+			It("Validation of Array and Map path elements", func() {
+				comp := jobSet(2, 100)
+				comp.PodSets[0].Path = "template.spec.replicatedJobs.template.spec.template"
+				aw := toAppWrapper(comp)
+				Expect(getClient(ctx).Create(ctx, aw)).ShouldNot(Succeed())
+
+				comp.PodSets[0].Path = "template.spec.replicatedJobs"
+				aw = toAppWrapper(comp)
+				Expect(getClient(ctx).Create(ctx, aw)).ShouldNot(Succeed())
+
+				comp.PodSets[0].Path = "template.spec.replicatedJobs[0].template[0].spec.template"
+				aw = toAppWrapper(comp)
+				Expect(getClient(ctx).Create(ctx, aw)).ShouldNot(Succeed())
+
+				comp.PodSets[0].Path = "template.spec.replicatedJobs[10].template.spec.template"
+				aw = toAppWrapper(comp)
+				Expect(getClient(ctx).Create(ctx, aw)).ShouldNot(Succeed())
+
+				comp.PodSets[0].Path = "template.spec.replicatedJobs[-1].template.spec.template"
+				aw = toAppWrapper(comp)
+				Expect(getClient(ctx).Create(ctx, aw)).ShouldNot(Succeed())
+
+				comp.PodSets[0].Path = "template.spec.replicatedJobs[a10].template.spec.template"
+				aw = toAppWrapper(comp)
+				Expect(getClient(ctx).Create(ctx, aw)).ShouldNot(Succeed())
+
+				comp.PodSets[0].Path = "template.spec.replicatedJobs[1"
+				aw = toAppWrapper(comp)
+				Expect(getClient(ctx).Create(ctx, aw)).ShouldNot(Succeed())
+
+				comp.PodSets[0].Path = "template.spec.replicatedJobs[1]].template.spec.template"
+				aw = toAppWrapper(comp)
+				Expect(getClient(ctx).Create(ctx, aw)).ShouldNot(Succeed())
+			})
+		})
+
+		It("Components in other namespaces are rejected", func() {
+			aw := toAppWrapper(namespacedPod("test", 100))
+			Expect(getClient(ctx).Create(ctx, aw)).ShouldNot(Succeed())
+		})
+
+		It("Nested AppWrappers are rejected", func() {
+			child := toAppWrapper(pod(100))
+			childBytes, err := json.Marshal(child)
+			Expect(err).ShouldNot(HaveOccurred())
+			aw := toAppWrapper(pod(100), workloadv1beta2.AppWrapperComponent{
+				PodSets:  []workloadv1beta2.AppWrapperPodSet{},
+				Template: runtime.RawExtension{Raw: childBytes},
+			})
+			Expect(getClient(ctx).Create(ctx, aw)).ShouldNot(Succeed())
+		})
+
+		It("Sensitive fields of aw.Spec.Components are immutable", func() {
+			aw := createAppWrapper(ctx, pod(1000), deployment(4, 1000))
+			appwrappers = append(appwrappers, aw)
+			awName := types.NamespacedName{Name: aw.Name, Namespace: aw.Namespace}
+
+			Expect(updateAppWrapper(ctx, awName, func(aw *workloadv1beta2.AppWrapper) {
+				aw.Spec.Components[0].Template = aw.Spec.Components[1].Template
+			})).ShouldNot(Succeed())
 
-		// TODO: Additional Kubeflow Training Operator GVKs of interest
+			Expect(updateAppWrapper(ctx, awName, func(aw *workloadv1beta2.AppWrapper) {
+				aw.Spec.Components = append(aw.Spec.Components, aw.Spec.Components[0])
+			})).ShouldNot(Succeed())
 
+			Expect(updateAppWrapper(ctx, awName, func(aw *workloadv1beta2.AppWrapper) {
+				aw.Spec.Components[0].PodSets = append(aw.Spec.Components[0].PodSets, aw.Spec.Components[0].PodSets...)
+			})).ShouldNot(Succeed())
+
+			Expect(updateAppWrapper(ctx, awName, func(aw *workloadv1beta2.AppWrapper) {
+				aw.Spec.Components[0].PodSets[0].Path = "bad"
+			})).ShouldNot(Succeed())
+
+			Expect(updateAppWrapper(ctx, awName, func(aw *workloadv1beta2.AppWrapper) {
+				aw.Spec.Components[0].PodSets[0].Replicas = ptr.To(int32(12))
+			})).ShouldNot(Succeed())
+		})
 	})
 
-	Describe("Error Handling for Invalid Resources", func() {
-		// TODO: Replicate scenarios from the AdmissionController unit tests
+	Describe("Webhook Enforces RBAC", Label("Webhook"), func() {
+		It("AppWrapper containing permitted resources can be created", func() {
+			aw := toAppWrapper(pod(100))
+			Expect(getLimitedClient(ctx).Create(ctx, aw)).To(Succeed(), "Limited user should be allowed to create AppWrapper containing Pods")
+			Expect(getClient(ctx).Delete(ctx, aw)).To(Succeed())
+		})
 
+		It("AppWrapper containing unpermitted resources cannot be created", func() {
+			aw := toAppWrapper(deployment(4, 100))
+			Expect(getLimitedClient(ctx).Create(ctx, aw)).NotTo(Succeed(), "Limited user should not be allowed to create AppWrapper containing Deployments")
+		})
 	})
 
 	Describe("Queueing and Preemption", Label("Kueue"), func() {
@@ -103,24 +233,111 @@ var _ = Describe("AppWrapper E2E Test", func() {
 			appwrappers = []*workloadv1beta2.AppWrapper{aw2, aw3}
 			Expect(waitAWPodsReady(ctx, aw3)).Should(Succeed())
 		})
-
 	})
 
+	// AppWrapper consumes the entire quota itself; tests verify that we don't double count children
 	Describe("Recognition of Child Jobs", Label("Kueue"), func() {
-		// TODO: Test scenarios where the AW "just fits" in the quota and
-		//       contains components that Kueue might try to queue
-		//       but should not in this case because they are using the parent workload's quota
-		//  1. batch v1 jobs
-		//  2. pytorch jobs (which themself contain child Jobs)
+		It("Batch Job", func() {
+			aw := createAppWrapper(ctx, batchjob(2000))
+			appwrappers = append(appwrappers, aw)
+			Expect(waitAWPodsReady(ctx, aw)).Should(Succeed())
+		})
+
+		It("PyTorch Job", func() {
+			aw := createAppWrapper(ctx, pytorchjob(2, 1000))
+			appwrappers = append(appwrappers, aw)
+			Expect(waitAWPodsReady(ctx, aw)).Should(Succeed())
+		})
 
+		It("Compound Workloads", func() {
+			aw := createAppWrapper(ctx, batchjob(500), pytorchjob(2, 500), deployment(2, 250))
+			appwrappers = append(appwrappers, aw)
+			Expect(waitAWPodsReady(ctx, aw)).Should(Succeed())
+		})
 	})
 
-	Describe("Detection of Completion Status", Label("Kueue", "Standalone"), func() {
+	Describe("Detection of Completion Status", Label("slow"), Label("Kueue", "Standalone"), func() {
+		It("A successful Batch Job yields a successful AppWrapper", func() {
+			aw := createAppWrapper(ctx, succeedingBatchjob(500))
+			appwrappers = append(appwrappers, aw)
+			Expect(waitAWPodsReady(ctx, aw)).Should(Succeed())
+			Eventually(AppWrapperPhase(ctx, aw), 60*time.Second).Should(Equal(workloadv1beta2.AppWrapperSucceeded))
+		})
 
+		It("A failed Batch Job yields a failed AppWrapper", func() {
+			aw := createAppWrapper(ctx, failingBatchjob(500))
+			appwrappers = append(appwrappers, aw)
+			Expect(waitAWPodsReady(ctx, aw)).Should(Succeed())
+			Eventually(AppWrapperPhase(ctx, aw), 90*time.Second).Should(Equal(workloadv1beta2.AppWrapperFailed))
+		})
 	})
 
 	Describe("Load Testing", Label("slow"), Label("Kueue", "Standalone"), func() {
+		It("Create 50 AppWrappers", func() {
+			const (
+				awCount   = 50
+				cpuDemand = 5
+			)
 
+			By("Creating 50 AppWrappers")
+			replicas := 2
+			for i := 0; i < awCount; i++ {
+				aw := createAppWrapper(ctx, deployment(replicas, cpuDemand))
+				appwrappers = append(appwrappers, aw)
+			}
+			nonRunningAWs := appwrappers
+
+			By("Polling for all AppWrappers to be Running")
+			err := wait.PollUntilContextTimeout(ctx, 500*time.Millisecond, 1*time.Minute, false, func(ctx context.Context) (done bool, err error) {
+				t := time.Now()
+				toCheckAWS := make([]*workloadv1beta2.AppWrapper, 0, len(appwrappers))
+				for _, aw := range nonRunningAWs {
+					if !checkAppWrapperRunning(ctx, aw) {
+						toCheckAWS = append(toCheckAWS, aw)
+					}
+				}
+				nonRunningAWs = toCheckAWS
+				if len(toCheckAWS) == 0 {
+					fmt.Fprintf(GinkgoWriter, "\tAll AppWrappers Running at time %s\n", t.Format(time.RFC3339))
+					return true, nil
+				}
+				fmt.Fprintf(GinkgoWriter, "\tThere are %d non-Running AppWrappers at time %s\n", len(toCheckAWS), t.Format(time.RFC3339))
+				return false, nil
+			})
+			if err != nil {
+				fmt.Fprintf(GinkgoWriter, "Load Testing - Create 50 AppWrappers - There are %d non-Running AppWrappers, err = %v\n", len(nonRunningAWs), err)
+				for _, uaw := range nonRunningAWs {
+					fmt.Fprintf(GinkgoWriter, "Load Testing - Create 50 AppWrappers - Non-Running AW '%s/%s'\n", uaw.Namespace, uaw.Name)
+				}
+			}
+			Expect(err).Should(Succeed(), "All AppWrappers should have ready Pods")
+
+			By("Polling for all pods to become ready")
+			nonReadyAWs := appwrappers
+			err = wait.PollUntilContextTimeout(ctx, 500*time.Millisecond, 3*time.Minute, false, func(ctx context.Context) (done bool, err error) {
+				t := time.Now()
+				toCheckAWS := make([]*workloadv1beta2.AppWrapper, 0, len(appwrappers))
+				for _, aw := range nonReadyAWs {
+					if !checkAllAWPodsReady(ctx, aw) {
+						toCheckAWS = append(toCheckAWS, aw)
+					}
+				}
+				nonReadyAWs = toCheckAWS
+				if len(toCheckAWS) == 0 {
+					fmt.Fprintf(GinkgoWriter, "\tAll pods ready at time %s\n", t.Format(time.RFC3339))
+					return true, nil
+				}
+				fmt.Fprintf(GinkgoWriter, "\tThere are %d app wrappers without ready pods at time %s\n", len(toCheckAWS), t.Format(time.RFC3339))
+				return false, nil
+			})
+			if err != nil {
+				fmt.Fprintf(GinkgoWriter, "Load Testing - Create 50 AppWrappers - There are %d app wrappers without ready pods, err = %v\n", len(nonReadyAWs), err)
+				for _, uaw := range nonReadyAWs {
+					fmt.Fprintf(GinkgoWriter, "Load Testing - Create 50 AppWrappers - Non-Ready AW '%s/%s'\n", uaw.Namespace, uaw.Name)
+				}
+			}
+			Expect(err).Should(Succeed(), "All AppWrappers should have ready Pods")
+		})
 	})
 
 })

test/e2e/e2e_test.go

@@ -33,6 +33,7 @@ var _ = BeforeSuite(func() {
 	log.SetLogger(zap.New(zap.WriteTo(GinkgoWriter), zap.UseDevMode(true)))
 	ctx = extendContextWithClient(context.Background())
 	ensureNamespaceExists(ctx)
+	ctx = extendContextWithLimitedClient(ctx)
 	if Label("Kueue").MatchesLabelFilter(GinkgoLabelFilter()) {
 		ensureTestQueuesExist(ctx)
 	}