add python docker-compose test

fix python integ tests

Signed-off-by: Shyam Jesal <s.jesalpura@gmail.com>

Author: shyamjesal

.github/workflows/xdt_docker_compose.yml

@@ -43,7 +43,7 @@ jobs:
         env:
           KO_DOCKER_REPO: ${{ secrets.DOCKER_HUB_USERNAME }}
         run: |
-          make docker-images-push
+          make all-images-push
 
   docker-compose-integ-test:
     name: Docker compose integ test
@@ -60,3 +60,5 @@ jobs:
         run: |
           cd ./user-functions
           docker-compose up --abort-on-container-exit --exit-code-from fx
+          docker-compose -f docker-compose-python.yml up --abort-on-container-exit --exit-code-from fx
+          

.github/workflows/xdt_tests.yml

@@ -24,7 +24,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        test-type: [unit, integ, timeout, parallel-invoke, fan-in, fan-out, get-put]
+        test-type: [unit, integ, parallel-invoke, fan-in, fan-out, get-put]
     steps:
     - name: Set up Go 1.x
       uses: actions/setup-go@v2
@@ -46,32 +46,37 @@ jobs:
         uses: golangci/golangci-lint-action@v2
         with:
           working-directory: sdk/golang
-          args: --timeout 5m
+          args: "--timeout 5m -e \"S1000: should use a simple channel send/receive instead of \\`select\\` with a single case\""
       - name: Check dQP code
         if: ${{ always() }}
-        env:
-          DIR: queue-proxy/dQP
-        run: cd $DIR && golangci-lint run --out-format=github-actions --path-prefix=$DIR --timeout 5m
+        uses: golangci/golangci-lint-action@v2
+        with:
+          working-directory: queue-proxy/dQP
+          args: "--timeout 5m -e \"S1000: should use a simple channel send/receive instead of \\`select\\` with a single case\""
       - name: Check plotter code
         if: ${{ always() }}
-        env:
-          DIR: plotter
-        run: cd $DIR && golangci-lint run --out-format=github-actions --path-prefix=$DIR --timeout 5m
+        uses: golangci/golangci-lint-action@v2
+        with:
+          working-directory: plotter
+          args: "--timeout 5m -e \"S1000: should use a simple channel send/receive instead of \\`select\\` with a single case\""
       - name: Check user-functions/fx code
         if: ${{ always() }}
-        env:
-          DIR: user-functions/fx
-        run: cd $DIR && golangci-lint run --out-format=github-actions --path-prefix=$DIR --timeout 5m
+        uses: golangci/golangci-lint-action@v2
+        with:
+          working-directory: user-functions/fx
+          args: "--timeout 5m -e \"S1000: should use a simple channel send/receive instead of \\`select\\` with a single case\""
       - name: Check user-functions/gx code
         if: ${{ always() }}
-        env:
-          DIR: user-functions/gx
-        run: cd $DIR && golangci-lint run --out-format=github-actions --path-prefix=$DIR --timeout 5m
+        uses: golangci/golangci-lint-action@v2
+        with:
+          working-directory: user-functions/gx
+          args: "--timeout 5m -e \"S1000: should use a simple channel send/receive instead of \\`select\\` with a single case\""
       - name: Check user-functions/dQP code
         if: ${{ always() }}
-        env:
-          DIR: user-functions/dQP
-        run: cd $DIR && golangci-lint run --out-format=github-actions --path-prefix=$DIR --timeout 5m
+        uses: golangci/golangci-lint-action@v2
+        with:
+          working-directory: user-functions/dQP
+          args: "--timeout 5m -e \"S1000: should use a simple channel send/receive instead of \\`select\\` with a single case\""
   Benchmark:
     name: Benchmark XDT
     runs-on: ubuntu-latest
@@ -98,7 +103,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        test-type: [ unit, integ, timeout, get-put]
+        test-type: [integ, timeout, get-put]
         python-version: [3.7, 3.8, 3.9]
     steps:
       - name: Set up Python
@@ -113,7 +118,7 @@ jobs:
         run: sudo apt-get install psmisc
       - name: Check out code into the Go module directory
         uses: actions/checkout@v2
-      - name: install environs package
-        run: pip install environs --user
       - name: Run test
-        run: make python-${{ matrix.test-type }}-test
+        run: |
+          cp proto/crossXDT/crossXDT.py.capnp ./sdk/python/crossXDT.py.capnp
+          make python-${{ matrix.test-type }}-test

Dockerfile

@@ -0,0 +1,31 @@
+FROM python:slim as pygx
+
+RUN apt update && \
+    apt install g++ make cmake -y && \
+    pip install pycapnp grpcio && \
+    apt remove g++ make cmake -y && \
+    apt autoremove -y
+RUN pip install environs grpcio-tools
+WORKDIR /app
+COPY sdk/python/ ./
+COPY user-functions/gx/gx.py ./
+COPY ./proto/*/*.py ./
+COPY ./proto/crossXDT/crossXDT.py.capnp ./
+
+ENTRYPOINT ["python3", "-u", "gx.py"]
+
+FROM python:slim as pyfx
+
+RUN apt update && \
+    apt install g++ make cmake -y && \
+    pip install pycapnp grpcio && \
+    apt remove g++ make cmake -y && \
+    apt autoremove -y
+RUN pip install environs grpcio-tools
+WORKDIR /app
+COPY sdk/python/ ./
+COPY user-functions/fx/fx.py ./
+COPY ./proto/*/*.py ./
+COPY ./proto/crossXDT/crossXDT.py.capnp ./
+
+ENTRYPOINT ["python3", "-u", "fx.py"]

Makefile

@@ -65,87 +65,26 @@ build_local:
 clean:
 	rm -rf bins
 
-unit-test: export ROUTING = Store&Forward
 unit-test:
 	cd tests && go test unit_test.go $(GO_TEST_FLAGS)
 
-integ-test_CT: export ROUTING = CutThrough
-integ-test_CT:
+integ-test:
 	cd tests && go test ./integration_test.go -run TestSdk_InvokeWithXDT $(GO_TEST_FLAGS)
 
-
-integ-test_SF: export ROUTING = Store&Forward
-integ-test_SF:
-	cd tests && go test ./integration_test.go -run TestSdk_InvokeWithXDT $(GO_TEST_FLAGS)
-
-integ-test-noCopy_CT: export ROUTING = CutThrough
-integ-test-noCopy_CT: export NO_COPY = true
-integ-test-noCopy_CT:
-	cd tests && go test ./integration_test.go -run TestSdk_InvokeWithXDTNoCopy $(GO_TEST_FLAGS)
-
-
-integ-test-noCopy_SF: export ROUTING = Store&Forward
-integ-test-noCopy_SF: export NO_COPY = true
-integ-test-noCopy_SF:
-	cd tests && go test ./integration_test.go -run TestSdk_InvokeWithXDTNoCopy $(GO_TEST_FLAGS)
-
-integ-test: integ-test_CT integ-test_SF
-integ-test-noCopy: integ-test-noCopy_CT integ-test-noCopy_SF
-
-timeout-test_CT: export ROUTING = CutThrough
-timeout-test_CT:
-	sleep 60
-	cd tests && go test ./integration_test.go -run TestErr_DQPTimeout $(GO_TEST_FLAGS)
-
-timeout-test_SF: export ROUTING = Store&Forward
-timeout-test_SF:
-	sleep 60
-	cd tests && go test ./integration_test.go -run TestErr_DQPTimeout $(GO_TEST_FLAGS)
-
-timeout-test: timeout-test_SF timeout-test_CT
-
-parallel-invoke-test_CT: export ROUTING = CutThrough
-parallel-invoke-test_CT:
-	cd tests && go test ./integration_test.go -run TestParallel_Invoke -concurrentCalls 1 $(GO_TEST_FLAGS)
-	cd tests && go test ./integration_test.go -run TestParallel_Invoke -concurrentCalls 2 $(GO_TEST_FLAGS)
-	cd tests && go test ./integration_test.go -run TestParallel_Invoke -concurrentCalls 5 $(GO_TEST_FLAGS)
-
-parallel-invoke-test_SF: export ROUTING = Store&Forward
-parallel-invoke-test_SF:
+parallel-invoke-test:
 	cd tests && go test ./integration_test.go -run TestParallel_Invoke -concurrentCalls 1 $(GO_TEST_FLAGS)
 	cd tests && go test ./integration_test.go -run TestParallel_Invoke -concurrentCalls 2 $(GO_TEST_FLAGS)
 	cd tests && go test ./integration_test.go -run TestParallel_Invoke -concurrentCalls 5 $(GO_TEST_FLAGS)
 
-parallel-invoke-test: parallel-invoke-test_SF parallel-invoke-test_CT
-fan-out-test: fan-out_SF fan-out_CT
-fan-in-test: fan-in_SF fan-in_CT
-
-fan-out_SF: export ROUTING = Store&Forward
-fan-out_SF:
-	cd tests && go test ./integration_test.go -run TestParallel_FanOut -concurrentCalls 1 $(GO_TEST_FLAGS)
-	sleep 2
-	cd tests && go test ./integration_test.go -run TestParallel_FanOut -concurrentCalls 2 $(GO_TEST_FLAGS)
-	sleep 2
-	cd tests && go test ./integration_test.go -run TestParallel_FanOut -concurrentCalls 5 $(GO_TEST_FLAGS)
-	sleep 2
-fan-out_CT: export ROUTING = CutThrough
-fan-out_CT:
+fan-out-test:
 	cd tests && go test ./integration_test.go -run TestParallel_FanOut -concurrentCalls 1 $(GO_TEST_FLAGS)
 	sleep 2
 	cd tests && go test ./integration_test.go -run TestParallel_FanOut -concurrentCalls 2 $(GO_TEST_FLAGS)
 	sleep 2
 	cd tests && go test ./integration_test.go -run TestParallel_FanOut -concurrentCalls 5 $(GO_TEST_FLAGS)
 	sleep 2
 
-fan-in_SF: export ROUTING = Store&Forward
-fan-in_SF:
-	cd tests && go test ./integration_test.go -run TestParallel_FanIn -concurrentCalls 1 $(GO_TEST_FLAGS)
-	sleep 2
-	cd tests && go test ./integration_test.go -run TestParallel_FanIn -concurrentCalls 2 $(GO_TEST_FLAGS)
-	sleep 2
-	cd tests && go test ./integration_test.go -run TestParallel_FanIn -concurrentCalls 5 $(GO_TEST_FLAGS)
-fan-in_CT: export ROUTING = CutThrough
-fan-in_CT:
+fan-in-test:
 	cd tests && go test ./integration_test.go -run TestParallel_FanIn -concurrentCalls 1 $(GO_TEST_FLAGS)
 	sleep 2
 	cd tests && go test ./integration_test.go -run TestParallel_FanIn -concurrentCalls 2 $(GO_TEST_FLAGS)
@@ -155,19 +94,10 @@ fan-in_CT:
 install_python_modules:
 	pip install grpcio --user
 	pip install grpcio-tools --user
+	pip install pycapnp --user
+	pip install environs --user
 
-python-unit-test: export ROUTING = Store&Forward
-python-unit-test: install_python_modules
-	cd tests && go test ./integration_test.go -run TestPython_SDK $(GO_TEST_FLAGS) &
-	sleep 60
-	cd sdk/python && python -m unittest -v test.UnitTest
-	# kill the process bound to the given port.
-	-fuser -k 50005/tcp
-
-python-integ-test: install_python_modules python-integ-test_CT python-integ-test_SF
-
-python-integ-test_CT: export ROUTING = CutThrough
-python-integ-test_CT:
+python-integ-test: install_python_modules
 	cd tests && go test ./integration_test.go -run TestPython_SDK $(GO_TEST_FLAGS) &
 	sleep 60
 	cd sdk/python && python destination.py &
@@ -176,75 +106,31 @@ python-integ-test_CT:
 	-fuser -k 50005/tcp
 	-fuser -k 50007/tcp
 
-python-integ-test_SF: export ROUTING = Store&Forward
-python-integ-test_SF:
-	cd tests && go test ./integration_test.go -run TestPython_SDK $(GO_TEST_FLAGS) &
-	sleep 60
-	cd sdk/python && python destination.py &
-	sleep 5
-	cd sdk/python && python -m unittest -v test.IntegTest.test_Invoke_XDT
-	-fuser -k 50005/tcp
-	-fuser -k 50007/tcp
-
-python-timeout-test: install_python_modules python-timeout-test_CT python-timeout-test_SF
-
-python-timeout-test_CT: export ROUTING = CutThrough
-python-timeout-test_CT:
-	cd tests && go test ./integration_test.go -run TestPython_SDKTimeout $(GO_TEST_FLAGS) &
-	sleep 60
-	cd sdk/python && python -m unittest -v test.IntegTest.test_Timeout
-	-fuser -k 50005/tcp
-
-python-timeout-test_SF: export ROUTING = Store&Forward
-python-timeout-test_SF:
+python-timeout-test: install_python_modules
 	cd tests && go test ./integration_test.go -run TestPython_SDKTimeout $(GO_TEST_FLAGS) &
 	sleep 60
 	cd sdk/python && python -m unittest -v test.IntegTest.test_Timeout
 	-fuser -k 50005/tcp
 
 python-get-put-test: install_python_modules
-python-get-put-test: export ROUTING = Store&Forward
-python-get-put-test:
 	cd tests && go test ./integration_test.go -run TestPython_SDK $(GO_TEST_FLAGS) &
 	sleep 60
 	cd sdk/python && python -m unittest -v test.IntegTest.test_GetPut
 	-fuser -k 50005/tcp
 	-fuser -k 50007/tcp
 
-python-broadcast-get-put-test: export ROUTING = Store&Forward
-python-broadcast-get-put-test:
+python-broadcast-get-put-test: install_python_modules
 	cd tests && go test ./integration_test.go -run TestPython_SDK $(GO_TEST_FLAGS) &
 	sleep 60
 	cd sdk/python && python -m unittest -v test.IntegTest.test_Broadcast_GetPut
 	-fuser -k 50005/tcp
 	-fuser -k 50007/tcp
 
-python-noCopy-get-put-test: export ROUTING = Store&Forward
-python-noCopy-get-put-test: export NO_COPY = true
-python-noCopy-get-put-test:
-	cd tests && go test ./integration_test.go -run TestPython_SDK $(GO_TEST_FLAGS) &
-	sleep 60
-	cd sdk/python && python -m unittest -v test.IntegTest.test_GetPut
-	-fuser -k 50005/tcp
-	-fuser -k 50006/tcp
-	-fuser -k 50007/tcp
+all-images-push:
+	cd user-functions/fx && ko publish ./ -B
+	cd user-functions/dQP && ko publish ./ -B
+	cd user-functions/gx && ko publish ./ -B
 
-python-noCopy-broadcast-get-put-test: export ROUTING = Store&Forward
-python-noCopy-broadcast-get-put-test: export NO_COPY = true
-python-noCopy-broadcast-get-put-test:
-	cd tests && go test ./integration_test.go -run TestPython_SDK $(GO_TEST_FLAGS) &
-	sleep 60
-	cd sdk/python && python -m unittest -v test.IntegTest.test_Broadcast_GetPut
-	-fuser -k 50005/tcp
-	-fuser -k 50006/tcp
-	-fuser -k 50007/tcp
-
-docker-images-push:
-	cd user-functions/fx && ko publish ./ -B --tags=capnp
-	cd user-functions/dQP && ko publish ./ -B --tags=capnp
-	cd user-functions/gx && ko publish ./ -B --tags=capnp
-
-python-images-push:
 	DOCKER_BUILDKIT=1 docker build \
         	-t vhiveease/pyfx:latest \
         	--target pyfx \
@@ -257,14 +143,12 @@ python-images-push:
 			$(ROOT)
 	docker push vhiveease/pygx:latest
 	docker push vhiveease/pyfx:latest
-benchmark-XDT: export ROUTING = CutThrough
+
 benchmark-XDT:
 		cd tests && go test ./integration_test.go -run TestBenchmark_XDT -v
 
-get-put-test: export ROUTING = Store&Forward
 get-put-test:
 	cd tests && go test ./integration_test.go -run TestGet_Put $(GO_TEST_FLAGS)
 
-broadcast-test: export ROUTING = Store&Forward
 broadcast-test:
 	cd tests && go test ./integration_test.go -run TestBroadcast_GetPut $(GO_TEST_FLAGS)

queue-proxy/dQP/dQP.go

@@ -42,13 +42,13 @@ type crossXDTServer struct {
 }
 
 // ServeBroadcastData is a gRPC server to serve data to the DstFn
-func (s crossXDTServer) ServeBroadcastData(ctx context.Context, req crossXDT.StreamData_serveBroadcastData) error {
+func (s crossXDTServer) ServeBroadcastData(_ context.Context, _ crossXDT.StreamData_serveBroadcastData) error {
 	log.Fatal("[dqp] No op. Should not have reached here")
 	return nil
 }
 
 // ServeData is a gRPC server to serve data to the DstFn
-func (s crossXDTServer) ServeData(ctx context.Context, req crossXDT.StreamData_serveData) error {
+func (s crossXDTServer) ServeData(_ context.Context, req crossXDT.StreamData_serveData) error {
 
 	res, err := req.AllocResults() // allocate the results struct
 	if err != nil {
@@ -93,7 +93,12 @@ func PullDataFromSrc(ctx context.Context) error {
 	if err != nil {
 		log.Fatal(err)
 	}
-	defer conn.Close()
+	defer func(conn net.Conn) {
+		err := conn.Close()
+		if err != nil {
+			log.Fatal(err)
+		}
+	}(conn)
 	capconn := rpc.NewConn(rpc.NewStreamTransport(conn), nil)
 	client := crossXDT.StreamData(capconn.Bootstrap(ctx))
 
@@ -110,7 +115,10 @@ func PullDataFromSrc(ctx context.Context) error {
 	//     the RPC call and its results.
 	// do release
 	f, _ := client.ServeData(ctx, func(ps crossXDT.StreamData_serveData_Params) error {
-		ps.SetKey(key)
+		err := ps.SetKey(key)
+		if err != nil {
+			log.Fatal(err)
+		}
 		return nil
 	})
 
@@ -158,7 +166,10 @@ func StartServer(config utils.Config) {
 			// Block until the connection terminates.
 			select {
 			case <-capnpconn.Done():
-				capnpconn.Close()
+				err := capnpconn.Close()
+				if err != nil {
+					log.Fatal(err)
+				}
 				return
 			}
 		}()