package responses import ( "bytes" "encoding/json" "net/http" "strings" "testing" "gitea.rutua.cn/kate522/CrossRelay/internal/chatcmpl" ) // captureWriter records everything the converter writes. type captureWriter struct { header http.Header body bytes.Buffer status int flushes int } func newCaptureWriter() *captureWriter { return &captureWriter{header: http.Header{}} } func (c *captureWriter) Header() http.Header { return c.header } func (c *captureWriter) Write(b []byte) (int, error) { return c.body.Write(b) } func (c *captureWriter) WriteHeader(status int) { c.status = status } func (c *captureWriter) Flush() { c.flushes++ } type sseEvent struct { name string data map[string]any } func parseSSE(t *testing.T, data string) []sseEvent { t.Helper() var events []sseEvent for _, block := range strings.Split(strings.TrimSpace(data), "\n\n") { if block == "" { continue } var name string var payload string for _, line := range strings.Split(block, "\n") { switch { case strings.HasPrefix(line, "event: "): name = strings.TrimPrefix(line, "event: ") case strings.HasPrefix(line, "data: "): payload = strings.TrimPrefix(line, "data: ") } } var parsed map[string]any if err := json.Unmarshal([]byte(payload), &parsed); err != nil { t.Fatalf("event %s payload is not JSON: %v (%s)", name, err, payload) } events = append(events, sseEvent{name: name, data: parsed}) } return events } func eventNames(events []sseEvent) []string { names := make([]string, len(events)) for i, e := range events { names[i] = e.name } return names } func textChunk(text string) *chatcmpl.StreamChunk { return &chatcmpl.StreamChunk{Choices: []chatcmpl.ChunkChoice{{Delta: chatcmpl.Delta{Content: &text}}}} } func finishChunk(reason string, usage *chatcmpl.Usage) *chatcmpl.StreamChunk { return &chatcmpl.StreamChunk{ Choices: []chatcmpl.ChunkChoice{{FinishReason: &reason}}, Usage: usage, } } func runConverter(t *testing.T, chunks []*chatcmpl.StreamChunk) []sseEvent { t.Helper() w := newCaptureWriter() conv := NewStreamConverter(w, "gpt-5-codex", 42) if err := conv.WriteStart(); err != nil { t.Fatalf("WriteStart: %v", err) } for _, c := range chunks { if err := conv.OnChunk(c); err != nil { t.Fatalf("OnChunk: %v", err) } } if err := conv.Finish(); err != nil { t.Fatalf("Finish: %v", err) } if w.status != http.StatusOK { t.Errorf("status = %d", w.status) } if ct := w.header.Get("Content-Type"); ct != "text/event-stream; charset=utf-8" { t.Errorf("content type = %q", ct) } return parseSSE(t, w.body.String()) } func TestStreamTextOnly(t *testing.T) { events := runConverter(t, []*chatcmpl.StreamChunk{ textChunk("Hello"), textChunk(", world"), finishChunk("stop", &chatcmpl.Usage{PromptTokens: 10, CompletionTokens: 3}), }) wantNames := []string{ "response.created", "response.in_progress", "response.output_item.added", "response.content_part.added", "response.output_text.delta", "response.output_text.delta", "response.output_text.done", "response.content_part.done", "response.output_item.done", "response.completed", } if got := eventNames(events); !equalStrings(got, wantNames) { t.Fatalf("events = %v, want %v", got, wantNames) } for i, e := range events { if seq, ok := e.data["sequence_number"].(float64); !ok || int(seq) != i+1 { t.Errorf("event %d (%s) sequence_number = %v, want %d", i, e.name, e.data["sequence_number"], i+1) } } created := events[0].data["response"].(map[string]any) if created["status"] != "in_progress" { t.Errorf("created status = %v", created["status"]) } if created["model"] != "gpt-5-codex" { t.Errorf("created model = %v", created["model"]) } added := events[2].data["item"].(map[string]any) if added["type"] != "message" || added["role"] != "assistant" { t.Errorf("added item = %v", added) } if events[4].data["delta"] != "Hello" || events[5].data["delta"] != ", world" { t.Errorf("deltas = %v, %v", events[4].data["delta"], events[5].data["delta"]) } completed := events[9].data["response"].(map[string]any) if completed["status"] != "completed" { t.Errorf("completed status = %v", completed["status"]) } if completed["output_text"] != "Hello, world" { t.Errorf("completed output_text = %v", completed["output_text"]) } usage := completed["usage"].(map[string]any) if usage["input_tokens"].(float64) != 10 || usage["output_tokens"].(float64) != 3 { t.Errorf("completed usage = %v", usage) } output := completed["output"].([]any) if len(output) != 1 { t.Fatalf("completed output = %v", output) } msg := output[0].(map[string]any) content := msg["content"].([]any) part := content[0].(map[string]any) if part["type"] != "output_text" || part["text"] != "Hello, world" { t.Errorf("completed message content = %v", part) } } func TestStreamToolCall(t *testing.T) { events := runConverter(t, []*chatcmpl.StreamChunk{ {Choices: []chatcmpl.ChunkChoice{{Delta: chatcmpl.Delta{ToolCalls: []chatcmpl.ToolCall{{ Index: 0, ID: "call_1", Type: "function", Function: chatcmpl.ToolCallFunction{Name: "read_file"}, }}}}}}, {Choices: []chatcmpl.ChunkChoice{{Delta: chatcmpl.Delta{ToolCalls: []chatcmpl.ToolCall{{ Index: 0, Function: chatcmpl.ToolCallFunction{Arguments: `{"path":`}, }}}}}}, {Choices: []chatcmpl.ChunkChoice{{Delta: chatcmpl.Delta{ToolCalls: []chatcmpl.ToolCall{{ Index: 0, Function: chatcmpl.ToolCallFunction{Arguments: `"a.txt"}`}, }}}}}}, finishChunk("tool_calls", &chatcmpl.Usage{PromptTokens: 7, CompletionTokens: 5}), }) wantNames := []string{ "response.created", "response.in_progress", "response.output_item.added", "response.function_call_arguments.delta", "response.function_call_arguments.delta", "response.function_call_arguments.done", "response.output_item.done", "response.completed", } if got := eventNames(events); !equalStrings(got, wantNames) { t.Fatalf("events = %v, want %v", got, wantNames) } added := events[2].data["item"].(map[string]any) if added["type"] != "function_call" || added["call_id"] != "call_1" || added["name"] != "read_file" { t.Errorf("added item = %v", added) } if events[3].data["delta"] != `{"path":` || events[4].data["delta"] != `"a.txt"}` { t.Errorf("argument deltas = %v, %v", events[3].data["delta"], events[4].data["delta"]) } if events[5].data["arguments"] != `{"path":"a.txt"}` { t.Errorf("arguments done = %v", events[5].data["arguments"]) } for _, e := range events[2:7] { if e.data["output_index"].(float64) != 0 { t.Errorf("%s output_index = %v, want 0", e.name, e.data["output_index"]) } } completed := events[7].data["response"].(map[string]any) output := completed["output"].([]any) call := output[0].(map[string]any) if call["call_id"] != "call_1" || call["arguments"] != `{"path":"a.txt"}` { t.Errorf("completed function_call = %v", call) } } func TestStreamParallelToolCalls(t *testing.T) { events := runConverter(t, []*chatcmpl.StreamChunk{ {Choices: []chatcmpl.ChunkChoice{{Delta: chatcmpl.Delta{ToolCalls: []chatcmpl.ToolCall{{ Index: 0, ID: "call_a", Type: "function", Function: chatcmpl.ToolCallFunction{Name: "tool_a"}, }}}}}}, {Choices: []chatcmpl.ChunkChoice{{Delta: chatcmpl.Delta{ToolCalls: []chatcmpl.ToolCall{{ Index: 1, ID: "call_b", Type: "function", Function: chatcmpl.ToolCallFunction{Name: "tool_b", Arguments: `{}`}, }}}}}}, {Choices: []chatcmpl.ChunkChoice{{Delta: chatcmpl.Delta{ToolCalls: []chatcmpl.ToolCall{{ Index: 0, Function: chatcmpl.ToolCallFunction{Arguments: `{"x":1}`}, }}}}}}, finishChunk("tool_calls", nil), }) names := map[float64]string{} args := map[float64]string{} for _, e := range events { if e.name == "response.output_item.added" { idx := e.data["output_index"].(float64) names[idx] = e.data["item"].(map[string]any)["name"].(string) } if e.name == "response.function_call_arguments.delta" { idx := e.data["output_index"].(float64) args[idx] += e.data["delta"].(string) } } if len(names) != 2 { t.Fatalf("output_item.added count = %d, want 2 (%v)", len(names), names) } if names[0] != "tool_a" || names[1] != "tool_b" { t.Errorf("tool names by output index = %v", names) } if args[0] != `{"x":1}` || args[1] != `{}` { t.Errorf("interleaved args = %v", args) } } func TestStreamSynthesizesCallID(t *testing.T) { events := runConverter(t, []*chatcmpl.StreamChunk{ {Choices: []chatcmpl.ChunkChoice{{Delta: chatcmpl.Delta{ToolCalls: []chatcmpl.ToolCall{{ Index: 0, Type: "function", Function: chatcmpl.ToolCallFunction{Name: "shell"}, }}}}}}, finishChunk("tool_calls", nil), }) added := events[2].data["item"].(map[string]any) callID, _ := added["call_id"].(string) if !strings.HasPrefix(callID, "call_") { t.Errorf("synthesized call_id = %q, want a call_ prefix", callID) } } func TestStreamLengthFinishIsIncomplete(t *testing.T) { events := runConverter(t, []*chatcmpl.StreamChunk{ textChunk("cut"), finishChunk("length", nil), }) completed := events[len(events)-1] if completed.name != "response.completed" { t.Fatalf("last event = %s", completed.name) } resp := completed.data["response"].(map[string]any) if resp["status"] != "incomplete" { t.Errorf("status = %v, want incomplete", resp["status"]) } details, ok := resp["incomplete_details"].(map[string]any) if !ok || details["reason"] != "max_output_tokens" { t.Errorf("incomplete_details = %v", resp["incomplete_details"]) } } func TestStreamEstimatedUsageWithoutUpstreamUsage(t *testing.T) { events := runConverter(t, []*chatcmpl.StreamChunk{ textChunk("some text"), finishChunk("stop", nil), }) completed := events[len(events)-1].data["response"].(map[string]any) usage := completed["usage"].(map[string]any) if usage["input_tokens"].(float64) != 42 { t.Errorf("input tokens = %v, want the 42 estimate", usage["input_tokens"]) } if usage["output_tokens"].(float64) <= 0 { t.Errorf("output tokens = %v, want a positive estimate", usage["output_tokens"]) } } func TestStreamEmptyUpstream(t *testing.T) { events := runConverter(t, nil) want := []string{"response.created", "response.in_progress", "response.completed"} if got := eventNames(events); !equalStrings(got, want) { t.Fatalf("events = %v, want %v", got, want) } } func TestStreamWriteError(t *testing.T) { w := newCaptureWriter() conv := NewStreamConverter(w, "gpt-5-codex", 1) if err := conv.WriteStart(); err != nil { t.Fatalf("WriteStart: %v", err) } conv.WriteError("api_error", "upstream exploded") events := parseSSE(t, w.body.String()) last := events[len(events)-1] if last.name != "response.failed" { t.Fatalf("last event = %s, want response.failed", last.name) } resp := last.data["response"].(map[string]any) if resp["status"] != "failed" { t.Errorf("status = %v", resp["status"]) } errObj := resp["error"].(map[string]any) if errObj["code"] != "api_error" || errObj["message"] != "upstream exploded" { t.Errorf("error = %v", errObj) } } func equalStrings(a, b []string) bool { if len(a) != len(b) { return false } for i := range a { if a[i] != b[i] { return false } } return true }