From 988d5f4540cf89fd56e083d99c6f5086616fea32 Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Thu, 16 Jul 2026 06:49:01 +0200 Subject: [PATCH 1/2] fix: enable nullable streaming flags --- internal/requestflag/requestflag.go | 14 ++ internal/requestflag/requestflag_test.go | 48 ++++++ pkg/cmd/audiotranscription.go | 2 +- pkg/cmd/betaresponse.go | 2 +- pkg/cmd/betathread.go | 2 +- pkg/cmd/betathreadrun.go | 4 +- pkg/cmd/chatcompletion.go | 2 +- pkg/cmd/completion.go | 2 +- pkg/cmd/image.go | 4 +- pkg/cmd/response.go | 2 +- pkg/cmd/streaming_test.go | 188 +++++++++++++++++++++++ 11 files changed, 260 insertions(+), 10 deletions(-) create mode 100644 pkg/cmd/streaming_test.go diff --git a/internal/requestflag/requestflag.go b/internal/requestflag/requestflag.go index 77c4f1f..1dcaada 100644 --- a/internal/requestflag/requestflag.go +++ b/internal/requestflag/requestflag.go @@ -251,6 +251,20 @@ func ExtractRequestContents(cmd *cli.Command) RequestContents { return res } +// FlagBool returns the boolean value of a request flag. Nullable boolean flags +// expose *bool through cli.Command.Value, while cli.Command.Bool only handles +// concrete bool values. +func FlagBool(cmd *cli.Command, name string) bool { + switch value := cmd.Value(name).(type) { + case bool: + return value + case *bool: + return value != nil && *value + default: + return false + } +} + func GetMissingRequiredFlags(cmd *cli.Command, body any) []cli.Flag { missing := []cli.Flag{} for _, flag := range cmd.Flags { diff --git a/internal/requestflag/requestflag_test.go b/internal/requestflag/requestflag_test.go index 779bd57..d515a5f 100644 --- a/internal/requestflag/requestflag_test.go +++ b/internal/requestflag/requestflag_test.go @@ -617,6 +617,54 @@ func TestYamlHandling(t *testing.T) { }) } +func TestFlagBool(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + flag cli.Flag + want bool + }{ + { + name: "concrete true", + flag: &Flag[bool]{Name: "stream", Default: true}, + want: true, + }, + { + name: "concrete false", + flag: &Flag[bool]{Name: "stream", Default: false}, + want: false, + }, + { + name: "nullable true", + flag: &Flag[*bool]{Name: "stream", Default: Ptr(true)}, + want: true, + }, + { + name: "nullable false", + flag: &Flag[*bool]{Name: "stream", Default: Ptr(false)}, + want: false, + }, + { + name: "nullable null", + flag: &Flag[*bool]{Name: "stream", Default: nil}, + want: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + assert.Equal(t, tt.want, FlagBool(&cli.Command{Flags: []cli.Flag{tt.flag}}, "stream")) + }) + } + + t.Run("unknown flag", func(t *testing.T) { + t.Parallel() + assert.False(t, FlagBool(&cli.Command{}, "stream")) + }) +} + // TestNullLiteralHandling pins how each Flag[T] type handles the literal value "null" // when passed via the CLI. Pointer-typed flags serialize nil as JSON null, which is how // nullable body fields (`anyOf: [T, null]` / `{nullable: true}`) let users clear a field diff --git a/pkg/cmd/audiotranscription.go b/pkg/cmd/audiotranscription.go index 6801ccf..b812ec7 100644 --- a/pkg/cmd/audiotranscription.go +++ b/pkg/cmd/audiotranscription.go @@ -119,7 +119,7 @@ func handleAudioTranscriptionsCreate(ctx context.Context, cmd *cli.Command) erro format := cmd.Root().String("format") explicitFormat := cmd.Root().IsSet("format") transform := cmd.Root().String("transform") - if cmd.Bool("stream") { + if requestflag.FlagBool(cmd, "stream") { stream := client.Audio.Transcriptions.NewStreaming(ctx, params, options...) maxItems := int64(-1) if cmd.IsSet("max-items") { diff --git a/pkg/cmd/betaresponse.go b/pkg/cmd/betaresponse.go index 3fe2ff7..c54642b 100644 --- a/pkg/cmd/betaresponse.go +++ b/pkg/cmd/betaresponse.go @@ -487,7 +487,7 @@ func handleBetaResponsesCreate(ctx context.Context, cmd *cli.Command) error { format := cmd.Root().String("format") explicitFormat := cmd.Root().IsSet("format") transform := cmd.Root().String("transform") - if cmd.Bool("stream") { + if requestflag.FlagBool(cmd, "stream") { stream := client.Beta.Responses.NewStreaming(ctx, params, options...) maxItems := int64(-1) if cmd.IsSet("max-items") { diff --git a/pkg/cmd/betathread.go b/pkg/cmd/betathread.go index 716f768..79aa761 100644 --- a/pkg/cmd/betathread.go +++ b/pkg/cmd/betathread.go @@ -474,7 +474,7 @@ func handleBetaThreadsCreateAndRun(ctx context.Context, cmd *cli.Command) error format := cmd.Root().String("format") explicitFormat := cmd.Root().IsSet("format") transform := cmd.Root().String("transform") - if cmd.Bool("stream") { + if requestflag.FlagBool(cmd, "stream") { stream := client.Beta.Threads.NewAndRunStreaming(ctx, params, options...) maxItems := int64(-1) if cmd.IsSet("max-items") { diff --git a/pkg/cmd/betathreadrun.go b/pkg/cmd/betathreadrun.go index 46d59eb..00de4ee 100644 --- a/pkg/cmd/betathreadrun.go +++ b/pkg/cmd/betathreadrun.go @@ -348,7 +348,7 @@ func handleBetaThreadsRunsCreate(ctx context.Context, cmd *cli.Command) error { format := cmd.Root().String("format") explicitFormat := cmd.Root().IsSet("format") transform := cmd.Root().String("transform") - if cmd.Bool("stream") { + if requestflag.FlagBool(cmd, "stream") { stream := client.Beta.Threads.Runs.NewStreaming( ctx, cmd.Value("thread-id").(string), @@ -645,7 +645,7 @@ func handleBetaThreadsRunsSubmitToolOutputs(ctx context.Context, cmd *cli.Comman format := cmd.Root().String("format") explicitFormat := cmd.Root().IsSet("format") transform := cmd.Root().String("transform") - if cmd.Bool("stream") { + if requestflag.FlagBool(cmd, "stream") { stream := client.Beta.Threads.Runs.SubmitToolOutputsStreaming( ctx, cmd.Value("thread-id").(string), diff --git a/pkg/cmd/chatcompletion.go b/pkg/cmd/chatcompletion.go index fbdbb0e..6b95f6d 100644 --- a/pkg/cmd/chatcompletion.go +++ b/pkg/cmd/chatcompletion.go @@ -433,7 +433,7 @@ func handleChatCompletionsCreate(ctx context.Context, cmd *cli.Command) error { format := cmd.Root().String("format") explicitFormat := cmd.Root().IsSet("format") transform := cmd.Root().String("transform") - if cmd.Bool("stream") { + if requestflag.FlagBool(cmd, "stream") { stream := client.Chat.Completions.NewStreaming(ctx, params, options...) maxItems := int64(-1) if cmd.IsSet("max-items") { diff --git a/pkg/cmd/completion.go b/pkg/cmd/completion.go index 761ea03..a1763dd 100644 --- a/pkg/cmd/completion.go +++ b/pkg/cmd/completion.go @@ -168,7 +168,7 @@ func handleCompletionsCreate(ctx context.Context, cmd *cli.Command) error { format := cmd.Root().String("format") explicitFormat := cmd.Root().IsSet("format") transform := cmd.Root().String("transform") - if cmd.Bool("stream") { + if requestflag.FlagBool(cmd, "stream") { stream := client.Completions.NewStreaming(ctx, params, options...) maxItems := int64(-1) if cmd.IsSet("max-items") { diff --git a/pkg/cmd/image.go b/pkg/cmd/image.go index 8691736..0325029 100644 --- a/pkg/cmd/image.go +++ b/pkg/cmd/image.go @@ -319,7 +319,7 @@ func handleImagesEdit(ctx context.Context, cmd *cli.Command) error { format := cmd.Root().String("format") explicitFormat := cmd.Root().IsSet("format") transform := cmd.Root().String("transform") - if cmd.Bool("stream") { + if requestflag.FlagBool(cmd, "stream") { stream := client.Images.EditStreaming(ctx, params, options...) maxItems := int64(-1) if cmd.IsSet("max-items") { @@ -375,7 +375,7 @@ func handleImagesGenerate(ctx context.Context, cmd *cli.Command) error { format := cmd.Root().String("format") explicitFormat := cmd.Root().IsSet("format") transform := cmd.Root().String("transform") - if cmd.Bool("stream") { + if requestflag.FlagBool(cmd, "stream") { stream := client.Images.GenerateStreaming(ctx, params, options...) maxItems := int64(-1) if cmd.IsSet("max-items") { diff --git a/pkg/cmd/response.go b/pkg/cmd/response.go index 08acd37..0e749b1 100644 --- a/pkg/cmd/response.go +++ b/pkg/cmd/response.go @@ -450,7 +450,7 @@ func handleResponsesCreate(ctx context.Context, cmd *cli.Command) error { format := cmd.Root().String("format") explicitFormat := cmd.Root().IsSet("format") transform := cmd.Root().String("transform") - if cmd.Bool("stream") { + if requestflag.FlagBool(cmd, "stream") { stream := client.Responses.NewStreaming(ctx, params, options...) maxItems := int64(-1) if cmd.IsSet("max-items") { diff --git a/pkg/cmd/streaming_test.go b/pkg/cmd/streaming_test.go new file mode 100644 index 0000000..68b4d6e --- /dev/null +++ b/pkg/cmd/streaming_test.go @@ -0,0 +1,188 @@ +package cmd + +import ( + "bytes" + "context" + "fmt" + "io" + "net/http" + "net/http/httptest" + "os" + "os/exec" + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestNullableStreamFlags(t *testing.T) { + cliPath := buildTestCLI(t) + inputFile := filepath.Join(t.TempDir(), "input.txt") + require.NoError(t, os.WriteFile(inputFile, []byte("input"), 0o600)) + + tests := []struct { + name string + args []string + }{ + { + name: "completions create", + args: []string{"completions", "create", "--model", "model", "--prompt", "prompt"}, + }, + { + name: "responses create", + args: []string{"responses", "create", "--model", "model", "--input", "input"}, + }, + { + name: "beta responses create", + args: []string{"beta:responses", "create", "--model", "model", "--input", "input"}, + }, + { + name: "chat completions create", + args: []string{"chat:completions", "create", "--model", "model", "--message", "{content: input, role: user}"}, + }, + { + name: "audio transcriptions create", + args: []string{"audio:transcriptions", "create", "--file", inputFile, "--model", "model"}, + }, + { + name: "beta thread runs create", + args: []string{"beta:threads:runs", "create", "--thread-id", "thread", "--assistant-id", "assistant"}, + }, + { + name: "beta thread runs submit tool outputs", + args: []string{"beta:threads:runs", "submit-tool-outputs", "--thread-id", "thread", "--run-id", "run", "--tool-output", "{output: output, tool_call_id: call}"}, + }, + { + name: "beta threads create and run", + args: []string{"beta:threads", "create-and-run", "--assistant-id", "assistant"}, + }, + { + name: "images edit", + args: []string{"images", "edit", "--image", inputFile, "--prompt", "prompt"}, + }, + { + name: "images generate", + args: []string{"images", "generate", "--prompt", "prompt"}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + fmt.Fprint(w, "data: {}\n\ndata: [DONE]\n\n") + })) + defer server.Close() + + args := append(baseTestCLIArgs(server.URL), tt.args...) + args = append(args, "--stream", "true", "--max-items", "1") + output, err := runTestCLI(t, cliPath, args...) + require.NoError(t, err, "output: %s", output) + require.Contains(t, string(output), "{}") + }) + } + + t.Run("null selects the non-streaming handler", func(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + fmt.Fprint(w, `{"id":"completion"}`) + })) + defer server.Close() + + args := append(baseTestCLIArgs(server.URL), "completions", "create", "--model", "model", "--prompt", "prompt", "--stream", "null") + output, err := runTestCLI(t, cliPath, args...) + require.NoError(t, err, "output: %s", output) + require.Contains(t, string(output), `"id":"completion"`) + }) + + t.Run("writes SSE output before the response completes", func(t *testing.T) { + eventsSent := make(chan struct{}) + releaseResponse := make(chan struct{}) + released := false + defer func() { + if !released { + close(releaseResponse) + } + }() + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + flusher, ok := w.(http.Flusher) + if !ok { + http.Error(w, "streaming unsupported", http.StatusInternalServerError) + return + } + for i := range 40 { + fmt.Fprintf(w, "data: {\"id\":\"event-%d\"}\n\n", i) + flusher.Flush() + } + close(eventsSent) + <-releaseResponse + fmt.Fprint(w, "data: [DONE]\n\n") + flusher.Flush() + })) + defer server.Close() + + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + args := append(baseTestCLIArgs(server.URL), "completions", "create", "--model", "model", "--prompt", "prompt", "--stream", "true") + command := exec.CommandContext(ctx, cliPath, args...) + command.Env = append(os.Environ(), "PAGER=cat") + stdout, err := command.StdoutPipe() + require.NoError(t, err) + var stderr bytes.Buffer + command.Stderr = &stderr + require.NoError(t, command.Start()) + + select { + case <-eventsSent: + case <-ctx.Done(): + require.FailNow(t, "server did not send events") + } + + outputStarted := make(chan error, 1) + go func() { + buffer := make([]byte, 1) + _, readErr := io.ReadFull(stdout, buffer) + outputStarted <- readErr + }() + + select { + case readErr := <-outputStarted: + require.NoError(t, readErr) + case <-ctx.Done(): + require.FailNow(t, "command did not emit streamed output before the response completed") + } + + close(releaseResponse) + released = true + require.NoError(t, command.Wait(), "stderr: %s", stderr.String()) + }) +} + +func buildTestCLI(t *testing.T) string { + t.Helper() + path := filepath.Join(t.TempDir(), "openai") + command := exec.Command("go", "build", "-o", path, "../../cmd/openai") + output, err := command.CombinedOutput() + require.NoError(t, err, "output: %s", output) + return path +} + +func baseTestCLIArgs(baseURL string) []string { + return []string{ + "--base-url", baseURL, + "--format", "raw", + "--api-key", "key", + } +} + +func runTestCLI(t *testing.T, path string, args ...string) ([]byte, error) { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + command := exec.CommandContext(ctx, path, args...) + command.Env = append(os.Environ(), "PAGER=cat") + return command.CombinedOutput() +} From ee56d7d6598d3d1843c9db99c79363f58e2bb5bc Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Thu, 16 Jul 2026 15:16:07 +0200 Subject: [PATCH 2/2] test: tighten streaming response coverage --- pkg/cmd/streaming_test.go | 34 ++++++++++++++-------------------- 1 file changed, 14 insertions(+), 20 deletions(-) diff --git a/pkg/cmd/streaming_test.go b/pkg/cmd/streaming_test.go index 68b4d6e..941658b 100644 --- a/pkg/cmd/streaming_test.go +++ b/pkg/cmd/streaming_test.go @@ -1,10 +1,10 @@ package cmd import ( + "bufio" "bytes" "context" "fmt" - "io" "net/http" "net/http/httptest" "os" @@ -96,8 +96,7 @@ func TestNullableStreamFlags(t *testing.T) { require.Contains(t, string(output), `"id":"completion"`) }) - t.Run("writes SSE output before the response completes", func(t *testing.T) { - eventsSent := make(chan struct{}) + t.Run("writes an SSE event before the response completes", func(t *testing.T) { releaseResponse := make(chan struct{}) released := false defer func() { @@ -113,11 +112,8 @@ func TestNullableStreamFlags(t *testing.T) { http.Error(w, "streaming unsupported", http.StatusInternalServerError) return } - for i := range 40 { - fmt.Fprintf(w, "data: {\"id\":\"event-%d\"}\n\n", i) - flusher.Flush() - } - close(eventsSent) + fmt.Fprint(w, "data: {\"id\":\"event\"}\n\n") + flusher.Flush() <-releaseResponse fmt.Fprint(w, "data: [DONE]\n\n") flusher.Flush() @@ -126,7 +122,7 @@ func TestNullableStreamFlags(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) defer cancel() - args := append(baseTestCLIArgs(server.URL), "completions", "create", "--model", "model", "--prompt", "prompt", "--stream", "true") + args := append(baseTestCLIArgs(server.URL), "completions", "create", "--model", "model", "--prompt", "prompt", "--stream", "true", "--max-items", "1") command := exec.CommandContext(ctx, cliPath, args...) command.Env = append(os.Environ(), "PAGER=cat") stdout, err := command.StdoutPipe() @@ -135,22 +131,20 @@ func TestNullableStreamFlags(t *testing.T) { command.Stderr = &stderr require.NoError(t, command.Start()) - select { - case <-eventsSent: - case <-ctx.Done(): - require.FailNow(t, "server did not send events") + type outputResult struct { + line string + err error } - - outputStarted := make(chan error, 1) + output := make(chan outputResult, 1) go func() { - buffer := make([]byte, 1) - _, readErr := io.ReadFull(stdout, buffer) - outputStarted <- readErr + line, readErr := bufio.NewReader(stdout).ReadString('\n') + output <- outputResult{line: line, err: readErr} }() select { - case readErr := <-outputStarted: - require.NoError(t, readErr) + case result := <-output: + require.NoError(t, result.err) + require.JSONEq(t, `{"id":"event"}`, result.line) case <-ctx.Done(): require.FailNow(t, "command did not emit streamed output before the response completed") }