package queue import ( "context" "sync" "testing" "time" ) func TestPriorityQueue_Submit(t *testing.T) { config := &Config{ Enabled: true, Workers: 5, CriticalBuffer: 10, HighBuffer: 20, NormalBuffer: 50, LowBuffer: 30, BatchBuffer: 100, } pq := NewPriorityQueue(config) if pq == nil { t.Fatal("expected queue to be created") } defer pq.Close() // Submit a request result, err := pq.Submit(context.Background(), PriorityNormal, func() (interface{}, error) { return "test-result", nil }) if err != nil { t.Fatalf("Submit failed: %v", err) } if result != "test-result" { t.Errorf("expected test-result, got %v", result) } } func TestPriorityQueue_PriorityOrder(t *testing.T) { config := &Config{ Enabled: true, Workers: 1, // Single worker to ensure ordering CriticalBuffer: 10, HighBuffer: 10, NormalBuffer: 10, LowBuffer: 10, BatchBuffer: 10, } pq := NewPriorityQueue(config) defer pq.Close() var order []string var orderMu sync.Mutex // Add requests in reverse priority order for _, p := range []Priority{PriorityBatch, PriorityLow, PriorityNormal, PriorityHigh, PriorityCritical} { priority := p req := &Request{ Priority: priority, Context: context.Background(), Handler: func() (interface{}, error) { time.Sleep(10 * time.Millisecond) orderMu.Lock() order = append(order, priority.String()) orderMu.Unlock() return nil, nil }, ResultCh: make(chan *Result, 1), } pq.Enqueue(req) } // Wait for processing time.Sleep(200 * time.Millisecond) // Critical should be processed first (or among first) t.Logf("processing order: %v", order) } func TestPriorityQueue_Stats(t *testing.T) { config := &Config{ Enabled: true, Workers: 2, CriticalBuffer: 10, HighBuffer: 10, NormalBuffer: 50, LowBuffer: 10, BatchBuffer: 10, } pq := NewPriorityQueue(config) defer pq.Close() // Submit some requests for i := 0; i < 5; i++ { pq.Submit(context.Background(), PriorityNormal, func() (interface{}, error) { return nil, nil }) } stats := pq.Stats() if stats.Processed[PriorityNormal] != 5 { t.Errorf("expected 5 normal requests processed, got %d", stats.Processed[PriorityNormal]) } } func TestDetectPriority(t *testing.T) { tests := []struct { streaming bool messageCount int headers map[string]string expected Priority }{ {true, 0, nil, PriorityCritical}, {false, 10, nil, PriorityHigh}, {false, 2, nil, PriorityNormal}, {false, 0, map[string]string{"X-Batch-Request": "true"}, PriorityBatch}, {false, 0, map[string]string{"X-Request-Priority": "low"}, PriorityLow}, } for _, tt := range tests { got := DetectPriority(tt.streaming, tt.messageCount, tt.headers) if got != tt.expected { t.Errorf("DetectPriority(%v, %d, %v) = %v, want %v", tt.streaming, tt.messageCount, tt.headers, got, tt.expected) } } }