wild-central/internal/sse/manager_test.go

package sse

import (
	"testing"
	"time"
)

func TestManagerClientRegistration(t *testing.T) {
	manager := NewManager()

	// Test client registration
	client := manager.RegisterClient("test-instance", EventFilters{})
	if client == nil {
		t.Fatal("Expected client to be registered")
	}

	// Give manager time to process registration
	time.Sleep(10 * time.Millisecond)

	// Check client is registered
	if count := manager.GetInstanceConnectionCount("test-instance"); count != 1 {
		t.Errorf("Expected 1 connection, got %d", count)
	}

	// Test client unregistration
	manager.UnregisterClient(client)
	time.Sleep(10 * time.Millisecond)

	if count := manager.GetInstanceConnectionCount("test-instance"); count != 0 {
		t.Errorf("Expected 0 connections, got %d", count)
	}
}

func TestManagerBroadcast(t *testing.T) {
	manager := NewManager()

	// Register two clients for same instance
	client1 := manager.RegisterClient("test-instance", EventFilters{
		EventTypes: []string{"pod.added"},
	})
	client2 := manager.RegisterClient("test-instance", EventFilters{})

	// Give manager time to process registrations
	time.Sleep(10 * time.Millisecond)

	// Broadcast an event
	event := &Event{
		Type:         "pod.added",
		InstanceName: "test-instance",
		Data:         map[string]any{"name": "test-pod"},
		Metadata:     map[string]any{},
	}
	manager.Broadcast(event)

	// Both clients should receive the event
	select {
	case receivedEvent := <-client1.Channel:
		if receivedEvent.Type != "pod.added" {
			t.Errorf("Expected event type 'pod.added', got '%s'", receivedEvent.Type)
		}
	case <-time.After(100 * time.Millisecond):
		t.Error("Client1 did not receive event")
	}

	select {
	case receivedEvent := <-client2.Channel:
		if receivedEvent.Type != "pod.added" {
			t.Errorf("Expected event type 'pod.added', got '%s'", receivedEvent.Type)
		}
	case <-time.After(100 * time.Millisecond):
		t.Error("Client2 did not receive event")
	}

	// Cleanup
	manager.UnregisterClient(client1)
	manager.UnregisterClient(client2)
}

func TestEventFiltering(t *testing.T) {
	manager := NewManager()

	// Register client with event type filter
	client := manager.RegisterClient("test-instance", EventFilters{
		EventTypes: []string{"pod.added", "pod.updated"},
	})

	// Give manager time to process registration
	time.Sleep(10 * time.Millisecond)

	// Send events of different types
	podAddedEvent := &Event{
		Type:         "pod.added",
		InstanceName: "test-instance",
		Data:         map[string]any{"name": "test-pod"},
	}
	podDeletedEvent := &Event{
		Type:         "pod.deleted",
		InstanceName: "test-instance",
		Data:         map[string]any{"name": "test-pod"},
	}

	manager.Broadcast(podAddedEvent)
	manager.Broadcast(podDeletedEvent)

	// Should only receive pod.added event
	select {
	case receivedEvent := <-client.Channel:
		if receivedEvent.Type != "pod.added" {
			t.Errorf("Expected event type 'pod.added', got '%s'", receivedEvent.Type)
		}
	case <-time.After(100 * time.Millisecond):
		t.Error("Client did not receive pod.added event")
	}

	// Should not receive pod.deleted event
	select {
	case receivedEvent := <-client.Channel:
		t.Errorf("Should not have received event of type '%s'", receivedEvent.Type)
	case <-time.After(100 * time.Millisecond):
		// Expected - no event should be received
	}

	// Cleanup
	manager.UnregisterClient(client)
}

func TestNamespaceFiltering(t *testing.T) {
	manager := NewManager()

	// Register client with namespace filter
	client := manager.RegisterClient("test-instance", EventFilters{
		Namespaces: []string{"default", "kube-system"},
	})

	// Give manager time to process registration
	time.Sleep(10 * time.Millisecond)

	// Send events from different namespaces
	defaultNsEvent := &Event{
		Type:         "pod.added",
		InstanceName: "test-instance",
		Data:         map[string]any{"name": "test-pod"},
		Metadata:     map[string]any{"namespace": "default"},
	}
	otherNsEvent := &Event{
		Type:         "pod.added",
		InstanceName: "test-instance",
		Data:         map[string]any{"name": "test-pod"},
		Metadata:     map[string]any{"namespace": "other-namespace"},
	}

	manager.Broadcast(defaultNsEvent)
	manager.Broadcast(otherNsEvent)

	// Should only receive event from default namespace
	select {
	case receivedEvent := <-client.Channel:
		if ns, ok := receivedEvent.Metadata["namespace"].(string); !ok || ns != "default" {
			t.Errorf("Expected event from 'default' namespace, got '%v'", receivedEvent.Metadata["namespace"])
		}
	case <-time.After(100 * time.Millisecond):
		t.Error("Client did not receive event from default namespace")
	}

	// Should not receive event from other namespace
	select {
	case receivedEvent := <-client.Channel:
		t.Errorf("Should not have received event from namespace '%v'", receivedEvent.Metadata["namespace"])
	case <-time.After(100 * time.Millisecond):
		// Expected - no event should be received
	}

	// Cleanup
	manager.UnregisterClient(client)
}

func TestMultipleInstanceIsolation(t *testing.T) {
	manager := NewManager()

	// Register clients for different instances
	client1 := manager.RegisterClient("instance-1", EventFilters{})
	client2 := manager.RegisterClient("instance-2", EventFilters{})

	// Give manager time to process registrations
	time.Sleep(10 * time.Millisecond)

	// Send event to instance-1
	event := &Event{
		Type:         "pod.added",
		InstanceName: "instance-1",
		Data:         map[string]any{"name": "test-pod"},
	}
	manager.Broadcast(event)

	// Only client1 should receive the event
	select {
	case receivedEvent := <-client1.Channel:
		if receivedEvent.InstanceName != "instance-1" {
			t.Errorf("Expected event for 'instance-1', got '%s'", receivedEvent.InstanceName)
		}
	case <-time.After(100 * time.Millisecond):
		t.Error("Client1 did not receive event")
	}

	// Client2 should not receive the event
	select {
	case <-client2.Channel:
		t.Error("Client2 should not have received event for instance-1")
	case <-time.After(100 * time.Millisecond):
		// Expected - no event should be received
	}

	// Cleanup
	manager.UnregisterClient(client1)
	manager.UnregisterClient(client2)
}

func TestRateLimiting(t *testing.T) {
	manager := NewManager()

	// Register a client
	client := manager.RegisterClient("test-instance", EventFilters{})

	// Give manager time to process registration
	time.Sleep(10 * time.Millisecond)

	// Send many events rapidly (more than rate limit allows)
	for i := 0; i < 300; i++ {
		event := &Event{
			Type:         "pod.added",
			InstanceName: "test-instance",
			Data:         map[string]any{"index": i},
		}
		manager.Broadcast(event)
	}

	// Count received events
	receivedCount := 0
	timeout := time.After(500 * time.Millisecond)

	for {
		select {
		case <-client.Channel:
			receivedCount++
		case <-timeout:
			// Rate limiter should have dropped some events
			// We allow 100/sec with burst of 200, so we should receive around 200
			if receivedCount >= 300 {
				t.Errorf("Rate limiting not working: received %d events (expected < 300)", receivedCount)
			}
			if receivedCount < 100 {
				t.Errorf("Too many events dropped: received only %d events", receivedCount)
			}
			// Cleanup
			manager.UnregisterClient(client)
			return
		}
	}
}

func TestClientContextCancellation(t *testing.T) {
	manager := NewManager()

	// Register a client
	client := manager.RegisterClient("test-instance", EventFilters{})

	// Cancel the client context
	client.Cancel()

	// Try to send an event
	event := &Event{
		Type:         "pod.added",
		InstanceName: "test-instance",
		Data:         map[string]any{"name": "test-pod"},
	}
	manager.Broadcast(event)

	// Client should not receive event after context cancellation
	select {
	case <-client.Channel:
		// Channel might still have buffered events, but should be closed soon
	case <-time.After(100 * time.Millisecond):
		// Expected - context cancelled
	}

	// Cleanup
	manager.UnregisterClient(client)
}

func TestConnectionCount(t *testing.T) {
	manager := NewManager()

	// Initially no connections
	if count := manager.GetConnectionCount(); count != 0 {
		t.Errorf("Expected 0 connections initially, got %d", count)
	}

	// Register clients for different instances
	client1 := manager.RegisterClient("instance-1", EventFilters{})
	client2 := manager.RegisterClient("instance-1", EventFilters{})
	client3 := manager.RegisterClient("instance-2", EventFilters{})

	// Give manager time to process registrations
	time.Sleep(10 * time.Millisecond)

	// Check total connection count
	if count := manager.GetConnectionCount(); count != 3 {
		t.Errorf("Expected 3 total connections, got %d", count)
	}

	// Check per-instance counts
	if count := manager.GetInstanceConnectionCount("instance-1"); count != 2 {
		t.Errorf("Expected 2 connections for instance-1, got %d", count)
	}
	if count := manager.GetInstanceConnectionCount("instance-2"); count != 1 {
		t.Errorf("Expected 1 connection for instance-2, got %d", count)
	}

	// Cleanup
	manager.UnregisterClient(client1)
	manager.UnregisterClient(client2)
	manager.UnregisterClient(client3)
}

func BenchmarkBroadcast(b *testing.B) {
	manager := NewManager()

	// Register 100 clients
	clients := make([]*Client, 100)
	for i := 0; i < 100; i++ {
		clients[i] = manager.RegisterClient("test-instance", EventFilters{})
	}

	// Give manager time to process registrations
	time.Sleep(100 * time.Millisecond)

	// Benchmark broadcasting events
	b.ResetTimer()
	for i := 0; i < b.N; i++ {
		event := &Event{
			Type:         "pod.added",
			InstanceName: "test-instance",
			Data:         map[string]any{"index": i},
		}
		manager.Broadcast(event)
	}

	// Cleanup
	for _, client := range clients {
		manager.UnregisterClient(client)
	}
}