From 3f4cc359ad0d8ae564dbc1ba8b6a980b325682e7 Mon Sep 17 00:00:00 2001 From: apstndb <803393+apstndb@users.noreply.github.com> Date: Tue, 8 Sep 2026 11:21:59 +0900 Subject: [PATCH] Refactor transport tests to inspect outgoing RPCs --- database_role_test.go | 54 +++++----- internal/grpctest/intercept.go | 58 +++++++++++ internal/grpctest/intercept_test.go | 68 +++++++++++++ jqresult/protojson_test.go | 43 ++++++++ main.go | 11 +- pdml_query_mode_test.go | 5 +- priority_transport_test.go | 67 +++++-------- query_stats_modes_test.go | 149 +++++++++++++++------------- 8 files changed, 306 insertions(+), 149 deletions(-) create mode 100644 internal/grpctest/intercept.go create mode 100644 internal/grpctest/intercept_test.go diff --git a/database_role_test.go b/database_role_test.go index b57e2b8..b6bf16f 100644 --- a/database_role_test.go +++ b/database_role_test.go @@ -2,29 +2,17 @@ package main import ( "context" - "net" + "testing" "time" sppb "cloud.google.com/go/spanner/apiv1/spannerpb" "github.com/alecthomas/kong" - "google.golang.org/grpc" + "github.com/apstndb/execspansql/internal/grpctest" + "github.com/apstndb/spanemuboost" + "google.golang.org/api/option" ) -type databaseRoleSpannerServer struct { - sppb.UnimplementedSpannerServer - createSession chan *sppb.CreateSessionRequest -} - -func (s *databaseRoleSpannerServer) CreateSession(_ context.Context, req *sppb.CreateSessionRequest) (*sppb.Session, error) { - s.createSession <- req - return &sppb.Session{ - Name: req.Database + "/sessions/test", - CreatorRole: req.GetSession().GetCreatorRole(), - Multiplexed: true, - }, nil -} - func TestDatabaseRoleFlag(t *testing.T) { t.Parallel() @@ -46,6 +34,13 @@ func TestDatabaseRoleFlag(t *testing.T) { } func TestNewClientSendsDatabaseRoleWhenCreatingSession(t *testing.T) { + env, err := spanemuboost.RunEmulatorWithClients(context.Background()) + if err != nil { + t.Fatal(err) + } + defer env.Close() //nolint:errcheck + t.Setenv("SPANNER_EMULATOR_HOST", env.Emulator().URI()) + for _, tt := range []struct { name string role string @@ -54,31 +49,28 @@ func TestNewClientSendsDatabaseRoleWhenCreatingSession(t *testing.T) { {name: "omitted"}, } { t.Run(tt.name, func(t *testing.T) { - server := &databaseRoleSpannerServer{createSession: make(chan *sppb.CreateSessionRequest, 1)} - grpcServer := grpc.NewServer() - sppb.RegisterSpannerServer(grpcServer, server) - listener, err := net.Listen("tcp", "127.0.0.1:0") - if err != nil { - t.Fatal(err) - } - t.Cleanup(func() { - grpcServer.Stop() - _ = listener.Close() + roles := make(chan string, 1) + dialOptions := grpctest.Inspect(func(_ string, req any) error { + if req, ok := req.(*sppb.CreateSessionRequest); ok { + select { + case roles <- req.GetSession().GetCreatorRole(): + default: + } + } + return nil }) - go func() { _ = grpcServer.Serve(listener) }() - t.Setenv("SPANNER_EMULATOR_HOST", listener.Addr().String()) ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() - client, err := newClient(ctx, "p", "i", "d", tt.role, logGrpcModeOff, false) + client, err := newClient(ctx, env.ProjectID, env.InstanceID, env.DatabaseID, tt.role, logGrpcModeOff, false, option.WithGRPCDialOption(dialOptions[0]), option.WithGRPCDialOption(dialOptions[1])) if err != nil { t.Fatal(err) } defer client.Close() select { - case req := <-server.createSession: - if got := req.GetSession().GetCreatorRole(); got != tt.role { + case got := <-roles: + if got != tt.role { t.Fatalf("session CreatorRole = %q, want %q", got, tt.role) } case <-ctx.Done(): diff --git a/internal/grpctest/intercept.go b/internal/grpctest/intercept.go new file mode 100644 index 0000000..affe91b --- /dev/null +++ b/internal/grpctest/intercept.go @@ -0,0 +1,58 @@ +// Package grpctest provides client-side RPC inspection for transport tests. +// It complements emulator tests without implementing Spanner server behavior. +package grpctest + +import ( + "context" + + "google.golang.org/grpc" +) + +// Inspect calls inspect before each unary request or stream message is sent. +// Returning a non-nil gRPC status error prevents that message from being sent. +// The callback may run concurrently; it must not mutate or retain the message +// without copying it. Returning nil preserves the actual server response. +func Inspect(inspect func(method string, request any) error) []grpc.DialOption { + return []grpc.DialOption{ + grpc.WithChainUnaryInterceptor(func(ctx context.Context, method string, req, reply any, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error { + if err := inspect(method, req); err != nil { + return err + } + return invoker(ctx, method, req, reply, cc, opts...) + }), + grpc.WithChainStreamInterceptor(func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) { + ctx, cancel := context.WithCancel(ctx) + stream, err := streamer(ctx, desc, cc, method, opts...) + if err != nil { + cancel() + return nil, err + } + return &inspectingStream{ClientStream: stream, inspect: inspect, method: method, cancel: cancel}, nil + }), + } +} + +type inspectingStream struct { + grpc.ClientStream + inspect func(string, any) error + method string + cancel context.CancelFunc +} + +func (s *inspectingStream) SendMsg(m any) error { + if err := s.inspect(s.method, m); err != nil { + s.cancel() + return err + } + // Preserve transport errors (including io.EOF); RecvMsg may still need + // to retrieve the server status. + return s.ClientStream.SendMsg(m) +} + +func (s *inspectingStream) RecvMsg(m any) error { + err := s.ClientStream.RecvMsg(m) + if err != nil { + s.cancel() + } + return err +} diff --git a/internal/grpctest/intercept_test.go b/internal/grpctest/intercept_test.go new file mode 100644 index 0000000..d5f5d93 --- /dev/null +++ b/internal/grpctest/intercept_test.go @@ -0,0 +1,68 @@ +package grpctest + +import ( + "context" + "io" + "testing" + + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +type transportStream struct { + grpc.ClientStream + sent any + sendErr error + recvErr error +} + +func (s *transportStream) SendMsg(m any) error { s.sent = m; return s.sendErr } +func (s *transportStream) RecvMsg(any) error { return s.recvErr } + +func TestInspectingStreamRejectsBeforeSending(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + transport := &transportStream{} + rejected := status.Error(codes.InvalidArgument, "captured") + stream := &inspectingStream{ClientStream: transport, method: "/test/Query", cancel: cancel, + inspect: func(method string, req any) error { + if method != "/test/Query" || req != "request" { + t.Fatalf("inspection = %s %v", method, req) + } + return rejected + }} + if err := stream.SendMsg("request"); err != rejected { + t.Fatalf("error = %v", err) + } + if transport.sent != nil { + t.Fatal("rejected request reached transport") + } + if ctx.Err() == nil { + t.Fatal("rejected stream was not cancelled") + } +} + +func TestInspectingStreamPreservesServerStatusAfterSendEOF(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + serverErr := status.Error(codes.PermissionDenied, "denied") + transport := &transportStream{sendErr: io.EOF, recvErr: serverErr} + stream := &inspectingStream{ClientStream: transport, cancel: cancel, + inspect: func(string, any) error { return nil }} + if err := stream.SendMsg("request"); err != io.EOF { + t.Fatalf("send error = %v", err) + } + if transport.sent != "request" { + t.Fatal("request was not forwarded") + } + if ctx.Err() != nil { + t.Fatal("cancelled before receiving server status") + } + if err := stream.RecvMsg(nil); err != serverErr { + t.Fatalf("receive error = %v", err) + } + if ctx.Err() == nil { + t.Fatal("finished stream was not cancelled") + } +} diff --git a/jqresult/protojson_test.go b/jqresult/protojson_test.go index 9e46947..e9deae0 100644 --- a/jqresult/protojson_test.go +++ b/jqresult/protojson_test.go @@ -1,6 +1,9 @@ package jqresult import ( + sppb "cloud.google.com/go/spanner/apiv1/spannerpb" + "github.com/apstndb/execspansql/resultset" + "reflect" "testing" "github.com/apstndb/spaniter" @@ -52,3 +55,43 @@ func TestResultSetMapFromRowIteratorNil(t *testing.T) { t.Fatal("error = nil, want nil row iterator error") } } + +// Both eager result materialization and lazy stats conversion must preserve +// arbitrary plan/stat contents, independently of RPC query mode support. +func TestPlanAndStatsAcrossResultPaths(t *testing.T) { + t.Parallel() + for _, withPlan := range []bool{false, true} { + for _, withStats := range []bool{false, true} { + stats := spaniter.Stats{} + want := map[string]any{} + if withPlan { + stats.QueryPlan = &sppb.QueryPlan{PlanNodes: []*sppb.PlanNode{{DisplayName: "Test Scan"}}} + want["queryPlan"] = map[string]any{"planNodes": []any{map[string]any{"displayName": "Test Scan"}}} + } + if withStats { + stats.QueryStats = map[string]any{"elapsed_time": "1 msecs", "nested": map[string]any{"complete": true}} + want["queryStats"] = stats.QueryStats + } + input := spaniter.RowIteratorResult{Stats: stats} + lazy, err := StatsMapFromResult(input) + if err != nil { + t.Fatal(err) + } + rs, err := resultset.FromIteratorResult(nil, input) + if err != nil { + t.Fatal(err) + } + eager, err := ResultSetMap(rs) + if err != nil { + t.Fatal(err) + } + if len(want) == 0 { + if lazy != nil || eager["stats"] != nil { + t.Fatalf("empty stats: lazy=%v eager=%v", lazy, eager) + } + } else if !reflect.DeepEqual(lazy, want) || !reflect.DeepEqual(eager["stats"], want) { + t.Fatalf("plan=%v stats=%v: lazy=%v eager=%v want=%v", withPlan, withStats, lazy, eager["stats"], want) + } + } + } +} diff --git a/main.go b/main.go index d3f1c1e..5e250dc 100644 --- a/main.go +++ b/main.go @@ -402,6 +402,11 @@ func runInNewTransaction(ctx context.Context, client *spanner.Client, stmt spann } func _main() error { + return runCLI() +} + +// runCLI accepts client options so transport tests can inspect outgoing RPCs. +func runCLI(clientOptions ...option.ClientOption) error { o, err := processFlags() if err != nil { os.Exit(1) @@ -469,7 +474,7 @@ func _main() error { }() } - client, err := newClient(ctx, o.Project, o.Instance, o.Database, o.DatabaseRole, o.LogGrpc, tracingEnabled(o)) + client, err := newClient(ctx, o.Project, o.Instance, o.Database, o.DatabaseRole, o.LogGrpc, tracingEnabled(o), clientOptions...) if err != nil { return err } @@ -598,7 +603,7 @@ func writeCsvFromResultSet(writer io.Writer, rs *sppb.ResultSet) error { return csvWriter.Flush() } -func newClient(ctx context.Context, project, instance, database, databaseRole string, logGrpcMode string, doTrace bool) (*spanner.Client, error) { +func newClient(ctx context.Context, project, instance, database, databaseRole string, logGrpcMode string, doTrace bool, clientOptions ...option.ClientOption) (*spanner.Client, error) { name, err := databaseResourceName(project, instance, database) if err != nil { return nil, err @@ -612,7 +617,7 @@ func newClient(ctx context.Context, project, instance, database, databaseRole st if doTrace { copts = append(copts, option.WithGRPCDialOption(grpc.WithChainStreamInterceptor(interceptor.StreamInterceptor(interceptor.WithDefaultDecorators())))) } - return spanner.NewClientWithConfig(ctx, name, spanner.ClientConfig{DatabaseRole: databaseRole}, copts...) + return spanner.NewClientWithConfig(ctx, name, spanner.ClientConfig{DatabaseRole: databaseRole}, append(copts, clientOptions...)...) } type encoder interface { diff --git a/pdml_query_mode_test.go b/pdml_query_mode_test.go index dc4f1e1..e845579 100644 --- a/pdml_query_mode_test.go +++ b/pdml_query_mode_test.go @@ -10,6 +10,7 @@ import ( "cloud.google.com/go/spanner" "github.com/apstndb/spanemuboost" + "google.golang.org/api/option" ) const pdmlQueryModeAuditSQL = "UPDATE PdmlQueryModeAudit SET V=99 WHERE TRUE" @@ -73,12 +74,12 @@ func TestPartitionedDMLQueryMode(t *testing.T) { } } -func runMain(t *testing.T, args []string) error { +func runMain(t *testing.T, args []string, options ...option.ClientOption) error { t.Helper() old := os.Args os.Args = append([]string{"execspansql"}, args...) defer func() { os.Args = old }() - return _main() + return runCLI(options...) } func readPdmlQueryModeV(t *testing.T, ctx context.Context, client *spanner.Client) int64 { diff --git a/priority_transport_test.go b/priority_transport_test.go index c6f64c8..916727c 100644 --- a/priority_transport_test.go +++ b/priority_transport_test.go @@ -2,58 +2,28 @@ package main import ( "context" - "net" + "strings" "testing" "time" sppb "cloud.google.com/go/spanner/apiv1/spannerpb" - "google.golang.org/grpc" + "github.com/apstndb/execspansql/internal/grpctest" + "github.com/apstndb/spanemuboost" + "google.golang.org/api/option" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" + "google.golang.org/protobuf/proto" ) -type priorityRecordingSpannerServer struct { - sppb.UnimplementedSpannerServer - requests chan *sppb.ExecuteSqlRequest -} - -func (s *priorityRecordingSpannerServer) CreateSession(_ context.Context, req *sppb.CreateSessionRequest) (*sppb.Session, error) { - return &sppb.Session{Name: req.Database + "/sessions/priority-test"}, nil -} - -func (*priorityRecordingSpannerServer) BeginTransaction(context.Context, *sppb.BeginTransactionRequest) (*sppb.Transaction, error) { - return &sppb.Transaction{Id: []byte("priority-test")}, nil -} - -func (s *priorityRecordingSpannerServer) ExecuteStreamingSql(req *sppb.ExecuteSqlRequest, _ sppb.Spanner_ExecuteStreamingSqlServer) error { - s.requests <- req - return status.Error(codes.InvalidArgument, "priority test capture") -} - -func (s *priorityRecordingSpannerServer) ExecuteSql(_ context.Context, req *sppb.ExecuteSqlRequest) (*sppb.ResultSet, error) { - s.requests <- req - return nil, status.Error(codes.InvalidArgument, "priority test capture") -} - -func startPriorityRecordingSpannerServer(t *testing.T) (*priorityRecordingSpannerServer, string) { - t.Helper() - listener, err := net.Listen("tcp", "127.0.0.1:0") +func TestMainSendsPriorityOnExecuteSQL(t *testing.T) { + env, err := spanemuboost.RunEmulatorWithClients(context.Background()) if err != nil { t.Fatal(err) } - server := grpc.NewServer() - recorder := &priorityRecordingSpannerServer{requests: make(chan *sppb.ExecuteSqlRequest, 1)} - sppb.RegisterSpannerServer(server, recorder) - go func() { _ = server.Serve(listener) }() - t.Cleanup(func() { - server.Stop() - _ = listener.Close() - }) - return recorder, listener.Addr().String() -} + defer env.Close() //nolint:errcheck + t.Setenv("SPANNER_EMULATOR_HOST", env.Emulator().URI()) -func TestMainSendsPriorityOnExecuteSQL(t *testing.T) { tests := []struct { name string sql string @@ -69,22 +39,31 @@ func TestMainSendsPriorityOnExecuteSQL(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - recorder, host := startPriorityRecordingSpannerServer(t) - t.Setenv("SPANNER_EMULATOR_HOST", host) + requests := make(chan *sppb.ExecuteSqlRequest, 1) + dialOptions := grpctest.Inspect(func(_ string, req any) error { + if req, ok := req.(*sppb.ExecuteSqlRequest); ok { + select { + case requests <- proto.Clone(req).(*sppb.ExecuteSqlRequest): + default: + } + return status.Error(codes.InvalidArgument, "priority test capture") + } + return nil + }) sql := tt.sql if sql == "" { sql = "SELECT 1" } - args := []string{"database", "--project", "project", "--instance", "instance", "--sql", sql, "--timeout", "5s"} + args := []string{env.DatabaseID, "--project", env.ProjectID, "--instance", env.InstanceID, "--sql", sql, "--timeout", "5s"} args = append(args, tt.args...) - err := runMain(t, args) + err := runMain(t, args, option.WithGRPCDialOption(dialOptions[0]), option.WithGRPCDialOption(dialOptions[1])) if err == nil || !strings.Contains(err.Error(), "priority test capture") { t.Fatalf("_main() error = %v, want priority test capture", err) } select { - case req := <-recorder.requests: + case req := <-requests: if got := req.GetRequestOptions().GetPriority(); got != tt.priority { t.Fatalf("request priority = %v, want %v", got, tt.priority) } diff --git a/query_stats_modes_test.go b/query_stats_modes_test.go index fb0822e..1a43fc7 100644 --- a/query_stats_modes_test.go +++ b/query_stats_modes_test.go @@ -2,10 +2,15 @@ package main import ( "context" + "github.com/apstndb/execspansql/internal/grpctest" + "github.com/apstndb/spanemuboost" + "google.golang.org/api/option" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" "net" "strings" - "sync" "testing" + "time" "cloud.google.com/go/spanner" sppb "cloud.google.com/go/spanner/apiv1/spannerpb" @@ -54,7 +59,7 @@ func TestQueryStatsModesPreserveDMLCounts(t *testing.T) { } } -func TestAdditionalQueryStatsModesReachSpannerAndProduceStats(t *testing.T) { +func TestQueryStatsResponseReachesOutput(t *testing.T) { server := &queryStatsModeServer{} listener, err := net.Listen("tcp", "127.0.0.1:0") if err != nil { @@ -69,71 +74,46 @@ func TestAdditionalQueryStatsModesReachSpannerAndProduceStats(t *testing.T) { }) t.Setenv("SPANNER_EMULATOR_HOST", listener.Addr().String()) - for _, queryMode := range []string{"WITH_PLAN_AND_STATS", "WITH_STATS"} { - for _, tc := range []struct { - name string - format string - lazy bool - }{ - {name: "json_eager", format: "json"}, - {name: "yaml_eager", format: "yaml"}, - {name: "json_lazy", format: "json", lazy: true}, - {name: "yaml_lazy", format: "yaml", lazy: true}, - } { - queryMode := queryMode - tc := tc - t.Run(queryMode+"_"+tc.name, func(t *testing.T) { - server.resetModes() - args := []string{ - "database", "--project", "project", "--instance", "instance", "--sql", "SELECT 'value'", - "--query-mode", queryMode, "--format", tc.format, "--timeout", "5s", - } - if tc.lazy { - args = append(args, "--jq-input-mode", "lazy", "--filter", ".stats.queryStats.mode") - } - out, err := captureStdout(t, func() error { return runMain(t, args) }) - if err != nil { - t.Fatal(err) - } - if !strings.Contains(out, queryMode) { - t.Fatalf("output = %q, want query stats containing %q", out, queryMode) - } - if !tc.lazy { - if !strings.Contains(out, "value") { - t.Fatalf("eager output = %q, want row value", out) - } - if queryMode == "WITH_PLAN_AND_STATS" && !strings.Contains(out, "Fake Scan") { - t.Fatalf("WITH_PLAN_AND_STATS output = %q, want fake query plan", out) - } - if queryMode == "WITH_STATS" && strings.Contains(out, "Fake Scan") { - t.Fatalf("WITH_STATS output = %q, got unexpected query plan", out) - } - } - modes := server.modes() - if len(modes) != 1 || modes[0].String() != queryMode { - t.Fatalf("received query modes = %v, want [%s]", modes, queryMode) - } - }) + // Two smoke cases cover the distinct eager and lazy SDK-to-output paths. + // Plan/stat combinations are covered without a server in jqresult tests. + for _, lazy := range []bool{false, true} { + name := "eager" + if lazy { + name = "lazy" } + t.Run(name, func(t *testing.T) { + args := []string{ + "database", "--project", "project", "--instance", "instance", "--sql", "SELECT 'value'", + "--query-mode", "WITH_PLAN_AND_STATS", "--format", "json", "--timeout", "5s", + } + if lazy { + args = append(args, "--jq-input-mode", "lazy", "--filter", ".stats") + } + out, err := captureStdout(t, func() error { return runMain(t, args) }) + if err != nil { + t.Fatal(err) + } + for _, want := range []string{"test stats", "Fake Scan"} { + if !strings.Contains(out, want) { + t.Fatalf("output = %q, want %q", out, want) + } + } + if !lazy && !strings.Contains(out, "value") { + t.Fatalf("eager output = %q, want row value", out) + } + }) } } type queryStatsModeServer struct { sppb.UnimplementedSpannerServer - - mu sync.Mutex - receivedModes []sppb.ExecuteSqlRequest_QueryMode } func (s *queryStatsModeServer) CreateSession(_ context.Context, req *sppb.CreateSessionRequest) (*sppb.Session, error) { return &sppb.Session{Name: req.GetDatabase() + "/sessions/test"}, nil } -func (s *queryStatsModeServer) ExecuteStreamingSql(req *sppb.ExecuteSqlRequest, stream sppb.Spanner_ExecuteStreamingSqlServer) error { - s.mu.Lock() - s.receivedModes = append(s.receivedModes, req.GetQueryMode()) - s.mu.Unlock() - +func (s *queryStatsModeServer) ExecuteStreamingSql(_ *sppb.ExecuteSqlRequest, stream sppb.Spanner_ExecuteStreamingSqlServer) error { if err := stream.Send(&sppb.PartialResultSet{ Metadata: &sppb.ResultSetMetadata{RowType: &sppb.StructType{Fields: []*sppb.StructType_Field{{ Name: "value", @@ -146,23 +126,54 @@ func (s *queryStatsModeServer) ExecuteStreamingSql(req *sppb.ExecuteSqlRequest, stats := &sppb.ResultSetStats{ QueryStats: &structpb.Struct{Fields: map[string]*structpb.Value{ - "mode": structpb.NewStringValue(req.GetQueryMode().String()), + "summary": structpb.NewStringValue("test stats"), }}, } - if req.GetQueryMode() == sppb.ExecuteSqlRequest_WITH_PLAN_AND_STATS { - stats.QueryPlan = &sppb.QueryPlan{PlanNodes: []*sppb.PlanNode{{DisplayName: "Fake Scan"}}} - } + stats.QueryPlan = &sppb.QueryPlan{PlanNodes: []*sppb.PlanNode{{DisplayName: "Fake Scan"}}} return stream.Send(&sppb.PartialResultSet{Stats: stats}) } -func (s *queryStatsModeServer) resetModes() { - s.mu.Lock() - defer s.mu.Unlock() - s.receivedModes = nil -} - -func (s *queryStatsModeServer) modes() []sppb.ExecuteSqlRequest_QueryMode { - s.mu.Lock() - defer s.mu.Unlock() - return append([]sppb.ExecuteSqlRequest_QueryMode(nil), s.receivedModes...) +// TestMainSendsAdditionalQueryStatsModes stops at the request boundary because +// emulator support for these modes is independent of CLI option propagation. +func TestMainSendsAdditionalQueryStatsModes(t *testing.T) { + env, err := spanemuboost.RunEmulatorWithClients(context.Background()) + if err != nil { + t.Fatal(err) + } + defer env.Close() //nolint:errcheck + t.Setenv("SPANNER_EMULATOR_HOST", env.Emulator().URI()) + for _, mode := range []string{"WITH_PLAN_AND_STATS", "WITH_STATS"} { + for _, args := range [][]string{ + {"--format", "json"}, + {"--format", "yaml", "--jq-input-mode", "lazy"}, + {"--format", "experimental_csv"}, + } { + t.Run(mode+"/"+strings.Join(args, "_"), func(t *testing.T) { + modes := make(chan string, 1) + dialOptions := grpctest.Inspect(func(_ string, req any) error { + if req, ok := req.(*sppb.ExecuteSqlRequest); ok { + select { + case modes <- req.GetQueryMode().String(): + default: + } + return status.Error(codes.InvalidArgument, "query mode test capture") + } + return nil + }) + cli := []string{env.DatabaseID, "--project", env.ProjectID, "--instance", env.InstanceID, "--sql", "SELECT 1", "--query-mode", mode, "--timeout", "5s"} + err := runMain(t, append(cli, args...), option.WithGRPCDialOption(dialOptions[0]), option.WithGRPCDialOption(dialOptions[1])) + if err == nil || !strings.Contains(err.Error(), "query mode test capture") { + t.Fatalf("error = %v, want capture error", err) + } + select { + case got := <-modes: + if got != mode { + t.Fatalf("query mode = %s, want %s", got, mode) + } + case <-time.After(5 * time.Second): + t.Fatal("no ExecuteSql request") + } + }) + } + } }