refactor(test): allow to set wait time for rule test

services/external_service_rule_test.go

@@ -252,7 +252,7 @@ func TestRestService(t *testing.T) {
 	topotest.DoRuleTest(t, tests, 0, &api.RuleOption{
 		BufferLength: 100,
 		SendError:    true,
-	})
+	}, 0)
 }
 
 type Resolver map[string]reflect.Value
@@ -433,7 +433,7 @@ func TestMsgpackService(t *testing.T) {
 	topotest.DoRuleTest(t, tests, 0, &api.RuleOption{
 		BufferLength: 100,
 		SendError:    true,
-	})
+	}, 0)
 }
 
 type server struct {
@@ -614,5 +614,5 @@ func TestGrpcService(t *testing.T) {
 	topotest.DoRuleTest(t, tests, 0, &api.RuleOption{
 		BufferLength: 100,
 		SendError:    true,
-	})
+	}, 0)
 }

xstream/topotest/mock_topo.go

@@ -973,11 +973,11 @@ func commonResultFunc(result [][]byte) interface{} {
 	return maps
 }
 
-func DoRuleTest(t *testing.T, tests []RuleTest, j int, opt *api.RuleOption) {
-	doRuleTestBySinkProps(t, tests, j, opt, nil, commonResultFunc)
+func DoRuleTest(t *testing.T, tests []RuleTest, j int, opt *api.RuleOption, wait int) {
+	doRuleTestBySinkProps(t, tests, j, opt, wait, nil, commonResultFunc)
 }
 
-func doRuleTestBySinkProps(t *testing.T, tests []RuleTest, j int, opt *api.RuleOption, sinkProps map[string]interface{}, resultFunc func(result [][]byte) interface{}) {
+func doRuleTestBySinkProps(t *testing.T, tests []RuleTest, j int, opt *api.RuleOption, w int, sinkProps map[string]interface{}, resultFunc func(result [][]byte) interface{}) {
 	fmt.Printf("The test bucket for option %d size is %d.\n\n", j, len(tests))
 	for i, tt := range tests {
 		datas, dataLength, tp, mockSink, errCh := createStream(t, tt, j, opt, sinkProps)
@@ -987,11 +987,15 @@ func doRuleTestBySinkProps(t *testing.T, tests []RuleTest, j int, opt *api.RuleO
 		}
 		wait := tt.W
 		if wait == 0 {
-			wait = 5
+			if w > 0 {
+				wait = w
+			} else {
+				wait = 5
+			}
 		}
 		switch opt.Qos {
 		case api.ExactlyOnce:
-			wait *= 4
+			wait *= 10
 		case api.AtLeastOnce:
 			wait *= 3
 		}

xstream/topotest/plugin_rule_test.go

@@ -213,7 +213,7 @@ func TestFuncState(t *testing.T) {
 	DoRuleTest(t, tests, 0, &api.RuleOption{
 		BufferLength: 100,
 		SendError:    true,
-	})
+	}, 0)
 }
 
 func TestFuncStateCheckpoint(t *testing.T) {

xstream/topotest/rule_test.go

@@ -483,7 +483,7 @@ func TestSingleSQL(t *testing.T) {
 		},
 	}
 	for j, opt := range options {
-		DoRuleTest(t, tests[9:10], j, opt)
+		DoRuleTest(t, tests[9:10], j, opt, 0)
 	}
 }
 
@@ -575,7 +575,7 @@ func TestSingleSQLError(t *testing.T) {
 	DoRuleTest(t, tests, 0, &api.RuleOption{
 		BufferLength: 100,
 		SendError:    true,
-	})
+	}, 0)
 }
 
 func TestSingleSQLOmitError(t *testing.T) {
@@ -660,7 +660,7 @@ func TestSingleSQLOmitError(t *testing.T) {
 	DoRuleTest(t, tests, 0, &api.RuleOption{
 		BufferLength: 100,
 		SendError:    false,
-	})
+	}, 0)
 }
 
 func TestSingleSQLTemplate(t *testing.T) {
@@ -719,7 +719,7 @@ func TestSingleSQLTemplate(t *testing.T) {
 	doRuleTestBySinkProps(t, tests, 0, &api.RuleOption{
 		BufferLength: 100,
 		SendError:    true,
-	}, map[string]interface{}{
+	}, 0, map[string]interface{}{
 		"dataTemplate": `{"wrapper":"w1", "c":"{{.color}}"}`,
 		"sendSingle":   true,
 	}, func(result [][]byte) interface{} {
@@ -778,7 +778,7 @@ func TestNoneSingleSQLTemplate(t *testing.T) {
 	doRuleTestBySinkProps(t, tests, 0, &api.RuleOption{
 		BufferLength: 100,
 		SendError:    true,
-	}, map[string]interface{}{
+	}, 0, map[string]interface{}{
 		"dataTemplate": `<div>results</div><ul>{{range .}}<li>{{.color}} - {{.size}}</li>{{end}}</ul>`,
 	}, func(result [][]byte) interface{} {
 		return result
@@ -859,6 +859,6 @@ func TestSingleSQLForBinary(t *testing.T) {
 		return maps
 	}
 	for j, opt := range options {
-		doRuleTestBySinkProps(t, tests, j, opt, nil, byteFunc)
+		doRuleTestBySinkProps(t, tests, j, opt, 0, nil, byteFunc)
 	}
 }

xstream/topotest/window_rule_test.go

@@ -284,7 +284,6 @@ func TestWindow(t *testing.T) {
 		}, {
 			Name: `TestWindowRule5`,
 			Sql:  `SELECT temp FROM sessionDemo GROUP BY SessionWindow(ss, 2, 1) `,
-			W:    10,
 			R: [][]map[string]interface{}{
 				{{
 					"temp": 25.5,
@@ -713,7 +712,7 @@ func TestWindow(t *testing.T) {
 		},
 	}
 	for j, opt := range options {
-		DoRuleTest(t, tests, j, opt)
+		DoRuleTest(t, tests, j, opt, 15)
 	}
 }
 
@@ -1149,7 +1148,7 @@ func TestEventWindow(t *testing.T) {
 		},
 	}
 	for j, opt := range options {
-		DoRuleTest(t, tests, j, opt)
+		DoRuleTest(t, tests, j, opt, 10)
 	}
 }
 
@@ -1402,5 +1401,5 @@ func TestWindowError(t *testing.T) {
 	DoRuleTest(t, tests, 0, &api.RuleOption{
 		BufferLength: 100,
 		SendError:    true,
-	})
+	}, 0)
 }