package sse import ( "fmt" "testing" "time" ) func TestKubectlWatcherJSONParsing(t *testing.T) { manager := NewManager() watcher := &KubectlWatcher{ instanceName: "test-instance", kubeconfig: "/tmp/test-kubeconfig", sseManager: manager, } // Test valid pod JSON parsing validPodJSON := `{ "type": "ADDED", "object": { "apiVersion": "v1", "kind": "Pod", "metadata": { "name": "test-pod", "namespace": "default", "labels": { "app": "test-app" } }, "status": { "phase": "Running", "conditions": [] } } }` // Register a client to receive events client := manager.RegisterClient("test-instance", EventFilters{}) defer manager.UnregisterClient(client) // Parse the JSON watcher.parsePodEvent([]byte(validPodJSON), "test-instance") // Should receive parsed event select { case event := <-client.Channel: if event.Type != "pod:added" { t.Errorf("Expected event type 'pod:added', got '%s'", event.Type) } if podData, ok := event.Data.(map[string]any); ok { if podData["name"] != "test-pod" { t.Errorf("Expected pod name 'test-pod', got '%v'", podData["name"]) } if podData["namespace"] != "default" { t.Errorf("Expected namespace 'default', got '%v'", podData["namespace"]) } } else { t.Error("Event data is not a map") } case <-time.After(100 * time.Millisecond): t.Error("Did not receive pod event") } } func TestKubectlWatcherEventTypeMapping(t *testing.T) { manager := NewManager() watcher := &KubectlWatcher{ instanceName: "test-instance", kubeconfig: "/tmp/test-kubeconfig", sseManager: manager, } testCases := []struct { watchType string expectedType string }{ {"ADDED", "pod:added"}, {"MODIFIED", "pod:modified"}, {"DELETED", "pod:deleted"}, } for _, tc := range testCases { t.Run(tc.watchType, func(t *testing.T) { // Register a client client := manager.RegisterClient("test-instance", EventFilters{}) defer manager.UnregisterClient(client) podJSON := fmt.Sprintf(`{ "type": "%s", "object": { "kind": "Pod", "metadata": { "name": "test-pod", "namespace": "default" } } }`, tc.watchType) // Parse the event watcher.parsePodEvent([]byte(podJSON), "test-instance") // Check received event type select { case event := <-client.Channel: if event.Type != tc.expectedType { t.Errorf("Expected event type '%s', got '%s'", tc.expectedType, event.Type) } case <-time.After(100 * time.Millisecond): t.Error("Did not receive event") } }) } } func TestDeploymentEventParsing(t *testing.T) { manager := NewManager() watcher := &KubectlWatcher{ instanceName: "test-instance", kubeconfig: "/tmp/test-kubeconfig", sseManager: manager, } deploymentJSON := `{ "type": "MODIFIED", "object": { "apiVersion": "apps/v1", "kind": "Deployment", "metadata": { "name": "test-deployment", "namespace": "default" }, "spec": { "replicas": 3 }, "status": { "replicas": 3, "readyReplicas": 2, "availableReplicas": 2 } } }` // Register a client client := manager.RegisterClient("test-instance", EventFilters{}) defer manager.UnregisterClient(client) // Parse deployment event watcher.parseDeploymentEvent([]byte(deploymentJSON), "test-instance") // Check received event select { case event := <-client.Channel: if event.Type != "deployment:modified" { t.Errorf("Expected event type 'deployment:modified', got '%s'", event.Type) } if deployData, ok := event.Data.(map[string]any); ok { if deployData["name"] != "test-deployment" { t.Errorf("Expected deployment name 'test-deployment', got '%v'", deployData["name"]) } if deployData["readyReplicas"] != int32(2) { t.Errorf("Expected 2 ready replicas, got '%v'", deployData["readyReplicas"]) } } case <-time.After(100 * time.Millisecond): t.Error("Did not receive deployment event") } } func TestServiceEventParsing(t *testing.T) { manager := NewManager() watcher := &KubectlWatcher{ instanceName: "test-instance", kubeconfig: "/tmp/test-kubeconfig", sseManager: manager, } serviceJSON := `{ "type": "ADDED", "object": { "apiVersion": "v1", "kind": "Service", "metadata": { "name": "test-service", "namespace": "default" }, "spec": { "type": "LoadBalancer", "clusterIP": "10.96.0.1", "ports": [ { "port": 80, "targetPort": 8080, "protocol": "TCP" } ] }, "status": { "loadBalancer": { "ingress": [ { "ip": "192.168.1.100" } ] } } } }` // Register a client client := manager.RegisterClient("test-instance", EventFilters{}) defer manager.UnregisterClient(client) // Parse service event watcher.parseServiceEvent([]byte(serviceJSON), "test-instance") // Check received event select { case event := <-client.Channel: if event.Type != "service:added" { t.Errorf("Expected event type 'service:added', got '%s'", event.Type) } if serviceData, ok := event.Data.(map[string]any); ok { if serviceData["name"] != "test-service" { t.Errorf("Expected service name 'test-service', got '%v'", serviceData["name"]) } if serviceData["type"] != "LoadBalancer" { t.Errorf("Expected service type 'LoadBalancer', got '%v'", serviceData["type"]) } } case <-time.After(100 * time.Millisecond): t.Error("Did not receive service event") } } // TestTalosEventParsing is commented out as TalosWatcher doesn't expose parsing methods // The actual parsing happens internally in the watchEvents method /* func TestTalosEventParsing(t *testing.T) { manager := NewManager() watcher := &TalosWatcher{ instanceName: "test-instance", talosconfig: "/tmp/test-talosconfig", nodeIP: "192.168.1.10", sseManager: manager, } // Test machine config event machineConfigEvent := `seq:task message:update successful node:192.168.1.10 type:MachineConfig` // Register a client client := manager.RegisterClient("test-instance", EventFilters{}) defer manager.UnregisterClient(client) // Parse talos event watcher.parseTalosEventLine(machineConfigEvent) // Check received event select { case event := <-client.Channel: if event.Type != "talos.MachineConfig" { t.Errorf("Expected event type 'talos.MachineConfig', got '%s'", event.Type) } if talosData, ok := event.Data.(map[string]any); ok { if talosData["message"] != "update successful" { t.Errorf("Expected message 'update successful', got '%v'", talosData["message"]) } if talosData["node"] != "192.168.1.10" { t.Errorf("Expected node '192.168.1.10', got '%v'", talosData["node"]) } } case <-time.After(100 * time.Millisecond): t.Error("Did not receive talos event") } } */ func TestWatcherManagerLifecycle(t *testing.T) { sseManager := NewManager() watcherManager := NewWatcherManager(sseManager) // Test starting watchers for an instance err := watcherManager.StartWatchers("test-instance", "/tmp/kubeconfig", "/tmp/talosconfig", "192.168.1.10") if err != nil { // This will likely fail because the configs don't exist, but we can test the manager state t.Logf("Expected error starting watchers with non-existent configs: %v", err) } // Check if instance is registered (even if watchers failed to start) watcherManager.mu.RLock() _, exists := watcherManager.watchers["test-instance"] watcherManager.mu.RUnlock() if !exists { t.Error("Instance should be registered even if watchers fail to start") } // Test stopping watchers watcherManager.StopWatchers("test-instance") // Check instance is removed watcherManager.mu.RLock() _, exists = watcherManager.watchers["test-instance"] watcherManager.mu.RUnlock() if exists { t.Error("Instance should be removed after stopping watchers") } } func TestWatcherManagerMultipleInstances(t *testing.T) { sseManager := NewManager() watcherManager := NewWatcherManager(sseManager) // Start watchers for multiple instances _ = watcherManager.StartWatchers("instance-1", "/tmp/kubeconfig1", "/tmp/talosconfig1", "192.168.1.11") _ = watcherManager.StartWatchers("instance-2", "/tmp/kubeconfig2", "/tmp/talosconfig2", "192.168.1.12") // Check both instances are registered watcherManager.mu.RLock() count := len(watcherManager.watchers) watcherManager.mu.RUnlock() if count != 2 { t.Errorf("Expected 2 instances registered, got %d", count) } // Stop watchers for each instance watcherManager.StopWatchers("instance-1") watcherManager.StopWatchers("instance-2") // Check all instances are removed watcherManager.mu.RLock() count = len(watcherManager.watchers) watcherManager.mu.RUnlock() if count != 0 { t.Errorf("Expected 0 instances after stopping all, got %d", count) } } func TestInvalidJSONHandling(t *testing.T) { manager := NewManager() watcher := &KubectlWatcher{ instanceName: "test-instance", kubeconfig: "/tmp/test-kubeconfig", sseManager: manager, } // Register a client client := manager.RegisterClient("test-instance", EventFilters{}) defer manager.UnregisterClient(client) // Try parsing invalid JSON invalidJSON := `{invalid json}` watcher.parsePodEvent([]byte(invalidJSON), "test-instance") // Should not receive any event select { case <-client.Channel: t.Error("Should not receive event for invalid JSON") case <-time.After(50 * time.Millisecond): // Expected - no event for invalid JSON } } func TestEmptyEventHandling(t *testing.T) { manager := NewManager() watcher := &KubectlWatcher{ instanceName: "test-instance", kubeconfig: "/tmp/test-kubeconfig", sseManager: manager, } // Register a client client := manager.RegisterClient("test-instance", EventFilters{}) defer manager.UnregisterClient(client) // Try parsing empty JSON emptyJSON := `{}` watcher.parsePodEvent([]byte(emptyJSON), "test-instance") // Should not crash, might not produce event select { case <-client.Channel: // If we get an event, make sure it doesn't crash t.Log("Received event for empty JSON (acceptable)") case <-time.After(50 * time.Millisecond): // Also acceptable - no event for empty JSON } } func BenchmarkJSONParsing(b *testing.B) { manager := NewManager() watcher := &KubectlWatcher{ instanceName: "test-instance", kubeconfig: "/tmp/test-kubeconfig", sseManager: manager, } podJSON := `{ "type": "ADDED", "object": { "apiVersion": "v1", "kind": "Pod", "metadata": { "name": "test-pod", "namespace": "default", "labels": { "app": "test-app" } }, "status": { "phase": "Running" } } }` // Register a client to consume events client := manager.RegisterClient("test-instance", EventFilters{}) defer manager.UnregisterClient(client) // Consume events in background go func() { for range client.Channel { // Consume events } }() // Benchmark JSON parsing and event creation b.ResetTimer() for i := 0; i < b.N; i++ { watcher.parsePodEvent([]byte(podJSON), "test-instance") } }