test(eventbus): add rotation, timeout, saturation, retention tests to raise coverage (PR #51)

Author: intel352

modules/eventbus/additional_eventbus_tests_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"testing"
 	"time"
+	"sync/atomic"
 )
 
 // Test basic publish/subscribe lifecycle using memory engine ensuring message receipt and stats increments.
@@ -129,3 +130,101 @@ func TestEventBusAsyncSubscription(t *testing.T) {
 }
 
 // Removed local mockApp (reuse the one defined in module_test.go)
+
+// TestMemoryEventBus_RotationFairness ensures subscriber ordering rotates when enabled.
+func TestMemoryEventBus_RotationFairness(t *testing.T) {
+	ctx := context.Background()
+	cfg := &EventBusConfig{WorkerCount: 1, DefaultEventBufferSize: 1, RotateSubscriberOrder: true, DeliveryMode: "drop"}
+	bus := NewMemoryEventBus(cfg)
+	if err := bus.Start(ctx); err != nil { t.Fatalf("start: %v", err) }
+	defer bus.Stop(ctx)
+
+	orderCh := make(chan string, 16)
+	mkHandler := func(id string) EventHandler { return func(ctx context.Context, evt Event) error { orderCh <- id; return nil } }
+	for i := 0; i < 3; i++ {
+		_, err := bus.Subscribe(ctx, "rot.topic", mkHandler(string(rune('A'+i))))
+		if err != nil { t.Fatalf("subscribe %d: %v", i, err) }
+	}
+
+	firsts := make(map[string]int)
+	for i := 0; i < 9; i++ {
+		_ = bus.Publish(ctx, Event{Topic: "rot.topic"})
+		select {
+		case id := <-orderCh:
+			firsts[id]++
+		case <-time.After(500 * time.Millisecond):
+			t.Fatalf("timeout waiting for first handler")
+		}
+		// Drain remaining handlers for this publish (best-effort)
+		for j := 0; j < 2; j++ {
+			select { case <-orderCh: default: }
+		}
+	}
+	if len(firsts) < 2 { t.Fatalf("expected rotation to vary first subscriber, got %v", firsts) }
+}
+
+// TestMemoryEventBus_PublishTimeoutImmediateDrop covers timeout mode with zero timeout resulting in immediate drop when subscriber buffer full.
+func TestMemoryEventBus_PublishTimeoutImmediateDrop(t *testing.T) {
+	ctx := context.Background()
+	cfg := &EventBusConfig{WorkerCount: 1, DefaultEventBufferSize: 1, DeliveryMode: "timeout", PublishBlockTimeout: 0}
+	bus := NewMemoryEventBus(cfg)
+	if err := bus.Start(ctx); err != nil { t.Fatalf("start: %v", err) }
+	defer bus.Stop(ctx)
+
+	// Manually construct a subscription with a full channel (no handler goroutine)
+	sub := &memorySubscription{
+		id:       "manual",
+		topic:    "t",
+		handler:  func(ctx context.Context, e Event) error { return nil },
+		isAsync:  false,
+		eventCh:  make(chan Event, 1),
+		done:     make(chan struct{}),
+		finished: make(chan struct{}),
+	}
+	// Fill the channel to force publish path into drop branch
+	sub.eventCh <- Event{Topic: "t"}
+	bus.topicMutex.Lock()
+	bus.subscriptions["t"] = map[string]*memorySubscription{sub.id: sub}
+	bus.topicMutex.Unlock()
+
+	before := atomic.LoadUint64(&bus.droppedCount)
+	_ = bus.Publish(ctx, Event{Topic: "t"})
+	after := atomic.LoadUint64(&bus.droppedCount)
+	if after != before+1 { t.Fatalf("expected exactly one drop, before=%d after=%d", before, after) }
+}
+
+// TestMemoryEventBus_AsyncWorkerSaturation ensures async drops when worker count is zero (no workers to consume tasks).
+func TestMemoryEventBus_AsyncWorkerSaturation(t *testing.T) {
+	ctx := context.Background()
+	cfg := &EventBusConfig{WorkerCount: 0, DefaultEventBufferSize: 1}
+	bus := NewMemoryEventBus(cfg)
+	if err := bus.Start(ctx); err != nil { t.Fatalf("start: %v", err) }
+	defer bus.Stop(ctx)
+
+	_, err := bus.SubscribeAsync(ctx, "a", func(ctx context.Context, e Event) error { return nil })
+	if err != nil { t.Fatalf("subscribe async: %v", err) }
+	before := atomic.LoadUint64(&bus.droppedCount)
+	for i := 0; i < 5; i++ { _ = bus.Publish(ctx, Event{Topic: "a"}) }
+	after := atomic.LoadUint64(&bus.droppedCount)
+	if after <= before { t.Fatalf("expected drops due to saturated worker pool, before=%d after=%d", before, after) }
+}
+
+// TestMemoryEventBus_RetentionCleanup verifies old events pruned.
+func TestMemoryEventBus_RetentionCleanup(t *testing.T) {
+	ctx := context.Background()
+	cfg := &EventBusConfig{WorkerCount: 1, DefaultEventBufferSize: 1, RetentionDays: 1}
+	bus := NewMemoryEventBus(cfg)
+	if err := bus.Start(ctx); err != nil { t.Fatalf("start: %v", err) }
+	defer bus.Stop(ctx)
+
+	old := Event{Topic: "old", CreatedAt: time.Now().AddDate(0,0,-2)}
+	recent := Event{Topic: "recent", CreatedAt: time.Now()}
+	bus.storeEventHistory(old)
+	bus.storeEventHistory(recent)
+	bus.cleanupOldEvents()
+	bus.historyMutex.RLock()
+	defer bus.historyMutex.RUnlock()
+	for _, evs := range bus.eventHistory {
+		for _, e := range evs { if e.Topic == "old" { t.Fatalf("old event not cleaned up") } }
+	}
+}