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

Code before:
		t.Fatalf("stream.Send(_) = %v, want <nil>", err)
	}
	if _, err := stream.CloseAndRecv(); status.Code(err) != codes.Internal {
		t.Fatalf("stream.CloseAndRecv() = %v, want error with status code %s", err, codes.Internal)
	}
}

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)

	_, err := tc.StreamingInputCall(te.ctx)
	if err != nil {
		t.Fatalf("%v.StreamingInputCall(_) = _, %v, want _, <nil>", tc, err)
	}
	// Loop until receiving the new max stream setting from the server.
	for {
		ctx, cancel := context.WithTimeout(context.Background(), defaultTestShortTimeout)
		defer cancel()
		_, err := tc.StreamingInputCall(ctx)
		if err == nil {
			time.Sleep(50 * time.Millisecond)
			continue
		}
		if status.Code(err) == codes.DeadlineExceeded {
			break
		}
		t.Fatalf("%v.StreamingInputCall(_) = _, %v, want _, %s", tc, err, codes.DeadlineExceeded)
	}
}

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

Review comment (suggestion):
It seems like we should have a test case that sets this `true`, also, and ensures that the server errors?  This is theoretically catching a client-side check that errors if zero requests are sent before `CloseSend` is called when the client knows it is not client-streaming.

(And if we don't have such a check we may want to add one.)

Code after:
		}
	}
}

// Tests the behavior of client for server-side streaming RPC when client sends zero request messages.
func (s) TestServerStreaming_ClientSendsZeroRequests(t *testing.T) {
	t.Skip("blocked on i/7286")
	// The initial call to recvMsg made by the generated code, will return the error.
	ss := stubserver.StubServer{}
	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()
	cc, err := grpc.NewClient(ss.Address, grpc.WithTransportCredentials(local.NewCredentials()))
	if err != nil {
		t.Fatalf("grpc.NewClient(%q) failed unexpectedly: %v", ss.Address, err)
	}
	defer cc.Close()

	desc := &grpc.StreamDesc{
		StreamName:    "StreamingOutputCall",
		ServerStreams: true,
		ClientStreams: false,
	}

	stream, err := cc.NewStream(ctx, desc, "/grpc.testing.TestService/StreamingOutputCall")
	if err != nil {
		t.Fatalf("cc.NewStream() failed unexpectedly: %v", err)
	}

	if err := stream.CloseSend(); status.Code(err) != codes.Internal {
		t.Errorf("stream.CloseSend() = %v, want error %v", status.Code(err), codes.Internal)
	}
}

// Tests that a client receives a cardinality violation error for client-streaming
// RPCs if the server call SendMsg multiple times.
func (s) TestClientStreaming_ServerHandlerSendMsgAfterSendMsg(t *testing.T) {
	ss := stubserver.StubServer{
		StreamingInputCallF: func(stream testgrpc.TestService_StreamingInputCallServer) error {
			if err := stream.SendMsg(&testpb.StreamingInputCallResponse{}); err != nil {
				t.Errorf("stream.SendMsg(_) = %v, want <nil>", err)
			}
			if err := stream.SendMsg(&testpb.StreamingInputCallResponse{}); err != nil {
				t.Errorf("stream.SendMsg(_) = %v, want <nil>", err)
			}
			return nil
		},