Integration test should check OpenFlow operations eventually take eff…

…ect (#3357)

The single flow/group add/modify/delete methods are asynchronous. It
may not take effect immediately after the methods return. The
integration test should take it into consideration and check the
operations eventually take effect.

Fixes #3213

Signed-off-by: Quan Tian <qtian@vmware.com>

test/integration/ovs/ofctrl_test.go

@@ -28,6 +28,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"golang.org/x/time/rate"
+	"k8s.io/apimachinery/pkg/util/wait"
 
 	binding "antrea.io/antrea/pkg/ovs/openflow"
 	"antrea.io/antrea/pkg/ovs/ovsconfig"
@@ -295,9 +296,12 @@ func TestOFctrlGroup(t *testing.T) {
 			}
 			// Check if the group could be added.
 			require.Nil(t, group.Add())
-			groups, err := OfCtlDumpGroups(ovsCtlClient)
-			require.Nil(t, err)
-			require.Len(t, groups, 1)
+			var groups [][]string
+			require.NoError(t, wait.PollImmediate(openFlowCheckInterval, openFlowCheckTimeout, func() (done bool, err error) {
+				groups, err = OfCtlDumpGroups(ovsCtlClient)
+				require.Nil(t, err)
+				return len(groups) == 1, nil
+			}), "Failed to install group")
 			dumpedGroup := groups[0]
 			for i, bucket := range buckets {
 				// Must have weight
@@ -317,9 +321,11 @@ func TestOFctrlGroup(t *testing.T) {
 			}
 			// Check if the group could be deleted.
 			require.Nil(t, group.Delete())
-			groups, err = OfCtlDumpGroups(ovsCtlClient)
-			require.Nil(t, err)
-			require.Len(t, groups, 0)
+			require.NoError(t, wait.PollImmediate(openFlowCheckInterval, openFlowCheckTimeout, func() (done bool, err error) {
+				groups, err = OfCtlDumpGroups(ovsCtlClient)
+				require.Nil(t, err)
+				return len(groups) == 0, nil
+			}), "Failed to delete group")
 		})
 	}
 }

test/integration/ovs/openflow_test_utils.go

@@ -19,11 +19,20 @@ import (
 	"os/exec"
 	"strings"
 	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"k8s.io/apimachinery/pkg/util/wait"
 
 	binding "antrea.io/antrea/pkg/ovs/openflow"
 	"antrea.io/antrea/pkg/ovs/ovsctl"
 )
 
+const (
+	openFlowCheckTimeout  = 500 * time.Millisecond
+	openFlowCheckInterval = 100 * time.Millisecond
+)
+
 func PrepareOVSBridge(brName string) error {
 	// using the netdev datapath type does not impact test coverage but
 	// ensures that the integration tests can be run with Docker Desktop on
@@ -55,53 +64,67 @@ func (f ExpectFlow) flowStr(name string) string {
 	return fmt.Sprintf("table=%s,%s actions=%s", name, f.MatchStr, f.ActStr)
 }
 
-func CheckFlowExists(t *testing.T, ovsCtlClient ovsctl.OVSCtlClient, tableName string, tableID uint8, exist bool, flows []*ExpectFlow) []string {
+func CheckFlowExists(t *testing.T, ovsCtlClient ovsctl.OVSCtlClient, tableName string, tableID uint8, expectFound bool, flows []*ExpectFlow) []string {
 	var flowList []string
-	if tableName != "" {
-		flowList, _ = OfctlDumpTableFlows(ovsCtlClient, tableName)
-	} else {
-		flowList, _ = OfctlDumpTableFlowsWithoutName(ovsCtlClient, tableID)
-		tableName = fmt.Sprintf("%d", tableID)
-	}
+	var unexpectedFlows []*ExpectFlow
+	if err := wait.PollImmediate(openFlowCheckInterval, openFlowCheckTimeout, func() (done bool, err error) {
+		unexpectedFlows = unexpectedFlows[:0]
+		if tableName != "" {
+			flowList, err = OfctlDumpTableFlows(ovsCtlClient, tableName)
+		} else {
+			flowList, err = OfctlDumpTableFlowsWithoutName(ovsCtlClient, tableID)
+			tableName = fmt.Sprintf("%d", tableID)
+		}
+		require.NoError(t, err, "Error dumping flows")
 
-	for _, flow := range flows {
-		found := OfctlFlowMatch(flowList, tableName, flow)
-		if exist && !found {
-			t.Errorf("Failed to install flow: %s", flow.flowStr(tableName))
+		for _, flow := range flows {
+			found := OfctlFlowMatch(flowList, tableName, flow)
+			if found != expectFound {
+				unexpectedFlows = append(unexpectedFlows, flow)
+			}
 		}
-		if !exist && found {
-			t.Errorf("Failed to uninstall flow: %s", flow.flowStr(tableName))
+		return len(unexpectedFlows) == 0, nil
+	}); err != nil {
+		for _, flow := range unexpectedFlows {
+			if expectFound {
+				t.Errorf("Failed to install flow: %s", flow.flowStr(tableName))
+			} else {
+				t.Errorf("Failed to uninstall flow: %s", flow.flowStr(tableName))
+			}
 		}
-	}
-	if t.Failed() {
-		t.Errorf("Existing flows:\n%s", strings.Join(flowList, "\n"))
+		t.Logf("Existing flows:\n%s", strings.Join(flowList, "\n"))
 	}
 	return flowList
 }
 
-func CheckGroupExists(t *testing.T, ovsCtlClient ovsctl.OVSCtlClient, groupID binding.GroupIDType, groupType string, buckets []string, expectExists bool) {
-	// dump groups
-	groupList, err := OfCtlDumpGroups(ovsCtlClient)
-	if err != nil {
-		t.Errorf("Error dumping flows: Err %v", err)
-	}
+func CheckGroupExists(t *testing.T, ovsCtlClient ovsctl.OVSCtlClient, groupID binding.GroupIDType, groupType string, buckets []string, expectFound bool) {
 	var bucketStrs []string
 	for _, bucket := range buckets {
 		bucketStr := fmt.Sprintf("bucket=%s", bucket)
 		bucketStrs = append(bucketStrs, bucketStr)
 	}
 	groupStr := fmt.Sprintf("group_id=%d,type=%s,%s", groupID, groupType, strings.Join(bucketStrs, ","))
-	found := false
-	for _, groupElems := range groupList {
-		groupEntry := fmt.Sprintf("%s,bucket=", groupElems[0])
-		groupEntry = fmt.Sprintf("%s%s", groupEntry, strings.Join(groupElems[1:], ",bucket="))
-		if strings.Contains(groupEntry, groupStr) {
-			found = true
-			break
+	var groupList [][]string
+	if err := wait.PollImmediate(openFlowCheckInterval, openFlowCheckTimeout, func() (done bool, err error) {
+		groupList, err = OfCtlDumpGroups(ovsCtlClient)
+		require.NoError(t, err, "Error dumping groups")
+		found := false
+		for _, groupElems := range groupList {
+			groupEntry := fmt.Sprintf("%s,bucket=", groupElems[0])
+			groupEntry = fmt.Sprintf("%s%s", groupEntry, strings.Join(groupElems[1:], ",bucket="))
+			if strings.Contains(groupEntry, groupStr) {
+				found = true
+				break
+			}
 		}
-	}
-	if found != expectExists {
-		t.Errorf("Failed to find group:\n%v\nExisting groups:\n%v", groupStr, groupList)
+		return found == expectFound, nil
+	}); err != nil {
+		if expectFound {
+			t.Errorf("Failed to install group: %s", groupStr)
+		} else {
+			t.Errorf("Failed to uninstall group: %s", groupStr)
+		}
+		t.Logf("Existing groups:\n%s", groupList)
 	}
 }