test: add unit test for pkg/remote/trans/nphttp2/grpc (cloudwego#462)

* remove meaningless ut and test method

* add ut for nphttp2/grpc

pkg/remote/trans/nphttp2/client_handler_test.go

@@ -45,31 +45,3 @@ func TestClientHandler(t *testing.T) {
 	err = cliTransHandler.Write(ctx, conn, msg)
 	test.Assert(t, err == nil, err)
 }
-
-func TestClientHandlerOnAction(t *testing.T) {
-	// init
-	opt := newMockClientOption()
-	ctx := newMockCtxWithRPCInfo()
-	conn, err := opt.ConnPool.Get(ctx, "tcp", mockAddr0, remote.ConnOption{Dialer: opt.Dialer, ConnectTimeout: time.Second})
-	test.Assert(t, err == nil, err)
-	defer conn.Close()
-	cliTransHandler, err := newCliTransHandler(opt)
-	test.Assert(t, err == nil, err)
-	// test OnConnection()
-	cliCtx, err := cliTransHandler.OnConnect(ctx)
-	test.Assert(t, err == nil, err)
-	test.Assert(t, cliCtx == ctx)
-
-	// test OnMessage()
-	cliCtx, err = cliTransHandler.OnMessage(ctx, new(mockMessage), new(mockMessage))
-	test.Assert(t, err == nil, err)
-	test.Assert(t, cliCtx == ctx)
-
-	// test OnRead()
-	setDontWaitHeader(conn.(*clientConn).s)
-	_, err = cliTransHandler.OnRead(ctx, conn)
-	test.Assert(t, err == nil, err)
-
-	// test SetPipeline()
-	cliTransHandler.SetPipeline(nil)
-}

pkg/remote/trans/nphttp2/grpc/bdp_estimator_test.go

@@ -0,0 +1,70 @@
+/*
+ * Copyright 2022 CloudWeGo Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package grpc
+
+import (
+	"testing"
+
+	"github.com/cloudwego/kitex/internal/test"
+)
+
+func TestBdp(t *testing.T) {
+	// init bdp estimator
+	bdpEst := &bdpEstimator{
+		bdp:               initialWindowSize,
+		updateFlowControl: func(n uint32) {},
+	}
+	size := 100
+
+	// mock data frame receive
+	sent := bdpEst.add(uint32(size))
+	test.Assert(t, sent)
+	bdpEst.timesnap(bdpPing.data)
+
+	// receive normal ping
+	bdpEst.timesnap([8]byte{0, 0, 0, 0, 0, 0, 0, 0})
+
+	// mock receiving data while bdp ping haven't been acked
+	for i := 0; i < 3; i++ {
+		sent = bdpEst.add(uint32(size))
+		test.Assert(t, !sent)
+	}
+
+	// now bdp estimator receive ping ack and do calculation
+	bdpEst.calculate(bdpPing.data)
+
+	// receive normal ping
+	bdpEst.calculate([8]byte{0, 0, 0, 0, 0, 0, 0, 0})
+
+	size = 10000
+	// calculate 15 times
+	for c := 0; c < 15; c++ {
+		sent = bdpEst.add(uint32(size))
+		test.Assert(t, sent)
+		bdpEst.timesnap(bdpPing.data)
+
+		// mock the situation that network delay is very long and data is very big
+		for i := 0; i < 15; i++ {
+			sent = bdpEst.add(uint32(size))
+			test.Assert(t, !sent)
+		}
+
+		// receive bdp ack and calculate again
+		bdpEst.calculate(bdpPing.data)
+
+	}
+}

pkg/remote/trans/nphttp2/grpc/controlbuf_test.go

@@ -0,0 +1,75 @@
+/*
+ * Copyright 2022 CloudWeGo Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package grpc
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/cloudwego/kitex/internal/test"
+)
+
+func TestControlBuf(t *testing.T) {
+	ctx := context.Background()
+	cb := newControlBuffer(ctx.Done())
+
+	// test put()
+	testItem := &ping{}
+	err := cb.put(testItem)
+	test.Assert(t, err == nil, err)
+
+	// test get() with block
+	item, err := cb.get(true)
+	test.Assert(t, err == nil, err)
+	test.Assert(t, item == testItem, err)
+
+	// test get() with no block
+	item, err = cb.get(false)
+	test.Assert(t, err == nil, err)
+	test.Assert(t, item == nil, err)
+
+	// test executeAndPut()
+	success, err := cb.execute(func(it interface{}) bool {
+		return false
+	}, &ping{})
+
+	test.Assert(t, err == nil, err)
+	test.Assert(t, !success, err)
+
+	// test throttle() mock a lot of response frame so throttle() will block current goroutine
+	for i := 0; i < maxQueuedTransportResponseFrames+5; i++ {
+		err := cb.put(&ping{})
+		test.Assert(t, err == nil, err)
+	}
+
+	// start a new goroutine to consume response frame
+	go func() {
+		time.Sleep(time.Millisecond * 100)
+		for {
+			it, err := cb.get(false)
+			if err != nil || it == nil {
+				break
+			}
+		}
+	}()
+
+	cb.throttle()
+
+	// test finish()
+	cb.finish()
+}

pkg/remote/trans/nphttp2/grpc/flowcontrol_test.go

@@ -0,0 +1,131 @@
+/*
+ * Copyright 2022 CloudWeGo Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package grpc
+
+import (
+	"math"
+	"testing"
+	"time"
+
+	"github.com/cloudwego/kitex/internal/test"
+)
+
+func TestWriteQuota(t *testing.T) {
+	// init
+	quotaLimit := int32(100)
+	done := make(chan struct{})
+
+	// test newWriteQuota()
+	wq := newWriteQuota(quotaLimit, done)
+
+	// test get() and less than quotaLimit
+	err := wq.get(80)
+	test.Assert(t, err == nil, err)
+
+	// test realReplenish() to refresh quotaLimit
+	wq.realReplenish(80)
+	test.Assert(t, wq.quota == quotaLimit)
+
+	// test get() in block situation
+	go func() {
+		time.Sleep(time.Millisecond * 50)
+		wq.replenish(40)
+	}()
+	err = wq.get(120)
+	test.Assert(t, err == nil, err)
+
+	// test get() after block and replenish
+	err = wq.get(20)
+	test.Assert(t, err == nil, err)
+
+	// test stop writeQuota by channel
+	go func() {
+		time.Sleep(time.Millisecond * 50)
+		var s struct{}
+		done <- s
+	}()
+	err = wq.get(20)
+	test.Assert(t, err == errStreamDone)
+}
+
+func TestTrInFlow(t *testing.T) {
+	// init
+	oldLimit := uint32(100)
+	trInFlow := &trInFlow{limit: oldLimit}
+	limit := 2 * oldLimit
+
+	// test newLimit()
+	delta := trInFlow.newLimit(limit)
+	test.Assert(t, delta == oldLimit)
+
+	// test onData() but unacked is less than 1/4 limit
+	increase := trInFlow.onData(limit / 10)
+	test.Assert(t, increase == 0)
+
+	// test onData() and unacked is more than 1/4 limit
+	increase = trInFlow.onData(limit / 2)
+	test.Assert(t, increase == limit/2+limit/10)
+
+	// test reset()
+	increase = trInFlow.onData(limit / 5)
+	test.Assert(t, increase == 0)
+	increase = trInFlow.reset()
+	test.Assert(t, increase == limit/5)
+}
+
+func TestInFlow(t *testing.T) {
+	// init
+	oldLimit := uint32(100)
+	inFlow := &inFlow{limit: oldLimit}
+	limit := oldLimit * 2
+
+	// test newLimit()
+	inFlow.newLimit(limit)
+	test.Assert(t, inFlow.limit == limit)
+
+	// test empty onData()
+	err := inFlow.onData(0)
+	test.Assert(t, err == nil, err)
+
+	// test empty onRead()
+	increase := inFlow.onRead(0)
+	test.Assert(t, increase == 0)
+
+	// test onData()
+	err = inFlow.onData(limit)
+	test.Assert(t, err == nil, err)
+
+	// test onRead()
+	size := inFlow.onRead(limit)
+	test.Assert(t, size == limit)
+
+	// test onData() illegal situation
+	err = inFlow.onData(limit * 2)
+	test.Assert(t, err != nil)
+
+	// test maybeAdjust() no adjust
+	adjustNum := inFlow.maybeAdjust(0)
+	test.Assert(t, adjustNum == 0)
+
+	// test maybeAdjust() do adjust
+	adjustNum = inFlow.maybeAdjust(limit * 2)
+	test.Assert(t, adjustNum == limit*2)
+
+	// test maybeAdjust() maxSize adjust (f.delta - limit)
+	adjustNum = inFlow.maybeAdjust(math.MaxInt32 + 1)
+	test.Assert(t, adjustNum == math.MaxInt32-limit)
+}