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,176 @@
+package eventbus
+
+import (
+ "log"
+ "testing"
+ "time"
+)
+
+type ModulelciHandler struct {
+ receivedEvents []*ObjectData
+}
+
+func (h *ModulelciHandler) HandleEvent(eventType string, objectData *ObjectData) {
+ 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)
+
+ if len(handler.receivedEvents) != 1 || handler.receivedEvents[0] != objectData {
+ t.Errorf("Event was not received by the handler as expected")
+ }
+}
+
+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)
+ }
+ }
+}