seewe/sub2api
1
0
Fork 0
mirror of https://github.com/Wei-Shaw/sub2api.git synced 2026-10-15 23:22:59 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' into fix/issue-5029-openai-passthrough-pool-auth-retry

Browse Source 
Resolve conflict in backend/internal/handler/openai_gateway_handler_test.go.

main and this branch each appended a passthrough upstream stub plus a test at
the same two insertion points:

  main   openAIHTTPPassthroughSSERateLimitUpstream
         TestOpenAIResponses_APIKeyPassthroughSSERateLimitUsesConfiguredPoolRetry
  branch openAIHTTPPassthroughAuthFailoverUpstream
         TestOpenAIResponses_APIKeyPassthroughPoolAuthFailureRetriesThenSwitchesToHealthyAccount

Both sides are kept verbatim; the only edit is giving each stub its own
calls() body instead of sharing the trailing one. No assertion was changed.

openai_gateway_passthrough.go and openai_oauth_passthrough_test.go merged
automatically.
This commit is contained in:
shaw
2026-08-11 14:09:07 +08:00
parent 7045f89de4 574dfad2dc
commit a3bbf35cbd
813 changed files with 72461 additions and 5157 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/internal/handler/openai_gateway_handler_test.go
+144 -3
View File
@@ -17,6 +17,7 @@ import (
"github.com/Wei-Shaw/sub2api/internal/config"
pkghttputil "github.com/Wei-Shaw/sub2api/internal/pkg/httputil"
"github.com/Wei-Shaw/sub2api/internal/pkg/pagination"
"github.com/Wei-Shaw/sub2api/internal/pkg/xai"
"github.com/Wei-Shaw/sub2api/internal/server/middleware"
"github.com/Wei-Shaw/sub2api/internal/service"
coderws "github.com/coder/websocket"
@@ -334,6 +335,26 @@ func TestOpenAIEnsureForwardErrorResponse_AfterDeltaAppendsSingleValidResponseFa
require.Equal(t, 1, errorEvents)
}
func TestOpenAIEnsureForwardErrorResponse_CompactKeepaliveOnlyWritesResponseFailed(t *testing.T) {
gin.SetMode(gin.TestMode)
w := httptest.NewRecorder()
c, _ := gin.CreateTestContext(w)
c.Request = httptest.NewRequest(http.MethodPost, EndpointResponses, nil)
service.MarkOpenAICompactClientStream(c)
stop := service.StartOpenAICompactSSEKeepalive(c, 5*time.Millisecond)
defer stop()
before := service.OpenAICompactKeepaliveAdjustedWrittenSize(c)
require.Eventually(t, c.Writer.Written, time.Second, time.Millisecond)
require.Equal(t, before, service.OpenAICompactKeepaliveAdjustedWrittenSize(c))
h := &OpenAIGatewayHandler{}
require.True(t, h.ensureForwardErrorResponse(c, false))
require.Equal(t, http.StatusOK, w.Code)
require.Contains(t, w.Body.String(), "event: response.failed\n")
require.NotContains(t, w.Body.String(), "event: error\n")
}
func TestOpenAIEnsureForwardErrorResponse_ImageJSONKeepaliveWritesSingleJSONFallback(t *testing.T) {
gin.SetMode(gin.TestMode)
w := httptest.NewRecorder()
@@ -643,6 +664,9 @@ func TestResolveOpenAIMessagesDispatchMappedModel(t *testing.T) {
})
t.Run("grok_group_maps_claude_cli_model_to_grok_default", func(t *testing.T) {
original := xai.RuntimeModelMappingOptions()
t.Cleanup(func() { xai.SetRuntimeModelMappingOptions(original) })
xai.SetRuntimeModelMappingOptions(xai.ModelMappingOptions{EnableCrossClientMap: true})
apiKey := &service.APIKey{
Group: &service.Group{
Platform: service.PlatformGrok,
@@ -1266,6 +1290,7 @@ func TestOpenAIResponsesWebSocket_ContentModerationBlocksFirstFrame(t *testing.T
nil,
nil,
nil,
nil,
)
decision, err := moderationSvc.Check(context.Background(), service.ContentModerationCheckInput{
UserID: 1,
@@ -1407,7 +1432,7 @@ func TestOpenAIResponsesWebSocket_PassthroughTracksModelPerTurn(t *testing.T) {
require.Equal(t, "gpt-5.6-terra", *got.logs[1].UpstreamModel)
require.NotNil(t, got.logs[1].ModelMappingChain)
require.Equal(t, "terra→terra-channel→gpt-5.6-terra", *got.logs[1].ModelMappingChain)
require.InDelta(t, got.logs[1].TotalCost*2, got.logs[0].TotalCost, 1e-12,
require.InDelta(t, got.logs[1].TotalCost*2.5, got.logs[0].TotalCost, 1e-12,
"each turn must be billed with its own channel-mapped model")
}
@@ -1477,7 +1502,7 @@ func TestOpenAIResponsesWebSocket_PassthroughKeepsTurnMappingSnapshot(t *testing
require.Equal(t, "gpt-5.6-terra", *got.logs[1].UpstreamModel)
require.NotNil(t, got.logs[1].ModelMappingChain)
require.Equal(t, "sol→gpt-5.6-terra", *got.logs[1].ModelMappingChain)
require.InDelta(t, got.logs[1].TotalCost*2, got.logs[0].TotalCost, 1e-12,
require.InDelta(t, got.logs[1].TotalCost*2.5, got.logs[0].TotalCost, 1e-12,
"the next turn must use the updated channel mapping")
}
@@ -1506,7 +1531,7 @@ func TestOpenAIResponsesWebSocket_CtxPoolAppliesPerTurnMappingAndPreservesReques
require.Equal(t, "gpt-5.6-terra", got.logs[1].RequestedModel)
require.NotNil(t, got.logs[1].ModelMappingChain)
require.Equal(t, "gpt-5.6-terra→gpt-5.6-sol", *got.logs[1].ModelMappingChain)
require.InDelta(t, 20e-6, got.logs[1].TotalCost, 1e-12,
require.InDelta(t, 16e-6, got.logs[1].TotalCost, 1e-12,
"BillingModelSourceRequested must use the client model before channel mapping")
}
@@ -1842,6 +1867,40 @@ func (u *openAIHTTPPassthroughAuthFailoverUpstream) calls() []int64 {
return append([]int64(nil), u.accountIDs...)
}
type openAIHTTPPassthroughSSERateLimitUpstream struct {
service.HTTPUpstream
mu sync.Mutex
accountIDs []int64
}
func (u *openAIHTTPPassthroughSSERateLimitUpstream) Do(_ *http.Request, _ string, accountID int64, _ int) (*http.Response, error) {
u.mu.Lock()
u.accountIDs = append(u.accountIDs, accountID)
u.mu.Unlock()
body := strings.Join([]string{
"event: response.created",
`data: {"type":"response.created","response":{"id":"resp_rate_limited"}}`,
"",
"event: response.failed",
`data: {"type":"response.failed","response":{"id":"resp_rate_limited","status":"failed","error":{"code":"rate_limit_exceeded","message":"Concurrency limit exceeded for account, please retry later"}}}`,
"",
}, "\n")
return &http.Response{
StatusCode: http.StatusOK,
Header: http.Header{
"Content-Type": []string{"text/event-stream"},
"Retry-After": []string{"1"},
},
Body: io.NopCloser(strings.NewReader(body)),
}, nil
}
func (u *openAIHTTPPassthroughSSERateLimitUpstream) calls() []int64 {
u.mu.Lock()
defer u.mu.Unlock()
return append([]int64(nil), u.accountIDs...)
}
func (s *openAIWSFailoverHandlerAccountRepoStub) ListSchedulableByPlatform(ctx context.Context, platform string) ([]service.Account, error) {
out := make([]service.Account, 0, len(s.accounts))
for _, account := range s.accounts {
@@ -2144,6 +2203,88 @@ func TestOpenAIResponses_APIKeyPassthroughPoolAuthFailureRetriesThenSwitchesToHe
}
}
func TestOpenAIResponses_APIKeyPassthroughSSERateLimitUsesConfiguredPoolRetry(t *testing.T) {
gin.SetMode(gin.TestMode)
groupID := int64(4204)
accounts := []service.Account{
{
ID: 9912, Name: "pool-sse-rate-limit", Platform: service.PlatformOpenAI,
Type: service.AccountTypeAPIKey, Status: service.StatusActive, Schedulable: true, Priority: 1,
Credentials: map[string]any{
"api_key": "sk-pool",
"base_url": "https://api.example.test",
"pool_mode": true,
"pool_mode_retry_count": float64(1),
"pool_mode_retry_status_codes": []any{float64(http.StatusTooManyRequests)},
},
Extra: map[string]any{"openai_passthrough": true},
},
}
cfg := &config.Config{RunMode: config.RunModeSimple}
cfg.Default.RateMultiplier = 1
cfg.Security.URLAllowlist.Enabled = false
cfg.Gateway.MaxAccountSwitches = 1
accountRepo := &openAIWSFailoverHandlerAccountRepoStub{accounts: accounts}
upstream := &openAIHTTPPassthroughSSERateLimitUpstream{}
billingCacheSvc := service.NewBillingCacheService(nil, nil, nil, nil, nil, nil, cfg, nil)
t.Cleanup(billingCacheSvc.Stop)
gatewaySvc := service.NewOpenAIGatewayService(
accountRepo,
nil,
nil,
nil,
nil,
nil,
nil,
cfg,
nil,
nil,
service.NewBillingService(cfg, nil),
nil,
billingCacheSvc,
upstream,
&service.DeferredService{},
nil,
nil,
nil,
nil,
nil,
nil,
nil,
)
h := NewOpenAIGatewayHandler(
gatewaySvc,
service.NewConcurrencyService(nil),
billingCacheSvc,
service.NewAPIKeyService(nil, nil, nil, nil, nil, nil, cfg),
nil,
nil,
nil,
nil,
cfg,
)
rec := httptest.NewRecorder()
c, _ := gin.CreateTestContext(rec)
c.Request = httptest.NewRequest(http.MethodPost, "/openai/v1/responses", strings.NewReader(`{"model":"gpt-5.6-sol","input":"hello","stream":true}`))
c.Request.Header.Set("Content-Type", "application/json")
c.Set(string(middleware.ContextKeyAPIKey), &service.APIKey{
ID: 1804, GroupID: &groupID,
User: &service.User{ID: 1704, Status: service.StatusActive},
Group: &service.Group{ID: groupID, Platform: service.PlatformOpenAI, Status: service.StatusActive},
})
c.Set(string(middleware.ContextKeyUser), middleware.AuthSubject{UserID: 1704, Concurrency: 0})
h.Responses(c)
require.Equal(t, []int64{9912, 9912}, upstream.calls())
require.Equal(t, http.StatusTooManyRequests, rec.Code)
require.Equal(t, "1", rec.Header().Get("Retry-After"))
require.Equal(t, "rate_limit_error", gjson.GetBytes(rec.Body.Bytes(), "error.type").String())
require.Equal(t, "Upstream rate limit exceeded, please retry later", gjson.GetBytes(rec.Body.Bytes(), "error.message").String())
}
func TestOpenAIResponsesWebSocket_FailoverOnUpstreamUsageLimitEvent(t *testing.T) {
gin.SetMode(gin.TestMode)