omes/scenarios/ebb_and_flow_test.go at main · temporalio/omes · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
package scenarios

import (
	"fmt"
	"testing"
	"time"

	"github.com/stretchr/testify/require"
	"github.com/temporalio/omes/cmd/clioptions"
	"github.com/temporalio/omes/loadgen"
	"github.com/temporalio/omes/workers"
)

func TestEbbAndFlow(t *testing.T) {
	t.Parallel()

	env := workers.SetupTestEnvironment(t,
		workers.WithExecutorTimeout(2*time.Minute))

	sleepActivityJson := `{
		"count": {
			"type": "fixed",
			"value": "5"
		},
		"groups": {
			"group_a": {
				"weight": 1,
				"sleepDuration": {
					"type": "fixed",
					"value": "1ms"
				},
				"fairnessKeys": {
					"type": "fixed",
					"value": "1"
				}
			},
			"group_b": {
				"weight": 1,
				"sleepDuration": {
					"type": "fixed",
					"value": "1ms"
				},
				"fairnessKeys": {
					"type": "fixed",
					"value": "2"
				}
			}
		}
	}`

	scenarioInfo := loadgen.ScenarioInfo{
		RunID: fmt.Sprintf("eaf-%d", time.Now().Unix()),
		Configuration: loadgen.RunConfiguration{
			Duration: 10 * time.Second,
		},
		ScenarioOptions: map[string]string{
			MinBacklogFlag:                    "0",
			MaxBacklogFlag:                    "1",
			PeriodFlag:                        "5s",
			BacklogLogIntervalFlag:            "5s",
			VisibilityVerificationTimeoutFlag: "5s",
			SleepActivityJsonFlag:             sleepActivityJson,
		},
	}

	var state ebbAndFlowState

	t.Run("Run executor", func(t *testing.T) {
		executor := newEbbAndFlowExecutor()

		_, err := env.RunExecutorTest(t, executor, scenarioInfo, clioptions.LangGo)
		require.NoError(t, err, "Executor should complete successfully")

		state = executor.Snapshot().(ebbAndFlowState)
		require.GreaterOrEqual(t, state.TotalCompletedWorkflows, int64(1))
	})

	t.Run("Run executor again, resuming from previous state", func(t *testing.T) {
		executor := newEbbAndFlowExecutor()

		// Load previous state
		previouState := state
		err := executor.LoadState(func(v any) error {
			s := v.(*ebbAndFlowState)
			*s = state
			return nil
		})
		require.NoError(t, err, "Should be able to load previous state")

		resumeScenarioInfo := scenarioInfo

		_, err = env.RunExecutorTest(t, executor, resumeScenarioInfo, clioptions.LangGo)
		require.NoError(t, err, "Executor should complete successfully")

		state = executor.Snapshot().(ebbAndFlowState)
		require.Greater(t, state.TotalCompletedWorkflows, previouState.TotalCompletedWorkflows)
	})

	t.Run("Run executor again, resuming from previous state but without any time left", func(t *testing.T) {
		executor := newEbbAndFlowExecutor()

		// Load previous state
		err := executor.LoadState(func(v any) error {
			s := v.(*ebbAndFlowState)
			*s = state
			return nil
		})
		require.NoError(t, err, "Should be able to load previous state")

		resumeScenarioInfo := scenarioInfo
		resumeScenarioInfo.Configuration.Duration = 0 // ie no more time left

		_, err = env.RunExecutorTest(t, executor, resumeScenarioInfo, clioptions.LangGo)
		require.NoError(t, err, "Executor should complete successfully")
	})
}