Repository: grpc/grpc-go
File: test/end2end_test.go (Go)

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

	_, err = stream.CloseAndRecv()
	if err == nil {
		t.Fatalf("%v.CloseAndRecv() = %v, want an error", stream, err)
	}
	if status.Code(err) != codes.Internal {
		t.Fatalf("%v.CloseAndRecv() = %v, want error %s", stream, err, codes.Internal)
	}
}

// Tests the scenario where a client-streaming server calls `Recv()` after calling
// `SendAndClose()`. And error is expected if `Recv()` is called after calling
// SendAndClose()` i.e the RPC should be terminated.
func (s) TestClientStreamingCardinalityViolation_ServerHandlerRecvAfterSendAndClose(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(stream testgrpc.TestService_StreamingInputCallServer) error {
			sum := 0
			in, _ := stream.Recv()
			p := in.GetPayload().GetBody()
			sum += len(p)
			stream.SendAndClose(&testpb.StreamingInputCallResponse{
				AggregatedPayloadSize: int32(sum),
			})
			_, err := stream.Recv()
			return err
		},
	}
	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)

Review comment (bug):

```suggestion
// Test to verify for client-streaming RPCs, when SendAndClose is called, the server 
// should reset stream after sending the response message successfully.
// If server calls Recv after, those operations should fail.
```

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

	_, err = stream.CloseAndRecv()
	if err == nil {
		t.Fatalf("stream.CloseAndRecv() = %v, want an error", err)
	}
	if status.Code(err) != codes.Internal {
		t.Fatalf("stream.CloseAndRecv() = %v, want error %s", err, codes.Internal)
	}
}

// Tests that the server can continue to receive messages after calling SendAndClose. Although
// this is unexpected, we retain it for backward compatibility.
func (s) TestClientStreaming_ServerHandlerRecvAfterSendAndClose(t *testing.T) {
	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)
			}
			if resp, err := stream.Recv(); err != nil || !proto.Equal(resp, &testpb.StreamingInputCallRequest{}) {
				t.Errorf("stream.Recv() = %s, %v, want non-nil empty response, <nil>", resp, err)
			}
			return nil
		},
	}
	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)
	}
	if err := stream.Send(&testpb.StreamingInputCallRequest{}); err != nil {