//go:build unit package service import ( "bytes" "context" "errors" "io" "net/http" "net/http/httptest" "strings" "testing" "time" "github.com/Wei-Shaw/sub2api/internal/config" "github.com/Wei-Shaw/sub2api/internal/pkg/openai_compat" "github.com/gin-gonic/gin" "github.com/stretchr/testify/require" "github.com/tidwall/gjson" ) func TestBuildOpenAIChatCompletionsURL(t *testing.T) { t.Parallel() tests := []struct { name string base string want string }{ // 已是 /chat/completions:原样返回 {"already chat/completions", "https://api.openai.com/v1/chat/completions", "https://api.openai.com/v1/chat/completions"}, // 以 /v1 结尾:追加 /chat/completions {"bare /v1", "https://api.openai.com/v1", "https://api.openai.com/v1/chat/completions"}, // 其他情况:追加 /v1/chat/completions {"bare domain", "https://api.openai.com", "https://api.openai.com/v1/chat/completions"}, {"domain with trailing slash", "https://api.openai.com/", "https://api.openai.com/v1/chat/completions"}, // 第三方上游常见形式 {"third-party bare domain", "https://api.deepseek.com", "https://api.deepseek.com/v1/chat/completions"}, {"third-party with path prefix", "https://api.gptgod.online/api", "https://api.gptgod.online/api/v1/chat/completions"}, {"third-party versioned path", "https://open.bigmodel.cn/api/paas/v4", "https://open.bigmodel.cn/api/paas/v4/chat/completions"}, // 带空白字符 {"whitespace trimmed", " https://api.openai.com/v1 ", "https://api.openai.com/v1/chat/completions"}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { t.Parallel() got := buildOpenAIChatCompletionsURL(tt.base) require.Equal(t, tt.want, got) }) } } // TestBuildOpenAIResponsesURL_ProbeURL 锁定 probe/测试端点使用的 URL 构建逻辑, // 确保 buildOpenAIResponsesURL 对标准 OpenAI base_url 格式均拼出 `/v1/responses`。 func TestBuildOpenAIResponsesURL_ProbeURL(t *testing.T) { t.Parallel() tests := []struct { name string base string want string }{ {"bare domain", "https://api.openai.com", "https://api.openai.com/v1/responses"}, {"domain trailing slash", "https://api.openai.com/", "https://api.openai.com/v1/responses"}, {"bare /v1", "https://api.openai.com/v1", "https://api.openai.com/v1/responses"}, {"already /responses", "https://api.openai.com/v1/responses", "https://api.openai.com/v1/responses"}, {"third-party bare domain", "https://api.deepseek.com", "https://api.deepseek.com/v1/responses"}, {"third-party versioned path", "https://open.bigmodel.cn/api/paas/v4", "https://open.bigmodel.cn/api/paas/v4/responses"}, {"only domain, no scheme", "api.gptgod.online", "api.gptgod.online/v1/responses"}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { t.Parallel() got := buildOpenAIResponsesURL(tt.base) require.Equal(t, tt.want, got) }) } } func TestForwardAsRawChatCompletions_ForcesStreamUsageUpstreamAndPassesUsageDownstream(t *testing.T) { gin.SetMode(gin.TestMode) body := []byte(`{"model":"gpt-5.4","messages":[{"role":"user","content":"hello"}],"stream":true}`) rec := httptest.NewRecorder() c, _ := gin.CreateTestContext(rec) c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)) c.Request.Header.Set("Content-Type", "application/json") upstreamBody := strings.Join([]string{ `data: {"id":"chatcmpl_1","object":"chat.completion.chunk","model":"gpt-5.4","choices":[{"index":0,"delta":{"content":"ok"}}]}`, "", `data: {"id":"chatcmpl_1","object":"chat.completion.chunk","model":"gpt-5.4","choices":[],"usage":{"prompt_tokens":9,"completion_tokens":4,"total_tokens":13,"prompt_tokens_details":{"cached_tokens":3}}}`, "", "data: [DONE]", "", }, "\n") upstream := &httpUpstreamRecorder{resp: &http.Response{ StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_raw_usage"}}, Body: io.NopCloser(strings.NewReader(upstreamBody)), }} svc := &OpenAIGatewayService{ cfg: rawChatCompletionsTestConfig(), httpUpstream: upstream, } account := rawChatCompletionsTestAccount() result, err := svc.forwardAsRawChatCompletions(context.Background(), c, account, body, "") require.NoError(t, err) require.NotNil(t, result) require.Equal(t, 9, result.Usage.InputTokens) require.Equal(t, 4, result.Usage.OutputTokens) require.Equal(t, 3, result.Usage.CacheReadInputTokens) require.NotNil(t, upstream.lastReq) require.NoError(t, upstream.lastReq.Context().Err()) require.Equal(t, HTTPUpstreamProfileOpenAI, HTTPUpstreamProfileFromContext(upstream.lastReq.Context())) require.True(t, gjson.GetBytes(upstream.lastBody, "stream_options.include_usage").Bool()) require.Contains(t, rec.Body.String(), `"usage"`) require.Contains(t, rec.Body.String(), "data: [DONE]") } func TestForwardAsChatCompletions_OpenAICompatibleGrokRawMissingUsageFailsBeforeWrite(t *testing.T) { gin.SetMode(gin.TestMode) body := []byte(`{"model":"grok-4.5","messages":[{"role":"user","content":"hello"}],"stream":false}`) recorder := httptest.NewRecorder() c, _ := gin.CreateTestContext(recorder) c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)) upstream := &httpUpstreamRecorder{resp: &http.Response{ StatusCode: http.StatusOK, Header: http.Header{ "Content-Type": []string{"application/json"}, "X-Request-Id": []string{"rid-openai-compat-grok-no-usage"}, }, Body: io.NopCloser(strings.NewReader( `{"id":"resp_missing_usage","object":"chat.completion","model":"grok-4.5","choices":[{"index":0,"message":{"role":"assistant","content":"hello"},"finish_reason":"stop"}]}`, )), }} svc := &OpenAIGatewayService{ cfg: rawChatCompletionsTestConfig(), httpUpstream: upstream, } account := rawChatCompletionsTestAccount() account.Name = "openai-compatible-grok" account.Extra = map[string]any{openai_compat.ExtraKeyResponsesSupported: false} result, err := svc.ForwardAsChatCompletions(context.Background(), c, account, body, "", "") require.Nil(t, result) var failoverErr *UpstreamFailoverError require.ErrorAs(t, err, &failoverErr) require.False(t, c.Writer.Written(), "unbilled Grok content must not be returned by an OpenAI-compatible account") require.Empty(t, recorder.Body.String()) } func TestForwardAsChatCompletions_OpenAICompatibleRawUsageGuard(t *testing.T) { tests := []struct { name string model string upstreamResponse string modelMapping map[string]any wantGuarded bool }{ { name: "Grok response without usage", model: "grok-4.5", upstreamResponse: `{"id":"resp_missing","object":"chat.completion","model":"grok-4.5","choices":[{"index":0,"message":{"role":"assistant","content":"hello"},"finish_reason":"stop"}]}`, wantGuarded: true, }, { name: "namespaced Grok response without usage", model: "x-ai/grok-4.5", upstreamResponse: `{"id":"resp_namespaced","object":"chat.completion","model":"x-ai/grok-4.5","choices":[{"index":0,"message":{"role":"assistant","content":"hello"},"finish_reason":"stop"}]}`, wantGuarded: true, }, { name: "Grok response with aggregate usage passes", model: "grok-4.5", upstreamResponse: `{"id":"resp_usage","object":"chat.completion","model":"grok-4.5","choices":[{"index":0,"message":{"role":"assistant","content":"hello"},"finish_reason":"stop"}],"usage":{"prompt_tokens":9,"completion_tokens":3,"total_tokens":12}}`, wantGuarded: false, }, { name: "Grok alias mapped to non-Grok remains unchanged", model: "grok-alias", upstreamResponse: `{"id":"resp_mapped","object":"chat.completion","model":"gpt-5.4","choices":[{"index":0,"message":{"role":"assistant","content":"hello"},"finish_reason":"stop"}]}`, modelMapping: map[string]any{"grok-alias": "gpt-5.4"}, wantGuarded: false, }, { name: "Grok response with detail-only usage", model: "grok-4.5", upstreamResponse: `{"id":"resp_detail_only","object":"chat.completion","model":"grok-4.5","choices":[{"index":0,"message":{"role":"assistant","content":"hello"},"finish_reason":"stop"}],"usage":{"input_tokens_details":{"text_tokens":9,"image_tokens":2},"output_tokens_details":{"image_tokens":1}}}`, wantGuarded: true, }, { name: "non-Grok response without usage remains unchanged", model: "gpt-5.4", upstreamResponse: `{"id":"resp_openai","object":"chat.completion","model":"gpt-5.4","choices":[{"index":0,"message":{"role":"assistant","content":"hello"},"finish_reason":"stop"}]}`, wantGuarded: false, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { gin.SetMode(gin.TestMode) body := []byte(`{"model":"` + tt.model + `","messages":[{"role":"user","content":"hello"}],"stream":false}`) recorder := httptest.NewRecorder() c, _ := gin.CreateTestContext(recorder) c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)) upstream := &httpUpstreamRecorder{resp: &http.Response{ StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"application/json"}, "X-Request-Id": []string{"rid-openai-compatible"}}, Body: io.NopCloser(strings.NewReader(tt.upstreamResponse)), }} svc := &OpenAIGatewayService{cfg: rawChatCompletionsTestConfig(), httpUpstream: upstream} account := rawChatCompletionsTestAccount() account.Name = "openai-compatible" account.Extra = map[string]any{openai_compat.ExtraKeyResponsesSupported: false} if tt.modelMapping != nil { account.Credentials["model_mapping"] = tt.modelMapping } result, err := svc.ForwardAsChatCompletions(context.Background(), c, account, body, "", "") if !tt.wantGuarded { require.NoError(t, err) require.NotNil(t, result) require.True(t, c.Writer.Written()) return } require.Nil(t, result) var failoverErr *UpstreamFailoverError require.ErrorAs(t, err, &failoverErr) require.Equal(t, http.StatusBadGateway, failoverErr.StatusCode) require.Equal(t, "grok_missing_usage", gjson.GetBytes(failoverErr.ResponseBody, "error.code").String()) require.False(t, c.Writer.Written(), "unbilled Grok content must not be returned") }) } } func TestForwardAsRawChatCompletions_PreservesMappedGPT56MaxEffort(t *testing.T) { gin.SetMode(gin.TestMode) body := []byte(`{"model":"sol","messages":[{"role":"user","content":"hello"}],"reasoning_effort":"max","stream":false}`) rec := httptest.NewRecorder() c, _ := gin.CreateTestContext(rec) c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)) c.Request.Header.Set("Content-Type", "application/json") upstream := &httpUpstreamRecorder{resp: &http.Response{ StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"application/json"}}, Body: io.NopCloser(strings.NewReader( `{"id":"chatcmpl_max","object":"chat.completion","model":"gpt-5.6-sol","choices":[{"index":0,"message":{"role":"assistant","content":"ok"},"finish_reason":"stop"}],"usage":{"prompt_tokens":3,"completion_tokens":2,"total_tokens":5}}`, )), }} svc := &OpenAIGatewayService{ cfg: rawChatCompletionsTestConfig(), httpUpstream: upstream, } account := rawChatCompletionsTestAccount() account.Credentials["model_mapping"] = map[string]any{"sol": "gpt-5.6-sol"} result, err := svc.forwardAsRawChatCompletions(context.Background(), c, account, body, "") require.NoError(t, err) require.NotNil(t, result) require.Equal(t, "gpt-5.6-sol", gjson.GetBytes(upstream.lastBody, "model").String()) require.Equal(t, "max", gjson.GetBytes(upstream.lastBody, "reasoning_effort").String()) require.NotNil(t, result.ReasoningEffort) require.Equal(t, "max", *result.ReasoningEffort) } func TestForwardAsRawChatCompletions_NonStreamingCapturesCacheWriteUsage(t *testing.T) { gin.SetMode(gin.TestMode) tests := []struct { name string usageJSON string wantWrite int }{ { name: "positive cache write", usageJSON: `{"prompt_tokens":12,"completion_tokens":3,"total_tokens":15,"prompt_tokens_details":{"cached_tokens":4,"cache_write_tokens":6}}`, wantWrite: 6, }, { name: "nested zero overrides legacy alias", usageJSON: `{"prompt_tokens":12,"completion_tokens":3,"total_tokens":15,"cache_creation_input_tokens":19,"prompt_tokens_details":{"cached_tokens":4,"cache_write_tokens":0}}`, wantWrite: 0, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { body := []byte(`{"model":"gpt-5.6","messages":[{"role":"user","content":"hello"}],"stream":false}`) rec := httptest.NewRecorder() c, _ := gin.CreateTestContext(rec) c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)) c.Request.Header.Set("Content-Type", "application/json") upstream := &httpUpstreamRecorder{resp: &http.Response{ StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"application/json"}}, Body: io.NopCloser(strings.NewReader( `{"id":"chatcmpl_cache","object":"chat.completion","model":"gpt-5.6","choices":[{"index":0,"message":{"role":"assistant","content":"ok"},"finish_reason":"stop"}],"usage":` + tt.usageJSON + `}`, )), }} svc := &OpenAIGatewayService{ cfg: rawChatCompletionsTestConfig(), httpUpstream: upstream, } result, err := svc.forwardAsRawChatCompletions(context.Background(), c, rawChatCompletionsTestAccount(), body, "") require.NoError(t, err) require.NotNil(t, result) require.Equal(t, 12, result.Usage.InputTokens) require.Equal(t, 4, result.Usage.CacheReadInputTokens) require.Equal(t, tt.wantWrite, result.Usage.CacheCreationInputTokens) }) } } func TestForwardAsRawChatCompletions_PreservesDeepSeekReasoningContentNonStreaming(t *testing.T) { gin.SetMode(gin.TestMode) body := []byte(`{"model":"deepseek-reasoner","messages":[{"role":"user","content":"hello"}],"stream":false}`) rec := httptest.NewRecorder() c, _ := gin.CreateTestContext(rec) c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)) c.Request.Header.Set("Content-Type", "application/json") upstreamJSON := `{"id":"chatcmpl_reasoning","object":"chat.completion","model":"deepseek-reasoner","choices":[{"index":0,"message":{"role":"assistant","reasoning_content":"think first","content":"final answer"},"finish_reason":"stop"}],"usage":{"prompt_tokens":3,"completion_tokens":5,"total_tokens":8}}` upstream := &httpUpstreamRecorder{resp: &http.Response{ StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"application/json"}, "x-request-id": []string{"rid_deepseek_reasoning_json"}}, Body: io.NopCloser(strings.NewReader(upstreamJSON)), }} svc := &OpenAIGatewayService{ cfg: rawChatCompletionsTestConfig(), httpUpstream: upstream, } account := rawChatCompletionsTestAccount() result, err := svc.forwardAsRawChatCompletions(context.Background(), c, account, body, "") require.NoError(t, err) require.NotNil(t, result) require.Equal(t, 3, result.Usage.InputTokens) require.Equal(t, 5, result.Usage.OutputTokens) require.Equal(t, "think first", gjson.Get(rec.Body.String(), "choices.0.message.reasoning_content").String()) require.Equal(t, "final answer", gjson.Get(rec.Body.String(), "choices.0.message.content").String()) } func TestForwardAsRawChatCompletions_PreservesDeepSeekReasoningContentStreaming(t *testing.T) { gin.SetMode(gin.TestMode) body := []byte(`{"model":"deepseek-reasoner","messages":[{"role":"user","content":"hello"}],"stream":true}`) rec := httptest.NewRecorder() c, _ := gin.CreateTestContext(rec) c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)) c.Request.Header.Set("Content-Type", "application/json") upstreamBody := strings.Join([]string{ `data: {"id":"chatcmpl_reasoning","object":"chat.completion.chunk","model":"deepseek-reasoner","choices":[{"index":0,"delta":{"role":"assistant"},"finish_reason":null}]}`, "", `data: {"id":"chatcmpl_reasoning","object":"chat.completion.chunk","model":"deepseek-reasoner","choices":[{"index":0,"delta":{"reasoning_content":"think first"},"finish_reason":null}]}`, "", `data: {"id":"chatcmpl_reasoning","object":"chat.completion.chunk","model":"deepseek-reasoner","choices":[{"index":0,"delta":{"content":"final answer"},"finish_reason":null}]}`, "", `data: {"id":"chatcmpl_reasoning","object":"chat.completion.chunk","model":"deepseek-reasoner","choices":[],"usage":{"prompt_tokens":3,"completion_tokens":5,"total_tokens":8}}`, "", "data: [DONE]", "", }, "\n") upstream := &httpUpstreamRecorder{resp: &http.Response{ StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_deepseek_reasoning_stream"}}, Body: io.NopCloser(strings.NewReader(upstreamBody)), }} svc := &OpenAIGatewayService{ cfg: rawChatCompletionsTestConfig(), httpUpstream: upstream, } account := rawChatCompletionsTestAccount() result, err := svc.forwardAsRawChatCompletions(context.Background(), c, account, body, "") require.NoError(t, err) require.NotNil(t, result) require.Equal(t, 3, result.Usage.InputTokens) require.Equal(t, 5, result.Usage.OutputTokens) require.Contains(t, rec.Body.String(), `"reasoning_content":"think first"`) require.Contains(t, rec.Body.String(), `"content":"final answer"`) require.Contains(t, rec.Body.String(), "data: [DONE]") } func TestForwardAsRawChatCompletions_PreservesDeepSeekReasoningContentInRequest(t *testing.T) { gin.SetMode(gin.TestMode) body := []byte(`{"model":"deepseek-v4-pro","messages":[{"role":"user","content":"weather"},{"role":"assistant","reasoning_content":"need tool","content":"","tool_calls":[{"id":"call_1","type":"function","function":{"name":"get_weather","arguments":"{}"}}]},{"role":"tool","tool_call_id":"call_1","content":"cloudy"}],"stream":false}`) rec := httptest.NewRecorder() c, _ := gin.CreateTestContext(rec) c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)) c.Request.Header.Set("Content-Type", "application/json") upstream := &httpUpstreamRecorder{resp: &http.Response{ StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"application/json"}, "x-request-id": []string{"rid_deepseek_reasoning_request"}}, Body: io.NopCloser(strings.NewReader(`{"id":"chatcmpl_request","object":"chat.completion","model":"deepseek-v4-pro","choices":[{"index":0,"message":{"role":"assistant","content":"done"},"finish_reason":"stop"}],"usage":{"prompt_tokens":4,"completion_tokens":2,"total_tokens":6}}`)), }} svc := &OpenAIGatewayService{ cfg: rawChatCompletionsTestConfig(), httpUpstream: upstream, } account := rawChatCompletionsTestAccount() result, err := svc.forwardAsRawChatCompletions(context.Background(), c, account, body, "") require.NoError(t, err) require.NotNil(t, result) require.Equal(t, "need tool", gjson.GetBytes(upstream.lastBody, "messages.1.reasoning_content").String()) require.Equal(t, "get_weather", gjson.GetBytes(upstream.lastBody, "messages.1.tool_calls.0.function.name").String()) } func TestForwardAsRawChatCompletions_NormalizesGLMReasoningEffortForUpstream(t *testing.T) { gin.SetMode(gin.TestMode) body := []byte(`{"model":"glm-5.2","messages":[{"role":"user","content":"hello"}],"reasoning_effort":"xhigh","stream":false}`) rec := httptest.NewRecorder() c, _ := gin.CreateTestContext(rec) c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)) c.Request.Header.Set("Content-Type", "application/json") upstream := &httpUpstreamRecorder{resp: &http.Response{ StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"application/json"}, "x-request-id": []string{"rid_glm_effort"}}, Body: io.NopCloser(strings.NewReader(`{"id":"chatcmpl_glm","object":"chat.completion","model":"glm-5.2","choices":[{"index":0,"message":{"role":"assistant","content":"ok"},"finish_reason":"stop"}],"usage":{"prompt_tokens":1,"completion_tokens":2,"total_tokens":3}}`)), }} svc := &OpenAIGatewayService{ cfg: rawChatCompletionsTestConfig(), httpUpstream: upstream, } account := rawChatCompletionsTestAccount() result, err := svc.forwardAsRawChatCompletions(context.Background(), c, account, body, "") require.NoError(t, err) require.NotNil(t, result) require.Equal(t, "max", gjson.GetBytes(upstream.lastBody, "reasoning_effort").String()) } func TestForwardAsRawChatCompletions_SilentRefusalTriggersFailover(t *testing.T) { gin.SetMode(gin.TestMode) body := largeRawChatCompletionsBody() rec := httptest.NewRecorder() c, _ := gin.CreateTestContext(rec) c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)) c.Request.Header.Set("Content-Type", "application/json") upstreamBody := strings.Join([]string{ `data: {"id":"chatcmpl_silent","object":"chat.completion.chunk","model":"gpt-5.5","choices":[{"index":0,"delta":{"role":"assistant"}}]}`, "", `data: {"id":"chatcmpl_silent","object":"chat.completion.chunk","model":"gpt-5.5","choices":[{"index":0,"delta":{"content":""},"finish_reason":"stop"}]}`, "", "data: [DONE]", "", }, "\n") upstream := &httpUpstreamRecorder{resp: &http.Response{ StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_silent"}}, Body: io.NopCloser(strings.NewReader(upstreamBody)), }} svc := &OpenAIGatewayService{ cfg: rawChatCompletionsTestConfig(), httpUpstream: upstream, } result, err := svc.forwardAsRawChatCompletions(context.Background(), c, rawChatCompletionsTestAccount(), body, "") require.Nil(t, result) var failoverErr *UpstreamFailoverError require.True(t, errors.As(err, &failoverErr)) require.Equal(t, http.StatusBadGateway, failoverErr.StatusCode) require.True(t, IsOpenAISilentRefusalErrorBody(failoverErr.ResponseBody)) require.False(t, c.Writer.Written(), "silent refusal must not commit a 200 response before failover") require.Empty(t, rec.Body.String()) } func TestForwardAsRawChatCompletions_SilentRefusalToolCallsExempt(t *testing.T) { gin.SetMode(gin.TestMode) body := largeRawChatCompletionsBody() rec := httptest.NewRecorder() c, _ := gin.CreateTestContext(rec) c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)) c.Request.Header.Set("Content-Type", "application/json") upstreamBody := strings.Join([]string{ `data: {"id":"chatcmpl_tool","object":"chat.completion.chunk","model":"gpt-5.5","choices":[{"index":0,"delta":{"role":"assistant"}}]}`, "", `data: {"id":"chatcmpl_tool","object":"chat.completion.chunk","model":"gpt-5.5","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"call_1","type":"function","function":{"name":"lookup","arguments":""}}]}}]}`, "", `data: {"id":"chatcmpl_tool","object":"chat.completion.chunk","model":"gpt-5.5","choices":[{"index":0,"delta":{},"finish_reason":"tool_calls"}]}`, "", "data: [DONE]", "", }, "\n") upstream := &httpUpstreamRecorder{resp: &http.Response{ StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_tool"}}, Body: io.NopCloser(strings.NewReader(upstreamBody)), }} svc := &OpenAIGatewayService{ cfg: rawChatCompletionsTestConfig(), httpUpstream: upstream, } result, err := svc.forwardAsRawChatCompletions(context.Background(), c, rawChatCompletionsTestAccount(), body, "") require.NoError(t, err) require.NotNil(t, result) require.Contains(t, rec.Body.String(), `"tool_calls"`) require.Contains(t, rec.Body.String(), `"finish_reason":"tool_calls"`) } func TestHandleChatStreamingResponse_SilentRefusalReasoningSummaryExempt(t *testing.T) { gin.SetMode(gin.TestMode) rec := httptest.NewRecorder() c, _ := gin.CreateTestContext(rec) c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", nil) upstreamBody := strings.Join([]string{ `data: {"type":"response.created","response":{"id":"resp_reasoning","model":"gpt-5.5"}}`, "", `data: {"type":"response.reasoning_summary_text.delta","delta":"thinking only"}`, "", `data: {"type":"response.completed","response":{"id":"resp_reasoning","model":"gpt-5.5","status":"completed"}}`, "", }, "\n") resp := &http.Response{ StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_reasoning"}}, Body: io.NopCloser(strings.NewReader(upstreamBody)), } svc := &OpenAIGatewayService{cfg: rawChatCompletionsTestConfig()} result, err := svc.handleChatStreamingResponse( resp, c, rawChatCompletionsTestAccount(), "gpt-5.5", "gpt-5.5", "gpt-5.5", time.Now(), openAISilentRefusalMinRequestBodyBytes, ) require.NoError(t, err) require.NotNil(t, result) require.Contains(t, rec.Body.String(), `"reasoning_content":"thinking only"`) require.Contains(t, rec.Body.String(), "data: [DONE]") } func TestForwardAsRawChatCompletions_SilentRefusalNormalContentExempt(t *testing.T) { gin.SetMode(gin.TestMode) body := largeRawChatCompletionsBody() rec := httptest.NewRecorder() c, _ := gin.CreateTestContext(rec) c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)) c.Request.Header.Set("Content-Type", "application/json") upstreamBody := strings.Join([]string{ `data: {"id":"chatcmpl_ok","object":"chat.completion.chunk","model":"gpt-5.5","choices":[{"index":0,"delta":{"role":"assistant"}}]}`, "", `data: {"id":"chatcmpl_ok","object":"chat.completion.chunk","model":"gpt-5.5","choices":[{"index":0,"delta":{"content":"ok"}}]}`, "", `data: {"id":"chatcmpl_ok","object":"chat.completion.chunk","model":"gpt-5.5","choices":[{"index":0,"delta":{"content":""},"finish_reason":"stop"}]}`, "", "data: [DONE]", "", }, "\n") upstream := &httpUpstreamRecorder{resp: &http.Response{ StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_ok"}}, Body: io.NopCloser(strings.NewReader(upstreamBody)), }} svc := &OpenAIGatewayService{ cfg: rawChatCompletionsTestConfig(), httpUpstream: upstream, } result, err := svc.forwardAsRawChatCompletions(context.Background(), c, rawChatCompletionsTestAccount(), body, "") require.NoError(t, err) require.NotNil(t, result) require.Contains(t, rec.Body.String(), `"content":"ok"`) require.Contains(t, rec.Body.String(), "data: [DONE]") } // TestForwardAsRawChatCompletions_StripsEmptyToolCallIdentity 端到端验证 raw // CC 流式直转路径剔除 DashScope/DeepSeek 后续参数 delta 的空 id/name: // 下游仍保留首包合法 id/name 与 arguments 碎片,但后续 delta 不再带 // `"id":""` / `"name":""`,避免 dsh 等客户端用 `!== undefined` 合并时把 // 首包合法值覆盖掉(ToolNotFoundError: unknown tool "")。 func TestForwardAsRawChatCompletions_StripsEmptyToolCallIdentity(t *testing.T) { gin.SetMode(gin.TestMode) body := []byte(`{"model":"deepseek-v4-flash","messages":[{"role":"user","content":"weather"}],"stream":true}`) rec := httptest.NewRecorder() c, _ := gin.CreateTestContext(rec) c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)) c.Request.Header.Set("Content-Type", "application/json") upstreamBody := strings.Join([]string{ `data: {"id":"chatcmpl_tool","object":"chat.completion.chunk","model":"deepseek-v4-flash","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"call_example","type":"function","function":{"name":"web_search","arguments":""}}]}}]}`, "", `data: {"id":"chatcmpl_tool","object":"chat.completion.chunk","model":"deepseek-v4-flash","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"","type":"function","function":{"name":"","arguments":"{\"query\":"}}]}}]}`, "", `data: {"id":"chatcmpl_tool","object":"chat.completion.chunk","model":"deepseek-v4-flash","choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"","type":"function","function":{"name":"","arguments":"\"example\"}"}}]}}]}`, "", `data: {"id":"chatcmpl_tool","object":"chat.completion.chunk","model":"deepseek-v4-flash","choices":[{"index":0,"delta":{},"finish_reason":"tool_calls"}]}`, "", "data: [DONE]", "", }, "\n") upstream := &httpUpstreamRecorder{resp: &http.Response{ StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_tool_identity"}}, Body: io.NopCloser(strings.NewReader(upstreamBody)), }} svc := &OpenAIGatewayService{ cfg: rawChatCompletionsTestConfig(), httpUpstream: upstream, } account := rawChatCompletionsTestAccount() result, err := svc.forwardAsRawChatCompletions(context.Background(), c, account, body, "") require.NoError(t, err) require.NotNil(t, result) downstream := rec.Body.String() require.Contains(t, downstream, `"id":"call_example"`) require.Contains(t, downstream, `"name":"web_search"`) require.Contains(t, downstream, `{\"query\":`) require.Contains(t, downstream, `\"example\"}`) require.Contains(t, downstream, "data: [DONE]") require.NotContains(t, downstream, `"id":""`) require.NotContains(t, downstream, `"name":""`) // 逐条扫下游 data payload:后续参数 delta 的 tool_calls.0.id / // function.name 必须已剔除(Exists() == false),首包合法值保留。 followUpSeen := false for _, line := range strings.Split(downstream, "\n") { payload, ok := extractOpenAISSEDataLine(line) if !ok { continue } trimmed := strings.TrimSpace(payload) if trimmed == "" || trimmed == "[DONE]" { continue } delta := gjson.Get(payload, "choices.0.delta") if !delta.Exists() || !delta.Get("tool_calls").Exists() { continue } id := delta.Get("tool_calls.0.id") if id.String() == "call_example" { require.Equal(t, "web_search", delta.Get("tool_calls.0.function.name").String()) continue } require.False(t, id.Exists(), "empty id must be stripped: %s", payload) require.False(t, delta.Get("tool_calls.0.function.name").Exists(), "empty name must be stripped: %s", payload) require.NotEmpty(t, delta.Get("tool_calls.0.function.arguments").String()) followUpSeen = true } require.True(t, followUpSeen) } // 上游在生成中途干净 EOF(无 [DONE]/usage/finish_reason)且已向客户端写出内容: // 不能再记成 HTTP 200 成功,必须回带类型化的上游截断错误,由 handler 补 SSE error // 帧并计入 SLA 失败。 func TestForwardAsRawChatCompletions_TruncatedStreamAfterOutputFailsRequest(t *testing.T) { gin.SetMode(gin.TestMode) body := []byte(`{"model":"deepseek-v4-pro","messages":[{"role":"user","content":"hello"}],"stream":true}`) rec := httptest.NewRecorder() c, _ := gin.CreateTestContext(rec) c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)) c.Request.Header.Set("Content-Type", "application/json") upstreamBody := strings.Join([]string{ `data: {"id":"chatcmpl_cut","object":"chat.completion.chunk","model":"deepseek-v4-pro","choices":[{"index":0,"delta":{"content":"half an ans"},"finish_reason":null}]}`, "", }, "\n") upstream := &httpUpstreamRecorder{resp: &http.Response{ StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_truncated"}}, Body: io.NopCloser(strings.NewReader(upstreamBody)), }} svc := &OpenAIGatewayService{ cfg: rawChatCompletionsTestConfig(), httpUpstream: upstream, } result, err := svc.forwardAsRawChatCompletions(context.Background(), c, rawChatCompletionsTestAccount(), body, "") require.Error(t, err) require.NotNil(t, result, "已收字节的用量仍需带回,供 ops 记录首 token 时延") var failoverErr *UpstreamFailoverError require.False(t, errors.As(err, &failoverErr), "已写出语义字节后不得再 failover") code, message, ok := OpenAIUpstreamStreamReadErrorDetails(err) require.True(t, ok) require.Equal(t, OpenAIUpstreamStreamTruncatedCode, code) require.NotEmpty(t, message) // 已写出的内容保持原样透传,客户端拿到的仍是它已经收到的那部分。 require.Contains(t, rec.Body.String(), `"content":"half an ans"`) require.NotContains(t, rec.Body.String(), "data: [DONE]") rawEvents, ok := c.Get(OpsUpstreamErrorsKey) require.True(t, ok) events := rawEvents.([]*OpsUpstreamErrorEvent) require.Len(t, events, 1) require.Equal(t, "http_error", events[0].Kind) require.Nil(t, events[0].ProxyID) require.Equal(t, opsProxyNameDirect, events[0].ProxyName) } // 截断 failover 事件必须带上本次传输真实使用的托管代理快照。 func TestForwardAsRawChatCompletions_TruncationFailoverAttributesManagedProxy(t *testing.T) { gin.SetMode(gin.TestMode) body := []byte(`{"model":"deepseek-v4-pro","messages":[{"role":"user","content":"hello"}],"stream":true}`) rec := httptest.NewRecorder() c, _ := gin.CreateTestContext(rec) c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)) c.Request.Header.Set("Content-Type", "application/json") upstream := &httpUpstreamRecorder{resp: &http.Response{ StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_empty_proxy"}}, Body: io.NopCloser(strings.NewReader("")), }} svc := &OpenAIGatewayService{cfg: rawChatCompletionsTestConfig(), httpUpstream: upstream} proxy := &Proxy{ID: 8001, Name: "oxylabs-uk-8001", Protocol: "http", Host: "proxy.example", Port: 8080} account := rawChatCompletionsTestAccount() account.ProxyID = &proxy.ID account.Proxy = proxy _, err := svc.forwardAsRawChatCompletions(context.Background(), c, account, body, "") var failoverErr *UpstreamFailoverError require.ErrorAs(t, err, &failoverErr) require.Equal(t, proxy.URL(), upstream.lastProxyURL) rawEvents, ok := c.Get(OpsUpstreamErrorsKey) require.True(t, ok) events := rawEvents.([]*OpsUpstreamErrorEvent) require.Len(t, events, 1) require.Equal(t, "failover", events[0].Kind) require.NotNil(t, events[0].ProxyID) require.Equal(t, proxy.ID, *events[0].ProxyID) require.Equal(t, proxy.Name, events[0].ProxyName) } // 上游 200 但一个 SSE 字节都没发:响应头尚未提交,应换号重试而不是回 200 空流。 func TestForwardAsRawChatCompletions_EmptyStreamBeforeOutputTriggersFailover(t *testing.T) { gin.SetMode(gin.TestMode) body := []byte(`{"model":"deepseek-v4-pro","messages":[{"role":"user","content":"hello"}],"stream":true}`) rec := httptest.NewRecorder() c, _ := gin.CreateTestContext(rec) c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)) c.Request.Header.Set("Content-Type", "application/json") upstream := &httpUpstreamRecorder{resp: &http.Response{ StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_empty"}}, Body: io.NopCloser(strings.NewReader("")), }} svc := &OpenAIGatewayService{ cfg: rawChatCompletionsTestConfig(), httpUpstream: upstream, } result, err := svc.forwardAsRawChatCompletions(context.Background(), c, rawChatCompletionsTestAccount(), body, "") require.Nil(t, result) var failoverErr *UpstreamFailoverError require.True(t, errors.As(err, &failoverErr)) require.Equal(t, http.StatusBadGateway, failoverErr.StatusCode) require.Equal(t, OpenAIUpstreamStreamTruncatedCode, gjson.GetBytes(failoverErr.ResponseBody, "error.code").String()) require.True(t, failoverErr.ShouldRetryNextAccount()) require.False(t, c.Writer.Written(), "换号重试前不得提交 200 响应头") require.Empty(t, rec.Body.String()) } // 传输层错误(Cloudflare edge reset 等)在写出后同样不能记成功,且分类要区别于 // 干净 EOF,便于 ops 分辨 reset 与静默截断。 func TestForwardAsRawChatCompletions_StreamReadErrorAfterOutputFailsRequest(t *testing.T) { gin.SetMode(gin.TestMode) body := []byte(`{"model":"deepseek-v4-pro","messages":[{"role":"user","content":"hello"}],"stream":true}`) rec := httptest.NewRecorder() c, _ := gin.CreateTestContext(rec) c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)) c.Request.Header.Set("Content-Type", "application/json") upstream := &httpUpstreamRecorder{resp: &http.Response{ StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_reset"}}, Body: &openAIChatStreamReadErrorCloser{ payload: []byte(`data: {"id":"chatcmpl_reset","object":"chat.completion.chunk","model":"deepseek-v4-pro","choices":[{"index":0,"delta":{"content":"partial"}}]}` + "\n\n"), err: errors.New("read tcp 172.18.0.4->172.65.90.23:443: read: connection reset by peer"), }, }} svc := &OpenAIGatewayService{ cfg: rawChatCompletionsTestConfig(), httpUpstream: upstream, } result, err := svc.forwardAsRawChatCompletions(context.Background(), c, rawChatCompletionsTestAccount(), body, "") require.Error(t, err) require.NotNil(t, result) code, _, ok := OpenAIUpstreamStreamReadErrorDetails(err) require.True(t, ok) require.Equal(t, OpenAIUpstreamStreamReadErrorCode, code) require.Contains(t, rec.Body.String(), `"content":"partial"`) } // 边界:缺 [DONE] 但收到了 usage 帧 —— 生成已完整,只是尾巴丢失。必须继续按成功 // 计费,否则会误伤那些跑完就直接 EOF 的兼容上游并白送 token。 func TestForwardAsRawChatCompletions_MissingDoneWithUsageStillSucceeds(t *testing.T) { gin.SetMode(gin.TestMode) body := []byte(`{"model":"deepseek-v4-pro","messages":[{"role":"user","content":"hello"}],"stream":true}`) rec := httptest.NewRecorder() c, _ := gin.CreateTestContext(rec) c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)) c.Request.Header.Set("Content-Type", "application/json") upstreamBody := strings.Join([]string{ `data: {"id":"chatcmpl_nodone","object":"chat.completion.chunk","model":"deepseek-v4-pro","choices":[{"index":0,"delta":{"content":"ok"}}]}`, "", `data: {"id":"chatcmpl_nodone","object":"chat.completion.chunk","model":"deepseek-v4-pro","choices":[],"usage":{"prompt_tokens":11,"completion_tokens":6,"total_tokens":17}}`, "", }, "\n") upstream := &httpUpstreamRecorder{resp: &http.Response{ StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_nodone_usage"}}, Body: io.NopCloser(strings.NewReader(upstreamBody)), }} svc := &OpenAIGatewayService{ cfg: rawChatCompletionsTestConfig(), httpUpstream: upstream, } result, err := svc.forwardAsRawChatCompletions(context.Background(), c, rawChatCompletionsTestAccount(), body, "") require.NoError(t, err) require.NotNil(t, result) require.Equal(t, 11, result.Usage.InputTokens) require.Equal(t, 6, result.Usage.OutputTokens) } // 边界:缺 [DONE] 与 usage,但末帧带 finish_reason —— 生成正常结束,同样不判截断。 func TestForwardAsRawChatCompletions_MissingDoneWithFinishReasonStillSucceeds(t *testing.T) { gin.SetMode(gin.TestMode) body := []byte(`{"model":"deepseek-v4-pro","messages":[{"role":"user","content":"hello"}],"stream":true}`) rec := httptest.NewRecorder() c, _ := gin.CreateTestContext(rec) c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)) c.Request.Header.Set("Content-Type", "application/json") upstreamBody := strings.Join([]string{ `data: {"id":"chatcmpl_finish","object":"chat.completion.chunk","model":"deepseek-v4-pro","choices":[{"index":0,"delta":{"content":"done"},"finish_reason":null}]}`, "", `data: {"id":"chatcmpl_finish","object":"chat.completion.chunk","model":"deepseek-v4-pro","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}`, "", }, "\n") upstream := &httpUpstreamRecorder{resp: &http.Response{ StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_nodone_finish"}}, Body: io.NopCloser(strings.NewReader(upstreamBody)), }} svc := &OpenAIGatewayService{ cfg: rawChatCompletionsTestConfig(), httpUpstream: upstream, } result, err := svc.forwardAsRawChatCompletions(context.Background(), c, rawChatCompletionsTestAccount(), body, "") require.NoError(t, err) require.NotNil(t, result) require.Contains(t, rec.Body.String(), `"finish_reason":"stop"`) } // openAIRawStreamDisconnectedWriter 模拟客户端已断开:raw 直转路径经 // WriteString 写出,故两个方法都必须失败(只覆盖 Write 会被内嵌 writer 绕过)。 type openAIRawStreamDisconnectedWriter struct { gin.ResponseWriter } func (w *openAIRawStreamDisconnectedWriter) Write([]byte) (int, error) { return 0, errors.New("write failed: client disconnected") } func (w *openAIRawStreamDisconnectedWriter) WriteString(string) (int, error) { return 0, errors.New("write failed: client disconnected") } // 客户端已断开时上游随后截断:两者不可区分,沿用既有语义按已收用量正常收尾计费, // 不得把客户端离场记成上游故障。 func TestForwardAsRawChatCompletions_ClientDisconnectTruncationStillBills(t *testing.T) { gin.SetMode(gin.TestMode) body := []byte(`{"model":"deepseek-v4-pro","messages":[{"role":"user","content":"hello"}],"stream":true}`) rec := httptest.NewRecorder() c, _ := gin.CreateTestContext(rec) c.Writer = &openAIRawStreamDisconnectedWriter{ResponseWriter: c.Writer} c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)) c.Request.Header.Set("Content-Type", "application/json") upstreamBody := strings.Join([]string{ `data: {"id":"chatcmpl_gone","object":"chat.completion.chunk","model":"deepseek-v4-pro","choices":[{"index":0,"delta":{"content":"ok"}}]}`, "", }, "\n") upstream := &httpUpstreamRecorder{resp: &http.Response{ StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_gone"}}, Body: io.NopCloser(strings.NewReader(upstreamBody)), }} svc := &OpenAIGatewayService{ cfg: rawChatCompletionsTestConfig(), httpUpstream: upstream, } result, err := svc.forwardAsRawChatCompletions(context.Background(), c, rawChatCompletionsTestAccount(), body, "") require.NoError(t, err) require.NotNil(t, result) } // 客户端取消会连带取消上游请求,上游读因此报 context.Canceled:同样不判为上游截断。 func TestForwardAsRawChatCompletions_ClientCancelTruncationStillBills(t *testing.T) { gin.SetMode(gin.TestMode) body := []byte(`{"model":"deepseek-v4-pro","messages":[{"role":"user","content":"hello"}],"stream":true}`) rec := httptest.NewRecorder() c, _ := gin.CreateTestContext(rec) c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)) c.Request.Header.Set("Content-Type", "application/json") upstream := &httpUpstreamRecorder{resp: &http.Response{ StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_cancel"}}, Body: &openAIChatStreamReadErrorCloser{ payload: []byte(`data: {"id":"chatcmpl_cancel","object":"chat.completion.chunk","model":"deepseek-v4-pro","choices":[{"index":0,"delta":{"content":"ok"}}]}` + "\n\n"), err: context.Canceled, }, }} svc := &OpenAIGatewayService{ cfg: rawChatCompletionsTestConfig(), httpUpstream: upstream, } result, err := svc.forwardAsRawChatCompletions(context.Background(), c, rawChatCompletionsTestAccount(), body, "") require.NoError(t, err) require.NotNil(t, result) } func TestOpenAIRawStreamTerminalState(t *testing.T) { t.Parallel() tests := []struct { name string payloads []string clientStarted bool wantTerminated bool wantTruncated bool }{ { name: "done sentinel", payloads: []string{`{"choices":[{"delta":{"content":"a"}}]}`, "[DONE]"}, clientStarted: true, wantTerminated: true, }, { name: "usage chunk", payloads: []string{`{"choices":[],"usage":{"prompt_tokens":1,"completion_tokens":1}}`}, clientStarted: true, wantTerminated: true, }, { name: "finish reason", payloads: []string{`{"choices":[{"delta":{},"finish_reason":"length"}]}`}, clientStarted: true, wantTerminated: true, }, { name: "null finish reason is not terminal", payloads: []string{`{"choices":[{"delta":{"content":"a"},"finish_reason":null}]}`}, clientStarted: true, wantTruncated: true, }, { name: "usage null is not terminal", payloads: []string{`{"choices":[{"delta":{"content":"a"}}],"usage":null}`}, clientStarted: true, wantTruncated: true, }, { // 上游对 stream 请求回了裸 JSON:无 data: 行,既有行为是原样透传。 name: "non-sse body already forwarded", payloads: nil, clientStarted: true, }, { name: "no bytes at all", payloads: nil, clientStarted: false, wantTruncated: true, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { t.Parallel() var state openAIRawStreamTerminalState for _, payload := range tt.payloads { state.ObserveDataLine(payload) } require.Equal(t, tt.wantTerminated, state.Terminated()) require.Equal(t, tt.wantTruncated, state.IsTruncated(tt.clientStarted)) }) } } func TestForwardAsRawChatCompletions_ClientDisconnectDrainsUsage(t *testing.T) { gin.SetMode(gin.TestMode) body := []byte(`{"model":"gpt-5.4","messages":[{"role":"user","content":"hello"}],"stream":true}`) rec := httptest.NewRecorder() c, _ := gin.CreateTestContext(rec) c.Writer = &openAIChatFailingWriter{ResponseWriter: c.Writer, failAfter: 0} c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)) c.Request.Header.Set("Content-Type", "application/json") upstreamBody := strings.Join([]string{ `data: {"id":"chatcmpl_1","object":"chat.completion.chunk","model":"gpt-5.4","choices":[{"index":0,"delta":{"content":"ok"}}]}`, "", `data: {"id":"chatcmpl_1","object":"chat.completion.chunk","model":"gpt-5.4","choices":[],"usage":{"prompt_tokens":17,"completion_tokens":8,"total_tokens":25,"prompt_tokens_details":{"cached_tokens":6}}}`, "", "data: [DONE]", "", }, "\n") upstream := &httpUpstreamRecorder{resp: &http.Response{ StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_raw_disconnect"}}, Body: io.NopCloser(strings.NewReader(upstreamBody)), }} svc := &OpenAIGatewayService{ cfg: rawChatCompletionsTestConfig(), httpUpstream: upstream, } account := rawChatCompletionsTestAccount() result, err := svc.forwardAsRawChatCompletions(context.Background(), c, account, body, "") require.NoError(t, err) require.NotNil(t, result) require.Equal(t, 17, result.Usage.InputTokens) require.Equal(t, 8, result.Usage.OutputTokens) require.Equal(t, 6, result.Usage.CacheReadInputTokens) require.True(t, gjson.GetBytes(upstream.lastBody, "stream_options.include_usage").Bool()) } func TestForwardAsRawChatCompletions_UpstreamRequestIgnoresClientCancel(t *testing.T) { gin.SetMode(gin.TestMode) reqCtx, cancel := context.WithCancel(context.Background()) body := []byte(`{"model":"gpt-5.4","messages":[{"role":"user","content":"hello"}],"stream":true}`) rec := httptest.NewRecorder() c, _ := gin.CreateTestContext(rec) c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)).WithContext(reqCtx) c.Request.Header.Set("Content-Type", "application/json") cancel() upstreamBody := strings.Join([]string{ `data: {"id":"chatcmpl_1","object":"chat.completion.chunk","model":"gpt-5.4","choices":[],"usage":{"prompt_tokens":5,"completion_tokens":2,"total_tokens":7}}`, "", "data: [DONE]", "", }, "\n") upstream := &httpUpstreamRecorder{resp: &http.Response{ StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"text/event-stream"}, "x-request-id": []string{"rid_raw_ctx"}}, Body: io.NopCloser(strings.NewReader(upstreamBody)), }} svc := &OpenAIGatewayService{ cfg: rawChatCompletionsTestConfig(), httpUpstream: upstream, } account := rawChatCompletionsTestAccount() result, err := svc.forwardAsRawChatCompletions(reqCtx, c, account, body, "") require.NoError(t, err) require.NotNil(t, result) require.NotNil(t, upstream.lastReq) require.NoError(t, upstream.lastReq.Context().Err()) } func TestForwardAsChatCompletions_UnknownResponsesSupportFallbackUsesVersionedChatURL(t *testing.T) { gin.SetMode(gin.TestMode) body := []byte(`{"model":"glm-4.5-air","messages":[{"role":"user","content":"hello"}],"stream":false}`) rec := httptest.NewRecorder() c, _ := gin.CreateTestContext(rec) c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", bytes.NewReader(body)) c.Request.Header.Set("Content-Type", "application/json") upstream := &httpUpstreamRecorder{responses: []*http.Response{ { StatusCode: http.StatusNotFound, Header: http.Header{"Content-Type": []string{"application/json"}}, Body: io.NopCloser(strings.NewReader(`{"error":{"message":"not found"}}`)), }, { StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"application/json"}, "x-request-id": []string{"rid_raw_fallback"}}, Body: io.NopCloser(strings.NewReader( `{"id":"chatcmpl_1","object":"chat.completion","model":"glm-4.5-air","choices":[{"index":0,"message":{"role":"assistant","content":"ok"},"finish_reason":"stop"}],"usage":{"prompt_tokens":1,"completion_tokens":2,"total_tokens":3}}`, )), }, }} svc := &OpenAIGatewayService{ cfg: rawChatCompletionsTestConfig(), httpUpstream: upstream, } account := rawChatCompletionsTestAccount() account.Credentials["base_url"] = "https://open.bigmodel.cn/api/paas/v4" result, err := svc.ForwardAsChatCompletions(context.Background(), c, account, body, "", "") require.NoError(t, err) require.NotNil(t, result) require.Equal(t, 1, result.Usage.InputTokens) require.Equal(t, 2, result.Usage.OutputTokens) require.Len(t, upstream.requests, 2) require.Equal(t, "https://open.bigmodel.cn/api/paas/v4/responses", upstream.requests[0].URL.String()) require.Equal(t, "https://open.bigmodel.cn/api/paas/v4/chat/completions", upstream.requests[1].URL.String()) require.Equal(t, http.StatusOK, rec.Code) require.Contains(t, rec.Body.String(), `"content":"ok"`) } func TestIsOpenAIChatUsageOnlyStreamChunk(t *testing.T) { t.Parallel() require.True(t, isOpenAIChatUsageOnlyStreamChunk(`{"choices":[],"usage":{"prompt_tokens":1,"completion_tokens":2}}`)) require.False(t, isOpenAIChatUsageOnlyStreamChunk(`{"choices":[{"index":0}],"usage":{"prompt_tokens":1,"completion_tokens":2}}`)) require.False(t, isOpenAIChatUsageOnlyStreamChunk(`{"choices":[]}`)) require.False(t, isOpenAIChatUsageOnlyStreamChunk(``)) } func TestEnsureOpenAIChatStreamUsage(t *testing.T) { t.Parallel() body, err := ensureOpenAIChatStreamUsage([]byte(`{"model":"gpt-5.4"}`)) require.NoError(t, err) require.True(t, gjson.GetBytes(body, "stream_options.include_usage").Bool()) body, err = ensureOpenAIChatStreamUsage([]byte(`{"model":"gpt-5.4","stream_options":{"include_usage":false}}`)) require.NoError(t, err) require.True(t, gjson.GetBytes(body, "stream_options.include_usage").Bool()) } func TestBufferRawChatCompletions_RejectsOversizedResponse(t *testing.T) { gin.SetMode(gin.TestMode) rec := httptest.NewRecorder() c, _ := gin.CreateTestContext(rec) c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", nil) resp := &http.Response{ StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"application/json"}}, Body: io.NopCloser(strings.NewReader("toolong")), } svc := &OpenAIGatewayService{cfg: rawChatCompletionsTestConfig()} svc.cfg.Gateway.UpstreamResponseReadMaxBytes = 3 result, err := svc.bufferRawChatCompletions(c, resp, rawChatCompletionsTestAccount(), "gpt-5.4", "gpt-5.4", "gpt-5.4", nil, nil, time.Now()) require.ErrorIs(t, err, ErrUpstreamResponseBodyTooLarge) require.Nil(t, result) require.Equal(t, http.StatusBadGateway, rec.Code) } func rawChatCompletionsTestConfig() *config.Config { return &config.Config{ Security: config.SecurityConfig{ URLAllowlist: config.URLAllowlistConfig{ Enabled: false, AllowInsecureHTTP: true, }, }, } } func rawChatCompletionsTestAccount() *Account { return &Account{ ID: 101, Name: "raw-openai-apikey", Platform: PlatformOpenAI, Type: AccountTypeAPIKey, Concurrency: 1, Credentials: map[string]any{ "api_key": "sk-test", "base_url": "http://upstream.example", }, } } func largeRawChatCompletionsBody() []byte { return []byte(`{"model":"gpt-5.5","messages":[{"role":"user","content":"` + strings.Repeat("x", openAISilentRefusalMinRequestBodyBytes) + `"}],"stream":true}`) }