Github/temporal
1
0
Fork 0
mirror of https://github.com/temporalio/temporal.git synced 2026-08-30 18:41:49 -07:00
Code Issues Packages Projects Releases Wiki Activity

Fix visibility tests (#10644)

Browse Source 
## What changed?
Fix visibility tests
Also refactored the tests to use `await.Require` instead of retrying
with sleep.

## Why?
PR https://github.com/temporalio/temporal/pull/10540 caused several
tests to fail.

## How did you test it?
- [ ] built
- [ ] run locally and tested manually
- [ ] covered by existing tests
- [ ] added new unit test(s)
- [ ] added new functional test(s)

## Potential risks
This commit is contained in:
Rodrigo Zhou
2026-06-10 18:01:42 -07:00
committed by GitHub
parent 921f9444c2
commit 87ece67d61
6 changed files with 303 additions and 269 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
service/history/queues/priority_assigner_test.go
View File

@@ -48,6 +48,7 @@ func (s *priorityAssignerSuite) TearDownTest() {
func (s *priorityAssignerSuite) TestAssign_SelectedTaskTypes() {
mockExecutable := NewMockExecutable(s.controller)
mockExecutable.EXPECT().GetNamespaceID().Return("test-namespace-id").Times(1)
mockExecutable.EXPECT().GetWorkflowID().Return("test-workflow-id").Times(1)
mockExecutable.EXPECT().GetType().Return(enumsspb.TASK_TYPE_DELETE_HISTORY_EVENT).Times(1)
// Create a local namespace that is active in "current" cluster
@@ -64,6 +65,7 @@ func (s *priorityAssignerSuite) TestAssign_SelectedTaskTypes() {
func (s *priorityAssignerSuite) TestAssign_UnknownTaskTypes() {
mockExecutable := NewMockExecutable(s.controller)
mockExecutable.EXPECT().GetNamespaceID().Return("test-namespace-id").Times(1)
mockExecutable.EXPECT().GetWorkflowID().Return("test-workflow-id").Times(1)
mockExecutable.EXPECT().GetType().Return(enumsspb.TaskType(1234)).Times(1)
// Create a local namespace that is active in "current" cluster
@@ -87,6 +89,7 @@ func (s *priorityAssignerSuite) TestAssign_HighPriorityTaskTypes() {
} {
mockExecutable := NewMockExecutable(s.controller)
mockExecutable.EXPECT().GetNamespaceID().Return("test-namespace-id").Times(1)
mockExecutable.EXPECT().GetWorkflowID().Return("test-workflow-id").Times(1)
mockExecutable.EXPECT().GetType().Return(taskType).Times(1)
// Create a local namespace that is active in "current" cluster
@@ -111,6 +114,7 @@ func (s *priorityAssignerSuite) TestAssign_BackgroundPriorityTaskTypes() {
} {
mockExecutable := NewMockExecutable(s.controller)
mockExecutable.EXPECT().GetNamespaceID().Return("test-namespace-id").Times(1)
mockExecutable.EXPECT().GetWorkflowID().Return("test-workflow-id").Times(1)
mockExecutable.EXPECT().GetType().Return(taskType).Times(1)
// Create a local namespace that is active in "current" cluster
@@ -134,6 +138,7 @@ func (s *priorityAssignerSuite) TestAssign_LowPriorityTaskTypes() {
} {
mockExecutable := NewMockExecutable(s.controller)
mockExecutable.EXPECT().GetNamespaceID().Return("test-namespace-id").Times(1)
mockExecutable.EXPECT().GetWorkflowID().Return("test-workflow-id").Times(1)
mockExecutable.EXPECT().GetType().Return(taskType).Times(1)
// Create a local namespace that is active in "current" cluster
@@ -151,6 +156,7 @@ func (s *priorityAssignerSuite) TestAssign_LowPriorityTaskTypes() {
func (s *priorityAssignerSuite) TestAssign_StandbyNamespace() {
mockExecutable := NewMockExecutable(s.controller)
mockExecutable.EXPECT().GetNamespaceID().Return("test-namespace-id").Times(1)
mockExecutable.EXPECT().GetWorkflowID().Return("test-workflow-id").Times(1)
// Create a global namespace with active cluster as "other", not "current"
ns := namespace.NewGlobalNamespaceForTest(
@@ -180,6 +186,7 @@ func (s *priorityAssignerSuite) TestAssign_NamespaceNotFound() {
func (s *priorityAssignerSuite) TestAssign_ActiveNamespaceHighPriority() {
mockExecutable := NewMockExecutable(s.controller)
mockExecutable.EXPECT().GetNamespaceID().Return("test-namespace-id").Times(1)
mockExecutable.EXPECT().GetWorkflowID().Return("test-workflow-id").Times(1)
mockExecutable.EXPECT().GetType().Return(enumsspb.TASK_TYPE_TRANSFER_WORKFLOW_TASK).Times(1)
// Create a global namespace with active cluster as "current"

2
service/history/queues/queue_base_test.go
View File

@@ -209,6 +209,7 @@ func (s *queueBaseSuite) TestStartStop() {
key := NewRandomKeyInRange(paginationRange)
mockTask.EXPECT().GetKey().Return(key).AnyTimes()
mockTask.EXPECT().GetNamespaceID().Return(uuid.NewString()).AnyTimes()
mockTask.EXPECT().GetWorkflowID().Return(uuid.NewString()).AnyTimes()
mockTask.EXPECT().GetVisibilityTime().Return(time.Now()).AnyTimes()
return []tasks.Task{mockTask}, nil, nil
}
@@ -553,6 +554,7 @@ func (s *queueBaseSuite) TestCheckPoint_MoveTaskGroupAction() {
mockTask := tasks.NewMockTask(s.controller)
mockTask.EXPECT().GetKey().Return(NewRandomKeyInRange(sliceRange)).AnyTimes()
mockTask.EXPECT().GetNamespaceID().Return(namespaceID).AnyTimes()
mockTask.EXPECT().GetWorkflowID().Return(uuid.NewString()).AnyTimes()
mockTask.EXPECT().GetVisibilityTime().Return(time.Now()).AnyTimes()
slice.(*SliceImpl).add(base.executableFactory.NewExecutable(mockTask, readerID))
}

5
service/history/queues/reader_test.go
View File

@@ -101,6 +101,7 @@ func (s *readerSuite) TestStartLoadStop() {
mockTask := tasks.NewMockTask(s.controller)
mockTask.EXPECT().GetKey().Return(NewRandomKeyInRange(r)).AnyTimes()
mockTask.EXPECT().GetNamespaceID().Return(uuid.NewString()).AnyTimes()
mockTask.EXPECT().GetWorkflowID().Return(uuid.NewString()).AnyTimes()
mockTask.EXPECT().GetVisibilityTime().Return(time.Now()).AnyTimes()
return []tasks.Task{mockTask}, nil, nil
}
@@ -292,6 +293,7 @@ func (s *readerSuite) TestPause() {
mockTask := tasks.NewMockTask(s.controller)
mockTask.EXPECT().GetKey().Return(NewRandomKeyInRange(scopes[0].Range)).AnyTimes()
mockTask.EXPECT().GetNamespaceID().Return(uuid.NewString()).AnyTimes()
mockTask.EXPECT().GetWorkflowID().Return(uuid.NewString()).AnyTimes()
mockTask.EXPECT().GetVisibilityTime().Return(time.Now()).AnyTimes()
return []tasks.Task{mockTask}, nil, nil
}
@@ -362,6 +364,7 @@ func (s *readerSuite) TestLoadAndSubmitTasks_MoreTasks() {
mockTask := tasks.NewMockTask(s.controller)
mockTask.EXPECT().GetKey().Return(NewRandomKeyInRange(scopes[0].Range)).AnyTimes()
mockTask.EXPECT().GetNamespaceID().Return(uuid.NewString()).AnyTimes()
mockTask.EXPECT().GetWorkflowID().Return(uuid.NewString()).AnyTimes()
mockTask.EXPECT().GetVisibilityTime().Return(time.Now()).AnyTimes()
result = append(result, mockTask)
}
@@ -398,6 +401,7 @@ func (s *readerSuite) TestLoadAndSubmitTasks_NoMoreTasks_HasNextSlice() {
mockTask := tasks.NewMockTask(s.controller)
mockTask.EXPECT().GetKey().Return(NewRandomKeyInRange(scopes[0].Range)).AnyTimes()
mockTask.EXPECT().GetNamespaceID().Return(uuid.NewString()).AnyTimes()
mockTask.EXPECT().GetWorkflowID().Return(uuid.NewString()).AnyTimes()
mockTask.EXPECT().GetVisibilityTime().Return(time.Now()).AnyTimes()
return []tasks.Task{mockTask}, nil, nil
}
@@ -431,6 +435,7 @@ func (s *readerSuite) TestLoadAndSubmitTasks_NoMoreTasks_NoNextSlice() {
mockTask := tasks.NewMockTask(s.controller)
mockTask.EXPECT().GetKey().Return(NewRandomKeyInRange(scopes[0].Range)).AnyTimes()
mockTask.EXPECT().GetNamespaceID().Return(uuid.NewString()).AnyTimes()
mockTask.EXPECT().GetWorkflowID().Return(uuid.NewString()).AnyTimes()
mockTask.EXPECT().GetVisibilityTime().Return(time.Now()).AnyTimes()
return []tasks.Task{mockTask}, nil, nil
}

3
service/history/queues/slice_test.go
View File

@@ -466,6 +466,7 @@ func (s *sliceSuite) TestSelectTasks_NoError() {
namespaceID = uuid.NewString() // should be filtered out
}
mockTask.EXPECT().GetNamespaceID().Return(namespaceID).AnyTimes()
mockTask.EXPECT().GetWorkflowID().Return(uuid.NewString()).AnyTimes()
mockTasks = append(mockTasks, mockTask)
}
@@ -515,6 +516,7 @@ func (s *sliceSuite) TestSelectTasks_Error_NoLoadedTasks() {
key := NewRandomKeyInRange(paginationRange)
mockTask.EXPECT().GetKey().Return(key).AnyTimes()
mockTask.EXPECT().GetNamespaceID().Return(uuid.NewString()).AnyTimes()
mockTask.EXPECT().GetWorkflowID().Return(uuid.NewString()).AnyTimes()
mockTask.EXPECT().GetVisibilityTime().Return(time.Now()).AnyTimes()
mockTasks = append(mockTasks, mockTask)
}
@@ -559,6 +561,7 @@ func (s *sliceSuite) TestSelectTasks_Error_WithLoadedTasks() {
key := NewRandomKeyInRange(paginationRange)
mockTask.EXPECT().GetKey().Return(key).AnyTimes()
mockTask.EXPECT().GetNamespaceID().Return(uuid.NewString()).AnyTimes()
mockTask.EXPECT().GetWorkflowID().Return(uuid.NewString()).AnyTimes()
mockTask.EXPECT().GetVisibilityTime().Return(time.Now()).AnyTimes()
mockTasks = append(mockTasks, mockTask)
}

553
tests/advanced_visibility_test.go
View File

@@ -11,7 +11,6 @@ import (
"time"
"github.com/google/uuid"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/stretchr/testify/suite"
commandpb "go.temporal.io/api/command/v1"
@@ -37,6 +36,7 @@ import (
"go.temporal.io/server/common/primitives"
"go.temporal.io/server/common/searchattribute"
"go.temporal.io/server/common/searchattribute/sadefs"
"go.temporal.io/server/common/testing/await"
"go.temporal.io/server/common/testing/protorequire"
"go.temporal.io/server/common/testing/testvars"
"go.temporal.io/server/common/worker_versioning"
@@ -48,8 +48,7 @@ import (
)
const (
numOfRetry = 50
waitTimeInMs = 400
esPollInterval = 300 * time.Millisecond
testSearchAttributeKey = "CustomTextField"
testSearchAttributeVal = "test value"
@@ -136,24 +135,28 @@ func (s *AdvancedVisibilitySuite) TestListOpenWorkflow() {
startFilter := &filterpb.StartTimeFilter{}
startFilter.EarliestTime = timestamppb.New(startTime)
var openExecution *workflowpb.WorkflowExecutionInfo
for range numOfRetry {
startFilter.LatestTime = timestamppb.New(time.Now().UTC())
resp, err := s.FrontendClient().ListOpenWorkflowExecutions(testcore.NewContext(), &workflowservice.ListOpenWorkflowExecutionsRequest{
Namespace: s.Namespace().String(),
MaximumPageSize: testcore.DefaultPageSize,
StartTimeFilter: startFilter,
Filters: &workflowservice.ListOpenWorkflowExecutionsRequest_ExecutionFilter{ExecutionFilter: &filterpb.WorkflowExecutionFilter{
WorkflowId: id,
}},
})
s.NoError(err)
if len(resp.GetExecutions()) == 1 {
await.Require(s.Context(), s.T(),
func(t *await.T) {
startFilter.LatestTime = timestamppb.New(time.Now().UTC())
resp, err := s.FrontendClient().ListOpenWorkflowExecutions(
testcore.NewContext(), &workflowservice.ListOpenWorkflowExecutionsRequest{
Namespace: s.Namespace().String(),
MaximumPageSize: testcore.DefaultPageSize,
StartTimeFilter: startFilter,
Filters: &workflowservice.ListOpenWorkflowExecutionsRequest_ExecutionFilter{
ExecutionFilter: &filterpb.WorkflowExecutionFilter{
WorkflowId: id,
},
},
},
)
require.NoError(t, err)
require.Len(t, resp.GetExecutions(), 1)
openExecution = resp.GetExecutions()[0]
s.Nil(resp.NextPageToken)
break
}
time.Sleep(waitTimeInMs * time.Millisecond) //nolint:forbidigo
}
},
testcore.WaitForESToSettle,
esPollInterval,
)
s.NotNil(openExecution)
s.Equal(we.GetRunId(), openExecution.GetExecution().GetRunId())
@@ -375,15 +378,16 @@ func (s *AdvancedVisibilitySuite) TestListWorkflow_OrQuery() {
PageSize: testcore.DefaultPageSize,
Query: query1,
}
for range numOfRetry {
resp, err := s.FrontendClient().ListWorkflowExecutions(testcore.NewContext(), listRequest)
s.NoError(err)
if len(resp.GetExecutions()) == 1 {
await.Require(s.Context(), s.T(),
func(t *await.T) {
resp, err := s.FrontendClient().ListWorkflowExecutions(testcore.NewContext(), listRequest)
require.NoError(t, err)
require.Len(t, resp.GetExecutions(), 1)
openExecution = resp.GetExecutions()[0]
break
}
time.Sleep(waitTimeInMs * time.Millisecond) //nolint:forbidigo
}
},
testcore.WaitForESToSettle,
esPollInterval,
)
s.NotNil(openExecution)
s.Equal(we1.GetRunId(), openExecution.GetExecution().GetRunId())
s.False(openExecution.GetExecutionTime().AsTime().Before(openExecution.GetStartTime().AsTime()))
@@ -396,15 +400,16 @@ func (s *AdvancedVisibilitySuite) TestListWorkflow_OrQuery() {
query2 := fmt.Sprintf(`CustomIntField = %d or CustomIntField = %d`, 1, 2)
listRequest.Query = query2
var openExecutions []*workflowpb.WorkflowExecutionInfo
for range numOfRetry {
resp, err := s.FrontendClient().ListWorkflowExecutions(testcore.NewContext(), listRequest)
s.NoError(err)
if len(resp.GetExecutions()) == 2 {
await.Require(s.Context(), s.T(),
func(t *await.T) {
resp, err := s.FrontendClient().ListWorkflowExecutions(testcore.NewContext(), listRequest)
require.NoError(t, err)
require.Len(t, resp.GetExecutions(), 2)
openExecutions = resp.GetExecutions()
break
}
time.Sleep(waitTimeInMs * time.Millisecond) //nolint:forbidigo
}
},
testcore.WaitForESToSettle,
esPollInterval,
)
s.Len(openExecutions, 2)
e1 := openExecutions[0]
e2 := openExecutions[1]
@@ -421,15 +426,16 @@ func (s *AdvancedVisibilitySuite) TestListWorkflow_OrQuery() {
// query for open
query3 := fmt.Sprintf(`(CustomIntField = %d or CustomIntField = %d) and ExecutionStatus = 'Running'`, 2, 3)
listRequest.Query = query3
for range numOfRetry {
resp, err := s.FrontendClient().ListWorkflowExecutions(testcore.NewContext(), listRequest)
s.NoError(err)
if len(resp.GetExecutions()) == 2 {
await.Require(s.Context(), s.T(),
func(t *await.T) {
resp, err := s.FrontendClient().ListWorkflowExecutions(testcore.NewContext(), listRequest)
require.NoError(t, err)
require.Len(t, resp.GetExecutions(), 2)
openExecutions = resp.GetExecutions()
break
}
time.Sleep(waitTimeInMs * time.Millisecond) //nolint:forbidigo
}
},
testcore.WaitForESToSettle,
esPollInterval,
)
s.Len(openExecutions, 2)
e1 = openExecutions[0]
e2 = openExecutions[1]
@@ -464,11 +470,11 @@ func (s *AdvancedVisibilitySuite) TestListWorkflow_KeywordQuery() {
PageSize: testcore.DefaultPageSize,
Query: `CustomKeywordField = "justice for all"`,
}
s.EventuallyWithT(
func(c *assert.CollectT) {
await.Require(s.Context(), s.T(),
func(t *await.T) {
resp, err := s.FrontendClient().ListWorkflowExecutions(testcore.NewContext(), listRequest)
require.NoError(c, err)
require.Len(c, resp.GetExecutions(), 1)
require.NoError(t, err)
require.Len(t, resp.GetExecutions(), 1)
openExecution = resp.GetExecutions()[0]
},
testcore.WaitForESToSettle,
@@ -553,15 +559,16 @@ func (s *AdvancedVisibilitySuite) TestListWorkflow_StringQuery() {
PageSize: testcore.DefaultPageSize,
Query: `CustomTextField = "nothing else matters"`,
}
for range numOfRetry {
resp, err := s.FrontendClient().ListWorkflowExecutions(testcore.NewContext(), listRequest)
s.NoError(err)
if len(resp.GetExecutions()) == 1 {
await.Require(s.Context(), s.T(),
func(t *await.T) {
resp, err := s.FrontendClient().ListWorkflowExecutions(testcore.NewContext(), listRequest)
require.NoError(t, err)
require.Len(t, resp.GetExecutions(), 1)
openExecution = resp.GetExecutions()[0]
break
}
time.Sleep(waitTimeInMs * time.Millisecond) //nolint:forbidigo
}
},
testcore.WaitForESToSettle,
esPollInterval,
)
s.NotNil(openExecution)
s.Equal(we1.GetRunId(), openExecution.GetExecution().GetRunId())
s.False(openExecution.GetExecutionTime().AsTime().Before(openExecution.GetStartTime().AsTime()))
@@ -618,15 +625,16 @@ func (s *AdvancedVisibilitySuite) TestListWorkflow_MaxWindowSize() {
Query: fmt.Sprintf(`WorkflowType = '%s' and ExecutionStatus = "Running"`, wt),
}
// get first page
for range numOfRetry {
resp, err := s.FrontendClient().ListWorkflowExecutions(testcore.NewContext(), listRequest)
s.NoError(err)
if len(resp.GetExecutions()) == testcore.DefaultPageSize {
await.Require(s.Context(), s.T(),
func(t *await.T) {
resp, err := s.FrontendClient().ListWorkflowExecutions(testcore.NewContext(), listRequest)
require.NoError(t, err)
require.Len(t, resp.GetExecutions(), testcore.DefaultPageSize)
listResp = resp
break
}
time.Sleep(waitTimeInMs * time.Millisecond) //nolint:forbidigo
}
},
testcore.WaitForESToSettle,
esPollInterval,
)
s.NotNil(listResp)
s.NotEmpty(listResp.GetNextPageToken())
@@ -682,8 +690,8 @@ func (s *AdvancedVisibilitySuite) TestListWorkflow_OrderBy() {
s.NoError(err)
}
s.EventuallyWithT(
func(c *assert.CollectT) {
await.Require(s.Context(), s.T(),
func(t *await.T) {
resp, err := s.FrontendClient().CountWorkflowExecutions(
ctx,
&workflowservice.CountWorkflowExecutionsRequest{
@@ -691,8 +699,8 @@ func (s *AdvancedVisibilitySuite) TestListWorkflow_OrderBy() {
Query: fmt.Sprintf(`WorkflowType = "%s"`, wt),
},
)
require.NoError(c, err)
require.EqualValues(c, 6, resp.GetCount())
require.NoError(t, err)
require.EqualValues(t, 6, resp.GetCount())
},
testcore.WaitForESToSettle,
100*time.Millisecond,
@@ -800,6 +808,9 @@ func (s *AdvancedVisibilitySuite) testListWorkflowHelper(
startRequest *workflowservice.StartWorkflowExecutionRequest,
wid, wType string,
) {
// this function assumes numOfWorkflows > pageSize
s.Greater(numOfWorkflows, pageSize, "numOfWorkflows must be greater than pageSize")
// start enough number of workflows
for i := range numOfWorkflows {
startRequest.RequestId = uuid.NewString()
@@ -808,11 +819,21 @@ func (s *AdvancedVisibilitySuite) testListWorkflowHelper(
s.NoError(err)
}
time.Sleep(testcore.WaitForESToSettle) //nolint:forbidigo
await.Require(s.Context(), s.T(),
func(t *await.T) {
countRequest := &workflowservice.CountWorkflowExecutionsRequest{
Namespace: s.Namespace().String(),
Query: fmt.Sprintf(`WorkflowType = '%s' and ExecutionStatus = 'Running'`, wType),
}
countResponse, err := s.FrontendClient().CountWorkflowExecutions(testcore.NewContext(), countRequest)
require.NoError(t, err)
require.Equal(t, int64(numOfWorkflows), countResponse.GetCount())
},
testcore.WaitForESToSettle,
esPollInterval,
)
var openExecutions []*workflowpb.WorkflowExecutionInfo
var nextPageToken []byte
listRequest := &workflowservice.ListWorkflowExecutionsRequest{
Namespace: s.Namespace().String(),
PageSize: int32(pageSize),
@@ -821,37 +842,47 @@ func (s *AdvancedVisibilitySuite) testListWorkflowHelper(
}
// test first page
for range numOfRetry {
listResponse, err := s.FrontendClient().ListWorkflowExecutions(testcore.NewContext(), listRequest)
s.NoError(err)
if len(listResponse.GetExecutions()) == pageSize {
openExecutions = listResponse.GetExecutions()
await.Require(s.Context(), s.T(),
func(t *await.T) {
listResponse, err := s.FrontendClient().ListWorkflowExecutions(testcore.NewContext(), listRequest)
require.NoError(t, err)
require.Len(t, listResponse.GetExecutions(), pageSize)
nextPageToken = listResponse.GetNextPageToken()
break
}
time.Sleep(waitTimeInMs * time.Millisecond) //nolint:forbidigo
}
s.NotNil(openExecutions)
s.NotNil(nextPageToken)
},
testcore.WaitForESToSettle,
esPollInterval,
)
s.NotEmpty(nextPageToken)
// test last page
listRequest.NextPageToken = nextPageToken
inIf := false
for range numOfRetry {
listResponse, err := s.FrontendClient().ListWorkflowExecutions(testcore.NewContext(), listRequest)
s.NoError(err)
if len(listResponse.GetExecutions()) == numOfWorkflows-pageSize {
inIf = true
openExecutions = listResponse.GetExecutions()
await.Require(s.Context(), s.T(),
func(t *await.T) {
listResponse, err := s.FrontendClient().ListWorkflowExecutions(testcore.NewContext(), listRequest)
require.NoError(t, err)
require.Len(t, listResponse.GetExecutions(), numOfWorkflows-pageSize)
nextPageToken = listResponse.GetNextPageToken()
break
}
time.Sleep(waitTimeInMs * time.Millisecond) //nolint:forbidigo
},
testcore.WaitForESToSettle,
esPollInterval,
)
// nextPageToken might be not be empty, depends on store implementation
if len(nextPageToken) > 0 {
// test page after last page is empty (true last page)
listRequest.NextPageToken = nextPageToken
await.Require(s.Context(), s.T(),
func(t *await.T) {
listResponse, err := s.FrontendClient().ListWorkflowExecutions(testcore.NewContext(), listRequest)
require.NoError(t, err)
require.Empty(t, listResponse.GetExecutions())
nextPageToken = listResponse.GetNextPageToken()
},
testcore.WaitForESToSettle,
esPollInterval,
)
s.Empty(nextPageToken)
}
s.True(inIf)
s.NotNil(openExecutions)
s.Nil(nextPageToken)
}
func (s *AdvancedVisibilitySuite) testHelperForReadOnce(expectedRunID string, query string) *workflowpb.WorkflowExecutionInfo {
@@ -862,16 +893,16 @@ func (s *AdvancedVisibilitySuite) testHelperForReadOnce(expectedRunID string, qu
Query: query,
}
for range numOfRetry {
listResponse, err := s.FrontendClient().ListWorkflowExecutions(testcore.NewContext(), listRequest)
s.NoError(err)
if len(listResponse.GetExecutions()) == 1 {
openExecution = listResponse.GetExecutions()[0]
s.Nil(listResponse.NextPageToken)
break
}
time.Sleep(waitTimeInMs * time.Millisecond) //nolint:forbidigo
}
await.Require(s.Context(), s.T(),
func(t *await.T) {
resp, err := s.FrontendClient().ListWorkflowExecutions(testcore.NewContext(), listRequest)
require.NoError(t, err)
require.Len(t, resp.GetExecutions(), 1)
openExecution = resp.GetExecutions()[0]
},
testcore.WaitForESToSettle,
esPollInterval,
)
s.NotNil(openExecution)
s.Equal(expectedRunID, openExecution.GetExecution().GetRunId())
s.False(openExecution.GetExecutionTime().AsTime().Before(openExecution.GetStartTime().AsTime()))
@@ -900,20 +931,19 @@ func (s *AdvancedVisibilitySuite) TestCountWorkflow() {
Namespace: s.Namespace().String(),
Query: query,
}
var resp *workflowservice.CountWorkflowExecutionsResponse
for range numOfRetry {
resp, err = s.FrontendClient().CountWorkflowExecutions(testcore.NewContext(), countRequest)
s.NoError(err)
if resp.GetCount() == int64(1) {
break
}
time.Sleep(waitTimeInMs * time.Millisecond) //nolint:forbidigo
}
s.Equal(int64(1), resp.GetCount())
await.Require(s.Context(), s.T(),
func(t *await.T) {
resp, err := s.FrontendClient().CountWorkflowExecutions(testcore.NewContext(), countRequest)
require.NoError(t, err)
require.Equal(t, int64(1), resp.GetCount())
},
testcore.WaitForESToSettle,
esPollInterval,
)
query = fmt.Sprintf(`WorkflowId = "%s" and %s = "%s"`, id, testSearchAttributeKey, "noMatch")
countRequest.Query = query
resp, err = s.FrontendClient().CountWorkflowExecutions(testcore.NewContext(), countRequest)
resp, err := s.FrontendClient().CountWorkflowExecutions(testcore.NewContext(), countRequest)
s.NoError(err)
s.Equal(int64(0), resp.GetCount())
}
@@ -950,18 +980,18 @@ func (s *AdvancedVisibilitySuite) TestCountGroupByWorkflow() {
Namespace: s.Namespace().String(),
Query: query,
}
var resp *workflowservice.CountWorkflowExecutionsResponse
var err error
for range numOfRetry {
resp, err = s.FrontendClient().CountWorkflowExecutions(testcore.NewContext(), countRequest)
s.NoError(err)
if resp.GetCount() == int64(numWorkflows) {
break
}
time.Sleep(waitTimeInMs * time.Millisecond) //nolint:forbidigo
}
s.Equal(int64(numWorkflows), resp.GetCount())
s.Len(resp.Groups, 2)
var countResp *workflowservice.CountWorkflowExecutionsResponse
await.Require(s.Context(), s.T(),
func(t *await.T) {
resp, err := s.FrontendClient().CountWorkflowExecutions(testcore.NewContext(), countRequest)
require.NoError(t, err)
require.Equal(t, int64(numWorkflows), resp.GetCount())
countResp = resp
},
testcore.WaitForESToSettle,
esPollInterval,
)
s.Len(countResp.Groups, 2)
runningStatusPayload, _ := sadefs.EncodeValue(
enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING.String(),
@@ -976,19 +1006,19 @@ func (s *AdvancedVisibilitySuite) TestCountGroupByWorkflow() {
GroupValues: []*commonpb.Payload{runningStatusPayload},
Count: int64(numWorkflows - numClosedWorkflows),
},
resp.Groups[0],
countResp.Groups[0],
)
s.ProtoEqual(
&workflowservice.CountWorkflowExecutionsResponse_AggregationGroup{
GroupValues: []*commonpb.Payload{terminatedStatusPayload},
Count: int64(numClosedWorkflows),
},
resp.Groups[1],
countResp.Groups[1],
)
query = `GROUP BY WorkflowType`
countRequest.Query = query
_, err = s.FrontendClient().CountWorkflowExecutions(testcore.NewContext(), countRequest)
_, err := s.FrontendClient().CountWorkflowExecutions(testcore.NewContext(), countRequest)
s.Error(err)
s.Contains(strings.ToLower(err.Error()), "'group by' clause is only supported for")
@@ -1136,26 +1166,22 @@ func (s *AdvancedVisibilitySuite) TestUpsertWorkflowExecutionSearchAttributes()
PageSize: int32(2),
Query: fmt.Sprintf(`WorkflowType = '%s' and ExecutionStatus = 'Running'`, wt),
}
verified := false
for range numOfRetry {
resp, err := s.FrontendClient().ListWorkflowExecutions(testcore.NewContext(), listRequest)
s.NoError(err)
if len(resp.GetExecutions()) == 1 {
await.Require(s.Context(), s.T(),
func(t *await.T) {
resp, err := s.FrontendClient().ListWorkflowExecutions(testcore.NewContext(), listRequest)
require.NoError(t, err)
require.Len(t, resp.GetExecutions(), 1)
execution := resp.GetExecutions()[0]
retrievedSearchAttr := execution.SearchAttributes
if retrievedSearchAttr != nil && len(retrievedSearchAttr.GetIndexedFields()) > 0 {
searchValBytes := retrievedSearchAttr.GetIndexedFields()[testSearchAttributeKey]
var searchVal string
err = payload.Decode(searchValBytes, &searchVal)
s.NoError(err)
s.Equal(testSearchAttributeVal, searchVal)
verified = true
break
}
}
time.Sleep(waitTimeInMs * time.Millisecond) //nolint:forbidigo
}
s.True(verified)
require.NotEmpty(t, execution.GetSearchAttributes().GetIndexedFields())
searchValBytes := execution.GetSearchAttributes().GetIndexedFields()[testSearchAttributeKey]
var searchVal string
err = payload.Decode(searchValBytes, &searchVal)
require.NoError(t, err)
require.Equal(t, testSearchAttributeVal, searchVal)
},
testcore.WaitForESToSettle,
esPollInterval,
)
// process 2nd workflow task and assert workflow task is handled correctly.
res, err = poller.PollAndProcessWorkflowTask(
@@ -1204,23 +1230,19 @@ func (s *AdvancedVisibilitySuite) TestUpsertWorkflowExecutionSearchAttributes()
PageSize: int32(2),
Query: fmt.Sprintf(`WorkflowType = '%s' and ExecutionStatus = 'Running'`, wt),
}
verified = false
for range numOfRetry {
resp, err := s.FrontendClient().ListWorkflowExecutions(testcore.NewContext(), listRequest)
s.NoError(err)
if len(resp.GetExecutions()) == 1 {
await.Require(s.Context(), s.T(),
func(t *await.T) {
resp, err := s.FrontendClient().ListWorkflowExecutions(testcore.NewContext(), listRequest)
require.NoError(t, err)
require.Len(t, resp.GetExecutions(), 1)
execution := resp.GetExecutions()[0]
retrievedSearchAttr := execution.SearchAttributes
if retrievedSearchAttr != nil && len(retrievedSearchAttr.GetIndexedFields()) > 0 {
s.NotContains(retrievedSearchAttr.GetIndexedFields(), "CustomTextField")
s.NotContains(retrievedSearchAttr.GetIndexedFields(), "CustomIntField")
verified = true
break
}
}
time.Sleep(waitTimeInMs * time.Millisecond) //nolint:forbidigo
}
s.True(verified)
require.NotEmpty(t, execution.GetSearchAttributes().GetIndexedFields())
require.NotContains(t, execution.GetSearchAttributes().GetIndexedFields(), "CustomTextField")
require.NotContains(t, execution.GetSearchAttributes().GetIndexedFields(), "CustomIntField")
},
testcore.WaitForESToSettle,
esPollInterval,
)
// verify query by unset search attribute
listRequest = &workflowservice.ListWorkflowExecutionsRequest{
@@ -1228,17 +1250,15 @@ func (s *AdvancedVisibilitySuite) TestUpsertWorkflowExecutionSearchAttributes()
PageSize: int32(2),
Query: fmt.Sprintf(`WorkflowType = '%s' and ExecutionStatus = 'Running' and CustomTextField is null and CustomIntField is null`, wt),
}
verified = false
for range numOfRetry {
resp, err := s.FrontendClient().ListWorkflowExecutions(testcore.NewContext(), listRequest)
s.NoError(err)
if len(resp.GetExecutions()) == 1 {
verified = true
break
}
time.Sleep(waitTimeInMs * time.Millisecond) //nolint:forbidigo
}
s.True(verified)
await.Require(s.Context(), s.T(),
func(t *await.T) {
resp, err := s.FrontendClient().ListWorkflowExecutions(testcore.NewContext(), listRequest)
require.NoError(t, err)
require.Len(t, resp.GetExecutions(), 1)
},
testcore.WaitForESToSettle,
esPollInterval,
)
// verify search attributes from DescribeWorkflowExecution
descRequest := &workflowservice.DescribeWorkflowExecutionRequest{
@@ -1431,17 +1451,16 @@ func (s *AdvancedVisibilitySuite) TestModifyWorkflowExecutionProperties() {
PageSize: int32(2),
Query: fmt.Sprintf(`WorkflowType = '%s' and ExecutionStatus = 'Running'`, wt),
}
verified := false
for range numOfRetry {
resp, err := s.FrontendClient().ListWorkflowExecutions(testcore.NewContext(), listRequest)
s.NoError(err)
if len(resp.GetExecutions()) == 1 {
s.True(proto.Equal(expectedMemo, resp.Executions[0].Memo))
verified = true
}
time.Sleep(waitTimeInMs * time.Millisecond) //nolint:forbidigo
}
s.True(verified)
await.Require(s.Context(), s.T(),
func(t *await.T) {
resp, err := s.FrontendClient().ListWorkflowExecutions(testcore.NewContext(), listRequest)
require.NoError(t, err)
require.Len(t, resp.GetExecutions(), 1)
require.True(t, proto.Equal(expectedMemo, resp.Executions[0].Memo))
},
testcore.WaitForESToSettle,
esPollInterval,
)
// process 2nd workflow task and assert workflow task is handled correctly.
res, err = poller.PollAndProcessWorkflowTask(
@@ -1477,17 +1496,16 @@ func (s *AdvancedVisibilitySuite) TestModifyWorkflowExecutionProperties() {
PageSize: int32(2),
Query: fmt.Sprintf(`WorkflowType = '%s' and ExecutionStatus = 'Running'`, wt),
}
verified = false
for range numOfRetry {
resp, err := s.FrontendClient().ListWorkflowExecutions(testcore.NewContext(), listRequest)
s.NoError(err)
if len(resp.GetExecutions()) == 1 {
s.True(proto.Equal(expectedMemo, resp.Executions[0].Memo))
verified = true
}
time.Sleep(waitTimeInMs * time.Millisecond) //nolint:forbidigo
}
s.True(verified)
await.Require(s.Context(), s.T(),
func(t *await.T) {
resp, err := s.FrontendClient().ListWorkflowExecutions(testcore.NewContext(), listRequest)
require.NoError(t, err)
require.Len(t, resp.GetExecutions(), 1)
require.True(t, proto.Equal(expectedMemo, resp.Executions[0].Memo))
},
testcore.WaitForESToSettle,
esPollInterval,
)
// process close workflow task and assert workflow task is handled correctly.
res, err = poller.PollAndProcessWorkflowTask(
@@ -1516,53 +1534,48 @@ func (s *AdvancedVisibilitySuite) TestModifyWorkflowExecutionProperties() {
}
func (s *AdvancedVisibilitySuite) testListResultForUpsertSearchAttributes(listRequest *workflowservice.ListWorkflowExecutionsRequest) {
verified := false
for range numOfRetry {
resp, err := s.FrontendClient().ListWorkflowExecutions(testcore.NewContext(), listRequest)
s.NoError(err)
if len(resp.GetExecutions()) == 1 {
s.Nil(resp.NextPageToken)
await.Require(s.Context(), s.T(),
func(t *await.T) {
resp, err := s.FrontendClient().ListWorkflowExecutions(testcore.NewContext(), listRequest)
require.NoError(t, err)
require.Len(t, resp.GetExecutions(), 1)
execution := resp.GetExecutions()[0]
retrievedSearchAttr := execution.SearchAttributes
if retrievedSearchAttr != nil && len(retrievedSearchAttr.GetIndexedFields()) == 5 {
fields := retrievedSearchAttr.GetIndexedFields()
searchValBytes := fields[testSearchAttributeKey]
var searchVal string
err := payload.Decode(searchValBytes, &searchVal)
s.NoError(err)
s.Equal("another string", searchVal)
require.NotEmpty(t, execution.GetSearchAttributes().GetIndexedFields())
fields := execution.GetSearchAttributes().GetIndexedFields()
searchValBytes2 := fields["CustomIntField"]
var searchVal2 int
err = payload.Decode(searchValBytes2, &searchVal2)
s.NoError(err)
s.Equal(123, searchVal2)
searchValBytes := fields[testSearchAttributeKey]
var searchVal string
err = payload.Decode(searchValBytes, &searchVal)
require.NoError(t, err)
s.Equal("another string", searchVal)
doublePayload := fields["CustomDoubleField"]
var doubleVal float64
err = payload.Decode(doublePayload, &doubleVal)
s.NoError(err)
s.Equal(22.0878, doubleVal)
searchValBytes2 := fields["CustomIntField"]
var searchVal2 int
err = payload.Decode(searchValBytes2, &searchVal2)
require.NoError(t, err)
require.Equal(t, 123, searchVal2)
binaryChecksumsBytes := fields[sadefs.BinaryChecksums]
var binaryChecksums []string
err = payload.Decode(binaryChecksumsBytes, &binaryChecksums)
s.NoError(err)
s.Equal([]string{"binary-v1", "binary-v2"}, binaryChecksums)
doublePayload := fields["CustomDoubleField"]
var doubleVal float64
err = payload.Decode(doublePayload, &doubleVal)
require.NoError(t, err)
require.InDelta(t, 22.0878, doubleVal, 1e-6)
buildIdsBytes := fields[sadefs.BuildIds]
var buildIds []string
err = payload.Decode(buildIdsBytes, &buildIds)
s.NoError(err)
s.Equal([]string{worker_versioning.UnversionedSearchAttribute}, buildIds)
binaryChecksumsBytes := fields[sadefs.BinaryChecksums]
var binaryChecksums []string
err = payload.Decode(binaryChecksumsBytes, &binaryChecksums)
require.NoError(t, err)
require.Equal(t, []string{"binary-v1", "binary-v2"}, binaryChecksums)
verified = true
break
}
}
time.Sleep(waitTimeInMs * time.Millisecond) //nolint:forbidigo
}
s.True(verified)
buildIdsBytes := fields[sadefs.BuildIds]
var buildIds []string
err = payload.Decode(buildIdsBytes, &buildIds)
require.NoError(t, err)
require.Equal(t, []string{worker_versioning.UnversionedSearchAttribute}, buildIds)
},
testcore.WaitForESToSettle,
esPollInterval,
)
}
func (s *AdvancedVisibilitySuite) createSearchAttributes() *commonpb.SearchAttributes {
@@ -1677,8 +1690,8 @@ func (s *AdvancedVisibilitySuite) TestChildWorkflow_ParentWorkflow() {
s.NoError(run.Get(ctx, nil))
// check main workflow doesn't have parent workflow and root is itself
s.EventuallyWithT(
func(c *assert.CollectT) {
await.Require(s.Context(), s.T(),
func(t *await.T) {
resp, err := s.FrontendClient().ListWorkflowExecutions(
ctx,
&workflowservice.ListWorkflowExecutionsRequest{
@@ -1687,13 +1700,13 @@ func (s *AdvancedVisibilitySuite) TestChildWorkflow_ParentWorkflow() {
PageSize: testcore.DefaultPageSize,
},
)
require.NoError(c, err)
require.Len(c, resp.Executions, 1)
require.NoError(t, err)
require.Len(t, resp.Executions, 1)
wfInfo := resp.Executions[0]
require.Nil(c, wfInfo.GetParentExecution())
require.NotNil(c, wfInfo.GetRootExecution())
require.Equal(c, wfID, wfInfo.RootExecution.GetWorkflowId())
require.Equal(c, run.GetRunID(), wfInfo.RootExecution.GetRunId())
require.Nil(t, wfInfo.GetParentExecution())
require.NotNil(t, wfInfo.GetRootExecution())
require.Equal(t, wfID, wfInfo.RootExecution.GetWorkflowId())
require.Equal(t, run.GetRunID(), wfInfo.RootExecution.GetRunId())
},
testcore.WaitForESToSettle,
100*time.Millisecond,
@@ -1701,8 +1714,8 @@ func (s *AdvancedVisibilitySuite) TestChildWorkflow_ParentWorkflow() {
// check child workflow has parent workflow and root is the parent
var childWfInfo *workflowpb.WorkflowExecutionInfo
s.EventuallyWithT(
func(c *assert.CollectT) {
await.Require(s.Context(), s.T(),
func(t *await.T) {
resp, err := s.FrontendClient().ListWorkflowExecutions(
ctx,
&workflowservice.ListWorkflowExecutionsRequest{
@@ -1711,15 +1724,15 @@ func (s *AdvancedVisibilitySuite) TestChildWorkflow_ParentWorkflow() {
PageSize: testcore.DefaultPageSize,
},
)
require.NoError(c, err)
require.Len(c, resp.Executions, 1)
require.NoError(t, err)
require.Len(t, resp.Executions, 1)
childWfInfo = resp.Executions[0]
require.NotNil(c, childWfInfo.GetParentExecution())
require.Equal(c, wfID, childWfInfo.ParentExecution.GetWorkflowId())
require.Equal(c, run.GetRunID(), childWfInfo.ParentExecution.GetRunId())
require.NotNil(c, childWfInfo.GetRootExecution())
require.Equal(c, wfID, childWfInfo.RootExecution.GetWorkflowId())
require.Equal(c, run.GetRunID(), childWfInfo.RootExecution.GetRunId())
require.NotNil(t, childWfInfo.GetParentExecution())
require.Equal(t, wfID, childWfInfo.ParentExecution.GetWorkflowId())
require.Equal(t, run.GetRunID(), childWfInfo.ParentExecution.GetRunId())
require.NotNil(t, childWfInfo.GetRootExecution())
require.Equal(t, wfID, childWfInfo.RootExecution.GetWorkflowId())
require.Equal(t, run.GetRunID(), childWfInfo.RootExecution.GetRunId())
},
testcore.WaitForESToSettle,
100*time.Millisecond,
@@ -2675,15 +2688,21 @@ func (s *AdvancedVisibilitySuite) updateMaxResultWindow() {
s.Require().NoError(err)
s.Require().True(acknowledged)
for range numOfRetry {
settings, err := esClient.IndexGetSettings(context.Background(), esConfig.GetVisibilityIndex())
s.Require().NoError(err)
if settings[esConfig.GetVisibilityIndex()].Settings["index"].(map[string]any)["max_result_window"].(string) == strconv.Itoa(testcore.DefaultPageSize) { //nolint:revive // unchecked-type-assertion
return
}
time.Sleep(waitTimeInMs * time.Millisecond) //nolint:forbidigo
}
s.Require().FailNowf("", "ES max result window size hasn't reach target size within %v", numOfRetry*waitTimeInMs*time.Millisecond)
await.Requiref(s.Context(), s.T(),
func(t *await.T) {
settings, err := esClient.IndexGetSettings(context.Background(), esConfig.GetVisibilityIndex())
require.NoError(t, err)
indexSettings, ok := settings[esConfig.GetVisibilityIndex()].Settings["index"].(map[string]any)
require.True(t, ok)
maxResultWindow, ok := indexSettings["max_result_window"].(string)
require.True(t, ok)
require.Equal(t, strconv.Itoa(testcore.DefaultPageSize), maxResultWindow)
},
testcore.WaitForESToSettle,
esPollInterval,
"ES max result window size hasn't reach target size within %v",
testcore.WaitForESToSettle,
)
}
func (s *AdvancedVisibilitySuite) addCustomKeywordSearchAttribute(ctx context.Context, attrName string) {

2
tests/workflow_visibility_test.go
View File

@@ -137,7 +137,6 @@ func (s *WorkflowVisibilityTestSuite) TestVisibility() {
closedCount = len(resp.Executions)
if closedCount == 1 {
historyLength = resp.Executions[0].HistoryLength
s.Nil(resp.NextPageToken)
return true
}
env.Logger.Info("Closed WorkflowExecution is not yet visible")
@@ -164,7 +163,6 @@ func (s *WorkflowVisibilityTestSuite) TestVisibility() {
s.NoError(err4)
openCount = len(resp.Executions)
if openCount == 1 {
s.Nil(resp.NextPageToken)
return true
}
env.Logger.Info("Open WorkflowExecution is not yet visible")