use watch to wait gateway and listeners set to not programmed

Author: randmonkey

test/envtest/watch.go

@@ -17,6 +17,8 @@ const (
 	clientWatchTimeout = 45 * time.Second
 )
 
+// TODO: migrate all usage of `watch` in envtest package to use `helper.Watch`.
+
 // watch is a wrapper around watch.Interface.
 // It is useful to provide a type-safe way to access the watch.Interface so that
 // the callers do not use an invalid watches when using watchFor().

test/helpers/watch.go

@@ -0,0 +1,62 @@
+package helpers
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/kr/pretty"
+	"github.com/stretchr/testify/require"
+	apiwatch "k8s.io/apimachinery/pkg/watch"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+// WatchFor watches for an event of type eventType using the provided watch.Interface.
+// It returns when either context is done - and then it marks the test as failed -
+// or when the event has been received and predicate returned true.
+func WatchFor[
+	T client.Object,
+](
+	t *testing.T,
+	ctx context.Context,
+	w apiwatch.Interface,
+	eventType apiwatch.EventType,
+	timeout time.Duration,
+	predicate func(T) bool,
+	failMsg string,
+) T {
+	t.Helper()
+
+	ctx, cancel := context.WithTimeout(ctx, timeout)
+	defer cancel()
+
+	var (
+		obj                   T
+		receivedAtLeastOneObj bool
+	)
+	for found := false; !found; {
+		select {
+		case <-ctx.Done():
+			if receivedAtLeastOneObj {
+				require.Failf(t, failMsg, "Last object received:\n%v", pretty.Sprint(obj))
+			} else {
+				require.Fail(t, failMsg)
+			}
+		case e := <-w.ResultChan():
+			if e.Type != eventType {
+				continue
+			}
+			var ok bool
+			obj, ok = e.Object.(T)
+			if !ok {
+				continue
+			}
+			receivedAtLeastOneObj = true
+			if !predicate(obj) {
+				continue
+			}
+			found = true
+		}
+	}
+	return obj
+}

test/helpers/watch_test.go

@@ -0,0 +1,39 @@
+package helpers
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	apiwatch "k8s.io/apimachinery/pkg/watch"
+	fakectrlruntimeclient "sigs.k8s.io/controller-runtime/pkg/client/fake"
+
+	configurationv1 "github.com/kong/kong-operator/api/configuration/v1"
+	"github.com/kong/kong-operator/modules/manager/scheme"
+)
+
+func TestWatch(t *testing.T) {
+	var (
+		ctx = t.Context()
+		cl  = fakectrlruntimeclient.NewClientBuilder().
+			WithScheme(scheme.Get()).
+			Build()
+		consumer = &configurationv1.KongConsumer{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "test-consumer",
+			},
+		}
+	)
+
+	wConsumer, err := cl.Watch(ctx, &configurationv1.KongConsumerList{})
+	require.NoError(t, err)
+	require.NoError(t, cl.Create(ctx, consumer))
+	WatchFor(t, ctx, wConsumer, apiwatch.Added,
+		time.Second,
+		func(c *configurationv1.KongConsumer) bool {
+			return c.Name == consumer.Name
+		},
+		"error",
+	)
+}

test/integration/gateway_test.go

@@ -19,6 +19,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/intstr"
+	apiwatch "k8s.io/apimachinery/pkg/watch"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	gatewayv1 "sigs.k8s.io/gateway-api/apis/v1"
 
@@ -27,6 +28,7 @@ import (
 	operatorv2beta1 "github.com/kong/kong-operator/api/gateway-operator/v2beta1"
 	konnectv1alpha1 "github.com/kong/kong-operator/api/konnect/v1alpha1"
 	konnectv1alpha2 "github.com/kong/kong-operator/api/konnect/v1alpha2"
+	"github.com/kong/kong-operator/modules/manager/scheme"
 	"github.com/kong/kong-operator/pkg/consts"
 	"github.com/kong/kong-operator/pkg/gatewayapi"
 	gatewayutils "github.com/kong/kong-operator/pkg/utils/gateway"
@@ -48,6 +50,14 @@ func TestGatewayEssentials(t *testing.T) {
 	require.NoError(t, err)
 	cleaner.Add(gatewayClass)
 
+	t.Log("setting up watch for Gateways")
+	cl, err := client.NewWithWatch(GetEnv().Cluster().Config(), client.Options{
+		Scheme: scheme.Get(),
+	})
+	require.NoError(t, err, "failed to setup a client for watching gateways")
+	wGateway, err := cl.Watch(GetCtx(), &gatewayv1.GatewayList{}, client.InNamespace(namespace.Name))
+	require.NoError(t, err, "failed to start watching gateways")
+
 	t.Log("deploying Gateway resource")
 	gatewayNN := types.NamespacedName{
 		Name:      uuid.NewString(),
@@ -110,9 +120,16 @@ func TestGatewayEssentials(t *testing.T) {
 	t.Log("deleting dataplane")
 	require.NoError(t, dataplaneClient.Delete(GetCtx(), dataplane.Name, metav1.DeleteOptions{}))
 
-	t.Log("verifying Gateway gets and its listeners are marked as not Programmed")
-	require.Eventually(t, testutils.Not(testutils.GatewayIsProgrammed(t, GetCtx(), gatewayNN, clients)), testutils.GatewayReadyTimeLimit, 100*time.Millisecond)
-	require.Eventually(t, testutils.Not(testutils.GatewayListenersAreProgrammed(t, GetCtx(), gatewayNN, clients)), testutils.GatewayReadyTimeLimit, 100*time.Millisecond)
+	t.Logf("verifying Gateway gets and its listeners are marked as not Programmed at %v", time.Now())
+	_ = helpers.WatchFor(t, GetCtx(), wGateway, apiwatch.Modified,
+		testutils.GatewayReadyTimeLimit,
+		func(gw *gatewayv1.Gateway) bool {
+			return gw.Name == gatewayNN.Name && gw.Namespace == gatewayNN.Namespace &&
+				!gatewayutils.IsProgrammed(gw) && !gatewayutils.AreListenersProgrammed(gw)
+		},
+		"Did not see gateway and all its listeners' Programmed condition set to False",
+	)
+	t.Logf("see gateway and all its listeners marked as not programmed at %v in watch", time.Now())
 
 	t.Log("verifying that the ControlPlane becomes provisioned again")
 	require.Eventually(t, testutils.GatewayControlPlaneIsProvisioned(t, GetCtx(), gateway, clients), 45*time.Second, time.Second)