fix(evpn): added unit test for taskMgr and eventbus

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

pkg/infradb/subscriberframework/eventbus/event_bus_test.go

@@ -0,0 +1,181 @@
+package eventbus
+
+import (
+	"log"
+	"sync"
+	"testing"
+	"time"
+)
+
+type ModulelciHandler struct {
+	receivedEvents []*ObjectData
+	sync.Mutex
+}
+
+func (h *ModulelciHandler) HandleEvent(eventType string, objectData *ObjectData) {
+	h.Lock()
+	defer h.Unlock()
+	h.receivedEvents = append(h.receivedEvents, objectData)
+	switch eventType {
+	case "testEvent":
+	case "testEventpriority":
+	case "testEventChBusy":
+	case "testEventUnsub":
+		log.Printf("received event type %s", eventType)
+	default:
+		log.Printf("error: Unknown event type %s", eventType)
+	}
+}
+
+func TestSubscribeAndPublish(t *testing.T) {
+	handler := &ModulelciHandler{}
+
+	EventBus := NewEventBus()
+	EventBus.StartSubscriber("testModule", "testEvent", 1, handler)
+
+	time.Sleep(100 * time.Millisecond)
+
+	objectData := &ObjectData{
+		ResourceVersion: "v1",
+		Name:            "testObject",
+		NotificationID:  "123",
+	}
+
+	subscribers := EventBus.GetSubscribers("testEvent")
+	if len(subscribers) == 0 {
+		t.Errorf("No subscribers found for event type 'testEvent'")
+	}
+	subscriber := subscribers[0]
+
+	err := EventBus.Publish(objectData, subscriber)
+	if err != nil {
+		t.Errorf("Publish() failed with error: %v", err)
+	}
+
+	time.Sleep(100 * time.Millisecond)
+	handler.Lock()
+	if len(handler.receivedEvents) != 1 || handler.receivedEvents[0] != objectData {
+		t.Errorf("Event was not received by the handler as expected")
+	}
+	handler.Unlock()
+}
+
+func TestPriorityOrderWithStartSubscriber(t *testing.T) {
+	handler1 := &ModulelciHandler{}
+	handler2 := &ModulelciHandler{}
+
+	EventBus := NewEventBus()
+
+	EventBus.StartSubscriber("testModule1", "testEventpriority", 2, handler1)
+	EventBus.StartSubscriber("testModule2", "testEventpriority", 1, handler2)
+
+	time.Sleep(100 * time.Millisecond)
+
+	subscribers := EventBus.GetSubscribers("testEventpriority")
+	if len(subscribers) != 2 {
+		t.Errorf("Expected 2 subscribers, got %d", len(subscribers))
+	}
+	if subscribers[0].Priority > subscribers[1].Priority {
+		t.Errorf("Subscribers are not sorted by priority")
+	}
+
+	for _, sub := range subscribers {
+		EventBus.Unsubscribe(sub)
+	}
+}
+func TestPublishChannelBusyWithStartSubscriber(t *testing.T) {
+	handler := &ModulelciHandler{}
+	EventBus := NewEventBus()
+	EventBus.StartSubscriber("testModuleChBusy", "testEventChBusy", 1, handler)
+
+	time.Sleep(100 * time.Millisecond)
+
+	subscribers := EventBus.GetSubscribers("testEventChBusy")
+	if len(subscribers) == 0 {
+		t.Errorf("No subscribers found for event type 'testEventChBusy'")
+	}
+	subscriber := subscribers[0]
+
+	subscriber.Ch <- &ObjectData{}
+
+	objectData := &ObjectData{
+		ResourceVersion: "v1",
+		Name:            "testObject",
+		NotificationID:  "123",
+	}
+	err := EventBus.Publish(objectData, subscriber)
+	if err == nil {
+		t.Errorf("Expected an error when publishing to a busy channel, but got nil")
+	}
+
+	EventBus.Unsubscribe(subscriber)
+}
+func TestUnsubscribeEvent(t *testing.T) {
+	handler := &ModulelciHandler{}
+	EventBus := NewEventBus()
+	EventBus.StartSubscriber("testModuleUnsub", "testEventUnsub", 1, handler)
+
+	time.Sleep(100 * time.Millisecond)
+
+	subscribers := EventBus.GetSubscribers("testEventUnsub")
+	if len(subscribers) == 0 {
+		t.Errorf("No subscribers found for event type 'testEventUnsub'")
+	}
+	subscriber := subscribers[0]
+
+	EventBus.UnsubscribeEvent(subscriber, "testEventUnsub")
+
+	time.Sleep(200 * time.Millisecond)
+
+	subscribers = EventBus.GetSubscribers("testEventUnsub")
+	for _, sub := range subscribers {
+		if sub == subscriber {
+			t.Errorf("Subscriber was not successfully unsubscribed")
+		}
+	}
+}
+
+func TestUnsubscribe(t *testing.T) {
+	handler := &ModulelciHandler{}
+	EventBus := NewEventBus()
+	EventBus.StartSubscriber("testModuleUnsub", "testEventUnsub", 1, handler)
+
+	time.Sleep(100 * time.Millisecond)
+
+	subscribers := EventBus.GetSubscribers("testEventUnsub")
+	if len(subscribers) == 0 {
+		t.Errorf("No subscribers found for event type 'testEventUnsub'")
+	}
+	subscriber := subscribers[0]
+
+	EventBus.Unsubscribe(subscriber)
+	time.Sleep(200 * time.Millisecond)
+
+	select {
+	case _, ok := <-subscriber.Ch:
+		if ok {
+			t.Errorf("Subscriber's channel should be closed, but it's not")
+		}
+	default:
+	}
+}
+
+func TestSubscriberExist(t *testing.T) {
+	handler := &ModulelciHandler{}
+	EventBus := NewEventBus()
+	EventBus.StartSubscriber("testModuleSubExist", "testEventSubExist", 3, handler)
+
+	time.Sleep(100 * time.Millisecond)
+
+	exists := EventBus.subscriberExist("testEventSubExist", "testModuleSubExist")
+	if !exists {
+		t.Errorf("subscriberExist should return true for existing subscriber")
+	}
+
+	subscribers := EventBus.GetSubscribers("testEventSubExist")
+	for _, sub := range subscribers {
+		if sub.Name == "testModuleSubExist" {
+			EventBus.Unsubscribe(sub)
+		}
+	}
+}