feat: add udsource python e2e (#1204)

Signed-off-by: Sidhant Kohli <[email protected]>

Makefile

@@ -106,14 +106,16 @@ test-coverage-with-isb:
 test-code:
 	go test -tags=isb_redis -race -v $(shell go list ./... | grep -v /vendor/ | grep -v /numaflow/test/) -timeout 120s
 
-test-e2e-suite-1:
-test-e2e-suite-2:
+test-e2e:
 test-kafka-e2e:
 test-http-e2e:
 test-nats-e2e:
 test-sdks-e2e:
 test-reduce-e2e:
 test-api-e2e:
+test-udsource-e2e:
+test-transformer-e2e:
+test-diamond-e2e:
 test-%: 
 	$(MAKE) cleanup-e2e
 	$(MAKE) image e2eapi-image

test/udsource-e2e/testdata/simple-source-go.yaml

@@ -1,7 +1,7 @@
 apiVersion: numaflow.numaproj.io/v1alpha1
 kind: Pipeline
 metadata:
-  name: simple-source
+  name: simple-source-go
 spec:
   vertices:
     - name: in

test/udsource-e2e/testdata/simple-source-java.yaml

@@ -1,7 +1,7 @@
 apiVersion: numaflow.numaproj.io/v1alpha1
 kind: Pipeline
 metadata:
-  name: simple-source
+  name: simple-source-java
 spec:
   vertices:
     - name: in

test/udsource-e2e/testdata/simple-source-python.yaml

@@ -0,0 +1,22 @@
+apiVersion: numaflow.numaproj.io/v1alpha1
+kind: Pipeline
+metadata:
+  name: simple-source-python
+spec:
+  vertices:
+    - name: in
+      source:
+        udsource:
+          container:
+            # A simple user-defined source for e2e testing
+            # See https://github.com/numaproj/numaflow-python/tree/main/examples/source
+            image: quay.io/numaio/numaflow-python/simple-source:v0.5.3
+            imagePullPolicy: Always
+      limits:
+        readBatchSize: 2
+    - name: out
+      sink:
+        log: {}
+  edges:
+    - from: in
+      to: out

test/udsource-e2e/udsource_test.go

@@ -20,6 +20,7 @@ package e2e
 
 import (
 	"fmt"
+	"sync"
 	"testing"
 
 	"github.com/stretchr/testify/suite"
@@ -31,14 +32,36 @@ type UserDefinedSourceSuite struct {
 	E2ESuite
 }
 
-func (s *UserDefinedSourceSuite) TestSimpleSourceGo() {
+func (s *UserDefinedSourceSuite) testSimpleSourceGo() {
 	s.testSimpleSource("go")
 }
 
-func (s *UserDefinedSourceSuite) TestSimpleSourceJava() {
+func (s *UserDefinedSourceSuite) testSimpleSourceJava() {
 	s.testSimpleSource("java")
 }
 
+func (s *UserDefinedSourceSuite) testSimpleSourcePython() {
+	s.testSimpleSource("python")
+}
+
+func (s *UserDefinedSourceSuite) TestUDSource() {
+	var wg sync.WaitGroup
+	wg.Add(3)
+	go func() {
+		defer wg.Done()
+		s.testSimpleSourcePython()
+	}()
+	go func() {
+		defer wg.Done()
+		s.testSimpleSourceJava()
+	}()
+	go func() {
+		defer wg.Done()
+		s.testSimpleSourceGo()
+	}()
+	wg.Wait()
+}
+
 func (s *UserDefinedSourceSuite) testSimpleSource(lang string) {
 	w := s.Given().Pipeline(fmt.Sprintf("@testdata/simple-source-%s.yaml", lang)).
 		When().