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

Code before:

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)
	}

	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()


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

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

	_, 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)