additiona testcases to validate watch functionality

examples/watch_resources/watch_test.go

@@ -19,6 +19,7 @@ package watch_resources
 import (
 	"context"
 	"testing"
+	"time"
 
 	appsv1 "k8s.io/api/apps/v1"
 	v1 "k8s.io/api/core/v1"
@@ -134,6 +135,95 @@ func TestWatchForResourcesWithStop(t *testing.T) {
 
 }
 
+func TestWatchForResources1(t *testing.T) {
+	addWait := make(chan struct{})
+	delWait := make(chan struct{})
+	onAddfunc := func(obj interface{}) {
+		dep := obj.(*appsv1.Deployment)
+		depName := dep.GetName()
+		if depName == "demo-app" {
+			addWait <- struct{}{}
+		}
+	}
+
+	onDelfunc := func(obj interface{}) {
+		dep := obj.(*appsv1.Deployment)
+		depName := dep.GetName()
+		if depName == "demo-app" {
+			delWait <- struct{}{}
+		}
+	}
+
+	watchFeature := features.New("test watcher with callback methods").WithLabel("env", "prod").
+		Setup(func(ctx context.Context, t *testing.T, cfg *envconf.Config) context.Context {
+			cl, err := cfg.NewClient()
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			dep := appsv1.Deployment{
+				ObjectMeta: metav1.ObjectMeta{Name: "demo-app", Namespace: cfg.Namespace()},
+			}
+
+			// watch for the deployment and triger action based on the event recieved.
+			w = cl.Resources().Watch(&appsv1.DeploymentList{}, resources.WithFieldSelector(labels.FormatLabels(map[string]string{"metadata.name": dep.Name}))).
+				WithAddFunc(onAddfunc).WithDeleteFunc(onDelfunc)
+
+			err = w.Start(ctx)
+			if err != nil {
+				t.Error(err)
+			}
+
+			return ctx
+		}).
+		Assess("create watch deployment", func(ctx context.Context, t *testing.T, cfg *envconf.Config) context.Context {
+			// create a deployment
+			deployment := newDeployment(cfg.Namespace(), "demo-app", 1)
+			client, err := cfg.NewClient()
+			if err != nil {
+				t.Fatal(err)
+			}
+			if err := client.Resources().Create(ctx, deployment); err != nil {
+				t.Fatal(err)
+			}
+
+			// After creation, wait for signal
+			select {
+			case <-time.After(3 * time.Second):
+				t.Error("Add callback not called")
+			case <-addWait:
+				klog.InfoS("recieved signal, closing add channel")
+				close(addWait)
+			}
+
+			return context.WithValue(ctx, "demo-app", deployment)
+		}).
+		Teardown(func(ctx context.Context, t *testing.T, cfg *envconf.Config) context.Context {
+			client, err := cfg.NewClient()
+			if err != nil {
+				t.Fatal(err)
+			}
+			depl := ctx.Value("demo-app").(*appsv1.Deployment)
+			if err := client.Resources().Delete(ctx, depl); err != nil {
+				t.Fatal(err)
+			}
+
+			// After deletion, wait for signal
+			select {
+			case <-time.After(3 * time.Second):
+				t.Error("Delete callback not called")
+			case <-delWait:
+				klog.InfoS("recieved signal, closing delete channel")
+				close(delWait)
+			}
+
+			w.Stop()
+
+			return ctx
+		}).Feature()
+	testenv.Test(t, watchFeature)
+}
+
 func newDeployment(namespace string, name string, replicas int32) *appsv1.Deployment {
 	return &appsv1.Deployment{
 		ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: namespace, Labels: map[string]string{"app": "watch-for-resources"}},