floxy/engine_join_notify_test.go at main · floxy-project/floxy · 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
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
package floxy

import (
	"context"
	"encoding/json"
	"testing"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
)

func Test_notifyJoinSteps_Ready_NoPending_Enqueues(t *testing.T) {
	ctx := context.Background()
	engine, store := newTestEngineWithStore(t)

	def := buildForkJoinDirectWaitForDef() // Fork F -> A,B ; Join J.WaitFor=[A,B]
	instanceID := int64(7001)

	store.EXPECT().GetInstance(mock.Anything, instanceID).
		Return(&WorkflowInstance{ID: instanceID, WorkflowID: def.ID, Status: StatusRunning}, nil).Maybe()
	store.EXPECT().GetWorkflowDefinition(mock.Anything, def.ID).Return(def, nil).Maybe()

	// Steps in the instance: A and B completed; no join step yet
	steps := []WorkflowStep{
		{InstanceID: instanceID, StepName: "A", StepType: StepTypeTask, Status: StepStatusCompleted, Input: json.RawMessage(`{"a":1}`)},
		{InstanceID: instanceID, StepName: "B", StepType: StepTypeTask, Status: StepStatusCompleted},
	}
	store.EXPECT().GetStepsByInstance(mock.Anything, instanceID).Return(steps, nil)

	// Join state not yet created -> rely on stepDef.WaitFor
	store.EXPECT().GetJoinState(mock.Anything, instanceID, "J").Return(nil, ErrEntityNotFound)

	// Mark join as ready for completed step A
	store.EXPECT().UpdateJoinState(mock.Anything, instanceID, "J", "A", true).Return(true, nil)

	// Log event for join state update
	store.EXPECT().LogEvent(mock.Anything, instanceID, (*int64)(nil), EventJoinUpdated, mock.Anything).Return(nil).Maybe()

	// Since ready and no pending extras, the engine should create and enqueue join step J
	store.EXPECT().CreateStep(mock.Anything, mock.MatchedBy(func(s *WorkflowStep) bool {
		return s != nil && s.InstanceID == instanceID && s.StepName == "J" && s.StepType == StepTypeJoin && s.Status == StepStatusPending
	})).Return(nil)
	store.EXPECT().EnqueueStep(mock.Anything, instanceID, mock.AnythingOfType("*int64"), PriorityNormal, timeZero()).Return(nil)
	store.EXPECT().LogEvent(mock.Anything, instanceID, mock.AnythingOfType("*int64"), EventJoinReady, mock.Anything).Return(nil).Maybe()

	err := engine.notifyJoinSteps(ctx, instanceID, "A", true)
	assert.NoError(t, err)
}

func Test_notifyJoinSteps_NotReady_DueToPendingExtra(t *testing.T) {
	ctx := context.Background()
	engine, store := newTestEngineWithStore(t)

	def := buildForkJoinDirectWaitForDef()
	instanceID := int64(7002)

	store.EXPECT().GetInstance(mock.Anything, instanceID).
		Return(&WorkflowInstance{ID: instanceID, WorkflowID: def.ID, Status: StatusRunning}, nil).Maybe()
	store.EXPECT().GetWorkflowDefinition(mock.Anything, def.ID).Return(def, nil).Maybe()

	// Steps include extra pending step A1 within parallel branch A to block join
	steps := []WorkflowStep{
		{InstanceID: instanceID, StepName: "A", StepType: StepTypeTask, Status: StepStatusCompleted},
		{InstanceID: instanceID, StepName: "B", StepType: StepTypeTask, Status: StepStatusCompleted},
		{InstanceID: instanceID, StepName: "A1", StepType: StepTypeTask, Status: StepStatusPending}, // pending extra in branch
	}
	store.EXPECT().GetStepsByInstance(mock.Anything, instanceID).Return(steps, nil)

	store.EXPECT().GetJoinState(mock.Anything, instanceID, "J").Return(nil, ErrEntityNotFound)

	// First call indicates ready according to join strategy
	store.EXPECT().UpdateJoinState(mock.Anything, instanceID, "J", "A", true).Return(true, nil)
	// Engine will reset readiness due to pending extras by calling UpdateJoinState again
	store.EXPECT().UpdateJoinState(mock.Anything, instanceID, "J", "A", true).Return(false, nil)

	store.EXPECT().LogEvent(mock.Anything, instanceID, (*int64)(nil), EventJoinUpdated, mock.Anything).Return(nil).Maybe()

	// No CreateStep/EnqueueStep expected due to not ready
	err := engine.notifyJoinSteps(ctx, instanceID, "A", true)
	assert.NoError(t, err)
}

func Test_notifyJoinStepsForStep_Ready_Enqueues(t *testing.T) {
	ctx := context.Background()
	engine, store := newTestEngineWithStore(t)

	def := buildForkJoinDirectWaitForDef()
	instanceID := int64(7003)

	store.EXPECT().GetInstance(mock.Anything, instanceID).
		Return(&WorkflowInstance{ID: instanceID, WorkflowID: def.ID, Status: StatusRunning}, nil)
	store.EXPECT().GetStepsByInstance(mock.Anything, instanceID).Return([]WorkflowStep{
		{InstanceID: instanceID, StepName: "A", StepType: StepTypeTask, Status: StepStatusCompleted, Input: json.RawMessage(`{"a":2}`)},
		{InstanceID: instanceID, StepName: "B", StepType: StepTypeTask, Status: StepStatusCompleted},
	}, nil)
	store.EXPECT().GetWorkflowDefinition(mock.Anything, def.ID).Return(def, nil)

	// Update join state for a specific join
	store.EXPECT().UpdateJoinState(mock.Anything, instanceID, "J", "A", true).Return(true, nil)
	store.EXPECT().LogEvent(mock.Anything, instanceID, (*int64)(nil), EventJoinUpdated, mock.Anything).Return(nil).Maybe()

	store.EXPECT().CreateStep(mock.Anything, mock.MatchedBy(func(s *WorkflowStep) bool {
		return s.StepName == "J" && s.Status == StepStatusPending
	})).Return(nil)
	store.EXPECT().EnqueueStep(mock.Anything, instanceID, mock.AnythingOfType("*int64"), PriorityNormal, timeZero()).Return(nil)
	store.EXPECT().LogEvent(mock.Anything, instanceID, mock.AnythingOfType("*int64"), EventJoinReady, mock.Anything).Return(nil).Maybe()

	err := engine.notifyJoinStepsForStep(ctx, instanceID, "J", "A", true)
	assert.NoError(t, err)
}

func Test_notifyJoinStepsForStep_DLQ_Pauses(t *testing.T) {
	ctx := context.Background()
	engine, store := newTestEngineWithStore(t)

	def := buildForkJoinDirectWaitForDef()
	instanceID := int64(7004)

	store.EXPECT().GetInstance(mock.Anything, instanceID).
		Return(&WorkflowInstance{ID: instanceID, WorkflowID: def.ID, Status: StatusDLQ}, nil)
	store.EXPECT().GetStepsByInstance(mock.Anything, instanceID).Return([]WorkflowStep{
		{InstanceID: instanceID, StepName: "A", StepType: StepTypeTask, Status: StepStatusCompleted},
		{InstanceID: instanceID, StepName: "B", StepType: StepTypeTask, Status: StepStatusCompleted},
	}, nil)
	store.EXPECT().GetWorkflowDefinition(mock.Anything, def.ID).Return(def, nil)

	store.EXPECT().UpdateJoinState(mock.Anything, instanceID, "J", "A", true).Return(true, nil)
	store.EXPECT().LogEvent(mock.Anything, instanceID, (*int64)(nil), EventJoinUpdated, mock.Anything).Return(nil).Maybe()

	// Should create paused step and NOT enqueue
	store.EXPECT().CreateStep(mock.Anything, mock.MatchedBy(func(s *WorkflowStep) bool {
		return s.StepName == "J" && s.Status == StepStatusPaused
	})).Return(nil)
	store.EXPECT().LogEvent(mock.Anything, instanceID, mock.AnythingOfType("*int64"), EventJoinReady, mock.Anything).Return(nil).Maybe()

	err := engine.notifyJoinStepsForStep(ctx, instanceID, "J", "A", true)
	assert.NoError(t, err)
}

// timeZero returns a time.Duration literal 0 in a way acceptable to testify's argument matching for EnqueueStep delay param.
// We use a helper to avoid importing time package in this test file explicitly for matching purposes.
func timeZero() interface{} { return mock.Anything } // keep matcher generic for delay