ronantakizawa/codereview-bench huggingface dataset:
Repository: grpc/grpc-go
File: test/end2end_test.go (Go)

Code before:

	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
		}
		if err != nil {
			t.Fatalf("Stream.Send(req) = %v, want <nil>", err)
		}
		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",
	)
	te.maxStream = 1 // Only allows 1 live stream per server transport.
	te.startServer(&testServer{security: e.security})
	defer te.tearDown()

	cc := te.clientConn()
	tc := testgrpc.NewTestServiceClient(cc)

Review comment (suggestion):
This can be replaced by doing:

```go
for ctx.Err() == nil {
	if err := stream.Send()....
}
```

Code after:
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)
	}
}

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",
	)
	te.maxStream = 1 // Only allows 1 live stream per server transport.
	te.startServer(&testServer{security: e.security})
	defer te.tearDown()

	cc := te.clientConn()
	tc := testgrpc.NewTestServiceClient(cc)
Repository: grpc/grpc-go
File: scripts/vet.sh (Shell)

Code before:
Metadata is deprecated: use Attributes instead.
NewSubConn is deprecated:
OverrideServerName is deprecated:
RemoveSubConn is deprecated:
SecurityVersion is deprecated:
Target is deprecated: Use the Target field in the BuildOptions instead.
UpdateAddresses is deprecated:
UpdateSubConnState is deprecated:
balancer.ErrTransientFailure is deprecated:
grpc/reflection/v1alpha/reflection.proto
SwitchTo is deprecated:
XXXXX xDS deprecated fields we support
.ExactMatch
.PrefixMatch
.SafeRegexMatch
.SuffixMatch
GetContainsMatch
GetExactMatch
GetMatchSubjectAltNames
GetPrefixMatch
GetSafeRegexMatch
GetSuffixMatch
GetTlsCertificateCertificateProviderInstance
GetValidationContextCertificateProviderInstance
XXXXX PleaseIgnoreUnused
NewAddress is deprecated:'
  popd
done

echo SUCCESS


Review comment (refactor):
```suggestion
NewAddress is deprecated:
XXXXX PleaseIgnoreUnused'
```

That last line is there to avoid need to move the `'` around.  Also, let's try to keep this list sorted, so actually, please move this to right before `NewSubConn is deprecated:`.

Code after:
Metadata is deprecated: use Attributes instead.
NewAddress is deprecated:
NewSubConn is deprecated:
OverrideServerName is deprecated:
RemoveSubConn is deprecated:
SecurityVersion is deprecated:
Target is deprecated: Use the Target field in the BuildOptions instead.
UpdateAddresses is deprecated:
UpdateSubConnState is deprecated:
balancer.ErrTransientFailure is deprecated:
grpc/reflection/v1alpha/reflection.proto
SwitchTo is deprecated:
XXXXX xDS deprecated fields we support
.ExactMatch
.PrefixMatch
.SafeRegexMatch
.SuffixMatch
GetContainsMatch
GetExactMatch
GetMatchSubjectAltNames
GetPrefixMatch
GetSafeRegexMatch
GetSuffixMatch
GetTlsCertificateCertificateProviderInstance
GetValidationContextCertificateProviderInstance
XXXXX PleaseIgnoreUnused'
  popd
done

echo SUCCESS
Repository: grpc/grpc-go
File: resolver/manual/manual.go (Go)

Code before:
	return r, nil
}

// Scheme returns the manual resolver's scheme.
func (r *Resolver) Scheme() string {
	return r.scheme
}

// ResolveNow is a noop for Resolver.
func (r *Resolver) ResolveNow(o resolver.ResolveNowOptions) {
	r.ResolveNowCallback(o)
}

// Close is a noop for Resolver.
func (r *Resolver) Close() {
	r.CloseCallback()
}

// UpdateState calls CC.UpdateState.
func (r *Resolver) UpdateState(s resolver.State) {
	r.mu.Lock()
	defer r.mu.Unlock()
	var err error
	if r.CC == nil {
		panic("cannot update state as channel has not exited IDLE state")
	}
	err = r.CC.UpdateState(s)
	r.lastSeenState = &s
	r.UpdateStateCallback(err)
}

// ReportError calls CC.ReportError.
func (r *Resolver) ReportError(err error) {
	r.mu.Lock()
	defer r.mu.Unlock()
	if r.CC == nil {
		panic("cannot report error as channel has not exited IDLE state")
	}
	r.CC.ReportError(err)
}


Review comment (suggestion):
That's clever, even if we call `UpdateState` before creating the resolver, it saves the state for later and returns. The saved state gets used when the resolver is built. 

Does this make `InitialState` redundant? IIUC every caller of `InitialState` should now be able to call `UpdateState`.

Code after:
	return r, nil
}

// Scheme returns the manual resolver's scheme.
func (r *Resolver) Scheme() string {
	return r.scheme
}

// ResolveNow is a noop for Resolver.
func (r *Resolver) ResolveNow(o resolver.ResolveNowOptions) {
	r.ResolveNowCallback(o)
}

// Close is a noop for Resolver.
func (r *Resolver) Close() {
	r.CloseCallback()
}

// UpdateState calls UpdateState(s) on the channel.  If the resolver has not
// been Built before, this instead sets the initial state of the resolver, like
// InitialState.
func (r *Resolver) UpdateState(s resolver.State) {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.lastSeenState = &s
	if r.cc == nil {
		return
	}
	err := r.cc.UpdateState(s)
	r.UpdateStateCallback(err)
}

// CC returns r's ClientConn when r was last Built.  Panics if the resolver has
// not been Built before.
func (r *Resolver) CC() resolver.ClientConn {
	r.mu.Lock()
	defer r.mu.Unlock()
	if r.cc == nil {
		panic("Manual resolver instance has not yet been built.")
	}
	return r.cc
}
Repository: grpc/grpc-go
File: health/server_internal_test.go (Go)

Code before:
		if !ok {
			t.Fatalf("List should have returned all resources, wanted %s, but it did not exist in the inner status map", key)
		}
		if v != healthpb.HealthCheckResponse_SERVING {
			t.Fatalf("%s returned the wrong status, wanted %d, got %d", key, healthpb.HealthCheckResponse_SERVING, v)
		}
	}
}

// TestListResourceExhausted verifies that the service status list returns an error when it exceeds
// maxServiceStatusListLength.
func (s) TestListResourceExhausted(t *testing.T) {
	// Setup
	s := NewServer()
	s.mu.Lock()
	// Remove the zero value
	delete(s.statusMap, "")

	// Fill out status map with service information, 101 elements will trigger an error.
	for i := 1; i <= 101; i++ {
		s.statusMap[fmt.Sprintf("%d", i)] = healthpb.HealthCheckResponse_SERVING
	}
	s.mu.Unlock()

	// Execution
	ctx := context.TODO()
	var in healthpb.HealthListRequest
	_, err := s.List(ctx, &in)

	// Assertions
	if err == nil {
		t.Fatalf("List should have failed, got %s", err)
	}
	if !errors.Is(err, status.Error(codes.ResourceExhausted, "server health list exceeds maximum capacity (100)")) {
		t.Fatal("List should have failed with resource exhausted")
	}
}


Review comment (suggestion):
Please cancelable context either `WithCancel` or `WIthTimout`

```
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
```

Code after:
	want := &healthpb.HealthListResponse{
		Statuses: map[string]*healthpb.HealthCheckResponse{
			"":  {Status: healthpb.HealthCheckResponse_SERVING},
			"0": {Status: healthpb.HealthCheckResponse_UNKNOWN},
			"1": {Status: healthpb.HealthCheckResponse_SERVING},
			"2": {Status: healthpb.HealthCheckResponse_NOT_SERVING},
		},
	}
	if diff := cmp.Diff(got, want, protocmp.Transform()); diff != "" {
		t.Fatalf("Health response did not match expectation.  Diff (-got, +want): %s", diff)
	}
}

// TestListResourceExhausted verifies that List()
// returns a ResourceExhausted error if no. of services are more than
// maxAllowedServices.
func (s) TestListResourceExhausted(t *testing.T) {
	s := NewServer()

	// Fill out status map with service information,
	// 101 (100 + 1 existing) elements will trigger an error.
	for i := 1; i <= maxAllowedServices; i++ {
		s.SetServingStatus(fmt.Sprintf("%d", i),
			healthpb.HealthCheckResponse_SERVING)
	}

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	var in healthpb.HealthListRequest
	_, err := s.List(ctx, &in)

	want := status.Errorf(codes.ResourceExhausted,
		"server health list exceeds maximum capacity: %d", maxAllowedServices)
	if !errors.Is(err, want) {
		t.Fatalf("s.List(ctx, &in) returned %v, want %v", err, want)
	}
}
Repository: grpc/grpc-go
File: health/server_internal_test.go (Go)

Code before:
	}
}

// TestListResourceExhausted verifies that the service status list returns an error when it exceeds
// maxServiceStatusListLength.
func (s) TestListResourceExhausted(t *testing.T) {
	// Setup
	s := NewServer()
	s.mu.Lock()
	// Remove the zero value
	delete(s.statusMap, "")

	// Fill out status map with service information, 101 elements will trigger an error.
	for i := 1; i <= 101; i++ {
		s.statusMap[fmt.Sprintf("%d", i)] = healthpb.HealthCheckResponse_SERVING
	}
	s.mu.Unlock()

	// Execution
	ctx := context.TODO()
	var in healthpb.HealthListRequest
	_, err := s.List(ctx, &in)

	// Assertions
	if err == nil {
		t.Fatalf("List should have failed, got %s", err)
	}
	if !errors.Is(err, status.Error(codes.ResourceExhausted, "server health list exceeds maximum capacity (100)")) {
		t.Fatal("List should have failed with resource exhausted")
	}
}


Review comment (suggestion):
```suggestion
		t.Fatalf("s.List(ctx, &in) return nil error, want non-nil")
```

Code after:
		},
	}
	if diff := cmp.Diff(got, want, protocmp.Transform()); diff != "" {
		t.Fatalf("Health response did not match expectation.  Diff (-got, +want): %s", diff)
	}
}

// TestListResourceExhausted verifies that List()
// returns a ResourceExhausted error if no. of services are more than
// maxAllowedServices.
func (s) TestListResourceExhausted(t *testing.T) {
	s := NewServer()

	// Fill out status map with service information,
	// 101 (100 + 1 existing) elements will trigger an error.
	for i := 1; i <= maxAllowedServices; i++ {
		s.SetServingStatus(fmt.Sprintf("%d", i),
			healthpb.HealthCheckResponse_SERVING)
	}

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	var in healthpb.HealthListRequest
	_, err := s.List(ctx, &in)

	want := status.Errorf(codes.ResourceExhausted,
		"server health list exceeds maximum capacity: %d", maxAllowedServices)
	if !errors.Is(err, want) {
		t.Fatalf("s.List(ctx, &in) returned %v, want %v", err, want)
	}
}
Repository: grpc/grpc-go
File: health/server_internal_test.go (Go)

Code before:
// TestListResourceExhausted verifies that the service status list returns an error when it exceeds
// maxAllowedServices.
func (s) TestListResourceExhausted(t *testing.T) {
	// Setup
	s := NewServer()
	s.mu.Lock()
	// Remove the zero value
	delete(s.statusMap, "")

	// Fill out status map with service information, 101 elements will trigger an error.
	for i := 1; i <= maxAllowedServices+1; i++ {
		s.statusMap[fmt.Sprintf("%d", i)] = healthpb.HealthCheckResponse_SERVING
	}
	s.mu.Unlock()

	// Execution
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	var in healthpb.HealthListRequest
	_, err := s.List(ctx, &in)

	// Assertions
	if err == nil {
		t.Fatalf("s.List(ctx, &in) return nil error, want non-nil")
	}
	if !errors.Is(err, status.Error(codes.ResourceExhausted, "server health list exceeds maximum capacity (100)")) {
		t.Fatal("List should have failed with resource exhausted")
	}
}


Review comment (suggestion):
```suggestion
	if !errors.Is(err, status.Error(codes.ResourceExhausted, fmt.Sprintf("server health list exceeds maximum capacity: %d", maxAllowedServices))) {
```

Code after:
	}
}

// TestListResourceExhausted verifies that List()
// returns a ResourceExhausted error if no. of services are more than
// maxAllowedServices.
func (s) TestListResourceExhausted(t *testing.T) {
	s := NewServer()

	// Fill out status map with service information,
	// 101 (100 + 1 existing) elements will trigger an error.
	for i := 1; i <= maxAllowedServices; i++ {
		s.SetServingStatus(fmt.Sprintf("%d", i),
			healthpb.HealthCheckResponse_SERVING)
	}

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	var in healthpb.HealthListRequest
	_, err := s.List(ctx, &in)

	want := status.Errorf(codes.ResourceExhausted,
		"server health list exceeds maximum capacity: %d", maxAllowedServices)
	if !errors.Is(err, want) {
		t.Fatalf("s.List(ctx, &in) returned %v, want %v", err, want)
	}
}
Repository: grpc/grpc-go
File: health/server_internal_test.go (Go)

Code before:
}

// TestListResourceExhausted verifies that List() returns a ResourceExhausted error if no. of services are more than
// maxAllowedServices.
func (s) TestListResourceExhausted(t *testing.T) {
	s := NewServer()
	s.mu.Lock()
	// Remove the zero value
	delete(s.statusMap, "")

	// Fill out status map with service information, 101 elements will trigger an error.
	for i := 1; i <= maxAllowedServices+1; i++ {
		s.statusMap[fmt.Sprintf("%d", i)] = healthpb.HealthCheckResponse_SERVING
	}
	s.mu.Unlock()

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	var in healthpb.HealthListRequest
	_, err := s.List(ctx, &in)

	if err == nil {
		t.Fatalf("s.List(ctx, &in) return nil error, want non-nil")
	}
	if !errors.Is(err, status.Errorf(codes.ResourceExhausted, "server health list exceeds maximum capacity: %d", maxAllowedServices)) {
		t.Fatal("List should have failed with resource exhausted")
	}
}


Review comment (suggestion):
```suggestion
		t.Fatal("s.List() = %v, want %v, err, status.Errorf(codes.ResourceExhausted, "server health list exceeds maximum capacity: %d", maxAllowedServices)")
```

Code after:
		},
	}
	if diff := cmp.Diff(got, want, protocmp.Transform()); diff != "" {
		t.Fatalf("Health response did not match expectation.  Diff (-got, +want): %s", diff)
	}
}

// TestListResourceExhausted verifies that List()
// returns a ResourceExhausted error if no. of services are more than
// maxAllowedServices.
func (s) TestListResourceExhausted(t *testing.T) {
	s := NewServer()

	// Fill out status map with service information,
	// 101 (100 + 1 existing) elements will trigger an error.
	for i := 1; i <= maxAllowedServices; i++ {
		s.SetServingStatus(fmt.Sprintf("%d", i),
			healthpb.HealthCheckResponse_SERVING)
	}

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	var in healthpb.HealthListRequest
	_, err := s.List(ctx, &in)

	want := status.Errorf(codes.ResourceExhausted,
		"server health list exceeds maximum capacity: %d", maxAllowedServices)
	if !errors.Is(err, want) {
		t.Fatalf("s.List(ctx, &in) returned %v, want %v", err, want)
	}
}
Repository: grpc/grpc-go
File: stats/stats_test.go (Go)

Code before:
	if te.compress == "gzip" {
		opts = append(opts,
			grpc.RPCCompressor(grpc.NewGZIPCompressor()),
			grpc.RPCDecompressor(grpc.NewGZIPDecompressor()),
		)
	}
	for _, sh := range te.serverStatsHandlers {
		opts = append(opts, grpc.StatsHandler(sh))
	}
	s := grpc.NewServer(opts...)
	te.srv = s
	if te.testServer != nil {
		testgrpc.RegisterTestServiceServer(s, te.testServer)
	}

	go s.Serve(lis)
	te.srvAddr = lis.Addr().String()
}

func (te *test) clientConn() *grpc.ClientConn {
	if te.cc != nil {
		return te.cc
	}
	opts := []grpc.DialOption{
		grpc.WithTransportCredentials(insecure.NewCredentials()),
		grpc.WithUserAgent("test/0.0.1"),
	}
	if te.compress == "gzip" {
		opts = append(opts,
			grpc.WithCompressor(grpc.NewGZIPCompressor()),
			grpc.WithDecompressor(grpc.NewGZIPDecompressor()),
		)
	}
	for _, sh := range te.clientStatsHandlers {
		opts = append(opts, grpc.WithStatsHandler(sh))
	}

	var err error
	te.cc, err = grpc.NewClient(te.srvAddr, opts...)
	if err != nil {
		te.t.Fatalf("grpc.NewClient(%q) = %v", te.srvAddr, err)
	}
	return te.cc
}

type rpcType int

const (
	unaryRPC rpcType = iota
	clientStreamRPC
	serverStreamRPC

Review comment (bug):
If you are removing `WithBlock`, you need to ensure the clientconn is ready before the function returns it. I ssupect this is the reason for the race conditions that you're seeing. Please use cc.Connect and `testutils.AwaitState` here to restore the behviour.

Code after:
	var opts []grpc.ServerOption
	if te.compress == "gzip" {
		opts = append(opts,
			grpc.RPCCompressor(grpc.NewGZIPCompressor()),
			grpc.RPCDecompressor(grpc.NewGZIPDecompressor()),
		)
	}
	for _, sh := range te.serverStatsHandlers {
		opts = append(opts, grpc.StatsHandler(sh))
	}
	s := grpc.NewServer(opts...)
	te.srv = s
	if te.testServer != nil {
		testgrpc.RegisterTestServiceServer(s, te.testServer)
	}

	go s.Serve(lis)
	te.srvAddr = lis.Addr().String()
}

func (te *test) clientConn(ctx context.Context) *grpc.ClientConn {
	if te.cc != nil {
		return te.cc
	}
	opts := []grpc.DialOption{
		grpc.WithTransportCredentials(insecure.NewCredentials()),
		grpc.WithUserAgent("test/0.0.1"),
	}
	if te.compress == "gzip" {
		opts = append(opts,
			grpc.WithCompressor(grpc.NewGZIPCompressor()),
			grpc.WithDecompressor(grpc.NewGZIPDecompressor()),
		)
	}
	for _, sh := range te.clientStatsHandlers {
		opts = append(opts, grpc.WithStatsHandler(sh))
	}

	var err error
	te.cc, err = grpc.NewClient(te.srvAddr, opts...)
	if err != nil {
		te.t.Fatalf("grpc.NewClient(%q) failed: %v", te.srvAddr, err)
	}
	te.cc.Connect()
	testutils.AwaitState(ctx, te.t, te.cc, connectivity.Ready)
	return te.cc
}

type rpcType int

const (
Repository: grpc/grpc-go
File: stats/stats_test.go (Go)

Code before:
func (te *test) clientConn() *grpc.ClientConn {
	if te.cc != nil {
		return te.cc
	}
	opts := []grpc.DialOption{
		grpc.WithTransportCredentials(insecure.NewCredentials()),
		grpc.WithUserAgent("test/0.0.1"),
	}
	if te.compress == "gzip" {
		opts = append(opts,
			grpc.WithCompressor(grpc.NewGZIPCompressor()),
			grpc.WithDecompressor(grpc.NewGZIPDecompressor()),
		)
	}
	for _, sh := range te.clientStatsHandlers {
		opts = append(opts, grpc.WithStatsHandler(sh))
	}

	var err error
	te.cc, err = grpc.NewClient(te.srvAddr, opts...)
	if err != nil {
		te.t.Fatalf("grpc.NewClient() failed(%q) = %v", te.srvAddr, err)
	}
	te.cc.Connect()
	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
	defer cancel()
	testutils.AwaitState(ctx, te.t, te.cc, connectivity.Ready)
	return te.cc
}

type rpcType int

const (
	unaryRPC rpcType = iota
	clientStreamRPC
	serverStreamRPC
	fullDuplexStreamRPC
)

type rpcConfig struct {
	count    int  // Number of requests and responses for streaming RPCs.
	success  bool // Whether the RPC should succeed or return error.
	failfast bool
	callType rpcType // Type of RPC.
}

func (te *test) doUnaryCall(c *rpcConfig) (*testpb.SimpleRequest, *testpb.SimpleResponse, error) {
	var (
		resp *testpb.SimpleResponse
		req  *testpb.SimpleRequest
		err  error

Review comment (suggestion):
We should not create a context in this method, instead pass a context created in the original test function. This ensures the entire test has 10 secs to run. Presently if a test created channels in a for loop, the overall test timeout will get multiplied.

Code after:
func (te *test) clientConn(ctx context.Context) *grpc.ClientConn {
	if te.cc != nil {
		return te.cc
	}
	opts := []grpc.DialOption{
		grpc.WithTransportCredentials(insecure.NewCredentials()),
		grpc.WithUserAgent("test/0.0.1"),
	}
	if te.compress == "gzip" {
		opts = append(opts,
			grpc.WithCompressor(grpc.NewGZIPCompressor()),
			grpc.WithDecompressor(grpc.NewGZIPDecompressor()),
		)
	}
	for _, sh := range te.clientStatsHandlers {
		opts = append(opts, grpc.WithStatsHandler(sh))
	}

	var err error
	te.cc, err = grpc.NewClient(te.srvAddr, opts...)
	if err != nil {
		te.t.Fatalf("grpc.NewClient(%q) failed: %v", te.srvAddr, err)
	}
	te.cc.Connect()
	testutils.AwaitState(ctx, te.t, te.cc, connectivity.Ready)
	return te.cc
}

type rpcType int

const (
	unaryRPC rpcType = iota
	clientStreamRPC
	serverStreamRPC
	fullDuplexStreamRPC
)

type rpcConfig struct {
	count    int  // Number of requests and responses for streaming RPCs.
	success  bool // Whether the RPC should succeed or return error.
	failfast bool
	callType rpcType // Type of RPC.
}

func (te *test) doUnaryCall(c *rpcConfig) (*testpb.SimpleRequest, *testpb.SimpleResponse, error) {
	var (
		resp *testpb.SimpleResponse
		req  *testpb.SimpleRequest
		err  error
	)
	tCtx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)