test: improve coverage

Author: elijah-rou

pkg/activator/handler/handler_test.go

@@ -39,6 +39,7 @@ import (
 	"knative.dev/serving/pkg/activator"
 	activatorconfig "knative.dev/serving/pkg/activator/config"
 	activatortest "knative.dev/serving/pkg/activator/testing"
+	apiconfig "knative.dev/serving/pkg/apis/config"
 	"knative.dev/serving/pkg/apis/serving"
 	v1 "knative.dev/serving/pkg/apis/serving/v1"
 	"knative.dev/serving/pkg/queue"
@@ -47,6 +48,7 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 
 	"knative.dev/pkg/logging"
+	ktesting "knative.dev/pkg/logging/testing"
 )
 
 const (
@@ -495,3 +497,72 @@ func (rr *responseRecorder) Write(p []byte) (int, error) {
 func (rr *responseRecorder) WriteHeader(code int) {
 	rr.code = code
 }
+
+func TestWrapActivatorHandlerWithFullDuplex(t *testing.T) {
+	logger := ktesting.TestLogger(t)
+	
+	tests := []struct {
+		name               string
+		annotation         string
+		expectFullDuplex   bool
+	}{
+		{
+			name:               "full duplex enabled",
+			annotation:         "Enabled",
+			expectFullDuplex:   true,
+		},
+		{
+			name:               "full duplex disabled",
+			annotation:         "Disabled",
+			expectFullDuplex:   false,
+		},
+		{
+			name:               "full duplex missing annotation",
+			annotation:         "",
+			expectFullDuplex:   false,
+		},
+		{
+			name:               "full duplex case insensitive",
+			annotation:         "enabled",
+			expectFullDuplex:   true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Create a test handler
+			testHandler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				// Just respond with OK
+				w.WriteHeader(http.StatusOK)
+			})
+			
+			// Wrap with full duplex handler
+			wrapped := WrapActivatorHandlerWithFullDuplex(testHandler, logger)
+			
+			// Create request with context
+			req := httptest.NewRequest(http.MethodGet, "/", nil)
+			
+			// Set up revision context with annotation
+			rev := &v1.Revision{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{},
+				},
+			}
+			if tt.annotation != "" {
+				rev.Annotations[apiconfig.AllowHTTPFullDuplexFeatureKey] = tt.annotation
+			}
+			
+			ctx := WithRevisionAndID(context.Background(), rev, types.NamespacedName{})
+			req = req.WithContext(ctx)
+			
+			// Execute request
+			resp := httptest.NewRecorder()
+			wrapped.ServeHTTP(resp, req)
+			
+			// Verify response code
+			if resp.Code != http.StatusOK {
+				t.Errorf("Expected status %d, got %d", http.StatusOK, resp.Code)
+			}
+		})
+	}
+}

pkg/activator/net/throttler_test.go

@@ -1230,3 +1230,297 @@ func TestResetTrackersRaceCondition(t *testing.T) {
 		rt.resetTrackers()
 	})
 }
+
+func TestPodTrackerStateTransitions(t *testing.T) {
+	t.Run("initial state is healthy", func(t *testing.T) {
+		tracker := newPodTracker("10.0.0.1:8012", nil)
+		
+		state := podState(tracker.state.Load())
+		if state != podHealthy {
+			t.Errorf("Expected initial state to be podHealthy, got %v", state)
+		}
+	})
+	
+	t.Run("tryDrain transitions from healthy to draining", func(t *testing.T) {
+		tracker := newPodTracker("10.0.0.1:8012", nil)
+		
+		// Should successfully transition to draining
+		if !tracker.tryDrain() {
+			t.Error("Expected tryDrain to succeed on healthy pod")
+		}
+		
+		state := podState(tracker.state.Load())
+		if state != podDraining {
+			t.Errorf("Expected state to be podDraining after tryDrain, got %v", state)
+		}
+		
+		// Should not transition again
+		if tracker.tryDrain() {
+			t.Error("Expected tryDrain to fail on already draining pod")
+		}
+		
+		// Verify draining start time was set
+		if tracker.drainingStartTime.Load() == 0 {
+			t.Error("Expected drainingStartTime to be set")
+		}
+	})
+	
+	t.Run("pending state allows reservation", func(t *testing.T) {
+		tracker := newPodTracker("10.0.0.1:8012", nil)
+		tracker.state.Store(uint32(podPending))
+		
+		// Should be able to reserve on pending pod
+		release, ok := tracker.Reserve(context.Background())
+		if !ok {
+			t.Error("Expected Reserve to succeed on pending pod")
+		}
+		if release != nil {
+			release()
+		}
+	})
+	
+	t.Run("draining state blocks new reservations", func(t *testing.T) {
+		tracker := newPodTracker("10.0.0.1:8012", nil)
+		tracker.tryDrain()
+		
+		// Should not be able to reserve on draining pod
+		release, ok := tracker.Reserve(context.Background())
+		if ok {
+			t.Error("Expected Reserve to fail on draining pod")
+		}
+		if release != nil {
+			release()
+		}
+	})
+	
+	t.Run("removed state blocks new reservations", func(t *testing.T) {
+		tracker := newPodTracker("10.0.0.1:8012", nil)
+		tracker.state.Store(uint32(podRemoved))
+		
+		// Should not be able to reserve on removed pod
+		release, ok := tracker.Reserve(context.Background())
+		if ok {
+			t.Error("Expected Reserve to fail on removed pod")
+		}
+		if release != nil {
+			release()
+		}
+	})
+}
+
+func TestPodTrackerReferenceCouting(t *testing.T) {
+	t.Run("reference counting on successful reserve", func(t *testing.T) {
+		tracker := newPodTracker("10.0.0.1:8012", nil)
+		
+		// Initial ref count should be 0
+		if tracker.getRefCount() != 0 {
+			t.Errorf("Expected initial ref count to be 0, got %d", tracker.getRefCount())
+		}
+		
+		// Reserve should increment ref count
+		release, ok := tracker.Reserve(context.Background())
+		if !ok {
+			t.Fatal("Expected Reserve to succeed")
+		}
+		
+		if tracker.getRefCount() != 1 {
+			t.Errorf("Expected ref count to be 1 after Reserve, got %d", tracker.getRefCount())
+		}
+		
+		// Release should decrement ref count
+		release()
+		
+		if tracker.getRefCount() != 0 {
+			t.Errorf("Expected ref count to be 0 after release, got %d", tracker.getRefCount())
+		}
+	})
+	
+	t.Run("reference counting on failed reserve", func(t *testing.T) {
+		tracker := newPodTracker("10.0.0.1:8012", nil)
+		tracker.state.Store(uint32(podDraining))
+		
+		// Initial ref count should be 0
+		if tracker.getRefCount() != 0 {
+			t.Errorf("Expected initial ref count to be 0, got %d", tracker.getRefCount())
+		}
+		
+		// Reserve should fail and not increment ref count
+		release, ok := tracker.Reserve(context.Background())
+		if ok {
+			t.Fatal("Expected Reserve to fail on draining pod")
+		}
+		if release != nil {
+			release()
+		}
+		
+		if tracker.getRefCount() != 0 {
+			t.Errorf("Expected ref count to remain 0 after failed Reserve, got %d", tracker.getRefCount())
+		}
+	})
+	
+	t.Run("multiple concurrent reservations", func(t *testing.T) {
+		tracker := newPodTracker("10.0.0.1:8012", nil)
+		
+		const numReservations = 10
+		var wg sync.WaitGroup
+		releases := make([]func(), numReservations)
+		
+		// Make concurrent reservations
+		for i := 0; i < numReservations; i++ {
+			wg.Add(1)
+			go func(idx int) {
+				defer wg.Done()
+				release, ok := tracker.Reserve(context.Background())
+				if ok {
+					releases[idx] = release
+				}
+			}(i)
+		}
+		
+		wg.Wait()
+		
+		// Check ref count
+		expectedCount := uint64(0)
+		for _, release := range releases {
+			if release != nil {
+				expectedCount++
+			}
+		}
+		
+		if tracker.getRefCount() != expectedCount {
+			t.Errorf("Expected ref count to be %d, got %d", expectedCount, tracker.getRefCount())
+		}
+		
+		// Release all
+		for _, release := range releases {
+			if release != nil {
+				release()
+			}
+		}
+		
+		if tracker.getRefCount() != 0 {
+			t.Errorf("Expected ref count to be 0 after all releases, got %d", tracker.getRefCount())
+		}
+	})
+	
+	t.Run("releaseRef with zero refcount", func(t *testing.T) {
+		tracker := newPodTracker("10.0.0.1:8012", nil)
+		
+		// Should handle gracefully (not panic)
+		tracker.releaseRef()
+		
+		// Ref count should remain 0
+		if tracker.getRefCount() != 0 {
+			t.Errorf("Expected ref count to remain 0, got %d", tracker.getRefCount())
+		}
+	})
+}
+
+func TestPodTrackerWeightOperations(t *testing.T) {
+	t.Run("weight increment and decrement", func(t *testing.T) {
+		tracker := newPodTracker("10.0.0.1:8012", nil)
+		
+		// Initial weight should be 0
+		if tracker.getWeight() != 0 {
+			t.Errorf("Expected initial weight to be 0, got %d", tracker.getWeight())
+		}
+		
+		// Increment weight
+		tracker.increaseWeight()
+		if tracker.getWeight() != 1 {
+			t.Errorf("Expected weight to be 1 after increase, got %d", tracker.getWeight())
+		}
+		
+		// Decrement weight
+		tracker.decreaseWeight()
+		if tracker.getWeight() != 0 {
+			t.Errorf("Expected weight to be 0 after decrease, got %d", tracker.getWeight())
+		}
+	})
+	
+	t.Run("weight underflow protection", func(t *testing.T) {
+		tracker := newPodTracker("10.0.0.1:8012", nil)
+		
+		// Decrement from 0 should not underflow
+		tracker.decreaseWeight()
+		
+		// Weight should remain 0 (not wrap around to max uint32)
+		weight := tracker.getWeight()
+		if weight != 0 && weight != ^uint32(0) {
+			// Allow either 0 or max uint32 based on implementation
+			t.Logf("Weight after underflow: %d", weight)
+		}
+	})
+}
+
+func TestPodTrackerWithBreaker(t *testing.T) {
+	t.Run("capacity with breaker", func(t *testing.T) {
+		breaker := queue.NewBreaker(queue.BreakerParams{
+			QueueDepth:      10,
+			MaxConcurrency:  5,
+			InitialCapacity: 5,
+		})
+		tracker := newPodTracker("10.0.0.1:8012", breaker)
+		
+		if tracker.Capacity() != 5 {
+			t.Errorf("Expected capacity to be 5, got %d", tracker.Capacity())
+		}
+	})
+	
+	t.Run("pending with breaker", func(t *testing.T) {
+		breaker := queue.NewBreaker(queue.BreakerParams{
+			QueueDepth:      10,
+			MaxConcurrency:  5,
+			InitialCapacity: 5,
+		})
+		tracker := newPodTracker("10.0.0.1:8012", breaker)
+		
+		// Initially should have 0 pending
+		if tracker.Pending() != 0 {
+			t.Errorf("Expected pending to be 0, got %d", tracker.Pending())
+		}
+	})
+	
+	t.Run("in-flight with breaker", func(t *testing.T) {
+		breaker := queue.NewBreaker(queue.BreakerParams{
+			QueueDepth:      10,
+			MaxConcurrency:  5,
+			InitialCapacity: 5,
+		})
+		tracker := newPodTracker("10.0.0.1:8012", breaker)
+		
+		// Initially should have 0 in-flight
+		if tracker.InFlight() != 0 {
+			t.Errorf("Expected in-flight to be 0, got %d", tracker.InFlight())
+		}
+		
+		// Reserve should increment in-flight
+		ctx := context.Background()
+		release, ok := breaker.Reserve(ctx)
+		if !ok {
+			t.Fatal("Expected Reserve to succeed")
+		}
+		defer release()
+		
+		if tracker.InFlight() != 1 {
+			t.Errorf("Expected in-flight to be 1, got %d", tracker.InFlight())
+		}
+	})
+	
+	t.Run("update concurrency with breaker", func(t *testing.T) {
+		breaker := queue.NewBreaker(queue.BreakerParams{
+			QueueDepth:      10,
+			MaxConcurrency:  5,
+			InitialCapacity: 5,
+		})
+		tracker := newPodTracker("10.0.0.1:8012", breaker)
+		
+		// Update concurrency
+		tracker.UpdateConcurrency(10)
+		
+		// Capacity should be updated
+		if tracker.Capacity() != 10 {
+			t.Errorf("Expected capacity to be 10 after update, got %d", tracker.Capacity())
+		}
+	})
+}