Create process instance for specific version (#246)

Allows creating a process instance for a specific version and BPMN
process id via the verify steady-state command.
This is useful for the deployment distribution command.

Furthermore, I also created the first version of a Zeebe Fake client, in
order to intercept values which are set on the client requests. This
allows us to create more unit tests where we can verify whether the
right properties and values are set.

Added some new tests for the process instance creation via version and
BPMN process id.

closes #244

-----


**Example:**

Deploy model
```sh
$ ./zbchaos deploy process -v
Connecting to zell-chaos
Running experiment in self-managed environment.
Successfully created port forwarding tunnel
Deploy 10 versions of different type of models.
Deployed [2/10] versions.
Deployed [4/10] versions.
Deployed [6/10] versions.
Deployed [8/10] versions.
Deployed [10/10] versions.
Deployed different process models of different types and versions to zeebe!

```

Non-verbose: Start instance for a specific version
```
$ ./zbchaos verify steady-state --version 10 --bpmnProcessId multiVersion
The steady-state was successfully verified!
```

We can see in operate:


![execution](https://user-images.githubusercontent.com/2758593/203263224-a24a9f70-c956-4718-8fe7-50cde4c7610b.png)


Verbose: Start instance for a specific version

```sh
$ ./zbchaos verify steady-state --version 10 --bpmnProcessId multiVersion -v
Connecting to zell-chaos
Running experiment in self-managed environment.
Successfully created port forwarding tunnel
Create process instance with BPMN process ID multiVersion and version 10 [variables: '', awaitResult: false]
Created process instance with key 4503599627370497 on partition 2, required partition 1.
Created process instance with key 6755399441055745 on partition 3, required partition 1.
Created process instance with key 2251799813685299 on partition 1, required partition 1.
The steady-state was successfully verified!
```



![execution2](https://user-images.githubusercontent.com/2758593/203263565-cd7822cf-d0e0-4976-95f5-aecc41bd7cdf.png)

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)
+}