package modeladapter import ( "context" "encoding/json" "fmt" "io" "net/http" "net/http/httptest" "strings" "testing" "time" ) func TestProvidersForwardAuthenticationOverridesAndFinishStream(t *testing.T) { for _, provider := range []string{"openai", "anthropic"} { t.Run(provider, func(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.Header.Get("X-Gateway-Key") != "fixture-gateway" || r.Host != "gateway.fixture" { t.Error("custom headers not applied") } if provider == "openai" && r.Header.Get("Authorization") != "Bearer fixture-key" { t.Error("OpenAI authentication missing") } if provider == "anthropic" && r.Header.Get("x-api-key") != "fixture-key" { t.Error("Anthropic authentication missing") } var body map[string]any if err := json.NewDecoder(r.Body).Decode(&body); err != nil { t.Error(err) } if body["model"] != "fixture-model" { t.Error("model mapping lost") } w.Header().Set("Content-Type", "text/event-stream") if provider == "openai" { fmt.Fprint(w, "data: {\"choices\":[{\"delta\":{\"content\":\"ok\"},\"finish_reason\":null}]}\n\ndata: {\"choices\":[{\"delta\":{},\"finish_reason\":\"stop\"}],\"usage\":{\"prompt_tokens\":1,\"completion_tokens\":1}}\n\ndata: [DONE]\n\n") } else { for _, event := range []string{`{"type":"message_start","message":{"id":"fixture","type":"message","role":"assistant","content":[],"model":"fixture-model","usage":{"input_tokens":1,"output_tokens":0}}}`, `{"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}`, `{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"ok"}}`, `{"type":"content_block_stop","index":0}`, `{"type":"message_delta","delta":{"stop_reason":"end_turn","stop_sequence":null},"usage":{"output_tokens":1}}`, `{"type":"message_stop"}`} { var envelope struct { Type string `json:"type"` } _ = json.Unmarshal([]byte(event), &envelope) fmt.Fprintf(w, "event: %s\ndata: %s\n\n", envelope.Type, event) } } })) defer server.Close() req := StreamRequest{BaseURL: server.URL, APIKey: "fixture-key", ProviderModelID: "fixture-model", ModelID: "fixture-model", OpenAIEndpoint: "/v1/chat/completions", Messages: []Message{{Role: "user", Content: "hello"}}, MaxTokens: 32, Stream: true, CustomHeadersEnabled: true, CustomHeadersJSON: `{"X-Gateway-Key":"fixture-gateway","Host":"gateway.fixture"}`} var text strings.Builder finished := false sink := func(event ModelEvent) error { if event.Kind == ModelEventKindTextDelta { text.WriteString(event.Text) } if event.Kind == ModelEventKindTurnFinished { finished = true } return nil } ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) defer cancel() var err error if provider == "openai" { adapter := NewOpenAIAdapter() adapter.client = server.Client() err = adapter.Stream(ctx, req, sink) } else { adapter := NewAnthropicAdapter() adapter.client = server.Client() err = adapter.Stream(ctx, req, sink) } if err != nil { t.Fatal(err) } if text.String() != "ok" || !finished { t.Fatalf("stream did not finish correctly: %q %t", text.String(), finished) } }) } } func TestProviderDeadlineCancelsPendingHTTPResponse(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { _, _ = io.Copy(io.Discard, r.Body); <-r.Context().Done() })) defer server.Close() ctx, cancel := context.WithTimeout(context.Background(), 25*time.Millisecond) defer cancel() _, err := DoProviderRequestWithRetry(ctx, server.Client(), "fixture", "", "", func(ctx context.Context) (*http.Request, error) { return http.NewRequestWithContext(ctx, http.MethodPost, server.URL, strings.NewReader("fixture")) }) if err == nil || ctx.Err() == nil { t.Fatal("provider ignored cancellation") } }