Add unit test for config map watcher

Signed-off-by: chandankumar4 <[email protected]>

cmd/main.go

@@ -58,10 +58,6 @@ func init() {
 	utilruntime.Must(apiv1.AddToScheme(scheme))
 
 	utilruntime.Must(numaflowv1.AddToScheme(scheme))
-
-	if err := kubernetes.StartConfigMapWatcher(context.Background(), ctrl.GetConfigOrDie()); err != nil {
-		numaLogger.Fatal(err, "Failed to start configmap watcher")
-	}
 }
 
 func main() {
@@ -153,6 +149,10 @@ func main() {
 		os.Exit(1)
 	}
 
+	if err := kubernetes.StartConfigMapWatcher(context.Background(), mgr.GetConfig()); err != nil {
+		numaLogger.Fatal(err, "Failed to start configmap watcher")
+	}
+
 	//+kubebuilder:scaffold:builder
 
 	pipelineRolloutReconciler := controller.NewPipelineRolloutReconciler(

internal/util/kubernetes/config_watcher.go

@@ -39,7 +39,7 @@ func StartConfigMapWatcher(ctx context.Context, config *rest.Config) error {
 }
 
 // watchConfigMaps watches for configmaps continuously and writes the data to the given directory based on an event type.
-func watchConfigMaps(ctx context.Context, client *kubernetes.Clientset, namespace string) {
+func watchConfigMaps(ctx context.Context, client kubernetes.Interface, namespace string) {
 	numaLogger := logger.FromContext(ctx)
 	watcher, err := client.CoreV1().ConfigMaps(namespace).Watch(ctx, metav1.ListOptions{})
 	if err != nil {

internal/util/kubernetes/config_watcher_test.go

@@ -0,0 +1,63 @@
+package kubernetes
+
+import (
+	"context"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/numaproj/numaplane/internal/controller/config"
+	"github.com/stretchr/testify/assert"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/kubernetes/fake"
+)
+
+func Test_watchConfigMaps(t *testing.T) {
+	ctx := context.TODO()
+	scheme := runtime.NewScheme()
+	err := corev1.AddToScheme(scheme)
+	assert.NoError(t, err)
+
+	clientSet := fake.NewSimpleClientset()
+	go watchConfigMaps(ctx, clientSet, "default")
+	time.Sleep(20 * time.Second)
+
+	data, err := os.ReadFile("../../../tests/config/controller-definitions-config.yaml")
+	assert.NoError(t, err)
+	// Create a new ConfigMap object
+	configMap := &corev1.ConfigMap{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "numaflow-controller-definitions-config",
+			Namespace: "default",
+			Labels: map[string]string{
+				"config": "numaflow-controller-rollout",
+			},
+		},
+		Data: map[string]string{
+			"controller_definitions.yaml": string(data),
+		},
+	}
+
+	// Create the ConfigMap object in the fake clientset
+	_, err = clientSet.CoreV1().ConfigMaps("default").Create(ctx, configMap, metav1.CreateOptions{})
+	assert.NoError(t, err)
+
+	// Wait for the controller to process the ConfigMap
+	time.Sleep(10 * time.Second)
+
+	// Validate the controller definition config is set correctly
+	definition := config.GetConfigManagerInstance().GetControllerDefinitionsConfig()
+	assert.Len(t, definition, 2)
+
+	// Create the ConfigMap object in the fake clientset
+	err = clientSet.CoreV1().ConfigMaps("default").Delete(ctx, configMap.Name, metav1.DeleteOptions{})
+	assert.NoError(t, err)
+
+	// Wait for the controller to process the ConfigMap
+	time.Sleep(10 * time.Second)
+
+	definition = config.GetConfigManagerInstance().GetControllerDefinitionsConfig()
+	assert.Len(t, definition, 0)
+}