fix: fix flaky test (#2239)

Signed-off-by: yisaer <disxiaofei@163.com>

internal/topo/rule/ruleState.go

@@ -137,6 +137,9 @@ func (rs *RuleState) UpdateTopo(rule *api.Rule) error {
 	return nil
 }
 
+// only used for unit test
+var ignoreSignal = false
+
 // Run start to run the two loops, do not access any changeable states
 func (rs *RuleState) run() {
 	var (
@@ -155,6 +158,11 @@ func (rs *RuleState) run() {
 				}
 				return
 			}
+
+			if ignoreSignal {
+				continue
+			}
+
 			switch s {
 			case ActionSignalStart:
 				if ctx != nil {

internal/topo/rule/ruleState_test.go

@@ -113,6 +113,10 @@ func TestCreate(t *testing.T) {
 }
 
 func TestUpdate(t *testing.T) {
+	ignoreSignal = true
+	defer func() {
+		ignoreSignal = false
+	}()
 	sp := processor.NewStreamProcessor()
 	sp.ExecStmt(`CREATE STREAM demo () WITH (DATASOURCE="users", FORMAT="JSON")`)
 	defer sp.ExecStmt(`DROP STREAM demo`)
@@ -184,7 +188,7 @@ func TestUpdate(t *testing.T) {
 	}
 	for i, tt := range tests {
 		rs, err := NewRuleState(&api.Rule{
-			Triggered: false,
+			Triggered: true,
 			Id:        "test",
 			Sql:       "SELECT ts FROM demo",
 			Actions: []map[string]interface{}{
@@ -197,7 +201,10 @@ func TestUpdate(t *testing.T) {
 		require.NoError(t, err)
 		err = rs.Start()
 		require.NoError(t, err)
+		time.Sleep(5 * time.Millisecond)
+		require.Equal(t, 1, rs.triggered, fmt.Sprintf("case %v failed", i))
 		err = rs.UpdateTopo(tt.r)
+		time.Sleep(5 * time.Millisecond)
 		require.Equal(t, tt.e, err, fmt.Sprintf("case %v failed", i))
 		require.Equal(t, tt.triggered, rs.triggered, fmt.Sprintf("case %v failed", i))
 		rs.Close()