Create process instance for specific version

go-chaos/cmd/verify.go

@@ -15,15 +15,18 @@
 package cmd
 
 import (
-	"context"
 	"fmt"
 	"time"
 
-	"github.com/camunda/zeebe/clients/go/v8/pkg/zbc"
 	"github.com/spf13/cobra"
 	"github.com/zeebe-io/zeebe-chaos/go-chaos/internal"
 )
 
+var (
+	version       int
+	bpmnProcessId string
+)
+
 func init() {
 	rootCmd.AddCommand(verifyCmd)
 	verifyCmd.AddCommand(verifyReadinessCmd)
@@ -33,6 +36,8 @@ func init() {
 	verifySteadyStateCmd.Flags().StringVar(&processModelPath, "processModelPath", "", "Specify the path to a BPMN process model, which should be deployed and an instance should be created of.")
 	verifySteadyStateCmd.Flags().StringVar(&variables, "variables", "", "Specify the variables for the process instance. Expect json string.")
 	verifySteadyStateCmd.Flags().BoolVar(&awaitResult, "awaitResult", false, "Specify whether the completion of the created process instance should be awaited.")
+	verifySteadyStateCmd.Flags().IntVar(&version, "version", -1, "Specify the version for which the instance should be created.")
+	verifySteadyStateCmd.Flags().StringVar(&bpmnProcessId, "bpmnProcessId", "", "Specify the BPMN process ID for which the instance should be created.")
 }
 
 var verifyCmd = &cobra.Command{
@@ -48,14 +53,10 @@ var verifyReadinessCmd = &cobra.Command{
 	Run: func(cmd *cobra.Command, args []string) {
 
 		k8Client, err := internal.CreateK8Client()
-		if err != nil {
-			panic(err)
-		}
+		ensureNoError(err)
 
 		err = k8Client.AwaitReadiness()
-		if err != nil {
-			panic(err.Error())
-		}
+		ensureNoError(err)
 
 		fmt.Printf("All Zeebe nodes are running.\n")
 	},
@@ -68,58 +69,27 @@ var verifySteadyStateCmd = &cobra.Command{
 A process model will be deployed and process instances are created until the required partition is reached.`,
 	Run: func(cmd *cobra.Command, args []string) {
 		k8Client, err := internal.CreateK8Client()
-		if err != nil {
-			panic(err)
-		}
+		ensureNoError(err)
 
 		port := 26500
 		closeFn := k8Client.MustGatewayPortForward(port, port)
 		defer closeFn()
 
 		zbClient, err := internal.CreateZeebeClient(port)
-		if err != nil {
-			panic(err.Error())
-		}
+		ensureNoError(err)
 		defer zbClient.Close()
 
-		processDefinitionKey, err := internal.DeployModel(zbClient, processModelPath)
-		if err != nil {
-			panic(err.Error())
-		}
-
-		err = internal.CreateProcessInstanceOnPartition(func() (int64, error) {
-			return createInstance(zbClient, processDefinitionKey)
-		}, int32(partitionId), 30*time.Second)
-		if err != nil {
-			panic(err.Error())
-		}
+		processInstanceCreator, err := internal.CreateProcessInstanceCreator(zbClient, internal.ProcessInstanceCreationOptions{
+			BpmnProcessId:    bpmnProcessId,
+			Version:          int32(version),
+			ProcessModelPath: processModelPath,
+			AwaitResult:      awaitResult,
+			Variables:        variables,
+		})
+		ensureNoError(err)
+		err = internal.CreateProcessInstanceOnPartition(processInstanceCreator, int32(partitionId), 30*time.Second)
+		ensureNoError(err)
 
 		fmt.Printf("The steady-state was successfully verified!\n")
 	},
 }
-
-func createInstance(zbClient zbc.Client, processDefinitionKey int64) (int64, error) {
-	if Verbose {
-		fmt.Printf("Create process instance with defition key %d [variables: '%s', awaitResult: %t]\n", processDefinitionKey, variables, awaitResult)
-	}
-
-	commandStep3 := zbClient.NewCreateInstanceCommand().ProcessDefinitionKey(processDefinitionKey)
-	if len(variables) != 0 {
-		_, err := commandStep3.VariablesFromString(variables)
-		if err != nil {
-			return 0, err
-		}
-	}
-	if awaitResult {
-		instanceWithResultResponse, err := commandStep3.WithResult().Send(context.TODO())
-		if err != nil {
-			return 0, err
-		}
-		return instanceWithResultResponse.ProcessInstanceKey, nil
-	}
-	instanceResponse, err := commandStep3.Send(context.TODO())
-	if err != nil {
-		return 0, err
-	}
-	return instanceResponse.ProcessInstanceKey, nil
-}

go-chaos/internal/fake.go

@@ -0,0 +1,81 @@
+// Copyright 2022 Camunda Services GmbH
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package internal
+
+import (
+	"context"
+
+	"github.com/camunda/zeebe/clients/go/v8/pkg/commands"
+	"github.com/camunda/zeebe/clients/go/v8/pkg/pb"
+	"github.com/camunda/zeebe/clients/go/v8/pkg/zbc"
+)
+
+/*
+Fake implementation of the Zeebe client.
+
+Can be used for unit tests to verify whether the right properties are set. Should be continously extended to
+increase test coverage.
+*/
+type FakeClient struct {
+	zbc.Client
+	commands.CreateInstanceCommandStep1
+	commands.CreateInstanceCommandStep2
+	commands.CreateInstanceCommandStep3
+	commands.DispatchCreateInstanceCommand
+
+	fakeResultCommand FakeResultCommand
+
+	processId   string
+	version     int32
+	vars        string
+	awaitResult bool
+}
+
+type FakeResultCommand struct {
+	commands.CreateInstanceWithResultCommandStep1
+	commands.DispatchCreateInstanceWithResultCommand
+}
+
+func (f *FakeClient) NewCreateInstanceCommand() commands.CreateInstanceCommandStep1 {
+	return f
+}
+
+func (f *FakeClient) BPMNProcessId(id string) commands.CreateInstanceCommandStep2 {
+	f.processId = id
+	return f
+}
+
+func (f *FakeClient) Version(v int32) commands.CreateInstanceCommandStep3 {
+	f.version = v
+	return f
+}
+
+func (f *FakeClient) VariablesFromString(json string) (commands.CreateInstanceCommandStep3, error) {
+	f.vars = json
+	return f, nil
+}
+
+func (f *FakeClient) WithResult() commands.CreateInstanceWithResultCommandStep1 {
+	f.awaitResult = true
+	return &f.fakeResultCommand
+}
+
+func (f *FakeClient) Send(ctx context.Context) (*pb.CreateProcessInstanceResponse, error) {
+	return &pb.CreateProcessInstanceResponse{ProcessInstanceKey: 0xCAFE}, nil
+}
+
+func (f *FakeResultCommand) Send(ctx context.Context) (*pb.CreateProcessInstanceWithResultResponse, error) {
+	return &pb.CreateProcessInstanceWithResultResponse{ProcessInstanceKey: 0xCAFE}, nil
+}

go-chaos/internal/zeebe.go

@@ -22,6 +22,7 @@ import (
 	"os"
 	"time"
 
+	"github.com/camunda/zeebe/clients/go/v8/pkg/commands"
 	"github.com/camunda/zeebe/clients/go/v8/pkg/pb"
 	"github.com/camunda/zeebe/clients/go/v8/pkg/zbc"
 	"google.golang.org/grpc"
@@ -257,6 +258,67 @@ func readBPMNFileOrDefault(fileName string) ([]byte, string, error) {
 	return bpmnBytes, fileName, nil
 }
 
+type ProcessInstanceCreationOptions struct {
+	Version          int32
+	BpmnProcessId    string
+	ProcessModelPath string
+	Variables        string
+	AwaitResult      bool
+}
+
+func CreateProcessInstanceCreator(zbClient zbc.Client, options ProcessInstanceCreationOptions) (ProcessInstanceCreator, error) {
+	var processInstanceCreator ProcessInstanceCreator
+	if options.Version > 0 {
+		if Verbosity {
+			fmt.Printf("Create process instance with BPMN process ID %s and version %d [variables: '%s', awaitResult: %t]\n",
+				options.BpmnProcessId, options.Version, options.Variables, options.AwaitResult)
+		}
+
+		processInstanceCreator = func() (int64, error) {
+			commandStep3 := zbClient.NewCreateInstanceCommand().BPMNProcessId(options.BpmnProcessId).Version(int32(options.Version))
+			return createInstanceWithCommand(commandStep3, options)
+		}
+	} else {
+		processDefinitionKey, err := DeployModel(zbClient, options.ProcessModelPath)
+		if err != nil {
+			return nil, err
+		}
+
+		if Verbosity {
+			fmt.Printf("Create process instance with defition key %d [variables: '%s', awaitResult: %t]\n", processDefinitionKey, options.Variables, options.AwaitResult)
+		}
+
+		processInstanceCreator = func() (int64, error) {
+			commandStep3 := zbClient.NewCreateInstanceCommand().ProcessDefinitionKey(processDefinitionKey)
+
+			return createInstanceWithCommand(commandStep3, options)
+		}
+	}
+	return processInstanceCreator, nil
+}
+
+func createInstanceWithCommand(commandStep3 commands.CreateInstanceCommandStep3, options ProcessInstanceCreationOptions) (int64, error) {
+	if len(options.Variables) != 0 {
+		_, err := commandStep3.VariablesFromString(options.Variables)
+		if err != nil {
+			return 0, err
+		}
+	}
+
+	if options.AwaitResult {
+		instanceWithResultResponse, err := commandStep3.WithResult().Send(context.TODO())
+		if err != nil {
+			return 0, err
+		}
+		return instanceWithResultResponse.ProcessInstanceKey, nil
+	}
+	instanceResponse, err := commandStep3.Send(context.TODO())
+	if err != nil {
+		return 0, err
+	}
+	return instanceResponse.ProcessInstanceKey, nil
+}
+
 type ProcessInstanceCreator func() (int64, error)
 
 func CreateProcessInstanceOnPartition(piCreator ProcessInstanceCreator, requiredPartition int32, timeout time.Duration) error {

go-chaos/internal/zeebe_test.go

@@ -22,6 +22,7 @@ import (
 
 	"github.com/camunda/zeebe/clients/go/v8/pkg/pb"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func Test_ExtractNodeId(t *testing.T) {
@@ -327,3 +328,54 @@ func Test_ShouldReadGivenFile(t *testing.T) {
 	err = os.RemoveAll(fileName)
 	assert.NoError(t, err)
 }
+
+func Test_ShouldSetVersionAndProcessIdWhenUsingPICreator(t *testing.T) {
+	// given
+	options := ProcessInstanceCreationOptions{BpmnProcessId: "processId", Version: 10}
+	fakeClient := &FakeClient{}
+	creator, err := CreateProcessInstanceCreator(fakeClient, options)
+	require.NoError(t, err)
+
+	// when
+	processInstanceKey, err := creator()
+
+	// then
+	assert.Equal(t, int64(0xCAFE), processInstanceKey)
+	assert.Equal(t, int32(10), fakeClient.version)
+	assert.Equal(t, "processId", fakeClient.processId)
+}
+
+func Test_ShouldSetVariablesForVersionAndProcessIdWhenUsingPICreator(t *testing.T) {
+	// given
+	options := ProcessInstanceCreationOptions{BpmnProcessId: "processId", Version: 10, Variables: "{\"foo\":123}"}
+	fakeClient := &FakeClient{}
+	creator, err := CreateProcessInstanceCreator(fakeClient, options)
+	require.NoError(t, err)
+
+	// when
+	processInstanceKey, err := creator()
+
+	// then
+	assert.Equal(t, int64(0xCAFE), processInstanceKey)
+	assert.Equal(t, int32(10), fakeClient.version)
+	assert.Equal(t, "processId", fakeClient.processId)
+	assert.Equal(t, "{\"foo\":123}", fakeClient.vars)
+}
+
+func Test_ShouldAwaitResultForProcessInstanceWithVersionAndProcessIdWhenUsingPICreator(t *testing.T) {
+	// given
+	options := ProcessInstanceCreationOptions{BpmnProcessId: "processId", Version: 10, Variables: "{\"foo\":123}", AwaitResult: true}
+	fakeClient := &FakeClient{}
+	creator, err := CreateProcessInstanceCreator(fakeClient, options)
+	require.NoError(t, err)
+
+	// when
+	processInstanceKey, err := creator()
+
+	// then
+	assert.Equal(t, int64(0xCAFE), processInstanceKey)
+	assert.Equal(t, int32(10), fakeClient.version)
+	assert.Equal(t, "processId", fakeClient.processId)
+	assert.Equal(t, "{\"foo\":123}", fakeClient.vars)
+	assert.True(t, fakeClient.awaitResult)
+}