ronantakizawa/codereview-bench huggingface dataset:
Repository: grpc/grpc-go
File: xds/internal/clients/grpctransport/grpc_transport_test.go (Go)

Code before:
		{
			name: "ServerConfigExtension Credentials is nil",
			serverCfg: clients.ServerConfig{
				ServerURI:  "server-address",
				Extensions: ServerConfigExtension{},
			},
			wantErr: true,
		},
		{
			name: "success",
			serverCfg: clients.ServerConfig{
				ServerURI:  "server-address",
				Extensions: ServerConfigExtension{Credentials: insecure.NewBundle()},
			},
			wantErr: false,
		},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			b := &Builder{}
			tr, err := b.Build(test.serverCfg)
			if (err != nil) != test.wantErr {
				t.Fatalf("Build() error = %v, wantErr %v", err, test.wantErr)
			}
			if tr != nil {
				defer tr.Close()
			}
			if !test.wantErr && tr == nil {
				t.Fatalf("got non-nil transport from Build(), want nil")
			}
		})
	}
}

// TestNewStream verifies that grpcTransport.NewStream() successfully creates a
// new client stream for the server.
func (s) TestNewStream(t *testing.T) {
	ts := setupTestServer(t)

	tests := []struct {
		name      string
		serverURI string
		wantErr   bool
	}{
		{
			name:      "success",
			serverURI: ts.lis.Addr().String(),
			wantErr:   false,
		},
		{
			name:      "error",

Review comment (suggestion):
I think this conditional only captures one case: the test does not expect an error, but transport is `nil`. We also want to catch the other case: the test expects an error, but transport is non-nil.

Code after:

// TestBuild_Failure verifies that the Builder returns error when incorrect
// ServerConfig is provided.
//
// It covers the following scenarios:
// - ServerURI is empty.
// - Extensions is nil.
// - Extensions is not ServerConfigExtension.
// - Credentials are nil.
func (s) TestBuild_Failure(t *testing.T) {
	tests := []struct {
		name      string
		serverCfg clients.ServerConfig
	}{
		{
			name: "ServerURI is empty",
			serverCfg: clients.ServerConfig{
				ServerURI:  "",
				Extensions: ServerConfigExtension{Credentials: insecure.NewBundle()},
			},
		},
		{
			name:      "Extensions is nil",
			serverCfg: clients.ServerConfig{ServerURI: "server-address"},
		},
		{
			name: "Extensions is not a ServerConfigExtension",
			serverCfg: clients.ServerConfig{
				ServerURI:  "server-address",
				Extensions: 1,
			},
		},
		{
			name: "ServerConfigExtension Credentials is nil",
			serverCfg: clients.ServerConfig{
				ServerURI:  "server-address",
				Extensions: ServerConfigExtension{},
			},
		},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			b := &Builder{}
			tr, err := b.Build(test.serverCfg)
			if err == nil {
				t.Fatalf("Build() succeeded, want error")
			}
			if tr != nil {
				t.Fatalf("Got non-nil transport from Build(), want nil")
			}
		})
Repository: grpc/grpc-go
File: xds/internal/clients/grpctransport/grpc_transport_test.go (Go)

Code before:
				Extensions: ServerConfigExtension{Credentials: insecure.NewBundle()},
			},
			wantErr: false,
		},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			b := &Builder{}
			tr, err := b.Build(test.serverCfg)
			if (err != nil) != test.wantErr {
				t.Fatalf("Build() error = %v, wantErr %v", err, test.wantErr)
			}
			if tr != nil {
				defer tr.Close()
			}
			if !test.wantErr && tr == nil {
				t.Fatalf("got non-nil transport from Build(), want nil")
			}
		})
	}
}

// TestNewStream verifies that grpcTransport.NewStream() successfully creates a
// new client stream for the server.
func (s) TestNewStream(t *testing.T) {
	ts := setupTestServer(t)

	tests := []struct {
		name      string
		serverURI string
		wantErr   bool
	}{
		{
			name:      "success",
			serverURI: ts.lis.Addr().String(),
			wantErr:   false,
		},
		{
			name:      "error",
			serverURI: "invalid-server-uri",
			wantErr:   true,
		},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			serverCfg := clients.ServerConfig{
				ServerURI:  test.serverURI,
				Extensions: ServerConfigExtension{Credentials: insecure.NewBundle()},
			}
			builder := Builder{}
			transport, err := builder.Build(serverCfg)

Review comment (refactor):
Please move this to be inside `t.Run`. As much as possible, sub-tests should not share state among each other.

Code after:
		serverCfg clients.ServerConfig
	}{
		{
			name: "ServerURI is empty",
			serverCfg: clients.ServerConfig{
				ServerURI:  "",
				Extensions: ServerConfigExtension{Credentials: insecure.NewBundle()},
			},
		},
		{
			name:      "Extensions is nil",
			serverCfg: clients.ServerConfig{ServerURI: "server-address"},
		},
		{
			name: "Extensions is not a ServerConfigExtension",
			serverCfg: clients.ServerConfig{
				ServerURI:  "server-address",
				Extensions: 1,
			},
		},
		{
			name: "ServerConfigExtension Credentials is nil",
			serverCfg: clients.ServerConfig{
				ServerURI:  "server-address",
				Extensions: ServerConfigExtension{},
			},
		},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			b := &Builder{}
			tr, err := b.Build(test.serverCfg)
			if err == nil {
				t.Fatalf("Build() succeeded, want error")
			}
			if tr != nil {
				t.Fatalf("Got non-nil transport from Build(), want nil")
			}
		})
	}
}

// TestNewStream_Success verifies that NewStream() successfully creates a new
// client stream for the server when provided a valid server URI.
func (s) TestNewStream_Success(t *testing.T) {
	ts := setupTestServer(t, &v3discoverypb.DiscoveryResponse{VersionInfo: "1"})

	serverCfg := clients.ServerConfig{
		ServerURI:  ts.address,
		Extensions: ServerConfigExtension{Credentials: insecure.NewBundle()},
	}
Repository: grpc/grpc-go
File: xds/internal/clients/grpctransport/grpc_transport_test.go (Go)

Code before:
			}
			if !test.wantErr && tr == nil {
				t.Fatalf("got non-nil transport from Build(), want nil")
			}
		})
	}
}

// TestNewStream verifies that grpcTransport.NewStream() successfully creates a
// new client stream for the server.
func (s) TestNewStream(t *testing.T) {
	ts := setupTestServer(t)

	tests := []struct {
		name      string
		serverURI string
		wantErr   bool
	}{
		{
			name:      "success",
			serverURI: ts.lis.Addr().String(),
			wantErr:   false,
		},
		{
			name:      "error",
			serverURI: "invalid-server-uri",
			wantErr:   true,
		},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			serverCfg := clients.ServerConfig{
				ServerURI:  test.serverURI,
				Extensions: ServerConfigExtension{Credentials: insecure.NewBundle()},
			}
			builder := Builder{}
			transport, err := builder.Build(serverCfg)
			if err != nil {
				t.Fatalf("failed to build transport: %v", err)
			}
			defer transport.Close()

			ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
			defer cancel()
			_, err = transport.NewStream(ctx, "/envoy.service.discovery.v3.AggregatedDiscoveryService/StreamAggregatedResources")
			if (err != nil) != test.wantErr {
				t.Fatalf("transport.NewStream() error = %v, wantErr %v", err, test.wantErr)
			}
		})
	}
}

Review comment (suggestion):
Hmm.. ok, i see what you are doing. Maybe you can just split it into two separate tests, each with no table. So, in the second test, you wont even start a server and instead you will pass an invalid server uri to `Build`.

Code after:
			name: "Extensions is not a ServerConfigExtension",
			serverCfg: clients.ServerConfig{
				ServerURI:  "server-address",
				Extensions: 1,
			},
		},
		{
			name: "ServerConfigExtension Credentials is nil",
			serverCfg: clients.ServerConfig{
				ServerURI:  "server-address",
				Extensions: ServerConfigExtension{},
			},
		},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			b := &Builder{}
			tr, err := b.Build(test.serverCfg)
			if err == nil {
				t.Fatalf("Build() succeeded, want error")
			}
			if tr != nil {
				t.Fatalf("Got non-nil transport from Build(), want nil")
			}
		})
	}
}

// TestNewStream_Success verifies that NewStream() successfully creates a new
// client stream for the server when provided a valid server URI.
func (s) TestNewStream_Success(t *testing.T) {
	ts := setupTestServer(t, &v3discoverypb.DiscoveryResponse{VersionInfo: "1"})

	serverCfg := clients.ServerConfig{
		ServerURI:  ts.address,
		Extensions: ServerConfigExtension{Credentials: insecure.NewBundle()},
	}
	builder := Builder{}
	transport, err := builder.Build(serverCfg)
	if err != nil {
		t.Fatalf("Failed to build transport: %v", err)
	}
	defer transport.Close()

	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
	defer cancel()
	if _, err = transport.NewStream(ctx, "/envoy.service.discovery.v3.AggregatedDiscoveryService/StreamAggregatedResources"); err != nil {
		t.Fatalf("transport.NewStream() failed: %v", err)
	}
}
Repository: grpc/grpc-go
File: xds/internal/clients/grpctransport/grpc_transport.go (Go)

Code before:
func (s *stream) Send(msg []byte) error {
	return s.stream.SendMsg(msg)
}

// Recv receives a message from the server.
func (s *stream) Recv() ([]byte, error) {
	var typedRes []byte

	if err := s.stream.RecvMsg(&typedRes); err != nil {
		return typedRes, err
	}
	return typedRes, nil
}

// Close closes the gRPC channel to the server.
func (g *grpcTransport) Close() error {
	return g.cc.Close()
}

type byteCodec struct{}

func (c *byteCodec) Marshal(v any) ([]byte, error) {
	if b, ok := v.([]byte); ok {
		return b, nil
	}
	return nil, fmt.Errorf("message is %T, but must be a byte slice ", v)
}

func (c *byteCodec) Unmarshal(data []byte, v any) error {
	if b, ok := v.(*[]byte); ok {
		*b = data
		return nil
	}
	return fmt.Errorf("target is %T, but must be a pointer to a byte slice", v)
}

func (c *byteCodec) Name() string {
	return "grpctransport.byteCodec"
}


Review comment (suggestion):
```suggestion
	return nil, fmt.Errorf("message is %T, but must be a byte slice", v)
```

or even

```suggestion
	return nil, fmt.Errorf("message is %T, but must be []byte", v)
```


Code after:
type stream struct {
	stream grpc.ClientStream
}

// Send sends a message to the server.
func (s *stream) Send(msg []byte) error {
	return s.stream.SendMsg(msg)
}

// Recv receives a message from the server.
func (s *stream) Recv() ([]byte, error) {
	var typedRes []byte

	if err := s.stream.RecvMsg(&typedRes); err != nil {
		return nil, err
	}
	return typedRes, nil
}

type byteCodec struct{}

func (c *byteCodec) Marshal(v any) ([]byte, error) {
	if b, ok := v.([]byte); ok {
		return b, nil
	}
	return nil, fmt.Errorf("grpctransport: message is %T, but must be a []byte", v)
}

func (c *byteCodec) Unmarshal(data []byte, v any) error {
	if b, ok := v.(*[]byte); ok {
		*b = data
		return nil
	}
	return fmt.Errorf("grpctransport: target is %T, but must be *[]byte", v)
}

func (c *byteCodec) Name() string {
	return "grpctransport.byteCodec"
}
Repository: grpc/grpc-go
File: xds/internal/clients/grpctransport/grpc_transport.go (Go)

Code before:
	if err := s.stream.RecvMsg(&typedRes); err != nil {
		return typedRes, err
	}
	return typedRes, nil
}

// Close closes the gRPC channel to the server.
func (g *grpcTransport) Close() error {
	return g.cc.Close()
}

type byteCodec struct{}

func (c *byteCodec) Marshal(v any) ([]byte, error) {
	if b, ok := v.([]byte); ok {
		return b, nil
	}
	return nil, fmt.Errorf("message is %T, but must be a byte slice ", v)
}

func (c *byteCodec) Unmarshal(data []byte, v any) error {
	if b, ok := v.(*[]byte); ok {
		*b = data
		return nil
	}
	return fmt.Errorf("target is %T, but must be a pointer to a byte slice", v)
}

func (c *byteCodec) Name() string {
	return "grpctransport.byteCodec"
}


Review comment (suggestion):
Maybe

```suggestion
	return fmt.Errorf("target is %T, but must be *[]byte", v)
```

Code after:

// Recv receives a message from the server.
func (s *stream) Recv() ([]byte, error) {
	var typedRes []byte

	if err := s.stream.RecvMsg(&typedRes); err != nil {
		return nil, err
	}
	return typedRes, nil
}

type byteCodec struct{}

func (c *byteCodec) Marshal(v any) ([]byte, error) {
	if b, ok := v.([]byte); ok {
		return b, nil
	}
	return nil, fmt.Errorf("grpctransport: message is %T, but must be a []byte", v)
}

func (c *byteCodec) Unmarshal(data []byte, v any) error {
	if b, ok := v.(*[]byte); ok {
		*b = data
		return nil
	}
	return fmt.Errorf("grpctransport: target is %T, but must be *[]byte", v)
}

func (c *byteCodec) Name() string {
	return "grpctransport.byteCodec"
}
Repository: grpc/grpc-go
File: internal/transport/http2_server.go (Go)

Code before:
			stat, ok := status.FromError(err)
			if !ok {
				stat = status.New(codes.PermissionDenied, err.Error())
			}
			t.controlBuf.put(&earlyAbortStream{
				httpStatus:     http.StatusOK,
				streamID:       s.id,
				contentSubtype: s.contentSubtype,
				status:         stat,
				rst:            !frame.StreamEnded(),
			})
			return nil
		}
	}
	t.activeStreams[streamID] = s
	if len(t.activeStreams) == 1 {
		t.idle = time.Time{}
	}
	// Start a timer to close the stream on reaching the deadline.
	if timeoutSet {
		timer := internal.TimeAfterFunc(timeout, func() {
			t.closeStream(s, true, http2.ErrCodeCancel, false)
		})
		s.deadlineTimerCancel = func() {
			timer.Stop()
		}
	}
	t.mu.Unlock()
	if channelz.IsOn() {
		t.channelz.SocketMetrics.StreamsStarted.Add(1)
		t.channelz.SocketMetrics.LastRemoteStreamCreatedTimestamp.Store(time.Now().UnixNano())
	}
	s.requestRead = func(n int) {
		t.adjustWindow(s, uint32(n))
	}
	s.ctxDone = s.ctx.Done()
	s.wq = newWriteQuota(defaultWriteQuota, s.ctxDone)
	s.trReader = &transportReader{
		reader: &recvBufferReader{
			ctx:     s.ctx,
			ctxDone: s.ctxDone,
			recv:    s.buf,
		},
		windowHandler: func(n int) {
			t.updateWindow(s, uint32(n))
		},
	}
	// Register the stream with loopy.
	t.controlBuf.put(&registerStream{
		streamID: s.id,
		wq:       s.wq,

Review comment (suggestion):
```suggestion
		s.deadlineTimerCancel = timer.Stop
```

Code after:
			stat, ok := status.FromError(err)
			if !ok {
				stat = status.New(codes.PermissionDenied, err.Error())
			}
			t.controlBuf.put(&earlyAbortStream{
				httpStatus:     http.StatusOK,
				streamID:       s.id,
				contentSubtype: s.contentSubtype,
				status:         stat,
				rst:            !frame.StreamEnded(),
			})
			return nil
		}
	}
	t.activeStreams[streamID] = s
	if len(t.activeStreams) == 1 {
		t.idle = time.Time{}
	}
	// Start a timer to close the stream on reaching the deadline.
	if timeoutSet {
		// We need to wait for s.cancel to be updated before calling
		// t.closeStream to avoid data races.
		cancelUpdated := make(chan struct{})
		timer := internal.TimeAfterFunc(timeout, func() {
			<-cancelUpdated
			t.closeStream(s, true, http2.ErrCodeCancel, false)
		})
		oldCancel := s.cancel
		s.cancel = func() {
			oldCancel()
			timer.Stop()
		}
		close(cancelUpdated)
	}
	t.mu.Unlock()
	if channelz.IsOn() {
		t.channelz.SocketMetrics.StreamsStarted.Add(1)
		t.channelz.SocketMetrics.LastRemoteStreamCreatedTimestamp.Store(time.Now().UnixNano())
	}
	s.requestRead = func(n int) {
		t.adjustWindow(s, uint32(n))
	}
	s.ctxDone = s.ctx.Done()
	s.wq = newWriteQuota(defaultWriteQuota, s.ctxDone)
	s.trReader = &transportReader{
		reader: &recvBufferReader{
			ctx:     s.ctx,
			ctxDone: s.ctxDone,
			recv:    s.buf,
		},
		windowHandler: func(n int) {
Repository: grpc/grpc-go
File: internal/transport/http2_server.go (Go)

Code before:
		return
	}

	if s.deadlineTimerCancel != nil {
		s.deadlineTimerCancel()
	}

	hdr.cleanup = &cleanupStream{
		streamID: s.id,
		rst:      rst,
		rstCode:  rstCode,
		onWrite: func() {
			t.deleteStream(s, eosReceived)
		},
	}
	t.controlBuf.put(hdr)
}

// closeStream clears the footprint of a stream when the stream is not needed any more.
func (t *http2Server) closeStream(s *ServerStream, rst bool, rstCode http2.ErrCode, eosReceived bool) {
	// In case stream sending and receiving are invoked in separate
	// goroutines (e.g., bi-directional streaming), cancel needs to be
	// called to interrupt the potential blocking on other goroutines.
	s.cancel()

	oldState := s.swapState(streamDone)
	if oldState == streamDone {
		return
	}
	if s.deadlineTimerCancel != nil {
		s.deadlineTimerCancel()
	}
	t.deleteStream(s, eosReceived)

	t.controlBuf.put(&cleanupStream{
		streamID: s.id,
		rst:      rst,
		rstCode:  rstCode,
		onWrite:  func() {},
	})
}

func (t *http2Server) Drain(debugData string) {
	t.mu.Lock()
	defer t.mu.Unlock()
	if t.drainEvent != nil {
		return
	}
	t.drainEvent = grpcsync.NewEvent()
	t.controlBuf.put(&goAway{code: http2.ErrCodeNo, debugData: []byte(debugData), headsUp: true})
}

Review comment (bug):
Was this a bug before?  Or did we have something that prevented it?  Or was the code tolerant of multiple calls?

Code after:
	oldState := s.swapState(streamDone)
	if oldState == streamDone {
		// If the stream was already done, return.
		return
	}

	hdr.cleanup = &cleanupStream{
		streamID: s.id,
		rst:      rst,
		rstCode:  rstCode,
		onWrite: func() {
			t.deleteStream(s, eosReceived)
		},
	}
	t.controlBuf.put(hdr)
}

// closeStream clears the footprint of a stream when the stream is not needed any more.
func (t *http2Server) closeStream(s *ServerStream, rst bool, rstCode http2.ErrCode, eosReceived bool) {
	// In case stream sending and receiving are invoked in separate
	// goroutines (e.g., bi-directional streaming), cancel needs to be
	// called to interrupt the potential blocking on other goroutines.
	s.cancel()

	oldState := s.swapState(streamDone)
	if oldState == streamDone {
		return
	}
	t.deleteStream(s, eosReceived)

	t.controlBuf.put(&cleanupStream{
		streamID: s.id,
		rst:      rst,
		rstCode:  rstCode,
		onWrite:  func() {},
	})
}

func (t *http2Server) Drain(debugData string) {
	t.mu.Lock()
	defer t.mu.Unlock()
	if t.drainEvent != nil {
		return
	}
	t.drainEvent = grpcsync.NewEvent()
	t.controlBuf.put(&goAway{code: http2.ErrCodeNo, debugData: []byte(debugData), headsUp: true})
}

var goAwayPing = &ping{data: [8]byte{1, 6, 1, 8, 0, 3, 3, 9}}

// Handles outgoing GoAway and returns true if loopy needs to put itself
Repository: grpc/grpc-go
File: test/end2end_test.go (Go)

Code before:
		t.Fatalf("%v.StreamingInputCall(_) = _, %v, want <nil>", tc, err)
	}
	payload, err := newPayload(testpb.PayloadType_COMPRESSABLE, 1)
	if err != nil {
		t.Fatal(err)
	}

	req := &testpb.StreamingInputCallRequest{
		Payload: payload,
	}
	// The 1st request should go through.
	if err := stream.Send(req); err != nil {
		t.Fatalf("%v.Send(%v) = %v, want <nil>", stream, req, err)
	}
	for {
		if err := stream.Send(req); err != io.EOF {
			continue
		}
		if _, err := stream.CloseAndRecv(); status.Code(err) != codes.NotFound {
			t.Fatalf("%v.CloseAndRecv() = %v, want error %s", stream, err, codes.NotFound)
		}
		break
	}
}

func (s) TestClientStreamingMissingSendAndCloseError(t *testing.T) {
	// TODO : https://github.com/grpc/grpc-go/issues/8119 - remove `t.Skip()`
	// after this is fixed.
	t.Skip()
	testClientStreamingMissingSendAndCloseError(t)
}

func testClientStreamingMissingSendAndCloseError(t *testing.T) {
	ss := &stubserver.StubServer{
		StreamingInputCallF: func(_ testgrpc.TestService_StreamingInputCallServer) error {
			return nil
		},
	}
	if err := ss.Start(nil); err != nil {
		t.Fatalf("Error starting endpoint server: %v", err)
	}
	defer ss.Stop()

	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
	defer cancel()

	stream, err := ss.Client.StreamingInputCall(ctx)
	if err != nil {
		t.Fatalf(".StreamingInputCall(_) = _, %v, want <nil>", err)
	}


Review comment (suggestion):
suggestion for name to be more explicit `TestClientStreamingCardinalityViolation_ServerHandlerMissingSendAndClose`

Code after:
		t.Fatalf("%v.StreamingInputCall(_) = _, %v, want <nil>", tc, err)
	}
	payload, err := newPayload(testpb.PayloadType_COMPRESSABLE, 1)
	if err != nil {
		t.Fatal(err)
	}

	req := &testpb.StreamingInputCallRequest{
		Payload: payload,
	}
	// The 1st request should go through.
	if err := stream.Send(req); err != nil {
		t.Fatalf("%v.Send(%v) = %v, want <nil>", stream, req, err)
	}
	for {
		if err := stream.Send(req); err != io.EOF {
			continue
		}
		if _, err := stream.CloseAndRecv(); status.Code(err) != codes.NotFound {
			t.Fatalf("%v.CloseAndRecv() = %v, want error %s", stream, err, codes.NotFound)
		}
		break
	}
}

// Tests that a client receives a cardinality violation error for client-streaming
// RPCs if the server doesn't send a message before returning status OK.
func (s) TestClientStreamingCardinalityViolation_ServerHandlerMissingSendAndClose(t *testing.T) {
	// TODO : https://github.com/grpc/grpc-go/issues/8119 - remove `t.Skip()`
	// after this is fixed.
	t.Skip()
	ss := &stubserver.StubServer{
		StreamingInputCallF: func(_ testgrpc.TestService_StreamingInputCallServer) error {
			// Returning status OK without sending a response message.This is a
			// cardinality violation.
			return nil
		},
	}
	if err := ss.Start(nil); err != nil {
		t.Fatalf("Error starting endpoint server: %v", err)
	}
	defer ss.Stop()

	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
	defer cancel()

	stream, err := ss.Client.StreamingInputCall(ctx)
	if err != nil {
		t.Fatalf(".StreamingInputCall(_) = _, %v, want <nil>", err)
	}
Repository: grpc/grpc-go
File: test/end2end_test.go (Go)

Code before:
			return status.Errorf(codes.Unimplemented, "error for testing")
		},
	}
	if err := ss.Start(nil); err != nil {
		t.Fatalf("Error starting endpoint server: %v", err)
	}
	defer ss.Stop()

	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
	defer cancel()

	stream, err := ss.Client.StreamingInputCall(ctx)
	if err != nil {
		t.Fatalf("StreamingInputCall(_) = _, %v, want <nil>", err)
	}
	payload, err := newPayload(testpb.PayloadType_COMPRESSABLE, 1)
	if err != nil {
		t.Fatal(err)
	}

	req := &testpb.StreamingInputCallRequest{
		Payload: payload,
	}

	for {
		if err = stream.Send(req); err == io.EOF {
			break
		}
		select {
		case <-ctx.Done():
			t.Fatal("timed out waiting for error from server")
		default:
		}
	}
	if _, err = stream.CloseAndRecv(); err != nil {
		t.Fatalf("stream.CloseAndRecv() = %v, want nil", err)
	}
}

func (s) TestExceedMaxStreamsLimit(t *testing.T) {
	for _, e := range listTestEnv() {
		testExceedMaxStreamsLimit(t, e)
	}
}

func testExceedMaxStreamsLimit(t *testing.T, e env) {
	te := newTest(t, e)
	te.declareLogNoise(
		"http2Client.notifyError got notified that the client transport was broken",
		"Conn.resetTransport failed to create client transport",
		"grpc: the connection is closing",

Review comment (suggestion):
same. t.fatalf if any other error apart from `io.EOF`

Code after:
	close(waitForReturn)

	// Recv() should return after the server handler returns.
	select {
	case err := <-errCh:
		if err != nil {
			t.Fatalf("stream.RecvMsg(_) = %v, want <nil>", err)
		}
	case <-time.After(defaultTestTimeout):
		t.Fatal("Timed out waiting for stream.RecvMsg(_) to return")
	}
}

// Tests the behavior where server handler returns an error after calling
// SendAndClose. It verifies the that client receives nil message and
// non-nil error.
func (s) TestClientStreaming_ReturnErrorAfterSendAndClose(t *testing.T) {
	wantError := status.Error(codes.Internal, "error for testing")
	ss := stubserver.StubServer{
		StreamingInputCallF: func(stream testgrpc.TestService_StreamingInputCallServer) error {
			if err := stream.SendAndClose(&testpb.StreamingInputCallResponse{}); err != nil {
				t.Errorf("stream.SendAndClose(_) = %v, want <nil>", err)
			}
			return wantError
		},
	}
	if err := ss.Start(nil); err != nil {
		t.Fatal("Error starting server:", err)
	}
	defer ss.Stop()

	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
	defer cancel()
	stream, err := ss.Client.StreamingInputCall(ctx)
	if err != nil {
		t.Fatalf(".StreamingInputCall(_) = _, %v, want <nil>", err)
	}
	resp, err := stream.CloseAndRecv()

	wantStatus, _ := status.FromError(wantError)
	gotStatus, _ := status.FromError(err)

	if gotStatus.Code() != wantStatus.Code() || gotStatus.Message() != wantStatus.Message() || resp != nil {
		t.Fatalf("stream.CloseSend() = %v , %v, want <nil>, %s", resp, err, wantError)
	}
}

func (s) TestExceedMaxStreamsLimit(t *testing.T) {
	for _, e := range listTestEnv() {
		testExceedMaxStreamsLimit(t, e)
	}