test: add ResourceGroup assertions to WatchForSync

This adds validation for the ResourceGroup status in the WatchForSync
function, which is used extensively in the test framework to validate
RSyncs and wait for them to finish syncing.

This will cause WatchForSync to now also wait for all synced objects to
reconcile and for the ResourceGroup to be current. This is intended to
provide test coverage for the resource-group-controller when integrated
with Config Sync.

Author: sdowell

e2e/nomostest/sync.go

@@ -21,8 +21,10 @@ import (
 	"kpt.dev/configsync/e2e/nomostest/testpredicates"
 	"kpt.dev/configsync/pkg/api/configsync"
 	"kpt.dev/configsync/pkg/api/configsync/v1beta1"
+	"kpt.dev/configsync/pkg/api/kpt.dev/v1alpha1"
 	"kpt.dev/configsync/pkg/parse"
 	"kpt.dev/configsync/pkg/reposync"
+	"kpt.dev/configsync/pkg/resourcegroup/controllers/status"
 	"kpt.dev/configsync/pkg/rootsync"
 	"kpt.dev/configsync/pkg/util/log"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -263,3 +265,53 @@ func statusHasSyncDirAndNoErrors(status v1beta1.Status, sourceType configsync.So
 	}
 	return nil
 }
+
+func resourceGroupHasReconciled(sourceHash string) testpredicates.Predicate {
+	return func(o client.Object) error {
+		if o == nil {
+			return testpredicates.ErrObjectNotFound
+		}
+		rg, ok := o.(*v1alpha1.ResourceGroup)
+		if !ok {
+			return testpredicates.WrongTypeErr(o, &v1alpha1.ResourceGroup{})
+		}
+
+		if len(rg.Spec.Resources) != len(rg.Status.ResourceStatuses) {
+			return fmt.Errorf("length of spec.resources (%d) does not equal length of status.resourceStatuses (%d)",
+				len(rg.Spec.Resources), len(rg.Status.ResourceStatuses))
+		}
+
+		resourceCount := make(map[v1alpha1.ObjMetadata]int)
+		for _, s := range rg.Status.ResourceStatuses {
+			if err := resourceStatusIsCurrent(s, sourceHash); err != nil {
+				return fmt.Errorf("object %s is not current: %w", s.ObjMetadata, err)
+			}
+			resourceCount[s.ObjMetadata]++
+		}
+
+		for _, o := range rg.Spec.Resources {
+			if count, ok := resourceCount[o]; ok && count > 0 {
+				resourceCount[o]--
+			} else {
+				return fmt.Errorf("spec.resources does not equal status.resourceStatuses")
+			}
+		}
+		return nil
+	}
+}
+
+func resourceStatusIsCurrent(rs v1alpha1.ResourceStatus, sourceHash string) error {
+	if rs.Status != v1alpha1.Current {
+		return fmt.Errorf("resourceStatus.status is not %s. Got %s", v1alpha1.Current, rs.Status)
+	}
+	if rs.Reconcile != v1alpha1.ReconcileSucceeded {
+		return fmt.Errorf("resourceStatus.reconcile is not %s. Got %s", v1alpha1.ReconcileSucceeded, rs.Reconcile)
+	}
+	if rs.Actuation != v1alpha1.ActuationSucceeded {
+		return fmt.Errorf("resourceStatus.actuation is not %s. Got %s", v1alpha1.ActuationSucceeded, rs.Actuation)
+	}
+	if rs.SourceHash != status.TruncateSourceHash(sourceHash) {
+		return fmt.Errorf("resourceStatus.sourceHash is not %s. Got %s", sourceHash, rs.SourceHash)
+	}
+	return nil
+}

e2e/nomostest/wait_for_sync.go

@@ -229,6 +229,17 @@ func (nt *NT) WatchForSync(
 		return fmt.Errorf("waiting for sync: %w", err)
 	}
 	nt.T.Logf("%s %s/%s is synced", gvk.Kind, namespace, name)
+	rgPredicates := []testpredicates.Predicate{
+		// Wait until status.observedGeneration matches metadata.generation
+		testpredicates.HasObservedLatestGeneration(nt.Scheme),
+		// Wait until metadata.deletionTimestamp is missing, and conditions do not iniclude Reconciling=True or Stalled=True
+		testpredicates.StatusEquals(nt.Scheme, kstatus.CurrentStatus),
+		// Wait until all resourceStatuses are consistent with spec and current/reconciled
+		resourceGroupHasReconciled(expectedCommit),
+	}
+	if err := nt.Watcher.WatchObject(kinds.ResourceGroup(), name, namespace, testwatcher.WatchPredicates(rgPredicates...)); err != nil {
+		return fmt.Errorf("waiting for ResourceGroup: %w", err)
+	}
 	return nil
 }
 

pkg/resourcegroup/controllers/status/status.go

@@ -133,7 +133,11 @@ func GetSourceHash(annotations map[string]string) string {
 	if len(annotations) == 0 {
 		return ""
 	}
-	sourceHash := annotations[SourceHashAnnotationKey]
+	return TruncateSourceHash(annotations[SourceHashAnnotationKey])
+}
+
+// TruncateSourceHash truncates the provided source hash after the first 7 characters.
+func TruncateSourceHash(sourceHash string) string {
 	if len(sourceHash) > 7 {
 		return sourceHash[0:7]
 	}