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

Code before:
		t.Fatalf(".StreamingInputCall(_) = _, %v, want <nil>", err)
	}

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

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

func testClientStreamingRecvAfterCloseError(t *testing.T) {
	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)
	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 {

Review comment (bug):
nit: do we need to do this? What if just return a static error with status code like `status.Error(codes.Internal, "something went wrong")).
`? Because calling stream.Recv() adding one more possibility to the test case

Code after:
		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 {
		t.Fatalf("stream.Send(_) = %v, want <nil>", err)
	}
	if resp, err := stream.CloseAndRecv(); err != nil || !proto.Equal(resp, &testpb.StreamingInputCallResponse{}) {
		t.Fatalf("stream.CloseSend() = %v , %v, want non-nil empty response, <nil>", resp, err)
	}
}

// Tests that Recv() on client streaming client blocks till the server handler
// returns even after calling SendAndClose from the server handler.
func (s) TestClientStreaming_RecvWaitsForServerHandlerRetrun(t *testing.T) {
	waitForReturn := make(chan struct{})
	ss := stubserver.StubServer{
		StreamingInputCallF: func(stream testgrpc.TestService_StreamingInputCallServer) error {