Something went wrong. Try again.
sloproxy
Something went wrong. Try again.
18 kB · 559 lines
Go
at commit e8e39526
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560package redisqueue
import ( "context" "encoding/json" "net/http" "net/http/httptest" "testing" "time"
"github.com/gin-gonic/gin" internallogging "github.com/router-for-me/CLIProxyAPI/v7/internal/logging" coreusage "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/usage")
func TestUsageQueuePluginPayloadIncludesStableFieldsAndSuccess(t *testing.T) { withEnabledQueue(t, func() { ctx := internallogging.WithRequestID(context.Background(), "ctx-request-id") ctx = internallogging.WithEndpoint(ctx, "POST /v1/chat/completions") ctx = internallogging.WithClientRequestMetadata(ctx, internallogging.ClientRequestMetadata{ ClientIP: "192.0.2.10", XForwardedFor: "203.0.113.5, 198.51.100.8", UserAgent: "test-client/1.0", }) ctx = internallogging.WithResponseStatusHolder(ctx) internallogging.SetResponseStatus(ctx, http.StatusOK) responseHeaders := http.Header{} responseHeaders.Add("X-Upstream-Request-Id", "upstream-req-1") responseHeaders.Add("Retry-After", "30")
plugin := &usageQueuePlugin{} plugin.HandleUsage(ctx, coreusage.Record{ Provider: "openai", ExecutorType: "KimiExecutor", Model: "gpt-5.4", Alias: "client-gpt", APIKey: "test-key", AuthIndex: "0", AuthType: "apikey", Source: "user@example.com", ReasoningEffort: "medium", ServiceTier: "auto", ResponseServiceTier: "default", Generate: coreusage.GenerateFlag(true), RequestedAt: time.Date(2026, 4, 25, 0, 0, 0, 0, time.UTC), Latency: 1500 * time.Millisecond, Detail: coreusage.Detail{ InputTokens: 10, OutputTokens: 20, TotalTokens: 30, }, ResponseHeaders: responseHeaders.Clone(), }) responseHeaders.Set("Retry-After", "999")
payload := popSinglePayload(t) requireStringField(t, payload, "provider", "openai") requireStringField(t, payload, "executor_type", "KimiExecutor") requireStringField(t, payload, "model", "gpt-5.4") requireStringField(t, payload, "alias", "client-gpt") requireStringField(t, payload, "endpoint", "POST /v1/chat/completions") requireStringField(t, payload, "auth_type", "apikey") requireMissingField(t, payload, "user_api_key") requireStringField(t, payload, "request_id", "ctx-request-id") requireStringField(t, payload, "client_ip", "192.0.2.10") requireStringField(t, payload, "x_forwarded_for", "203.0.113.5, 198.51.100.8") requireStringField(t, payload, "user_agent", "test-client/1.0") requireStringField(t, payload, "reasoning_effort", "medium") requireStringField(t, payload, "service_tier", "auto") requireMissingField(t, payload, "request_service_tier") requireStringField(t, payload, "response_service_tier", "default") requireIntField(t, payload, "accounting_version", coreusage.TokenAccountingSchemaVersion) requireTokenBreakdown(t, payload, coreusage.TokenAccountingQualityComplete, 30) requireTokensBoolField(t, payload, "cache_read_tokens_present", true) requireHeaderField(t, payload, "response_headers", "X-Upstream-Request-Id", []string{"upstream-req-1"}) requireHeaderField(t, payload, "response_headers", "Retry-After", []string{"30"}) requireBoolField(t, payload, "failed", false) requireBoolField(t, payload, "generate", true) requireFailField(t, payload, http.StatusOK, "") })}
func TestUsageQueuePluginNormalizesDirectSDKUsageByProvider(t *testing.T) { tests := []struct { provider string wantTotal int }{ {provider: "openai", wantTotal: 130}, {provider: "gemini", wantTotal: 142}, } for _, tt := range tests { t.Run(tt.provider, func(t *testing.T) { withEnabledQueue(t, func() { ctx := internallogging.WithResponseStatusHolder(context.Background()) internallogging.SetResponseStatus(ctx, http.StatusOK)
(&usageQueuePlugin{}).HandleUsage(ctx, coreusage.Record{ Provider: tt.provider, Model: "direct-sdk-model", Detail: coreusage.Detail{ InputTokens: 100, OutputTokens: 30, ReasoningTokens: 12, }, })
payload := popSinglePayload(t) requireIntField(t, requireTokensPayload(t, payload), "total_tokens", tt.wantTotal) requireTokenBreakdown(t, payload, coreusage.TokenAccountingQualityComplete, int64(tt.wantTotal)) }) }) }}
func TestUsageQueuePluginPayloadIncludesGenerateFalse(t *testing.T) { withEnabledQueue(t, func() { ctx := internallogging.WithResponseStatusHolder(context.Background()) internallogging.SetResponseStatus(ctx, http.StatusOK)
(&usageQueuePlugin{}).HandleUsage(ctx, coreusage.Record{ Provider: "openai", Model: "gpt-5.4", Generate: coreusage.GenerateFlag(false), Detail: coreusage.Detail{ InputTokens: 1, TotalTokens: 1, }, })
payload := popSinglePayload(t) requireBoolField(t, payload, "generate", false) })}
func TestUsageQueuePluginPayloadDefaultsGenerateTrueWhenOmitted(t *testing.T) { withEnabledQueue(t, func() { ctx := internallogging.WithResponseStatusHolder(context.Background()) internallogging.SetResponseStatus(ctx, http.StatusOK)
// Legacy callers construct usage.Record without Generate; omission must publish as true. (&usageQueuePlugin{}).HandleUsage(ctx, coreusage.Record{ Provider: "openai", Model: "gpt-5.4", Detail: coreusage.Detail{ InputTokens: 1, TotalTokens: 1, }, })
payload := popSinglePayload(t) requireBoolField(t, payload, "generate", true) })}
func TestUsageQueuePluginPreservesLegacyCachedOnlyUsage(t *testing.T) { withEnabledQueue(t, func() { ctx := internallogging.WithResponseStatusHolder(context.Background()) internallogging.SetResponseStatus(ctx, http.StatusOK)
(&usageQueuePlugin{}).HandleUsage(ctx, coreusage.Record{ Provider: "openai", Model: "gpt-5.4", Detail: coreusage.Detail{ CachedTokens: 13, }, })
payload := popSinglePayload(t) requireTokensBoolField(t, payload, "cache_read_tokens_present", true) tokens := requireTokensPayload(t, payload) requireIntField(t, tokens, "cache_read_tokens", 13) requireIntField(t, tokens, "total_tokens", 13) requireTokenBreakdown(t, payload, coreusage.TokenAccountingQualityUnclassified, 13) })}
func TestUsageQueuePluginEmitsSingleCanonicalAutoTier(t *testing.T) { withEnabledQueue(t, func() { ctx := coreusage.WithServiceTier(context.Background(), coreusage.AutoServiceTier) ctx = internallogging.WithResponseStatusHolder(ctx) internallogging.SetResponseStatus(ctx, http.StatusOK)
(&usageQueuePlugin{}).HandleUsage(ctx, coreusage.Record{ Provider: "openai", Model: "gpt-5.4", Detail: coreusage.Detail{ InputTokens: 1, TotalTokens: 1, }, })
payload := popSinglePayload(t) requireStringField(t, payload, "service_tier", "auto") requireMissingField(t, payload, "request_service_tier") })}
func TestUsageQueuePluginAcceptsDeprecatedRequestTierRecordField(t *testing.T) { withEnabledQueue(t, func() { ctx := internallogging.WithResponseStatusHolder(context.Background()) internallogging.SetResponseStatus(ctx, http.StatusOK)
(&usageQueuePlugin{}).HandleUsage(ctx, coreusage.Record{ Provider: "openai", Model: "gpt-5.4", RequestServiceTier: "priority", Detail: coreusage.Detail{InputTokens: 1, TotalTokens: 1}, })
payload := popSinglePayload(t) requireStringField(t, payload, "service_tier", "priority") requireMissingField(t, payload, "request_service_tier") })}
func TestUsageQueuePluginAsyncUsesRecordResponseHeaders(t *testing.T) { withEnabledQueue(t, func() { ctx := internallogging.WithRequestID(context.Background(), "ctx-request-id") ctx = internallogging.WithEndpoint(ctx, "POST /v1/chat/completions") ctx = internallogging.WithResponseStatusHolder(ctx) ctx = internallogging.WithResponseHeadersHolder(ctx) internallogging.SetResponseStatus(ctx, http.StatusOK) initialHeaders := http.Header{} initialHeaders.Set("X-Upstream-Request-Id", "upstream-req-1") internallogging.SetResponseHeaders(ctx, initialHeaders)
mgr := coreusage.NewManager(16) defer mgr.Stop()
mgr.Register(pluginFunc(func(ctx context.Context, _ coreusage.Record) { nextHeaders := http.Header{} nextHeaders.Set("X-Upstream-Request-Id", "upstream-req-2") internallogging.SetResponseHeaders(ctx, nextHeaders) })) mgr.Register(&usageQueuePlugin{})
mgr.Publish(ctx, coreusage.Record{ Provider: "openai", Model: "gpt-5.4", Alias: "client-gpt", APIKey: "test-key", AuthIndex: "0", AuthType: "apikey", Source: "user@example.com", RequestedAt: time.Date(2026, 4, 25, 0, 0, 0, 0, time.UTC), Latency: 1500 * time.Millisecond, Detail: coreusage.Detail{ InputTokens: 10, OutputTokens: 20, TotalTokens: 30, }, ResponseHeaders: internallogging.GetResponseHeaders(ctx), })
payload := waitForSinglePayload(t, 2*time.Second) requireHeaderField(t, payload, "response_headers", "X-Upstream-Request-Id", []string{"upstream-req-1"}) })}
func TestUsageQueuePluginPayloadIncludesStableFieldsAndFailureAndGinRequestID(t *testing.T) { withEnabledQueue(t, func() { ctx := internallogging.WithRequestID(context.Background(), "gin-request-id") ctx = internallogging.WithEndpoint(ctx, "GET /v1/responses") ctx = internallogging.WithResponseStatusHolder(ctx) internallogging.SetResponseStatus(ctx, http.StatusInternalServerError)
plugin := &usageQueuePlugin{} plugin.HandleUsage(ctx, coreusage.Record{ Provider: "openai", Model: "gpt-5.4-mini", Alias: "client-mini", APIKey: "test-key", AuthIndex: "0", AuthType: "apikey", Source: "user@example.com", RequestedAt: time.Date(2026, 4, 25, 0, 0, 0, 0, time.UTC), Latency: 2500 * time.Millisecond, Fail: coreusage.Failure{ StatusCode: http.StatusInternalServerError, Body: "upstream failed", }, Detail: coreusage.Detail{ InputTokens: 10, OutputTokens: 20, TotalTokens: 30, }, })
payload := popSinglePayload(t) requireStringField(t, payload, "provider", "openai") requireStringField(t, payload, "model", "gpt-5.4-mini") requireStringField(t, payload, "alias", "client-mini") requireStringField(t, payload, "endpoint", "GET /v1/responses") requireStringField(t, payload, "auth_type", "apikey") requireMissingField(t, payload, "user_api_key") requireStringField(t, payload, "request_id", "gin-request-id") requireBoolField(t, payload, "failed", true) requireFailField(t, payload, http.StatusInternalServerError, "upstream failed") })}
func TestUsageQueuePluginAsyncIgnoresRecycledGinContext(t *testing.T) { withEnabledQueue(t, func() { ginCtx := newTestGinContext(t, http.MethodPost, "/v1/chat/completions", http.StatusOK) ctx := context.WithValue(context.Background(), "gin", ginCtx) ctx = internallogging.WithRequestID(ctx, "ctx-request-id") ctx = internallogging.WithEndpoint(ctx, "POST /v1/chat/completions") ctx = internallogging.WithResponseStatusHolder(ctx) internallogging.SetResponseStatus(ctx, http.StatusInternalServerError)
mgr := coreusage.NewManager(16) defer mgr.Stop()
mgr.Register(pluginFunc(func(_ context.Context, _ coreusage.Record) { ginCtx.Request = httptest.NewRequest(http.MethodGet, "http://example.com/v1/responses", nil) ginCtx.Status(http.StatusOK) })) mgr.Register(&usageQueuePlugin{})
mgr.Publish(ctx, coreusage.Record{ Provider: "openai", Model: "gpt-5.4", Alias: "client-gpt", APIKey: "test-key", AuthIndex: "0", AuthType: "apikey", Source: "user@example.com", RequestedAt: time.Date(2026, 4, 25, 0, 0, 0, 0, time.UTC), Latency: 1500 * time.Millisecond, Fail: coreusage.Failure{ StatusCode: http.StatusBadGateway, Body: "bad gateway", }, Detail: coreusage.Detail{ InputTokens: 10, OutputTokens: 20, TotalTokens: 30, }, })
payload := waitForSinglePayload(t, 2*time.Second) requireStringField(t, payload, "endpoint", "POST /v1/chat/completions") requireStringField(t, payload, "alias", "client-gpt") requireMissingField(t, payload, "user_api_key") requireStringField(t, payload, "request_id", "ctx-request-id") requireBoolField(t, payload, "failed", true) requireFailField(t, payload, http.StatusBadGateway, "bad gateway") })}
func withEnabledQueue(t *testing.T, fn func()) { t.Helper()
prevQueueEnabled := Enabled() prevUsageEnabled := UsageStatisticsEnabled()
SetEnabled(false) SetEnabled(true) SetUsageStatisticsEnabled(true)
defer func() { SetEnabled(false) SetEnabled(prevQueueEnabled) SetUsageStatisticsEnabled(prevUsageEnabled) }()
fn()}
func newTestGinContext(t *testing.T, method, path string, status int) *gin.Context { t.Helper()
gin.SetMode(gin.TestMode) recorder := httptest.NewRecorder() ginCtx, _ := gin.CreateTestContext(recorder) ginCtx.Request = httptest.NewRequest(method, "http://example.com"+path, nil) if status != 0 { ginCtx.Status(status) } return ginCtx}
func popSinglePayload(t *testing.T) map[string]json.RawMessage { t.Helper()
items := PopOldest(10) if len(items) != 1 { t.Fatalf("PopOldest() items = %d, want 1", len(items)) }
var payload map[string]json.RawMessage if err := json.Unmarshal(items[0], &payload); err != nil { t.Fatalf("unmarshal payload: %v", err) } return payload}
func waitForSinglePayload(t *testing.T, timeout time.Duration) map[string]json.RawMessage { t.Helper()
deadline := time.Now().Add(timeout) for time.Now().Before(deadline) { items := PopOldest(10) if len(items) == 0 { time.Sleep(10 * time.Millisecond) continue } if len(items) != 1 { t.Fatalf("PopOldest() items = %d, want 1", len(items)) } var payload map[string]json.RawMessage if err := json.Unmarshal(items[0], &payload); err != nil { t.Fatalf("unmarshal payload: %v", err) } return payload } t.Fatalf("timeout waiting for queued payload") return nil}
func requireStringField(t *testing.T, payload map[string]json.RawMessage, key, want string) { t.Helper()
raw, ok := payload[key] if !ok { t.Fatalf("payload missing %q", key) } var got string if err := json.Unmarshal(raw, &got); err != nil { t.Fatalf("unmarshal %q: %v", key, err) } if got != want { t.Fatalf("%s = %q, want %q", key, got, want) }}
func requireIntField(t *testing.T, payload map[string]json.RawMessage, key string, want int) { t.Helper()
raw, ok := payload[key] if !ok { t.Fatalf("payload missing %q", key) } var got int if err := json.Unmarshal(raw, &got); err != nil { t.Fatalf("unmarshal %q: %v", key, err) } if got != want { t.Fatalf("%s = %d, want %d", key, got, want) }}
func requireTokenBreakdown(t *testing.T, payload map[string]json.RawMessage, quality coreusage.TokenAccountingQuality, total int64) { t.Helper()
raw, ok := payload["token_breakdown"] if !ok { t.Fatal("payload missing token_breakdown") } var breakdown coreusage.TokenBreakdown if err := json.Unmarshal(raw, &breakdown); err != nil { t.Fatalf("unmarshal token_breakdown: %v", err) } if !breakdown.Valid() || breakdown.Quality != quality || breakdown.TotalTokens != total { t.Fatalf("token_breakdown = %+v, want quality=%s total=%d", breakdown, quality, total) }}
func requireMissingField(t *testing.T, payload map[string]json.RawMessage, key string) { t.Helper()
if _, ok := payload[key]; ok { t.Fatalf("payload unexpectedly contains %q", key) }}
type pluginFunc func(context.Context, coreusage.Record)
func (fn pluginFunc) HandleUsage(ctx context.Context, record coreusage.Record) { fn(ctx, record)}
func requireBoolField(t *testing.T, payload map[string]json.RawMessage, key string, want bool) { t.Helper()
raw, ok := payload[key] if !ok { t.Fatalf("payload missing %q", key) } var got bool if err := json.Unmarshal(raw, &got); err != nil { t.Fatalf("unmarshal %q: %v", key, err) } if got != want { t.Fatalf("%s = %t, want %t", key, got, want) }}
func requireTokensPayload(t *testing.T, payload map[string]json.RawMessage) map[string]json.RawMessage { t.Helper() raw, ok := payload["tokens"] if !ok { t.Fatal("payload missing tokens") } var tokens map[string]json.RawMessage if errUnmarshal := json.Unmarshal(raw, &tokens); errUnmarshal != nil { t.Fatalf("unmarshal tokens: %v", errUnmarshal) } return tokens}
func requireTokensBoolField(t *testing.T, payload map[string]json.RawMessage, key string, want bool) { t.Helper() requireBoolField(t, requireTokensPayload(t, payload), key, want)}
func requireFailField(t *testing.T, payload map[string]json.RawMessage, wantStatus int, wantBody string) { t.Helper()
raw, ok := payload["fail"] if !ok { t.Fatalf("payload missing %q", "fail") } var got struct { StatusCode int `json:"status_code"` Body string `json:"body"` } if err := json.Unmarshal(raw, &got); err != nil { t.Fatalf("unmarshal fail: %v", err) } if got.StatusCode != wantStatus || got.Body != wantBody { t.Fatalf("fail = {status_code:%d body:%q}, want {status_code:%d body:%q}", got.StatusCode, got.Body, wantStatus, wantBody) }}
func requireHeaderField(t *testing.T, payload map[string]json.RawMessage, field, key string, want []string) { t.Helper()
raw, ok := payload[field] if !ok { t.Fatalf("payload missing %q", field) } var headers map[string][]string if err := json.Unmarshal(raw, &headers); err != nil { t.Fatalf("unmarshal %q: %v", field, err) } got, ok := headers[key] if !ok { t.Fatalf("%s missing header %q", field, key) } if len(got) != len(want) { t.Fatalf("%s[%q] = %v, want %v", field, key, got, want) } for i := range want { if got[i] != want[i] { t.Fatalf("%s[%q] = %v, want %v", field, key, got, want) } }}