fix test failed

tests/e2e/v3_curl_test.go

@@ -23,6 +23,7 @@ import (
 	"strconv"
 	"sync"
 	"testing"
+	"time"
 
 	"github.com/grpc-ecosystem/grpc-gateway/runtime"
 	"github.com/stretchr/testify/assert"
@@ -37,7 +38,9 @@ import (
 
 var (
 	apiPrefix            = []string{"/v3"}
-	maxConcurrent uint32 = 3
+	maxConcurrent uint32 = 6
+	// 1 stream for PING/PONG and 1 stream for KV put
+	basicConcurrent uint32 = 2
 )
 
 func TestV3CurlPutGetNoTLS(t *testing.T) {
@@ -68,12 +71,11 @@ func TestV3CurlPutGetClientTLS(t *testing.T) {
 func TestV3CurlWatchConcurrent(t *testing.T) {
 	cfg := *e2e.NewConfigNoTLS()
 	cfg.ClusterSize = 1
-	cfg.MaxConcurrentStreams = maxConcurrent
+	cfg.MaxConcurrentStreams = maxConcurrent + basicConcurrent
 	for _, p := range apiPrefix {
 		testCtl(t, testV3CurlWatchConcurrent, withApiPrefix(p), withCfg(cfg))
 	}
 }
-
 func TestV3CurlWatchConcurrentClientTLS(t *testing.T) {
 	cfg := *e2e.NewConfigClientTLS()
 	cfg.ClusterSize = 1
@@ -187,7 +189,11 @@ func testV3CurlWatchConcurrent(cx ctlCtx) {
 
 	// concurrent watch
 	var wg sync.WaitGroup
-	excessConcurrent := 2
+	excessConcurrent := 0
+	// test exceeds the max concurrent limit
+	if cx.cfg.MaxConcurrentStreams == maxConcurrent {
+		excessConcurrent = 2
+	}
 	wg.Add(int(maxConcurrent) + excessConcurrent)
 	timeoutConcurrent := 0
 	for i := 0; i < int(maxConcurrent)+excessConcurrent; i++ {
@@ -201,14 +207,20 @@ func testV3CurlWatchConcurrent(cx ctlCtx) {
 			}
 		}()
 	}
-	// update
+	// wait for watch connection establishment
+	time.Sleep(20 * time.Millisecond)
+	// put KV
 	if err = e2e.CURLPost(cx.epc, e2e.CURLReq{Endpoint: path.Join(p, "/kv/put"), Value: string(putreq), Expected: "revision"}); err != nil {
 		cx.t.Fatalf("failed testV3CurlWatch put with curl using prefix (%s) (%v)", p, err)
 	}
-	// wait for watch
+	// wait for watch timeout
 	wg.Wait()
 	// assert that timeoutConcurrent should equal excessConcurrent
-	assert.Equal(cx.t, excessConcurrent, timeoutConcurrent)
+	if cx.cfg.MaxConcurrentStreams == maxConcurrent {
+		assert.GreaterOrEqual(cx.t, timeoutConcurrent, excessConcurrent)
+	} else {
+		assert.Equal(cx.t, excessConcurrent, timeoutConcurrent)
+	}
 }
 
 func testV3CurlTxn(cx ctlCtx) {