Migrate Zeebe tests to certification tests (#3112)

Signed-off-by: Christian Kaps <[email protected]>
Signed-off-by: Bernd Verst <[email protected]>
Co-authored-by: Bernd Verst <[email protected]>

Author: akkie

.github/scripts/test-info.mjs

@@ -120,6 +120,12 @@ const components = {
     'bindings.dubbo': {
         certification: true,
     },
+    'bindings.zeebe.command': {
+        certification: true,
+    },
+    'bindings.zeebe.jobworker': {
+        certification: true,
+    },
     'bindings.http': {
         conformance: true,
     },

Makefile

@@ -249,9 +249,4 @@ prettier-format:
 ################################################################################
 .PHONY: conf-tests
 conf-tests:
-	CGO_ENABLED=$(CGO) go test -v -tags=conftests -count=1 ./tests/conformance
-
-################################################################################
-# Target: e2e                                                                #
-################################################################################
-include tests/e2e/e2e_tests.mk
+	CGO_ENABLED=$(CGO) go test -v -tags=conftests -count=1 ./tests/conformance

bindings/zeebe/command/activate_jobs.go

@@ -20,6 +20,8 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/camunda/zeebe/clients/go/v8/pkg/entities"
+
 	"github.com/dapr/components-contrib/bindings"
 	"github.com/dapr/components-contrib/metadata"
 )
@@ -35,6 +37,7 @@ type activateJobsPayload struct {
 	Timeout           metadata.Duration `json:"timeout"`
 	WorkerName        string            `json:"workerName"`
 	FetchVariables    []string          `json:"fetchVariables"`
+	RequestTimeout    metadata.Duration `json:"requestTimeout"`
 }
 
 func (z *ZeebeCommand) activateJobs(ctx context.Context, req *bindings.InvokeRequest) (*bindings.InvokeResponse, error) {
@@ -68,7 +71,15 @@ func (z *ZeebeCommand) activateJobs(ctx context.Context, req *bindings.InvokeReq
 		cmd = cmd.FetchVariables(payload.FetchVariables...)
 	}
 
-	response, err := cmd.Send(ctx)
+	var response []entities.Job
+	if payload.RequestTimeout.Duration != time.Duration(0) {
+		ctxWithTimeout, cancel := context.WithTimeout(ctx, payload.RequestTimeout.Duration)
+		defer cancel()
+		response, err = cmd.Send(ctxWithTimeout)
+	} else {
+		response, err = cmd.Send(ctx)
+	}
+
 	if err != nil {
 		return nil, fmt.Errorf("cannot activate jobs for type %s: %w", payload.JobType, err)
 	}

bindings/zeebe/command/command.go

@@ -30,7 +30,8 @@ import (
 const (
 	// operations.
 	TopologyOperation         bindings.OperationKind = "topology"
-	DeployProcessOperation    bindings.OperationKind = "deploy-process"
+	DeployResourceOperation   bindings.OperationKind = "deploy-resource"
+	DeployProcessOperation    bindings.OperationKind = "deploy-process" // Deprecated, kept for backward compatibility
 	CreateInstanceOperation   bindings.OperationKind = "create-instance"
 	CancelInstanceOperation   bindings.OperationKind = "cancel-instance"
 	SetVariablesOperation     bindings.OperationKind = "set-variables"
@@ -78,6 +79,7 @@ func (z *ZeebeCommand) Operations() []bindings.OperationKind {
 	return []bindings.OperationKind{
 		TopologyOperation,
 		DeployProcessOperation,
+		DeployResourceOperation,
 		CreateInstanceOperation,
 		CancelInstanceOperation,
 		SetVariablesOperation,
@@ -95,8 +97,8 @@ func (z *ZeebeCommand) Invoke(ctx context.Context, req *bindings.InvokeRequest)
 	switch req.Operation {
 	case TopologyOperation:
 		return z.topology(ctx)
-	case DeployProcessOperation:
-		return z.deployProcess(ctx, req)
+	case DeployResourceOperation, DeployProcessOperation:
+		return z.deployResource(ctx, req)
 	case CreateInstanceOperation:
 		return z.createInstance(ctx, req)
 	case CancelInstanceOperation:

bindings/zeebe/command/command_test.go

@@ -93,17 +93,18 @@ func TestInvoke(t *testing.T) {
 func TestOperations(t *testing.T) {
 	testBinding := ZeebeCommand{logger: logger.NewLogger("test")}
 	operations := testBinding.Operations()
-	require.Equal(t, 12, len(operations))
+	require.Equal(t, 13, len(operations))
 	assert.Equal(t, TopologyOperation, operations[0])
 	assert.Equal(t, DeployProcessOperation, operations[1])
-	assert.Equal(t, CreateInstanceOperation, operations[2])
-	assert.Equal(t, CancelInstanceOperation, operations[3])
-	assert.Equal(t, SetVariablesOperation, operations[4])
-	assert.Equal(t, ResolveIncidentOperation, operations[5])
-	assert.Equal(t, PublishMessageOperation, operations[6])
-	assert.Equal(t, ActivateJobsOperation, operations[7])
-	assert.Equal(t, CompleteJobOperation, operations[8])
-	assert.Equal(t, FailJobOperation, operations[9])
-	assert.Equal(t, UpdateJobRetriesOperation, operations[10])
-	assert.Equal(t, ThrowErrorOperation, operations[11])
+	assert.Equal(t, DeployResourceOperation, operations[2])
+	assert.Equal(t, CreateInstanceOperation, operations[3])
+	assert.Equal(t, CancelInstanceOperation, operations[4])
+	assert.Equal(t, SetVariablesOperation, operations[5])
+	assert.Equal(t, ResolveIncidentOperation, operations[6])
+	assert.Equal(t, PublishMessageOperation, operations[7])
+	assert.Equal(t, ActivateJobsOperation, operations[8])
+	assert.Equal(t, CompleteJobOperation, operations[9])
+	assert.Equal(t, FailJobOperation, operations[10])
+	assert.Equal(t, UpdateJobRetriesOperation, operations[11])
+	assert.Equal(t, ThrowErrorOperation, operations[12])
 }

bindings/zeebe/command/create_instance.go

@@ -18,10 +18,12 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"time"
 
 	"github.com/camunda/zeebe/clients/go/v8/pkg/commands"
 
 	"github.com/dapr/components-contrib/bindings"
+	"github.com/dapr/components-contrib/metadata"
 )
 
 var (
@@ -30,10 +32,13 @@ var (
 )
 
 type createInstancePayload struct {
-	BpmnProcessID        string      `json:"bpmnProcessId"`
-	ProcessDefinitionKey *int64      `json:"processDefinitionKey"`
-	Version              *int32      `json:"version"`
-	Variables            interface{} `json:"variables"`
+	BpmnProcessID        string            `json:"bpmnProcessId"`
+	ProcessDefinitionKey *int64            `json:"processDefinitionKey"`
+	Version              *int32            `json:"version"`
+	Variables            interface{}       `json:"variables"`
+	WithResult           bool              `json:"withResult"`
+	FetchVariables       []string          `json:"fetchVariables"`
+	RequestTimeout       metadata.Duration `json:"requestTimeout"`
 }
 
 func (z *ZeebeCommand) createInstance(ctx context.Context, req *bindings.InvokeRequest) (*bindings.InvokeResponse, error) {
@@ -75,9 +80,25 @@ func (z *ZeebeCommand) createInstance(ctx context.Context, req *bindings.InvokeR
 		}
 	}
 
-	response, err := cmd3.Send(ctx)
+	var response interface{}
+	// The request timeout has only an affect if WithResult is used. Using WithResult means that the operation is
+	// synchronous instead of asynchronous, and the request timeout defines how long the client should wait for the
+	// workflow/process to finish to get the result.
+	//
+	// From a code perspective, there are two Send methods in the Zeebe client. One if WithResult was used and
+	// which extracts the request timeout from the context and another one which will not use any timeout.
+	if payload.WithResult && payload.RequestTimeout.Duration != time.Duration(0) {
+		ctxWithTimeout, cancel := context.WithTimeout(ctx, payload.RequestTimeout.Duration)
+		defer cancel()
+		response, err = cmd3.WithResult().FetchVariables(payload.FetchVariables...).Send(ctxWithTimeout)
+	} else if payload.WithResult {
+		response, err = cmd3.WithResult().FetchVariables(payload.FetchVariables...).Send(ctx)
+	} else {
+		response, err = cmd3.Send(ctx)
+	}
+
 	if err != nil {
-		return nil, fmt.Errorf("cannot create instane for %s: %w", errorDetail, err)
+		return nil, fmt.Errorf("cannot create instance for %s: %w", errorDetail, err)
 	}
 
 	jsonResponse, err := json.Marshal(response)

bindings/zeebe/command/deploy_process.go renamed to bindings/zeebe/command/deploy_resource.go

@@ -29,7 +29,7 @@ const (
 
 var ErrMissingFileName = errors.New("fileName is a required attribute")
 
-func (z *ZeebeCommand) deployProcess(ctx context.Context, req *bindings.InvokeRequest) (*bindings.InvokeResponse, error) {
+func (z *ZeebeCommand) deployResource(ctx context.Context, req *bindings.InvokeRequest) (*bindings.InvokeResponse, error) {
 	var deployFileName string
 
 	if val, ok := req.Metadata[fileName]; ok && val != "" {
@@ -42,7 +42,7 @@ func (z *ZeebeCommand) deployProcess(ctx context.Context, req *bindings.InvokeRe
 		AddResource(req.Data, deployFileName).
 		Send(ctx)
 	if err != nil {
-		return nil, fmt.Errorf("cannot deploy process with fileName %s: %w", deployFileName, err)
+		return nil, fmt.Errorf("cannot deploy resource with fileName %s: %w", deployFileName, err)
 	}
 
 	jsonResponse, err := json.Marshal(response)

bindings/zeebe/command/deploy_process_test.go renamed to bindings/zeebe/command/deploy_resource_test.go

@@ -23,12 +23,12 @@ import (
 	"github.com/dapr/kit/logger"
 )
 
-func TestDeployProcess(t *testing.T) {
+func TestDeployResource(t *testing.T) {
 	testLogger := logger.NewLogger("test")
 
 	t.Run("fileName is mandatory", func(t *testing.T) {
 		cmd := ZeebeCommand{logger: testLogger}
-		req := &bindings.InvokeRequest{Operation: DeployProcessOperation}
+		req := &bindings.InvokeRequest{Operation: DeployResourceOperation}
 		_, err := cmd.Invoke(context.TODO(), req)
 		assert.Error(t, err, ErrMissingFileName)
 	})

bindings/zeebe/command/fail_job.go

@@ -18,16 +18,22 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"time"
+
+	"github.com/camunda/zeebe/clients/go/v8/pkg/commands"
 
 	"github.com/dapr/components-contrib/bindings"
+	"github.com/dapr/components-contrib/metadata"
 )
 
 var ErrMissingRetries = errors.New("retries is a required attribute")
 
 type failJobPayload struct {
-	JobKey       *int64 `json:"jobKey"`
-	Retries      *int32 `json:"retries"`
-	ErrorMessage string `json:"errorMessage"`
+	JobKey       *int64            `json:"jobKey"`
+	Retries      *int32            `json:"retries"`
+	ErrorMessage string            `json:"errorMessage"`
+	RetryBackOff metadata.Duration `json:"retryBackOff"`
+	Variables    interface{}       `json:"variables"`
 }
 
 func (z *ZeebeCommand) failJob(ctx context.Context, req *bindings.InvokeRequest) (*bindings.InvokeResponse, error) {
@@ -53,7 +59,19 @@ func (z *ZeebeCommand) failJob(ctx context.Context, req *bindings.InvokeRequest)
 		cmd = cmd.ErrorMessage(payload.ErrorMessage)
 	}
 
-	_, err = cmd.Send(ctx)
+	if payload.RetryBackOff.Duration != time.Duration(0) {
+		cmd = cmd.RetryBackoff(payload.RetryBackOff.Duration)
+	}
+
+	var cmdDispatch commands.DispatchFailJobCommand = cmd
+	if payload.Variables != nil {
+		cmdDispatch, err = cmd.VariablesFromObject(payload.Variables)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	_, err = cmdDispatch.Send(ctx)
 	if err != nil {
 		return nil, fmt.Errorf("cannot fail job for key %d: %w", payload.JobKey, err)
 	}

bindings/zeebe/command/metadata.yaml

@@ -15,7 +15,9 @@ binding:
     - name: topology
       description: "Obtains the current topology of the cluster the gateway is part of."
     - name: deploy-process
-      description: "Deploys a single process to Zeebe."
+      description: "Deprecated alias of 'deploy-resource'."
+    - name: deploy-resource
+      description: "Deploys a single resource to Zeebe."
     - name: create-instance
       description: "Creates and starts an instance of the specified process."
     - name: cancel-instance