chore(test): unskip previously skipped tests (#9537)

Author: shiva-istari

dgraph/cmd/dgraphimport/import_test.go

@@ -75,8 +75,6 @@ func TestEmptyBulkOutDir(t *testing.T) {
 }
 
 func TestDrainModeAfterStartSnapshotStream(t *testing.T) {
-	t.Skip("Skipping... sometimes the query for schema succeeds even when the server is in draining mode")
-
 	tests := []struct {
 		name        string
 		numAlphas   int
@@ -123,8 +121,6 @@ func TestDrainModeAfterStartSnapshotStream(t *testing.T) {
 }
 
 func TestImportApis(t *testing.T) {
-	t.Skip("Skipping import tests due to persistent flakiness with container networking and Raft leadership issues")
-
 	tests := []testcase{
 		{
 			name:           "SingleGroupShutTwoAlphasPerGroup",
@@ -391,12 +387,12 @@ func verifyImportResults(t *testing.T, gc *dgraphapi.GrpcClient, downAlphas int)
 	}
 
 	retryDelay := time.Second
-	hasAllPredicates := true
 
 	// Get expected predicates first
 	var expectedSchemaObj map[string]interface{}
 	require.NoError(t, json.Unmarshal([]byte(expectedSchema), &expectedSchemaObj))
 	expectedPredicates := getPredicateMap(expectedSchemaObj)
+	var hasAllPredicates bool
 
 	for i := 0; i < maxRetries; i++ {
 		// Checking client connection again here because an import operation may be in progress on the rejoined alpha
@@ -412,7 +408,7 @@ func verifyImportResults(t *testing.T, gc *dgraphapi.GrpcClient, downAlphas int)
 		// Get actual predicates
 		actualPredicates := getPredicateMap(actualSchema)
 
-		// Check if all expected predicates are present
+		hasAllPredicates = true
 		for predName := range expectedPredicates {
 			if _, exists := actualPredicates[predName]; !exists {
 				hasAllPredicates = false

dgraphtest/load.go

@@ -580,8 +580,8 @@ func downloadFile(fname, url string) error {
 	cmd := exec.Command("wget", "-O", fname, url)
 	cmd.Dir = datasetFilesPath
 
-	if out, err := cmd.CombinedOutput(); err != nil {
-		return fmt.Errorf("error downloading file %s: %s", fname, string(out))
+	if _, err := cmd.CombinedOutput(); err != nil {
+		return fmt.Errorf("error downloading file %s: %w", fname, err)
 	}
 	return nil
 }

dgraphtest/local_cluster.go

@@ -930,8 +930,7 @@ func (c *LocalCluster) Client() (*dgraphapi.GrpcClient, func(), error) {
 	var conns []*grpc.ClientConn
 	for _, aa := range c.alphas {
 		if !aa.isRunning {
-			// QUESTIONS(shivaji): Should this be 'continue' instead of a break from the loop
-			break
+			continue
 		}
 		url, err := aa.alphaURL(c)
 		if err != nil {

worker/import.go

@@ -298,7 +298,7 @@ func InStream(stream api.Dgraph_StreamExtSnapshotServer) error {
 		return fmt.Errorf("failed to establish stream with leader: %v", err)
 	}
 	glog.Infof("[import] [forward %d -> %d] start", groups().Node.gid, groupId)
-	glog.Infof("[import] [forward %d -> %d] start", groups().Node.MyAddr, groups().Leader(groupId).Addr)
+	glog.Infof("[import] [forward %v -> %d] start", groups().Node.MyAddr, groups().Leader(groupId).Addr)
 
 	glog.Infof("[import] sending forward true to leader of group [%v]", groupId)
 	forwardReq := &api.StreamExtSnapshotRequest{Forward: true}
@@ -313,12 +313,6 @@ func InStream(stream api.Dgraph_StreamExtSnapshotServer) error {
 func pipeTwoStream(in api.Dgraph_StreamExtSnapshotServer, out pb.Worker_StreamExtSnapshotClient, groupId uint32) error {
 	currentGroup := groups().Node.gid
 	ctx := in.Context()
-	if err := out.Send(&api.StreamExtSnapshotRequest{GroupId: groupId}); err != nil {
-		return fmt.Errorf("send groupId downstream(%d): %w", groupId, err)
-	}
-	if _, err := out.Recv(); err != nil {
-		return fmt.Errorf("ack groupId downstream(%d): %w", groupId, err)
-	}
 
 	for {
 		if err := ctx.Err(); err != nil {
@@ -487,7 +481,7 @@ func streamInGroup(stream api.Dgraph_StreamExtSnapshotServer, forward bool) erro
 		if forward {
 			// We are not going to return any error from here because we care about the majority of nodes.
 			// If the majority of nodes are able to receive the data, the remaining ones can catch up later.
-			glog.Infof("[import] Streaming external snapshot to [%v] from [%v] forward [%v]", member.Addr, node.MyAddr)
+			glog.Infof("[import] Streaming external snapshot to [%v] from [%v]", member.Addr, node.MyAddr)
 			eg.Go(func() error {
 				glog.Infof(`[import:forward] streaming external snapshot to [%v] from [%v]`, member.Addr, node.MyAddr)
 				if member.AmDead {