Revert "Revert "test: add expand_step_test.go. Fixes #131""

This reverts commit d5cb429.

test/e2e/expand_step_test.go

@@ -0,0 +1,42 @@
+// +build test
+
+package e2e
+
+import (
+	. "github.com/argoproj-labs/argo-dataflow/api/v1alpha1"
+	. "github.com/argoproj-labs/argo-dataflow/test"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"testing"
+)
+
+func TestExpandStep(t *testing.T) {
+	defer Setup(t)()
+
+	CreatePipeline(Pipeline{
+		ObjectMeta: metav1.ObjectMeta{Name: "expand"},
+		Spec: PipelineSpec{
+			Steps: []StepSpec{
+				{
+					Name:    "main",
+					Expand:  &Expand{},
+					Sources: []Source{{HTTP: &HTTPSource{}}},
+					Sinks:   []Sink{{Log: &Log{}}},
+				},
+			},
+		},
+	})
+
+	WaitForPod()
+
+	defer StartPortForward("expand-main-0")()
+
+	SendMessageViaHTTP(`{"foo.bar": "baz"}`)
+
+	WaitForPipeline(UntilMessagesSunk)
+	WaitForStep(TotalSunkMessages(1))
+
+	ExpectLogLine("main", `"foo":`)
+
+	DeletePipelines()
+	WaitForPodsToBeDeleted()
+}

test/log.go

@@ -52,7 +52,7 @@ func podsLogContains(ctx context.Context, podList *corev1.PodList, containerName
 	resultChan := make(chan bool)
 	for _, p := range podList.Items {
 		go func(podName string) {
-			log.Printf("Watching pod: %s\n", podName)
+			log.Printf("watching pod: %s\n", podName)
 			contains, err := podLogContains(cctx, podName, containerName, pattern)
 			if err != nil {
 				errChan <- err