fix(openai): wire local observer service tier in WS ingress; bound handler tests

- openai_ws_forwarder_ingress: resolve billing tier from the local
  upstreamResponseModelObserver (upstream echo first) instead of the raw
  request payload, matching the HTTP->WS bridge and WS v2 forwarder.
- openai_ws_http_bridge_test: add fast-alias + upstream default case
  proving the local observer's echoed tier wins.
- handler tests: keep only invalid service_tier -> 400 (short-circuits at
  validation); valid/omitted semantics covered by the pure service-level
  validation tests, avoiding real account selection in tests.
This commit is contained in:
alfadb
2026-08-24 11:52:48 +08:00
parent f06bf181d2
commit c0c3e1cb47
3 changed files with 44 additions and 40 deletions
@@ -14,7 +14,12 @@ import (
)
// 非法 service_tier 必须在两个 OpenAI 端点(/v1/responses、/v1/chat/completions)
// 上以 OpenAI 兼容错误结构返回 HTTP 400;合法值(fast/priority)不被拒绝。
// 上以 OpenAI 兼容错误结构返回 HTTP 400。这些用例在 handler 的 service_tier
// 校验处短路,不会进入账号选择/重试。
//
// 合法值(fast/priority/flex/auto/default/scale)与省略/null 的接受语义由
// service 层纯校验函数 TestValidateOpenAIServiceTierField 覆盖,避免 handler
// 测试走入真实账号选择/重试路径。
func newServiceTierHandlerTest(t *testing.T) *OpenAIGatewayHandler {
t.Helper()
@@ -72,25 +77,6 @@ func TestOpenAIGatewayHandlerResponses_InvalidServiceTierRejected400(t *testing.
}
}
func TestOpenAIGatewayHandlerResponses_ValidServiceTierNotRejected(t *testing.T) {
for _, tier := range []string{"fast", "priority", "flex"} {
body := `{"model":"gpt-5.5","input":"hi","service_tier":"` + tier + `"}`
rec := runOpenAIHandlerServiceTierTest(t, "/v1/responses", body, func(h *OpenAIGatewayHandler, c *gin.Context) {
h.Responses(c)
})
require.NotEqual(t, http.StatusBadRequest, rec.Code, "tier=%s must not be rejected as invalid", tier)
require.NotContains(t, rec.Body.String(), "invalid service_tier", "tier=%s", tier)
}
}
func TestOpenAIGatewayHandlerResponses_ServiceTierOmittedKeepsCurrentBehavior(t *testing.T) {
body := `{"model":"gpt-5.5","input":"hi"}`
rec := runOpenAIHandlerServiceTierTest(t, "/v1/responses", body, func(h *OpenAIGatewayHandler, c *gin.Context) {
h.Responses(c)
})
require.NotContains(t, rec.Body.String(), "invalid service_tier")
}
func TestOpenAIGatewayHandlerChatCompletions_InvalidServiceTierRejected400(t *testing.T) {
for _, body := range []string{
`{"model":"gpt-5.5","messages":[{"role":"user","content":"hi"}],"service_tier":"turbo"}`,
@@ -106,22 +92,3 @@ func TestOpenAIGatewayHandlerChatCompletions_InvalidServiceTierRejected400(t *te
require.Contains(t, rec.Body.String(), "invalid service_tier", "body=%s", body)
}
}
func TestOpenAIGatewayHandlerChatCompletions_ValidServiceTierNotRejected(t *testing.T) {
for _, tier := range []string{"fast", "priority", "auto", "default", "scale", "flex"} {
body := `{"model":"gpt-5.5","messages":[{"role":"user","content":"hi"}],"service_tier":"` + tier + `"}`
rec := runOpenAIHandlerServiceTierTest(t, "/v1/chat/completions", body, func(h *OpenAIGatewayHandler, c *gin.Context) {
h.ChatCompletions(c)
})
require.NotEqual(t, http.StatusBadRequest, rec.Code, "tier=%q must not be rejected as invalid", tier)
require.NotContains(t, rec.Body.String(), "invalid service_tier", "tier=%q", tier)
}
}
func TestOpenAIGatewayHandlerChatCompletions_ServiceTierOmittedKeepsCurrentBehavior(t *testing.T) {
body := `{"model":"gpt-5.5","messages":[{"role":"user","content":"hi"}]}`
rec := runOpenAIHandlerServiceTierTest(t, "/v1/chat/completions", body, func(h *OpenAIGatewayHandler, c *gin.Context) {
h.ChatCompletions(c)
})
require.NotContains(t, rec.Body.String(), "invalid service_tier")
}
@@ -1164,7 +1164,7 @@ func (s *OpenAIGatewayService) ProxyResponsesWebSocketFromClient(
UpstreamResponseModel: responseModelObserver.Model(),
UpstreamResponseModelConflict: responseModelObserver.Conflict(),
UpstreamResponseServiceTier: responseModelObserver.ServiceTier(),
ServiceTier: extractOpenAIServiceTierFromBody(payload),
ServiceTier: resolvedOpenAIUpstreamServiceTierFromObserver(responseModelObserver, extractOpenAIServiceTierFromBody(payload)),
ReasoningEffort: ApplyThinkingEnabledFallback(extractOpenAIReasoningEffortFromBody(payload, mappedModel, originalModel), payload, mappedModel),
Stream: reqStream,
OpenAIWSMode: true,
@@ -83,6 +83,43 @@ func TestProxyOpenAIWSHTTPBridgeTurn_UpstreamDefaultServiceTierWinsOverRequest(t
require.Equal(t, "default", *result.ServiceTier)
}
func TestProxyOpenAIWSHTTPBridgeTurn_UpstreamDefaultWinsOverFastAlias(t *testing.T) {
gin.SetMode(gin.TestMode)
// 客户端别名 fast 同样被上游回显的 default 覆盖:局部 observer 的
// ServiceTier() 是唯一计费依据,绝不回退到请求侧 fast。
sse := strings.Join([]string{
`data: {"type":"response.completed","response":{"id":"resp_tier2","status":"completed","service_tier":"default","usage":{"input_tokens":1,"output_tokens":1}}}`,
``,
}, "\n")
upstream := &httpUpstreamRecorder{resp: &http.Response{
StatusCode: http.StatusOK,
Header: http.Header{"Content-Type": []string{"text/event-stream"}},
Body: io.NopCloser(strings.NewReader(sse)),
}}
svc := &OpenAIGatewayService{
cfg: &config.Config{Gateway: config.GatewayConfig{MaxLineSize: defaultMaxLineSize}},
httpUpstream: upstream,
}
account := &Account{ID: 5882, Platform: PlatformOpenAI, Type: AccountTypeAPIKey, Concurrency: 1}
payload := []byte(`{"type":"response.create","model":"gpt-5.5","stream":true,"service_tier":"fast","input":"hi"}`)
recorder := httptest.NewRecorder()
c, _ := gin.CreateTestContext(recorder)
c.Request = httptest.NewRequest(http.MethodGet, "/v1/responses", nil)
result, err := svc.proxyOpenAIWSHTTPBridgeTurn(
context.Background(), c, account, "test-token", payload, len(payload),
"gpt-5.5", "", "", "", "", 1,
func([]byte) error { return nil },
)
require.NoError(t, err)
require.NotNil(t, result)
require.NotNil(t, result.ServiceTier)
require.Equal(t, "default", *result.ServiceTier,
"local observer's upstream-echoed default must win over the fast alias")
}
func TestProxyOpenAIWSHTTPBridgeTurnAPIKeyAdaptsClientTools(t *testing.T) {
gin.SetMode(gin.TestMode)