1package provider
2
3import (
4 "context"
5 "net/http"
6 "net/http/httptest"
7 "strings"
8 "sync/atomic"
9 "testing"
10)
11
12// TestOpenAIStreamRetriesOn503 verifies the OpenAI/Kimi stream client
13// silently retries when the upstream returns 503 a couple of times,
14// then succeeds. The user-visible error path should NOT fire because
15// a later attempt landed on a 200.
16func TestOpenAIStreamRetriesOn503(t *testing.T) {
17 var attempts atomic.Int32
18 srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
19 n := attempts.Add(1)
20 if n < 3 {
21 w.WriteHeader(http.StatusServiceUnavailable)
22 _, _ = w.Write([]byte("upstream connect error"))
23 return
24 }
25 w.Header().Set("content-type", "text/event-stream")
26 w.WriteHeader(http.StatusOK)
27 fl, _ := w.(http.Flusher)
28 write := func(s string) {
29 _, _ = w.Write([]byte(s))
30 if fl != nil {
31 fl.Flush()
32 }
33 }
34 write("data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"ok\"},\"finish_reason\":\"stop\"}]}\n\n")
35 write("data: [DONE]\n\n")
36 }))
37 defer srv.Close()
38
39 c := NewOpenAI("x", srv.URL)
40 evs, err := c.Stream(context.Background(), Request{Model: "gpt-5"})
41 if err != nil {
42 t.Fatalf("stream returned error: %v", err)
43 }
44 gotText := ""
45 for ev := range evs {
46 if td, ok := ev.(EventTextDelta); ok {
47 gotText += td.Delta
48 }
49 }
50 if gotText != "ok" {
51 t.Fatalf("text=%q", gotText)
52 }
53 if got := attempts.Load(); got != 3 {
54 t.Fatalf("attempts=%d want 3", got)
55 }
56}
57
58// TestOpenAIStreamSurfaces503AfterRetriesExhausted verifies that when
59// every retry attempt returns 503 the caller sees a normal "http 503"
60// error, not a retry-internal placeholder. The body bytes must be
61// preserved so the rescue picker / red banner can show them.
62func TestOpenAIStreamSurfaces503AfterRetriesExhausted(t *testing.T) {
63 var attempts atomic.Int32
64 srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
65 attempts.Add(1)
66 w.WriteHeader(http.StatusServiceUnavailable)
67 _, _ = w.Write([]byte("upstream connect error"))
68 }))
69 defer srv.Close()
70
71 c := NewOpenAI("x", srv.URL)
72 _, err := c.Stream(context.Background(), Request{Model: "gpt-5"})
73 if err == nil {
74 t.Fatalf("expected error after exhausted retries")
75 }
76 if !strings.Contains(err.Error(), "503") {
77 t.Fatalf("error %q should mention 503", err)
78 }
79 if got := attempts.Load(); got != streamRetryAttempts+1 {
80 t.Fatalf("attempts=%d want %d", got, streamRetryAttempts+1)
81 }
82}
83
84// TestOpenAIStreamDoesNotRetryOn400 ensures non-transient errors are
85// returned immediately without retrying. We don't want to amplify
86// load on a provider that's actively rejecting our request shape.
87func TestOpenAIStreamDoesNotRetryOn400(t *testing.T) {
88 var attempts atomic.Int32
89 srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
90 attempts.Add(1)
91 w.WriteHeader(http.StatusBadRequest)
92 _, _ = w.Write([]byte(`{"error":{"message":"nope"}}`))
93 }))
94 defer srv.Close()
95
96 c := NewOpenAI("x", srv.URL)
97 _, err := c.Stream(context.Background(), Request{Model: "gpt-5"})
98 if err == nil || !strings.Contains(err.Error(), "400") {
99 t.Fatalf("want 400 err got %v", err)
100 }
101 if got := attempts.Load(); got != 1 {
102 t.Fatalf("attempts=%d want 1", got)
103 }
104}
105
106func TestIsTransientHTTPStatus(t *testing.T) {
107 for _, code := range []int{502, 503, 504} {
108 if !isTransientHTTPStatus(code) {
109 t.Errorf("%d should be transient", code)
110 }
111 }
112 for _, code := range []int{200, 400, 401, 403, 404, 429, 500, 501} {
113 if isTransientHTTPStatus(code) {
114 t.Errorf("%d should NOT be transient", code)
115 }
116 }
117}
118
119func TestIsTransientConnectError(t *testing.T) {
120 good := []string{
121 "read tcp: connection reset by peer",
122 "dial tcp: connection refused",
123 "unexpected EOF",
124 "upstream connect error or disconnect/reset before headers",
125 "tls handshake error: bad cert",
126 "i/o timeout",
127 }
128 for _, m := range good {
129 if !isTransientConnectError(stringErr(m)) {
130 t.Errorf("%q should be transient", m)
131 }
132 }
133 bad := []string{
134 "context canceled",
135 "json: cannot unmarshal",
136 }
137 for _, m := range bad {
138 if isTransientConnectError(stringErr(m)) {
139 t.Errorf("%q should NOT be transient", m)
140 }
141 }
142}
143
144type stringErr string
145
146func (s stringErr) Error() string { return string(s) }