package traefikllmgateway import ( "context" "encoding/json" "errors" "fmt" "net/http" "net/http/httptest" "testing" "time" ) // newEventsTestConfig builds a Config with Admin enabled, one openai-type // provider pointed at upstreamURL, and two users: "alice" (an ordinary // caller, Limits left to the test to set) and "admin1" (Admin: true, for // reading GET /admin/api/events). Callers may mutate the returned Config // (in particular cfg.Users.Inline[0].Limits) before calling New. func newEventsTestConfig(upstreamURL string) *Config { cfg := CreateConfig() cfg.Admin = &AdminConfig{Enabled: true} cfg.Providers = map[string]*ProviderConfig{ "openai": {Type: "openai", BaseURL: upstreamURL, APIKey: "sk-up", Models: []string{"gpt-test"}}, } cfg.Groups = map[string]*GroupConfig{"default": {}} cfg.Users = &UsersConfig{Inline: []*UserConfig{ {Name: "alice", Group: "default", APIKey: "sk-alice", Limits: &LimitsConfig{}}, {Name: "admin1", Group: "default", APIKey: "sk-admin1", Admin: true}, }} return cfg } // getAdminEvents drives GET /admin/api/events?limit=... as sk-admin1 and // decodes the response. func getAdminEvents(t *testing.T, h http.Handler, query string) adminEventsResponse { t.Helper() req := httptest.NewRequest(http.MethodGet, adminEventsPath+query, nil) req.Header.Set("Authorization", "Bearer sk-admin1") rec := httptest.NewRecorder() h.ServeHTTP(rec, req) if rec.Code != http.StatusOK { t.Fatalf("GET %s%s: status = %d, want 200, body=%s", adminEventsPath, query, rec.Code, rec.Body.String()) } var got adminEventsResponse if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil { t.Fatalf("decode: %v", err) } return got } // --- gate (spec §4, same as every other /admin/api/* route) --- func TestAdminEvents_Gate(t *testing.T) { t.Parallel() srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) })) defer srv.Close() cfg := newEventsTestConfig(srv.URL) next := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}) h, err := New(context.Background(), next, cfg, "llmgw") if err != nil { t.Fatalf("New: %v", err) } rec := httptest.NewRecorder() h.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, adminEventsPath, nil)) if rec.Code != http.StatusUnauthorized { t.Errorf("unauthenticated: status = %d, want 401", rec.Code) } req := httptest.NewRequest(http.MethodGet, adminEventsPath, nil) req.Header.Set("Authorization", "Bearer sk-alice") rec = httptest.NewRecorder() h.ServeHTTP(rec, req) if rec.Code != http.StatusForbidden { t.Errorf("non-admin: status = %d, want 403", rec.Code) } got := getAdminEvents(t, h, "") if got.Events == nil { t.Error("Events must never be nil") } if got.Capacity != eventRingCap { t.Errorf("Capacity = %d, want %d", got.Capacity, eventRingCap) } } // --- limit validation --- func TestAdminEvents_LimitValidation(t *testing.T) { t.Parallel() srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) })) defer srv.Close() cfg := newEventsTestConfig(srv.URL) next := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}) h, err := New(context.Background(), next, cfg, "llmgw") if err != nil { t.Fatalf("New: %v", err) } for _, tc := range []struct{ name, query string }{ {"zero", "?limit=0"}, {"negative", "?limit=-1"}, {"over max", fmt.Sprintf("?limit=%d", adminEventsMaxLimit+1)}, {"not a number", "?limit=lots"}, } { t.Run(tc.name, func(t *testing.T) { req := httptest.NewRequest(http.MethodGet, adminEventsPath+tc.query, nil) req.Header.Set("Authorization", "Bearer sk-admin1") rec := httptest.NewRecorder() h.ServeHTTP(rec, req) if rec.Code != http.StatusBadRequest { t.Errorf("status = %d, want 400, body=%s", rec.Code, rec.Body.String()) } }) } got := getAdminEvents(t, h, fmt.Sprintf("?limit=%d", adminEventsMaxLimit)) if got.Events == nil { t.Error("Events must never be nil at the max limit") } } // --- hook 1: rate_limit / budget events via a real admission rejection --- func TestAdminEvents_RateLimitEvent(t *testing.T) { t.Parallel() srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") w.WriteHeader(http.StatusOK) _, _ = w.Write([]byte(`{"choices":[{"message":{"role":"assistant","content":"hi"}}],"usage":{"prompt_tokens":1,"completion_tokens":1}}`)) })) defer srv.Close() cfg := newEventsTestConfig(srv.URL) cfg.Users.Inline[0].Limits = &LimitsConfig{RequestsPerMinute: 1} next := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}) h, err := New(context.Background(), next, cfg, "llmgw") if err != nil { t.Fatalf("New: %v", err) } body := map[string]any{"model": "gpt-test", "messages": []any{map[string]any{"role": "user", "content": "hi"}}} rec1 := httptest.NewRecorder() h.ServeHTTP(rec1, newUnifiedRequest(t, http.MethodPost, "/v1/chat/completions", "sk-alice", body)) if rec1.Code != http.StatusOK { t.Fatalf("1st call: status = %d, want 200, body=%s", rec1.Code, rec1.Body.String()) } rec2 := httptest.NewRecorder() h.ServeHTTP(rec2, newUnifiedRequest(t, http.MethodPost, "/v1/chat/completions", "sk-alice", body)) if rec2.Code != http.StatusTooManyRequests { t.Fatalf("2nd call: status = %d, want 429 (over requestsPerMinute:1), body=%s", rec2.Code, rec2.Body.String()) } got := getAdminEvents(t, h, "?limit=10") if len(got.Events) == 0 { t.Fatal("want at least one event") } ev := got.Events[0] // newest first if ev.Kind != eventKindRateLimit { t.Errorf("Kind = %q, want %q", ev.Kind, eventKindRateLimit) } if ev.Route != routeChatCompletions { t.Errorf("Route = %q, want %q", ev.Route, routeChatCompletions) } if ev.User != "alice" { t.Errorf("User = %q, want %q", ev.User, "alice") } if ev.Status != http.StatusTooManyRequests { t.Errorf("Status = %d, want 429", ev.Status) } } func TestAdminEvents_BudgetEvent(t *testing.T) { t.Parallel() srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") w.WriteHeader(http.StatusOK) _, _ = w.Write([]byte(`{"choices":[{"message":{"role":"assistant","content":"hi"}}],"usage":{"prompt_tokens":5,"completion_tokens":5}}`)) })) defer srv.Close() cfg := newEventsTestConfig(srv.URL) cfg.Users.Inline[0].Limits = &LimitsConfig{TokensPerDay: 1} // the 1st call's own 10 reported tokens already exceeds this next := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}) h, err := New(context.Background(), next, cfg, "llmgw") if err != nil { t.Fatalf("New: %v", err) } body := map[string]any{"model": "gpt-test", "messages": []any{map[string]any{"role": "user", "content": "hi"}}} rec1 := httptest.NewRecorder() h.ServeHTTP(rec1, newUnifiedRequest(t, http.MethodPost, "/v1/chat/completions", "sk-alice", body)) if rec1.Code != http.StatusOK { t.Fatalf("1st call: status = %d, want 200, body=%s", rec1.Code, rec1.Body.String()) } rec2 := httptest.NewRecorder() h.ServeHTTP(rec2, newUnifiedRequest(t, http.MethodPost, "/v1/chat/completions", "sk-alice", body)) if rec2.Code != http.StatusTooManyRequests { t.Fatalf("2nd call: status = %d, want 429 (over tokensPerDay:1 budget), body=%s", rec2.Code, rec2.Body.String()) } got := getAdminEvents(t, h, "?limit=10") if len(got.Events) == 0 { t.Fatal("want at least one event") } if got.Events[0].Kind != eventKindBudget { t.Errorf("Kind = %q, want %q", got.Events[0].Kind, eventKindBudget) } } // --- hook 2: upstream event via a real 500 from the provider --- func TestAdminEvents_UpstreamEvent(t *testing.T) { t.Parallel() srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusInternalServerError) _, _ = w.Write([]byte(`{"error":"boom"}`)) })) defer srv.Close() cfg := newEventsTestConfig(srv.URL) next := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}) h, err := New(context.Background(), next, cfg, "llmgw") if err != nil { t.Fatalf("New: %v", err) } body := map[string]any{"model": "gpt-test", "messages": []any{map[string]any{"role": "user", "content": "hi"}}} rec := httptest.NewRecorder() h.ServeHTTP(rec, newUnifiedRequest(t, http.MethodPost, "/v1/chat/completions", "sk-alice", body)) if rec.Code != http.StatusInternalServerError { t.Fatalf("status = %d, want 500 (provider's own status forwarded), body=%s", rec.Code, rec.Body.String()) } got := getAdminEvents(t, h, "?limit=10") if len(got.Events) == 0 { t.Fatal("want at least one event") } ev := got.Events[0] if ev.Kind != eventKindUpstream { t.Errorf("Kind = %q, want %q", ev.Kind, eventKindUpstream) } if ev.Status != http.StatusInternalServerError { t.Errorf("Status = %d, want 500", ev.Status) } if ev.Provider != "openai" { t.Errorf("Provider = %q, want %q", ev.Provider, "openai") } if ev.User != "alice" { t.Errorf("User = %q, want %q", ev.User, "alice") } } // --- hook 4: unpriced event via pricingWarn --- func TestAdminEvents_UnpricedEvent(t *testing.T) { // warnedModels is a process-global dedup set (pricing.go) — reset // around this test exactly like TestGateway_PricingWarn_LogsUnderOwnInstance // (llmgateway_test.go) does, so a prior test's own warning for the // same model id can't suppress this one. warnedModelsMu.Lock() prevWarned, prevCapNotified := warnedModels, warnCapNotified warnedModels = map[string]bool{} warnCapNotified = false warnedModelsMu.Unlock() t.Cleanup(func() { warnedModelsMu.Lock() warnedModels, warnCapNotified = prevWarned, prevCapNotified warnedModelsMu.Unlock() }) srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) })) defer srv.Close() cfg := newEventsTestConfig(srv.URL) next := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}) h, err := New(context.Background(), next, cfg, "llmgw") if err != nil { t.Fatalf("New: %v", err) } gw := h.(*Gateway) const model = "totally-unpriced-model-for-admin-events-test" costMicrosKnownFor(model, usage{prompt: 1}, nil, gw.pricingWarn) got := getAdminEvents(t, h, "?limit=10") if len(got.Events) == 0 { t.Fatal("want at least one event") } ev := got.Events[0] if ev.Kind != eventKindUnpriced { t.Errorf("Kind = %q, want %q", ev.Kind, eventKindUnpriced) } if ev.Route != routePricing { t.Errorf("Route = %q, want %q", ev.Route, routePricing) } if ev.Model != model { t.Errorf("Model = %q, want %q", ev.Model, model) } } // --- source/degraded: local ring vs Redis-backed list --- // TestAdminEvents_RingFallback_NoRedisConfigured proves the local ring // always answers, labeled source "replica", not degraded, when Redis is // simply not configured (the common single-replica/no-Redis deployment). func TestAdminEvents_RingFallback_NoRedisConfigured(t *testing.T) { t.Parallel() srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) })) defer srv.Close() cfg := newEventsTestConfig(srv.URL) next := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}) h, err := New(context.Background(), next, cfg, "llmgw") if err != nil { t.Fatalf("New: %v", err) } gw := h.(*Gateway) gw.events.ring.add(gatewayEvent{Route: routeCapacity, Kind: eventKindCapacity, Message: "seed"}) got := getAdminEvents(t, h, "") if got.Source != "replica" { t.Errorf("Source = %q, want %q", got.Source, "replica") } if got.Degraded { t.Error("Degraded must be false when Redis was never configured") } if len(got.Events) != 1 || got.Events[0].Message != "seed" { t.Errorf("Events = %+v, want the ring's own seeded entry", got.Events) } } // TestAdminEvents_RedisSource drives a real GET /admin/api/events against // a scripted fake RESP server (F3, v0.3 dashboard task): the response // carries a fixture event whose Replica ("other-pod") differs from this // process's own — proving the multi-replica point of the feature: a // dashboard reading through ANY replica sees every replica's events via // the shared Redis-backed list, not just its own local ring. func TestAdminEvents_RedisSource(t *testing.T) { t.Parallel() fixture := gatewayEvent{ Time: time.Date(2026, 9, 23, 9, 0, 0, 0, time.UTC), Replica: "other-pod", Instance: "llmgw-other", User: "bob", Route: routeChatCompletions, Kind: eventKindRateLimit, Message: `user "bob" exceeded requests-per-minute limit (1)`, Status: http.StatusTooManyRequests, } payload, err := json.Marshal(fixture) if err != nil { t.Fatalf("marshal fixture: %v", err) } ln := newFakeListener(t) runFakeRESPServer(t, ln, []respStep{ {wantArgs: []string{"SELECT", "0"}, reply: []byte("+OK\r\n")}, {wantArgs: []string{"LRANGE", eventsRedisKey, "0", "49"}, reply: []byte(fmt.Sprintf("*1\r\n$%d\r\n%s\r\n", len(payload), payload))}, }) srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) })) defer srv.Close() cfg := newEventsTestConfig(srv.URL) cfg.Redis = &RedisConfig{Address: ln.Addr().String()} next := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}) h, err := New(context.Background(), next, cfg, "llmgw") if err != nil { t.Fatalf("New: %v", err) } got := getAdminEvents(t, h, "") if got.Source != "redis" { t.Errorf("Source = %q, want %q", got.Source, "redis") } if got.Degraded { t.Error("Degraded must be false on a successful redis read") } if len(got.Events) != 1 { t.Fatalf("Events = %+v, want exactly the 1 fixture entry", got.Events) } if got.Events[0].Replica != "other-pod" || got.Events[0].Message != fixture.Message { t.Errorf("Events[0] = %+v, want the fixture read back from redis", got.Events[0]) } } // TestAdminEvents_Degraded_OnLRangeError proves a Redis error reply to // LRANGE (not just a dead connection) falls back to the local ring, // labeled source "replica" with Degraded true — Redis IS configured, this // particular read just failed, which is exactly what Degraded means // (never "Redis not configured at all", eventLog.read's own doc comment). func TestAdminEvents_Degraded_OnLRangeError(t *testing.T) { t.Parallel() ln := newFakeListener(t) runFakeRESPServer(t, ln, []respStep{ {wantArgs: []string{"SELECT", "0"}, reply: []byte("+OK\r\n")}, {wantArgs: []string{"LRANGE", eventsRedisKey, "0", "49"}, reply: []byte("-ERR wrongtype or similar\r\n")}, }) srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) })) defer srv.Close() cfg := newEventsTestConfig(srv.URL) cfg.Redis = &RedisConfig{Address: ln.Addr().String()} next := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}) h, err := New(context.Background(), next, cfg, "llmgw") if err != nil { t.Fatalf("New: %v", err) } gw := h.(*Gateway) gw.events.ring.add(gatewayEvent{Route: routeCapacity, Kind: eventKindCapacity, Message: "local-fallback-seed"}) got := getAdminEvents(t, h, "") if got.Source != "replica" { t.Errorf("Source = %q, want %q (LRANGE errored, fall back to the local ring)", got.Source, "replica") } if !got.Degraded { t.Error("Degraded must be true: redis IS configured, but this read failed") } if len(got.Events) != 1 || got.Events[0].Message != "local-fallback-seed" { t.Errorf("Events = %+v, want the ring's own seeded entry", got.Events) } } // TestAdminEvents_Latched_SkipsLRANGE proves GET /admin/api/events answers // from the local ring (source "replica", Degraded true) WITHOUT ever // issuing LRANGE while the limiter's own store-down latch (limits.go's // storeLatched) is open — item 3 fix (verify-dash-backend.md's lower- // severity observation): before this fix, every admin poll during a // store outage still paid LRANGE's own respCallTimeout before falling // back, exactly like recordEvent already avoids paying for its own // Redis mirror push while latched. The fake Redis server is scripted to // answer LRANGE SUCCESSFULLY with a distinguishable fixture event, so a // wrongly-issued LRANGE call would surface as source "redis" carrying // that fixture — not silently indistinguishable from the correct skip, // which also happens to report source "replica"/Degraded true. func TestAdminEvents_Latched_SkipsLRANGE(t *testing.T) { t.Parallel() fixture := gatewayEvent{ Time: time.Now(), Replica: "other-pod", Route: routeChatCompletions, Kind: eventKindRateLimit, Message: "SHOULD-NOT-BE-SEEN: LRANGE was issued while the store was latched down", } payload, err := json.Marshal(fixture) if err != nil { t.Fatalf("marshal fixture: %v", err) } ln := newFakeListener(t) runFakeRESPServer(t, ln, []respStep{ {wantArgs: []string{"SELECT", "0"}, reply: []byte("+OK\r\n")}, {wantArgs: []string{"LRANGE", eventsRedisKey, "0", "49"}, reply: []byte(fmt.Sprintf("*1\r\n$%d\r\n%s\r\n", len(payload), payload))}, }) srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) })) defer srv.Close() cfg := newEventsTestConfig(srv.URL) cfg.Redis = &RedisConfig{Address: ln.Addr().String()} next := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}) h, err := New(context.Background(), next, cfg, "llmgw") if err != nil { t.Fatalf("New: %v", err) } gw := h.(*Gateway) gw.events.ring.add(gatewayEvent{Route: routeCapacity, Kind: eventKindCapacity, Message: "local-fallback-seed"}) // Opens the SAME latch storeIncrMulti/storeGetMulti already respect // (limits.go's own storeLatched doc comment) — recordStoreFailure is // the real trigger a live store outage uses, not a hand-set field. gw.limiter.recordStoreFailure(errors.New("simulated store outage for this test")) got := getAdminEvents(t, h, "") if got.Source != "replica" { t.Errorf("Source = %q, want %q (latched: LRANGE must be skipped entirely, not merely tolerated if it failed)", got.Source, "replica") } if !got.Degraded { t.Error("Degraded must be true while the store is latched down") } if len(got.Events) != 1 || got.Events[0].Message != "local-fallback-seed" { t.Errorf("Events = %+v, want only the ring's own seeded entry — the fixture LRANGE would have answered must never appear", got.Events) } }