tests: add map test. Fixes #126

docs/FEATURES.md

@@ -32,7 +32,7 @@ See [Wikipedia](https://en.wikipedia.org/wiki/Software_release_life_cycle#Stages
 | Kafka sink | | v0.0.59 | | |
 | Kafka source | | v0.0.59 | | |
 | Log sink | | |  v0.0.59 |  |
-| Map step | | v0.0.59 | | |
+| Map step | | v0.0.59 | v0.0.70 | |
 | NATS Streaming sink | | v0.0.59 | | |
 | NATS Streaming source | | v0.0.59 | | |
 | Non-terminating pipelines | | | v0.0.59 | |

test/e2e/map_test.go

@@ -0,0 +1,45 @@
+// +build test
+
+package e2e
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	. "github.com/argoproj-labs/argo-dataflow/api/v1alpha1"
+	. "github.com/argoproj-labs/argo-dataflow/test"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func TestMap(t *testing.T) {
+	defer Setup(t)()
+
+	CreatePipeline(Pipeline{
+		ObjectMeta: metav1.ObjectMeta{Name: "map"},
+		Spec: PipelineSpec{
+			Steps: []StepSpec{
+				{
+					Name:    "main",
+					Map:     "bytes('hi! ' + string(msg))",
+					Sources: []Source{{HTTP: &HTTPSource{}}},
+					Sinks:   []Sink{{Log: &Log{}}},
+				},
+			},
+		},
+	})
+
+	WaitForPod()
+
+	defer StartPortForward("map-main-0")()
+
+	SendMessageViaHTTP("foo-bar")
+
+	WaitForPipeline(UntilMessagesSunk)
+	WaitForStep(TotalSunkMessages(1))
+
+	ExpectStepLogLine(context.Background(), "map", "main", "sidecar", `hi! foo-bar`, 1*time.Minute)
+
+	DeletePipelines()
+	WaitForPodsToBeDeleted()
+}