From d716daca8d53d61e3e809c5d5135f32965e6ef19 Mon Sep 17 00:00:00 2001 From: apstndb <803393+apstndb@users.noreply.github.com> Date: Thu, 17 Sep 2026 01:41:30 +0900 Subject: [PATCH] refactor: share in-memory gRPC bufconn helpers across fake-RPC tests Centralize listener, server, dialer, and cleanup for in-memory Spanner and DatabaseAdmin fakes. Fake-server behavior stays in each test file. Serve errors are reported inside Cleanup after Stop returns, and the dialer keeps listener.Dial because no migrated test covers cancelled dials. --- internal/mycli/bufconn_test.go | 171 ++++++++++++++++++ internal/mycli/ddl_in_transaction_rpc_test.go | 36 +--- internal/mycli/directed_read_matrix_test.go | 29 +-- internal/mycli/directed_read_wire_test.go | 39 +--- internal/mycli/execute_ddl_rpc_test.go | 43 +---- .../execute_ddl_sequence_kind_rpc_test.go | 37 +--- internal/mycli/execute_partitioned_test.go | 32 +--- internal/mycli/execute_sql_test.go | 42 +---- .../mycli/fuzzy_finder_candidates_test.go | 65 +------ internal/mycli/heartbeat_owner_test.go | 29 +-- internal/mycli/query_cache_test.go | 44 +---- internal/mycli/show_operation_test.go | 40 +--- .../mycli/statements_query_profile_test.go | 42 +---- .../mycli/statements_run_partition_test.go | 24 +-- internal/mycli/stream_width_test.go | 44 +---- internal/mycli/string_quote_stream_test.go | 43 +---- internal/mycli/sync_proto_bundle_test.go | 34 +--- 17 files changed, 204 insertions(+), 590 deletions(-) create mode 100644 internal/mycli/bufconn_test.go diff --git a/internal/mycli/bufconn_test.go b/internal/mycli/bufconn_test.go new file mode 100644 index 00000000..ede71037 --- /dev/null +++ b/internal/mycli/bufconn_test.go @@ -0,0 +1,171 @@ +// Copyright 2026 apstndb +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package mycli + +import ( + "context" + "errors" + "net" + "testing" + + "cloud.google.com/go/longrunning/autogen/longrunningpb" + "cloud.google.com/go/spanner" + adminapi "cloud.google.com/go/spanner/admin/database/apiv1" + "cloud.google.com/go/spanner/admin/database/apiv1/databasepb" + sppb "cloud.google.com/go/spanner/apiv1/spannerpb" + "google.golang.org/api/option" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/test/bufconn" +) + +// Shared in-memory gRPC transport for fake Spanner and DatabaseAdmin servers. +// +// These helpers own only the listener, server, dialer, and cleanup order. +// Each test keeps its own fake server implementation, response fixtures, and +// Session wiring so that the scenario-specific parts stay next to the test. + +// bufconnDialer serves the registered services on an in-memory listener and +// returns a gRPC context dialer for it. The server and listener are stopped +// via t.Cleanup, after any client connections registered later. +func bufconnDialer(t *testing.T, register func(*grpc.Server)) func(context.Context, string) (net.Conn, error) { + t.Helper() + listener := bufconn.Listen(1 << 20) + grpcServer := grpc.NewServer() + register(grpcServer) + serveDone := make(chan error, 1) + go func() { + serveDone <- grpcServer.Serve(listener) + }() + t.Cleanup(func() { + grpcServer.Stop() + err := <-serveDone + _ = listener.Close() + if err != nil && !errors.Is(err, grpc.ErrServerStopped) { + t.Errorf("serve: %v", err) + } + }) + // Keep Dial rather than DialContext: no migrated test covers cancelled + // dials, and RecreateClient / USE / DETACH still dial the shared listener + // after t.Context() is cancelled during cleanup. + return func(context.Context, string) (net.Conn, error) { return listener.Dial() } +} + +// bufconnClientOptions returns per-client dial options for tests that let +// NewSession, USE, DETACH, or RecreateClient build their own clients against +// the same in-memory server without closing the shared listener. +func bufconnClientOptions(t *testing.T, register func(*grpc.Server)) []option.ClientOption { + t.Helper() + return []option.ClientOption{ + option.WithoutAuthentication(), + option.WithEndpoint("bufnet"), + option.WithGRPCDialOption(grpc.WithContextDialer(bufconnDialer(t, register))), + option.WithGRPCDialOption(grpc.WithTransportCredentials(insecure.NewCredentials())), + } +} + +// dialBufconn returns a client connection to the in-memory server. The +// connection is closed via t.Cleanup before the server is stopped. +func dialBufconn(t *testing.T, register func(*grpc.Server)) *grpc.ClientConn { + t.Helper() + conn, err := grpc.NewClient("passthrough:///bufconn", + grpc.WithContextDialer(bufconnDialer(t, register)), + grpc.WithTransportCredentials(insecure.NewCredentials()), + ) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = conn.Close() }) + return conn +} + +// bufconnSpannerClientConfig is the client configuration every in-memory +// Spanner client and its TransactionManager use. +var bufconnSpannerClientConfig = spanner.ClientConfig{DisableNativeMetrics: true} + +// newBufconnSpannerClient returns a Spanner client for the fake Spanner +// service, with native metrics disabled. Extra services (for example a +// DatabaseAdmin fake) may be registered through register. +func newBufconnSpannerClient(t *testing.T, database string, server sppb.SpannerServer, register ...func(*grpc.Server)) *spanner.Client { + t.Helper() + conn := dialBufconn(t, func(s *grpc.Server) { + sppb.RegisterSpannerServer(s, server) + for _, r := range register { + r(s) + } + }) + client, err := spanner.NewClientWithConfig(t.Context(), database, bufconnSpannerClientConfig, option.WithGRPCConn(conn)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(client.Close) + return client +} + +// bufconnAdminServer is the pair of services a fake DatabaseAdmin needs to +// serve so that long-running operations can be polled. +type bufconnAdminServer interface { + databasepb.DatabaseAdminServer + longrunningpb.OperationsServer +} + +// newBufconnAdminClient returns a DatabaseAdmin client for the fake admin +// service, registering both the admin and operations services. +func newBufconnAdminClient(t *testing.T, server bufconnAdminServer) *adminapi.DatabaseAdminClient { + t.Helper() + conn := dialBufconn(t, func(s *grpc.Server) { + databasepb.RegisterDatabaseAdminServer(s, server) + longrunningpb.RegisterOperationsServer(s, server) + }) + adminClient, err := adminapi.NewDatabaseAdminClient(t.Context(), option.WithGRPCConn(conn)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = adminClient.Close() }) + return adminClient +} + +// bufconnTestIdentity is the connection identity used by admin-only sessions. +var bufconnTestIdentity = ConnectionVars{Project: "test", Instance: "test", Database: "test"} + +// newBufconnAdminSession returns a Session whose only backend is the fake +// DatabaseAdmin service, with default system variables and a test identity. +func newBufconnAdminSession(t *testing.T, server bufconnAdminServer) *Session { + t.Helper() + sysVars := newSystemVariablesWithDefaultsForTest() + sysVars.Connection = bufconnTestIdentity + return &Session{ + adminClient: newBufconnAdminClient(t, server), + systemVariables: sysVars, + connection: bufconnTestIdentity, + } +} + +// newBufconnQuerySession returns a DatabaseConnected Session backed by the +// fake Spanner service, with default system variables and a live +// TransactionManager, for statement-level query tests. +func newBufconnQuerySession(t *testing.T, server sppb.SpannerServer) (*Session, *systemVariables) { + t.Helper() + client := newBufconnSpannerClient(t, "projects/test/instances/test/databases/test", server) + live := newSystemVariablesWithDefaultsForTest() + session := &Session{ + mode: DatabaseConnected, + client: client, + systemVariables: live, + txn: NewTransactionManager(client, live, bufconnSpannerClientConfig), + } + live.inTransaction = session.txn.InTransaction + return session, live +} diff --git a/internal/mycli/ddl_in_transaction_rpc_test.go b/internal/mycli/ddl_in_transaction_rpc_test.go index 14580dbb..218bd38a 100644 --- a/internal/mycli/ddl_in_transaction_rpc_test.go +++ b/internal/mycli/ddl_in_transaction_rpc_test.go @@ -17,22 +17,14 @@ package mycli import ( "context" "errors" - "net" "strings" "testing" "time" - longrunningpb "cloud.google.com/go/longrunning/autogen/longrunningpb" "cloud.google.com/go/spanner" - adminapi "cloud.google.com/go/spanner/admin/database/apiv1" - "cloud.google.com/go/spanner/admin/database/apiv1/databasepb" "github.com/apstndb/spanner-mycli/enums" - "google.golang.org/api/option" - "google.golang.org/grpc" "google.golang.org/grpc/codes" - "google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/status" - "google.golang.org/grpc/test/bufconn" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/descriptorpb" ) @@ -76,33 +68,7 @@ func newDDLTxnHarness(t *testing.T) *ddlTxnHarness { func attachDDLAdmin(t *testing.T, session *Session, server *ddlAdminTestServer) { t.Helper() - listener := bufconn.Listen(1 << 20) - grpcServer := grpc.NewServer() - databasepb.RegisterDatabaseAdminServer(grpcServer, server) - longrunningpb.RegisterOperationsServer(grpcServer, server) - go func() { - if err := grpcServer.Serve(listener); err != nil && !errors.Is(err, grpc.ErrServerStopped) { - t.Errorf("serve ddl admin: %v", err) - } - }() - t.Cleanup(func() { - grpcServer.Stop() - _ = listener.Close() - }) - conn, err := grpc.NewClient("passthrough:///ddl-in-txn-admin", - grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) { return listener.Dial() }), - grpc.WithTransportCredentials(insecure.NewCredentials()), - ) - if err != nil { - t.Fatal(err) - } - t.Cleanup(func() { _ = conn.Close() }) - adminClient, err := adminapi.NewDatabaseAdminClient(t.Context(), option.WithGRPCConn(conn)) - if err != nil { - t.Fatal(err) - } - t.Cleanup(func() { _ = adminClient.Close() }) - session.adminClient = adminClient + session.adminClient = newBufconnAdminClient(t, server) } func (h *ddlTxnHarness) adminCalled() bool { diff --git a/internal/mycli/directed_read_matrix_test.go b/internal/mycli/directed_read_matrix_test.go index 4f42c20d..408cf972 100644 --- a/internal/mycli/directed_read_matrix_test.go +++ b/internal/mycli/directed_read_matrix_test.go @@ -17,11 +17,9 @@ package mycli import ( "bytes" "context" - "errors" "fmt" "io" "log/slog" - "net" "slices" "strings" "sync" @@ -36,37 +34,18 @@ import ( "github.com/apstndb/spanner-mycli/internal/mycli/streamio" "google.golang.org/api/option" "google.golang.org/grpc" - "google.golang.org/grpc/credentials/insecure" - "google.golang.org/grpc/test/bufconn" "google.golang.org/protobuf/proto" ) func startDirectedReadDial(t *testing.T) (*directedReadWireServer, []option.ClientOption) { t.Helper() srv := &directedReadWireServer{partitionFanInServer: partitionFanInServer{nPartitions: 1, rowsPer: 1}} - listener := bufconn.Listen(1 << 20) - grpcServer := grpc.NewServer() - sppb.RegisterSpannerServer(grpcServer, srv) - adminpb.RegisterDatabaseAdminServer(grpcServer, &directedReadAdminServer{}) - go func() { - if err := grpcServer.Serve(listener); err != nil && !errors.Is(err, grpc.ErrServerStopped) { - t.Errorf("serve: %v", err) - } - }() - t.Cleanup(func() { - grpcServer.Stop() - _ = listener.Close() - }) // Per-client dialer options so USE/DETACH/RecreateClient can close a // session without shutting down the shared in-memory listener. - opts := []option.ClientOption{ - option.WithoutAuthentication(), - option.WithEndpoint("bufnet"), - option.WithGRPCDialOption(grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) { - return listener.Dial() - })), - option.WithGRPCDialOption(grpc.WithTransportCredentials(insecure.NewCredentials())), - } + opts := bufconnClientOptions(t, func(s *grpc.Server) { + sppb.RegisterSpannerServer(s, srv) + registerDirectedReadAdmin(s) + }) return srv, opts } diff --git a/internal/mycli/directed_read_wire_test.go b/internal/mycli/directed_read_wire_test.go index e34c0e91..0169bd59 100644 --- a/internal/mycli/directed_read_wire_test.go +++ b/internal/mycli/directed_read_wire_test.go @@ -16,8 +16,6 @@ package mycli import ( "context" - "errors" - "net" "strings" "sync" "sync/atomic" @@ -27,12 +25,9 @@ import ( "cloud.google.com/go/spanner" adminpb "cloud.google.com/go/spanner/admin/database/apiv1/databasepb" sppb "cloud.google.com/go/spanner/apiv1/spannerpb" - "google.golang.org/api/option" "google.golang.org/grpc" "google.golang.org/grpc/codes" - "google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/status" - "google.golang.org/grpc/test/bufconn" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/emptypb" "google.golang.org/protobuf/types/known/structpb" @@ -218,36 +213,16 @@ func directedReadFakeRow(sql string, cycleFK bool) (*sppb.ResultSetMetadata, []* func startDirectedReadWire(t *testing.T) (*directedReadWireServer, *spanner.Client) { t.Helper() srv := &directedReadWireServer{partitionFanInServer: partitionFanInServer{nPartitions: 1, rowsPer: 1}} - listener := bufconn.Listen(1 << 20) - grpcServer := grpc.NewServer() - sppb.RegisterSpannerServer(grpcServer, srv) - adminpb.RegisterDatabaseAdminServer(grpcServer, &directedReadAdminServer{}) - go func() { - if err := grpcServer.Serve(listener); err != nil && !errors.Is(err, grpc.ErrServerStopped) { - t.Errorf("serve: %v", err) - } - }() - t.Cleanup(func() { - grpcServer.Stop() - _ = listener.Close() - }) - conn, err := grpc.NewClient("passthrough:///directed-read-wire", - grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) { return listener.Dial() }), - grpc.WithTransportCredentials(insecure.NewCredentials()), - ) - if err != nil { - t.Fatal(err) - } - t.Cleanup(func() { _ = conn.Close() }) - client, err := spanner.NewClientWithConfig(t.Context(), "projects/test/instances/test/databases/test", - spanner.ClientConfig{DisableNativeMetrics: true}, option.WithGRPCConn(conn)) - if err != nil { - t.Fatal(err) - } - t.Cleanup(client.Close) + client := newBufconnSpannerClient(t, "projects/test/instances/test/databases/test", srv, registerDirectedReadAdmin) return srv, client } +// registerDirectedReadAdmin adds the stub DatabaseAdmin service the directed +// read fixtures need next to the fake Spanner service. +func registerDirectedReadAdmin(s *grpc.Server) { + adminpb.RegisterDatabaseAdminServer(s, &directedReadAdminServer{}) +} + func consumeDirectedReadRequest(t *testing.T, srv *directedReadWireServer, it *spanner.RowIterator, want *sppb.DirectedReadOptions) *sppb.ExecuteSqlRequest { t.Helper() n := 0 diff --git a/internal/mycli/execute_ddl_rpc_test.go b/internal/mycli/execute_ddl_rpc_test.go index e555e7c1..f8697837 100644 --- a/internal/mycli/execute_ddl_rpc_test.go +++ b/internal/mycli/execute_ddl_rpc_test.go @@ -18,7 +18,6 @@ import ( "context" "errors" "io" - "net" "os" "strings" "sync" @@ -27,17 +26,12 @@ import ( "time" "cloud.google.com/go/longrunning/autogen/longrunningpb" - adminapi "cloud.google.com/go/spanner/admin/database/apiv1" "cloud.google.com/go/spanner/admin/database/apiv1/databasepb" "github.com/apstndb/spanner-mycli/enums" "github.com/apstndb/spanner-mycli/internal/mycli/streamio" - "google.golang.org/api/option" statuspb "google.golang.org/genproto/googleapis/rpc/status" - "google.golang.org/grpc" "google.golang.org/grpc/codes" - "google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/status" - "google.golang.org/grpc/test/bufconn" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/anypb" "google.golang.org/protobuf/types/known/emptypb" @@ -683,40 +677,5 @@ func (s *ddlAdminTestServer) GetOperation(ctx context.Context, _ *longrunningpb. func newDDLAdminSession(t *testing.T, server *ddlAdminTestServer) *Session { t.Helper() - listener := bufconn.Listen(1 << 20) - grpcServer := grpc.NewServer() - databasepb.RegisterDatabaseAdminServer(grpcServer, server) - longrunningpb.RegisterOperationsServer(grpcServer, server) - go func() { - if err := grpcServer.Serve(listener); err != nil && !errors.Is(err, grpc.ErrServerStopped) { - t.Errorf("serve ddl admin: %v", err) - } - }() - t.Cleanup(func() { - grpcServer.Stop() - _ = listener.Close() - }) - conn, err := grpc.NewClient("passthrough:///ddl-admin", - grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) { return listener.Dial() }), - grpc.WithTransportCredentials(insecure.NewCredentials()), - ) - if err != nil { - t.Fatal(err) - } - t.Cleanup(func() { _ = conn.Close() }) - adminClient, err := adminapi.NewDatabaseAdminClient(t.Context(), option.WithGRPCConn(conn)) - if err != nil { - t.Fatal(err) - } - t.Cleanup(func() { _ = adminClient.Close() }) - - sysVars := newSystemVariablesWithDefaultsForTest() - identity := ConnectionVars{Project: "test", Instance: "test", Database: "test"} - sysVars.Connection = identity - session := &Session{ - adminClient: adminClient, - systemVariables: sysVars, - connection: identity, - } - return session + return newBufconnAdminSession(t, server) } diff --git a/internal/mycli/execute_ddl_sequence_kind_rpc_test.go b/internal/mycli/execute_ddl_sequence_kind_rpc_test.go index 7f4f6daa..7e51c4de 100644 --- a/internal/mycli/execute_ddl_sequence_kind_rpc_test.go +++ b/internal/mycli/execute_ddl_sequence_kind_rpc_test.go @@ -17,7 +17,6 @@ package mycli import ( "context" "errors" - "net" "slices" "strings" "sync" @@ -25,16 +24,11 @@ import ( "time" "cloud.google.com/go/longrunning/autogen/longrunningpb" - adminapi "cloud.google.com/go/spanner/admin/database/apiv1" "cloud.google.com/go/spanner/admin/database/apiv1/databasepb" "github.com/apstndb/spanner-mycli/enums" - "google.golang.org/api/option" statuspb "google.golang.org/genproto/googleapis/rpc/status" - "google.golang.org/grpc" "google.golang.org/grpc/codes" - "google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/status" - "google.golang.org/grpc/test/bufconn" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/descriptorpb" "google.golang.org/protobuf/types/known/anypb" @@ -188,36 +182,7 @@ func mustEmptyAny() *anypb.Any { func newScriptedDDLSession(t *testing.T, step func(*databasepb.UpdateDatabaseDdlRequest, int) scriptedDDLStep) (*Session, *scriptedDDLServer) { t.Helper() server := &scriptedDDLServer{step: step} - lis := bufconn.Listen(1 << 20) - gs := grpc.NewServer() - databasepb.RegisterDatabaseAdminServer(gs, server) - longrunningpb.RegisterOperationsServer(gs, server) - go func() { - if err := gs.Serve(lis); err != nil && !errors.Is(err, grpc.ErrServerStopped) { - t.Errorf("serve: %v", err) - } - }() - t.Cleanup(func() { - gs.Stop() - _ = lis.Close() - }) - conn, err := grpc.NewClient("passthrough:///seq-kind", - grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) { return lis.Dial() }), - grpc.WithTransportCredentials(insecure.NewCredentials()), - ) - if err != nil { - t.Fatal(err) - } - t.Cleanup(func() { _ = conn.Close() }) - admin, err := adminapi.NewDatabaseAdminClient(t.Context(), option.WithGRPCConn(conn)) - if err != nil { - t.Fatal(err) - } - t.Cleanup(func() { _ = admin.Close() }) - sysVars := newSystemVariablesWithDefaultsForTest() - identity := ConnectionVars{Project: "test", Instance: "test", Database: "test"} - sysVars.Connection = identity - return &Session{adminClient: admin, systemVariables: sysVars, connection: identity}, server + return newBufconnAdminSession(t, server), server } func enableKind(session *Session) { diff --git a/internal/mycli/execute_partitioned_test.go b/internal/mycli/execute_partitioned_test.go index 86f03586..053f6ea7 100644 --- a/internal/mycli/execute_partitioned_test.go +++ b/internal/mycli/execute_partitioned_test.go @@ -21,7 +21,6 @@ import ( "fmt" "io" "iter" - "net" "strings" "sync" "sync/atomic" @@ -33,12 +32,8 @@ import ( "github.com/apstndb/spanner-mycli/enums" "github.com/apstndb/spanner-mycli/internal/mycli/decoder" "github.com/apstndb/spanner-mycli/internal/mycli/format" - "google.golang.org/api/option" - "google.golang.org/grpc" "google.golang.org/grpc/codes" - "google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/status" - "google.golang.org/grpc/test/bufconn" "google.golang.org/protobuf/types/known/emptypb" "google.golang.org/protobuf/types/known/structpb" "google.golang.org/protobuf/types/known/timestamppb" @@ -108,32 +103,7 @@ func (s *partitionFanInServer) ExecuteStreamingSql(r *sppb.ExecuteSqlRequest, st func startPartitionFanIn(t *testing.T, server *partitionFanInServer) (*spanner.BatchReadOnlyTransaction, []*spanner.Partition) { t.Helper() - listener := bufconn.Listen(1 << 20) - grpcServer := grpc.NewServer() - sppb.RegisterSpannerServer(grpcServer, server) - go func() { - if err := grpcServer.Serve(listener); err != nil && !errors.Is(err, grpc.ErrServerStopped) { - t.Errorf("serve: %v", err) - } - }() - t.Cleanup(func() { - grpcServer.Stop() - _ = listener.Close() - }) - conn, err := grpc.NewClient("passthrough:///fanin", - grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) { return listener.Dial() }), - grpc.WithTransportCredentials(insecure.NewCredentials()), - ) - if err != nil { - t.Fatal(err) - } - t.Cleanup(func() { _ = conn.Close() }) - client, err := spanner.NewClientWithConfig(t.Context(), "projects/test/instances/test/databases/test", - spanner.ClientConfig{DisableNativeMetrics: true}, option.WithGRPCConn(conn)) - if err != nil { - t.Fatal(err) - } - t.Cleanup(client.Close) + client := newBufconnSpannerClient(t, "projects/test/instances/test/databases/test", server) tx, err := client.BatchReadOnlyTransaction(t.Context(), spanner.StrongRead()) if err != nil { t.Fatal(err) diff --git a/internal/mycli/execute_sql_test.go b/internal/mycli/execute_sql_test.go index 65b5bf0a..1a9d319e 100644 --- a/internal/mycli/execute_sql_test.go +++ b/internal/mycli/execute_sql_test.go @@ -20,7 +20,6 @@ import ( "errors" "io" "math" - "net" "strings" "testing" @@ -31,12 +30,8 @@ import ( "github.com/apstndb/spanner-mycli/internal/mycli/streamio" "github.com/apstndb/spanvalue" "github.com/google/go-cmp/cmp" - "google.golang.org/api/option" - "google.golang.org/grpc" "google.golang.org/grpc/codes" - "google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/status" - "google.golang.org/grpc/test/bufconn" "google.golang.org/protobuf/types/known/structpb" ) @@ -746,43 +741,8 @@ func (s *emptySQLRPCServer) ExecuteStreamingSql(_ *sppb.ExecuteSqlRequest, strea func newEmptySQLRPCSession(t *testing.T, plan *sppb.QueryPlan, stats map[string]any) (*Session, *systemVariables) { t.Helper() - listener := bufconn.Listen(1 << 20) - grpcServer := grpc.NewServer() - sppb.RegisterSpannerServer(grpcServer, &emptySQLRPCServer{queryCacheRPCServer: queryCacheRPCServer{ + return newBufconnQuerySession(t, &emptySQLRPCServer{queryCacheRPCServer: queryCacheRPCServer{ plan: plan, stats: mustNewStruct(stats), }}) - go func() { - if err := grpcServer.Serve(listener); err != nil && !errors.Is(err, grpc.ErrServerStopped) { - t.Errorf("serve: %v", err) - } - }() - t.Cleanup(func() { - grpcServer.Stop() - _ = listener.Close() - }) - conn, err := grpc.NewClient("passthrough:///empty-sql", - grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) { return listener.Dial() }), - grpc.WithTransportCredentials(insecure.NewCredentials()), - ) - if err != nil { - t.Fatal(err) - } - t.Cleanup(func() { _ = conn.Close() }) - client, err := spanner.NewClientWithConfig(t.Context(), "projects/test/instances/test/databases/test", - spanner.ClientConfig{DisableNativeMetrics: true}, option.WithGRPCConn(conn)) - if err != nil { - t.Fatal(err) - } - t.Cleanup(client.Close) - - live := newSystemVariablesWithDefaultsForTest() - session := &Session{ - mode: DatabaseConnected, - client: client, - systemVariables: live, - txn: NewTransactionManager(client, live, spanner.ClientConfig{DisableNativeMetrics: true}), - } - live.inTransaction = session.txn.InTransaction - return session, live } diff --git a/internal/mycli/fuzzy_finder_candidates_test.go b/internal/mycli/fuzzy_finder_candidates_test.go index db5f4538..6b0a7108 100644 --- a/internal/mycli/fuzzy_finder_candidates_test.go +++ b/internal/mycli/fuzzy_finder_candidates_test.go @@ -18,16 +18,12 @@ import ( "bufio" "bytes" "context" - "errors" "fmt" "io" - "net" "strings" "testing" "cloud.google.com/go/longrunning/autogen/longrunningpb" - "cloud.google.com/go/spanner" - adminapi "cloud.google.com/go/spanner/admin/database/apiv1" "cloud.google.com/go/spanner/admin/database/apiv1/databasepb" sppb "cloud.google.com/go/spanner/apiv1/spannerpb" "github.com/cloudspannerecosystem/memefish" @@ -35,12 +31,8 @@ import ( "github.com/google/go-cmp/cmp" "github.com/hymkor/go-multiline-ny" readline "github.com/nyaosorg/go-readline-ny" - "google.golang.org/api/option" - "google.golang.org/grpc" "google.golang.org/grpc/codes" - "google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/status" - "google.golang.org/grpc/test/bufconn" "google.golang.org/protobuf/types/known/anypb" "google.golang.org/protobuf/types/known/emptypb" "google.golang.org/protobuf/types/known/structpb" @@ -564,37 +556,11 @@ func (s *fuzzyAdminServer) ListOperations(_ context.Context, req *longrunningpb. func newFuzzyAdminSession(t *testing.T, server *fuzzyAdminServer) *Session { t.Helper() - listener := bufconn.Listen(1 << 20) - grpcServer := grpc.NewServer() - databasepb.RegisterDatabaseAdminServer(grpcServer, server) - longrunningpb.RegisterOperationsServer(grpcServer, server) - go func() { - if err := grpcServer.Serve(listener); err != nil && !errors.Is(err, grpc.ErrServerStopped) { - t.Errorf("serve: %v", err) - } - }() - t.Cleanup(func() { - grpcServer.Stop() - _ = listener.Close() - }) - conn, err := grpc.NewClient("passthrough:///fuzzy-admin", - grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) { return listener.Dial() }), - grpc.WithTransportCredentials(insecure.NewCredentials()), - ) - if err != nil { - t.Fatal(err) - } - t.Cleanup(func() { _ = conn.Close() }) - adminClient, err := adminapi.NewDatabaseAdminClient(t.Context(), option.WithGRPCConn(conn)) - if err != nil { - t.Fatal(err) - } - t.Cleanup(func() { _ = adminClient.Close() }) sv := newSystemVariablesWithDefaultsForTest() identity := ConnectionVars{Project: "p", Instance: "i", Database: "db"} sv.Connection = identity return &Session{ - adminClient: adminClient, + adminClient: newBufconnAdminClient(t, server), systemVariables: sv, connection: identity, } @@ -687,32 +653,7 @@ func fuzzySchemaFixture(sql string) ([]string, [][]string) { func newFuzzySchemaSession(t *testing.T, server *fuzzySchemaServer) *Session { t.Helper() - listener := bufconn.Listen(1 << 20) - grpcServer := grpc.NewServer() - sppb.RegisterSpannerServer(grpcServer, server) - go func() { - if err := grpcServer.Serve(listener); err != nil && !errors.Is(err, grpc.ErrServerStopped) { - t.Errorf("serve: %v", err) - } - }() - t.Cleanup(func() { - grpcServer.Stop() - _ = listener.Close() - }) - conn, err := grpc.NewClient("passthrough:///fuzzy-schema", - grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) { return listener.Dial() }), - grpc.WithTransportCredentials(insecure.NewCredentials()), - ) - if err != nil { - t.Fatal(err) - } - t.Cleanup(func() { _ = conn.Close() }) - client, err := spanner.NewClientWithConfig(t.Context(), "projects/p/instances/i/databases/db", - spanner.ClientConfig{DisableNativeMetrics: true}, option.WithGRPCConn(conn)) - if err != nil { - t.Fatal(err) - } - t.Cleanup(client.Close) + client := newBufconnSpannerClient(t, "projects/p/instances/i/databases/db", server) sv := newSystemVariablesWithDefaultsForTest() sv.Connection = ConnectionVars{Project: "p", Instance: "i", Database: "db"} return &Session{ @@ -720,6 +661,6 @@ func newFuzzySchemaSession(t *testing.T, server *fuzzySchemaServer) *Session { client: client, systemVariables: sv, connection: sv.Connection, - txn: NewTransactionManager(client, sv, spanner.ClientConfig{DisableNativeMetrics: true}), + txn: NewTransactionManager(client, sv, bufconnSpannerClientConfig), } } diff --git a/internal/mycli/heartbeat_owner_test.go b/internal/mycli/heartbeat_owner_test.go index 9d98ecbe..00861ab8 100644 --- a/internal/mycli/heartbeat_owner_test.go +++ b/internal/mycli/heartbeat_owner_test.go @@ -16,10 +16,8 @@ package mycli import ( "context" - "errors" "fmt" "maps" - "net" "slices" "sync" "sync/atomic" @@ -31,8 +29,6 @@ import ( "google.golang.org/api/option" statuspb "google.golang.org/genproto/googleapis/rpc/status" "google.golang.org/grpc" - "google.golang.org/grpc/credentials/insecure" - "google.golang.org/grpc/test/bufconn" "google.golang.org/protobuf/types/known/emptypb" "google.golang.org/protobuf/types/known/structpb" "google.golang.org/protobuf/types/known/timestamppb" @@ -907,36 +903,17 @@ func newHeartbeatHarness(t *testing.T) *heartbeatHarness { server := &heartbeatRPCServer{ heartbeatStarted: make(chan struct{}), } - listener := bufconn.Listen(1 << 20) - grpcServer := grpc.NewServer() - sppb.RegisterSpannerServer(grpcServer, server) - go func() { - if err := grpcServer.Serve(listener); err != nil && !errors.Is(err, grpc.ErrServerStopped) { - t.Errorf("serve: %v", err) - } - }() - t.Cleanup(func() { - grpcServer.Stop() - _ = listener.Close() - }) - conn, err := grpc.NewClient("passthrough:///heartbeat", - grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) { return listener.Dial() }), - grpc.WithTransportCredentials(insecure.NewCredentials()), - ) - if err != nil { - t.Fatal(err) - } - t.Cleanup(func() { _ = conn.Close() }) + conn := dialBufconn(t, func(s *grpc.Server) { sppb.RegisterSpannerServer(s, server) }) clientOpts := []option.ClientOption{option.WithGRPCConn(conn)} client, err := spanner.NewClientWithConfig(t.Context(), "projects/test/instances/test/databases/test", - spanner.ClientConfig{DisableNativeMetrics: true}, clientOpts...) + bufconnSpannerClientConfig, clientOpts...) if err != nil { t.Fatal(err) } t.Cleanup(client.Close) sysVars := newSystemVariablesWithDefaultsForTest() - tm := NewTransactionManager(client, sysVars, spanner.ClientConfig{DisableNativeMetrics: true}) + tm := NewTransactionManager(client, sysVars, bufconnSpannerClientConfig) ticks := make(chan time.Time) h := &heartbeatHarness{ tm: tm, diff --git a/internal/mycli/query_cache_test.go b/internal/mycli/query_cache_test.go index 931dd2f5..9e3024f7 100644 --- a/internal/mycli/query_cache_test.go +++ b/internal/mycli/query_cache_test.go @@ -17,26 +17,19 @@ package mycli import ( "bytes" "context" - "errors" "fmt" "io" - "net" "strings" "testing" "time" - "cloud.google.com/go/spanner" sppb "cloud.google.com/go/spanner/apiv1/spannerpb" "github.com/apstndb/spanner-mycli/enums" "github.com/apstndb/spanner-mycli/internal/mycli/streamio" "github.com/google/go-cmp/cmp" "github.com/google/go-cmp/cmp/cmpopts" - "google.golang.org/api/option" - "google.golang.org/grpc" "google.golang.org/grpc/codes" - "google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/status" - "google.golang.org/grpc/test/bufconn" "google.golang.org/protobuf/testing/protocmp" "google.golang.org/protobuf/types/known/emptypb" "google.golang.org/protobuf/types/known/structpb" @@ -410,44 +403,9 @@ func (s *queryCacheRPCServer) resultSet() *sppb.ResultSet { func newQueryCacheRPCSession(t *testing.T, plan *sppb.QueryPlan, stats map[string]any, execErr error) (*Session, *systemVariables) { t.Helper() - listener := bufconn.Listen(1 << 20) - grpcServer := grpc.NewServer() - sppb.RegisterSpannerServer(grpcServer, &queryCacheRPCServer{ + return newBufconnQuerySession(t, &queryCacheRPCServer{ plan: plan, stats: mustNewStruct(stats), execErr: execErr, }) - go func() { - if err := grpcServer.Serve(listener); err != nil && !errors.Is(err, grpc.ErrServerStopped) { - t.Errorf("serve: %v", err) - } - }() - t.Cleanup(func() { - grpcServer.Stop() - _ = listener.Close() - }) - conn, err := grpc.NewClient("passthrough:///qcache", - grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) { return listener.Dial() }), - grpc.WithTransportCredentials(insecure.NewCredentials()), - ) - if err != nil { - t.Fatal(err) - } - t.Cleanup(func() { _ = conn.Close() }) - client, err := spanner.NewClientWithConfig(t.Context(), "projects/test/instances/test/databases/test", - spanner.ClientConfig{DisableNativeMetrics: true}, option.WithGRPCConn(conn)) - if err != nil { - t.Fatal(err) - } - t.Cleanup(client.Close) - - live := newSystemVariablesWithDefaultsForTest() - session := &Session{ - mode: DatabaseConnected, - client: client, - systemVariables: live, - txn: NewTransactionManager(client, live, spanner.ClientConfig{DisableNativeMetrics: true}), - } - live.inTransaction = session.txn.InTransaction - return session, live } diff --git a/internal/mycli/show_operation_test.go b/internal/mycli/show_operation_test.go index 0537157d..631d04be 100644 --- a/internal/mycli/show_operation_test.go +++ b/internal/mycli/show_operation_test.go @@ -4,7 +4,6 @@ import ( "context" "errors" "fmt" - "net" "strings" "testing" "time" @@ -16,9 +15,7 @@ import ( "google.golang.org/api/option" "google.golang.org/grpc" "google.golang.org/grpc/codes" - "google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/status" - "google.golang.org/grpc/test/bufconn" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/anypb" ) @@ -535,46 +532,15 @@ func (s *showOperationTestServer) GetOperation(_ context.Context, _ *longrunning func newShowOperationTestSession(t *testing.T, server longrunningpb.OperationsServer) *Session { t.Helper() - - listener := bufconn.Listen(1 << 20) - grpcServer := grpc.NewServer() - longrunningpb.RegisterOperationsServer(grpcServer, server) - go func() { - if err := grpcServer.Serve(listener); err != nil && !errors.Is(err, grpc.ErrServerStopped) { - t.Errorf("serve operation test server: %v", err) - } - }() - t.Cleanup(func() { - grpcServer.Stop() - _ = listener.Close() - }) - - conn, err := grpc.NewClient( - "passthrough:///show-operation-test", - grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) { - return listener.Dial() - }), - grpc.WithTransportCredentials(insecure.NewCredentials()), - ) - if err != nil { - t.Fatalf("create gRPC test client: %v", err) - } - t.Cleanup(func() { _ = conn.Close() }) - + conn := dialBufconn(t, func(s *grpc.Server) { longrunningpb.RegisterOperationsServer(s, server) }) adminClient, err := adminapi.NewDatabaseAdminClient(t.Context(), option.WithGRPCConn(conn)) if err != nil { t.Fatalf("create database admin client: %v", err) } t.Cleanup(func() { _ = adminClient.Close() }) - - identity := ConnectionVars{ - Project: "test", - Instance: "test", - Database: "test", - } return &Session{ adminClient: adminClient, - systemVariables: &systemVariables{Connection: identity}, - connection: identity, + systemVariables: &systemVariables{Connection: bufconnTestIdentity}, + connection: bufconnTestIdentity, } } diff --git a/internal/mycli/statements_query_profile_test.go b/internal/mycli/statements_query_profile_test.go index 9719fb2c..c338fcc5 100644 --- a/internal/mycli/statements_query_profile_test.go +++ b/internal/mycli/statements_query_profile_test.go @@ -17,9 +17,7 @@ package mycli import ( "context" "encoding/json" - "errors" "fmt" - "net" "strings" "sync" "testing" @@ -29,12 +27,8 @@ import ( sppb "cloud.google.com/go/spanner/apiv1/spannerpb" "github.com/google/go-cmp/cmp" "github.com/google/go-cmp/cmp/cmpopts" - "google.golang.org/api/option" - "google.golang.org/grpc" "google.golang.org/grpc/codes" - "google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/status" - "google.golang.org/grpc/test/bufconn" "google.golang.org/protobuf/encoding/protojson" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/testing/protocmp" @@ -617,42 +611,8 @@ func flattenResultSetValues(rs *sppb.ResultSet) []*structpb.Value { func newQueryProfileRPCSession(t *testing.T, rows []queryProfileRPCRow) (*Session, *queryProfileRPCServer) { t.Helper() - listener := bufconn.Listen(1 << 20) - grpcServer := grpc.NewServer() server := &queryProfileRPCServer{rows: rows} - sppb.RegisterSpannerServer(grpcServer, server) - go func() { - if err := grpcServer.Serve(listener); err != nil && !errors.Is(err, grpc.ErrServerStopped) { - t.Errorf("serve: %v", err) - } - }() - t.Cleanup(func() { - grpcServer.Stop() - _ = listener.Close() - }) - conn, err := grpc.NewClient("passthrough:///qprofile", - grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) { return listener.Dial() }), - grpc.WithTransportCredentials(insecure.NewCredentials()), - ) - if err != nil { - t.Fatal(err) - } - t.Cleanup(func() { _ = conn.Close() }) - client, err := spanner.NewClientWithConfig(t.Context(), "projects/test/instances/test/databases/test", - spanner.ClientConfig{DisableNativeMetrics: true}, option.WithGRPCConn(conn)) - if err != nil { - t.Fatal(err) - } - t.Cleanup(client.Close) - - live := newSystemVariablesWithDefaultsForTest() - session := &Session{ - mode: DatabaseConnected, - client: client, - systemVariables: live, - txn: NewTransactionManager(client, live, spanner.ClientConfig{DisableNativeMetrics: true}), - } - live.inTransaction = session.txn.InTransaction + session, _ := newBufconnQuerySession(t, server) return session, server } diff --git a/internal/mycli/statements_run_partition_test.go b/internal/mycli/statements_run_partition_test.go index 8ec895d3..a306860f 100644 --- a/internal/mycli/statements_run_partition_test.go +++ b/internal/mycli/statements_run_partition_test.go @@ -41,7 +41,6 @@ import ( "google.golang.org/api/option" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" - "google.golang.org/grpc/test/bufconn" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/emptypb" ) @@ -87,27 +86,10 @@ func startRunPartitionWire(t *testing.T, server *runPartitionWireServer) []optio if server.rowsPer == 0 { server.rowsPer = 1 } - listener := bufconn.Listen(1 << 20) - grpcServer := grpc.NewServer() - sppb.RegisterSpannerServer(grpcServer, server) - adminpb.RegisterDatabaseAdminServer(grpcServer, &directedReadAdminServer{}) - go func() { - if err := grpcServer.Serve(listener); err != nil && !errors.Is(err, grpc.ErrServerStopped) { - t.Errorf("serve: %v", err) - } - }() - t.Cleanup(func() { - grpcServer.Stop() - _ = listener.Close() + return bufconnClientOptions(t, func(s *grpc.Server) { + sppb.RegisterSpannerServer(s, server) + registerDirectedReadAdmin(s) }) - return []option.ClientOption{ - option.WithoutAuthentication(), - option.WithEndpoint("bufnet"), - option.WithGRPCDialOption(grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) { - return listener.Dial() - })), - option.WithGRPCDialOption(grpc.WithTransportCredentials(insecure.NewCredentials())), - } } func newRunPartitionVars(t *testing.T) *systemVariables { diff --git a/internal/mycli/stream_width_test.go b/internal/mycli/stream_width_test.go index 47ac9929..1b2198e8 100644 --- a/internal/mycli/stream_width_test.go +++ b/internal/mycli/stream_width_test.go @@ -17,21 +17,14 @@ package mycli import ( "bytes" "context" - "errors" "io" - "net" "strings" "testing" - "cloud.google.com/go/spanner" sppb "cloud.google.com/go/spanner/apiv1/spannerpb" "github.com/apstndb/spanner-mycli/enums" "github.com/apstndb/spanner-mycli/internal/mycli/format" "github.com/apstndb/spanner-mycli/internal/mycli/streamio" - "google.golang.org/api/option" - "google.golang.org/grpc" - "google.golang.org/grpc/credentials/insecure" - "google.golang.org/grpc/test/bufconn" "google.golang.org/protobuf/types/known/structpb" ) @@ -112,48 +105,13 @@ func (s *streamWidthRPCServer) ExecuteStreamingSql(_ *sppb.ExecuteSqlRequest, st func newStreamWidthRPCSession(t *testing.T, value string) (*Session, *systemVariables) { t.Helper() - listener := bufconn.Listen(1 << 20) - grpcServer := grpc.NewServer() - sppb.RegisterSpannerServer(grpcServer, &streamWidthRPCServer{ + return newBufconnQuerySession(t, &streamWidthRPCServer{ queryCacheRPCServer: queryCacheRPCServer{ plan: testQueryPlan(t), stats: mustNewStruct(map[string]any{"elapsed_time": "1 msec", "query": "stream-width"}), }, value: value, }) - go func() { - if err := grpcServer.Serve(listener); err != nil && !errors.Is(err, grpc.ErrServerStopped) { - t.Errorf("serve: %v", err) - } - }() - t.Cleanup(func() { - grpcServer.Stop() - _ = listener.Close() - }) - conn, err := grpc.NewClient("passthrough:///stream-width", - grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) { return listener.Dial() }), - grpc.WithTransportCredentials(insecure.NewCredentials()), - ) - if err != nil { - t.Fatal(err) - } - t.Cleanup(func() { _ = conn.Close() }) - client, err := spanner.NewClientWithConfig(t.Context(), "projects/test/instances/test/databases/test", - spanner.ClientConfig{DisableNativeMetrics: true}, option.WithGRPCConn(conn)) - if err != nil { - t.Fatal(err) - } - t.Cleanup(client.Close) - - live := newSystemVariablesWithDefaultsForTest() - session := &Session{ - mode: DatabaseConnected, - client: client, - systemVariables: live, - txn: NewTransactionManager(client, live, spanner.ClientConfig{DisableNativeMetrics: true}), - } - live.inTransaction = session.txn.InTransaction - return session, live } func TestExecuteSQLStreamingTablePreservesValue(t *testing.T) { diff --git a/internal/mycli/string_quote_stream_test.go b/internal/mycli/string_quote_stream_test.go index f7143d1c..9c3ba84f 100644 --- a/internal/mycli/string_quote_stream_test.go +++ b/internal/mycli/string_quote_stream_test.go @@ -17,9 +17,7 @@ package mycli import ( "bytes" "context" - "errors" "io" - "net" "strconv" "strings" "testing" @@ -29,10 +27,6 @@ import ( "github.com/apstndb/spancodec" "github.com/apstndb/spanner-mycli/enums" "github.com/apstndb/spanner-mycli/internal/mycli/streamio" - "google.golang.org/api/option" - "google.golang.org/grpc" - "google.golang.org/grpc/credentials/insecure" - "google.golang.org/grpc/test/bufconn" "google.golang.org/protobuf/types/known/structpb" ) @@ -130,47 +124,12 @@ func (s *stringQuoteStreamRPCServer) ExecuteStreamingSql(_ *sppb.ExecuteSqlReque func newStringQuoteStreamRPCSession(t *testing.T) (*Session, *systemVariables) { t.Helper() - listener := bufconn.Listen(1 << 20) - grpcServer := grpc.NewServer() - sppb.RegisterSpannerServer(grpcServer, &stringQuoteStreamRPCServer{ + return newBufconnQuerySession(t, &stringQuoteStreamRPCServer{ queryCacheRPCServer: queryCacheRPCServer{ plan: testQueryPlan(t), stats: mustNewStruct(map[string]any{"elapsed_time": "1 msec", "query": "string-quote"}), }, }) - go func() { - if err := grpcServer.Serve(listener); err != nil && !errors.Is(err, grpc.ErrServerStopped) { - t.Errorf("serve: %v", err) - } - }() - t.Cleanup(func() { - grpcServer.Stop() - _ = listener.Close() - }) - conn, err := grpc.NewClient("passthrough:///string-quote", - grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) { return listener.Dial() }), - grpc.WithTransportCredentials(insecure.NewCredentials()), - ) - if err != nil { - t.Fatal(err) - } - t.Cleanup(func() { _ = conn.Close() }) - client, err := spanner.NewClientWithConfig(t.Context(), "projects/test/instances/test/databases/test", - spanner.ClientConfig{DisableNativeMetrics: true}, option.WithGRPCConn(conn)) - if err != nil { - t.Fatal(err) - } - t.Cleanup(client.Close) - - live := newSystemVariablesWithDefaultsForTest() - session := &Session{ - mode: DatabaseConnected, - client: client, - systemVariables: live, - txn: NewTransactionManager(client, live, spanner.ClientConfig{DisableNativeMetrics: true}), - } - live.inTransaction = session.txn.InTransaction - return session, live } func joinedStreamingQuoteText(out string) string { diff --git a/internal/mycli/sync_proto_bundle_test.go b/internal/mycli/sync_proto_bundle_test.go index fe3f7676..d9b268cc 100644 --- a/internal/mycli/sync_proto_bundle_test.go +++ b/internal/mycli/sync_proto_bundle_test.go @@ -17,7 +17,6 @@ package mycli import ( "context" "errors" - "net" "slices" "strings" "sync" @@ -25,14 +24,9 @@ import ( "time" "cloud.google.com/go/longrunning/autogen/longrunningpb" - adminapi "cloud.google.com/go/spanner/admin/database/apiv1" "cloud.google.com/go/spanner/admin/database/apiv1/databasepb" "github.com/google/go-cmp/cmp" - "google.golang.org/api/option" statuspb "google.golang.org/genproto/googleapis/rpc/status" - "google.golang.org/grpc" - "google.golang.org/grpc/credentials/insecure" - "google.golang.org/grpc/test/bufconn" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/descriptorpb" "google.golang.org/protobuf/types/known/anypb" @@ -796,33 +790,7 @@ func (s *protoBundleAdminServer) GetOperation(_ context.Context, req *longrunnin func newProtoBundleAdminSession(t *testing.T, schema *databasepb.GetDatabaseDdlResponse) (*Session, *protoBundleAdminServer) { t.Helper() server := &protoBundleAdminServer{schema: schema} - lis := bufconn.Listen(1 << 20) - gs := grpc.NewServer() - databasepb.RegisterDatabaseAdminServer(gs, server) - longrunningpb.RegisterOperationsServer(gs, server) - go func() { - if err := gs.Serve(lis); err != nil && !errors.Is(err, grpc.ErrServerStopped) { - t.Errorf("serve: %v", err) - } - }() - t.Cleanup(func() { - gs.Stop() - _ = lis.Close() - }) - conn, err := grpc.NewClient("passthrough:///proto-bundle", - grpc.WithContextDialer(func(context.Context, string) (net.Conn, error) { return lis.Dial() }), - grpc.WithTransportCredentials(insecure.NewCredentials()), - ) - if err != nil { - t.Fatal(err) - } - t.Cleanup(func() { _ = conn.Close() }) - admin, err := adminapi.NewDatabaseAdminClient(t.Context(), option.WithGRPCConn(conn)) - if err != nil { - t.Fatal(err) - } - t.Cleanup(func() { _ = admin.Close() }) session := newSessionForLocalVarTest(t) - session.adminClient = admin + session.adminClient = newBufconnAdminClient(t, server) return session, server }