mirror of
https://github.com/temporalio/temporal.git
synced 2026-08-31 02:51:51 -07:00
256 lines
9.4 KiB
Go
256 lines
9.4 KiB
Go
package tests
|
|
|
|
import (
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/google/uuid"
|
|
"github.com/stretchr/testify/assert"
|
|
"github.com/stretchr/testify/require"
|
|
batchpb "go.temporal.io/api/batch/v1"
|
|
commonpb "go.temporal.io/api/common/v1"
|
|
"go.temporal.io/api/serviceerror"
|
|
"go.temporal.io/api/workflowservice/v1"
|
|
sdkclient "go.temporal.io/sdk/client"
|
|
"go.temporal.io/sdk/workflow"
|
|
"go.temporal.io/server/api/adminservice/v1"
|
|
"go.temporal.io/server/common/dynamicconfig"
|
|
"go.temporal.io/server/common/payloads"
|
|
"go.temporal.io/server/common/testing/parallelsuite"
|
|
"go.temporal.io/server/tests/testcore"
|
|
"google.golang.org/grpc/codes"
|
|
)
|
|
|
|
type AdminBatchRefreshWorkflowTasksTestSuite struct {
|
|
parallelsuite.Suite[*AdminBatchRefreshWorkflowTasksTestSuite]
|
|
}
|
|
|
|
func TestAdminBatchRefreshWorkflowTasksTestSuite(t *testing.T) {
|
|
parallelsuite.Run(t, &AdminBatchRefreshWorkflowTasksTestSuite{})
|
|
}
|
|
|
|
// newTestEnv creates a TestEnv with the dynamic config this suite needs.
|
|
// Additional per-test options may be passed in opts.
|
|
func (s *AdminBatchRefreshWorkflowTasksTestSuite) newTestEnv(opts ...testcore.TestOption) *testcore.TestEnv {
|
|
// Use a higher limit for general tests to avoid interference from batch operations
|
|
// that haven't completed yet. The isolation test (A_SeparateLimitFromFrontendBatchOperation)
|
|
// explicitly sets limit to 1 to verify frontend and admin batch ops use separate limits.
|
|
baseOpts := []testcore.TestOption{
|
|
testcore.WithDynamicConfig(dynamicconfig.FrontendMaxConcurrentAdminBatchOperationPerNamespace, 10),
|
|
}
|
|
return testcore.NewEnv(s.T(), append(baseOpts, opts...)...)
|
|
}
|
|
|
|
func (s *AdminBatchRefreshWorkflowTasksTestSuite) simpleWorkflow(ctx workflow.Context) (string, error) {
|
|
// Simple workflow that just returns
|
|
return "done", nil
|
|
}
|
|
|
|
func (s *AdminBatchRefreshWorkflowTasksTestSuite) createWorkflow(env *testcore.TestEnv, workflowFn any) sdkclient.WorkflowRun {
|
|
workflowOptions := sdkclient.StartWorkflowOptions{
|
|
ID: testcore.RandomizeStr("wf_id-" + s.T().Name()),
|
|
TaskQueue: env.WorkerTaskQueue(),
|
|
}
|
|
workflowRun, err := env.SdkClient().ExecuteWorkflow(s.Context(), workflowOptions, workflowFn)
|
|
s.NoError(err)
|
|
s.NotNil(workflowRun)
|
|
return workflowRun
|
|
}
|
|
|
|
func (s *AdminBatchRefreshWorkflowTasksTestSuite) TestStartAdminBatchOperation_RefreshWorkflowTasks_Success() {
|
|
env := s.newTestEnv()
|
|
|
|
env.SdkWorker().RegisterWorkflow(s.simpleWorkflow)
|
|
|
|
// Create two workflows
|
|
workflowRun1 := s.createWorkflow(env, s.simpleWorkflow)
|
|
workflowRun2 := s.createWorkflow(env, s.simpleWorkflow)
|
|
|
|
// Wait for workflows to complete
|
|
var out string
|
|
err := workflowRun1.Get(s.Context(), &out)
|
|
s.NoError(err)
|
|
err = workflowRun2.Get(s.Context(), &out)
|
|
s.NoError(err)
|
|
|
|
// Start admin batch operation to refresh workflow tasks using executions list
|
|
resp, err := env.AdminClient().StartAdminBatchOperation(s.Context(), &adminservice.StartAdminBatchOperationRequest{
|
|
Namespace: env.Namespace().String(),
|
|
JobId: uuid.NewString(),
|
|
Reason: "test refresh workflow tasks",
|
|
Identity: "test-identity",
|
|
Executions: []*commonpb.WorkflowExecution{
|
|
{WorkflowId: workflowRun1.GetID(), RunId: workflowRun1.GetRunID()},
|
|
{WorkflowId: workflowRun2.GetID(), RunId: workflowRun2.GetRunID()},
|
|
},
|
|
Operation: &adminservice.StartAdminBatchOperationRequest_RefreshTasksOperation{
|
|
RefreshTasksOperation: &adminservice.BatchOperationRefreshTasks{},
|
|
},
|
|
})
|
|
s.NoError(err)
|
|
s.NotNil(resp)
|
|
}
|
|
|
|
func (s *AdminBatchRefreshWorkflowTasksTestSuite) TestStartAdminBatchOperation_RefreshWorkflowTasks_WithVisibilityQuery() {
|
|
env := s.newTestEnv()
|
|
|
|
env.SdkWorker().RegisterWorkflow(s.simpleWorkflow)
|
|
|
|
// Create workflows
|
|
workflowRun1 := s.createWorkflow(env, s.simpleWorkflow)
|
|
workflowRun2 := s.createWorkflow(env, s.simpleWorkflow)
|
|
|
|
// Wait for workflows to complete
|
|
var out string
|
|
err := workflowRun1.Get(s.Context(), &out)
|
|
s.NoError(err)
|
|
err = workflowRun2.Get(s.Context(), &out)
|
|
s.NoError(err)
|
|
|
|
// Wait for workflows to be visible
|
|
s.EventuallyWithT(func(t *assert.CollectT) {
|
|
resp, err := env.FrontendClient().CountWorkflowExecutions(s.Context(), &workflowservice.CountWorkflowExecutionsRequest{
|
|
Namespace: env.Namespace().String(),
|
|
Query: "WorkflowType='simpleWorkflow'",
|
|
})
|
|
require.NoError(t, err)
|
|
require.GreaterOrEqual(t, resp.GetCount(), int64(2))
|
|
}, 10*time.Second, 500*time.Millisecond)
|
|
|
|
// Start admin batch operation using visibility query
|
|
resp, err := env.AdminClient().StartAdminBatchOperation(s.Context(), &adminservice.StartAdminBatchOperationRequest{
|
|
Namespace: env.Namespace().String(),
|
|
VisibilityQuery: "WorkflowType='simpleWorkflow'",
|
|
JobId: uuid.NewString(),
|
|
Reason: "test refresh workflow tasks with query",
|
|
Identity: "test-identity",
|
|
Operation: &adminservice.StartAdminBatchOperationRequest_RefreshTasksOperation{
|
|
RefreshTasksOperation: &adminservice.BatchOperationRefreshTasks{},
|
|
},
|
|
})
|
|
s.NoError(err)
|
|
s.NotNil(resp)
|
|
}
|
|
|
|
func (s *AdminBatchRefreshWorkflowTasksTestSuite) TestStartAdminBatchOperation_InvalidArgument_NoOperation() {
|
|
env := s.newTestEnv()
|
|
|
|
// Request without operation should fail
|
|
_, err := env.AdminClient().StartAdminBatchOperation(s.Context(), &adminservice.StartAdminBatchOperationRequest{
|
|
Namespace: env.Namespace().String(),
|
|
JobId: uuid.NewString(),
|
|
Reason: "test",
|
|
Executions: []*commonpb.WorkflowExecution{
|
|
{WorkflowId: "test-wf-id", RunId: "test-run-id"},
|
|
},
|
|
})
|
|
s.Error(err)
|
|
s.Equal(codes.InvalidArgument, serviceerror.ToStatus(err).Code())
|
|
}
|
|
|
|
func (s *AdminBatchRefreshWorkflowTasksTestSuite) TestStartAdminBatchOperation_InvalidArgument_NoNamespace() {
|
|
env := s.newTestEnv()
|
|
|
|
// Request without namespace should fail
|
|
_, err := env.AdminClient().StartAdminBatchOperation(s.Context(), &adminservice.StartAdminBatchOperationRequest{
|
|
JobId: uuid.NewString(),
|
|
Reason: "test",
|
|
Identity: "test-identity",
|
|
Executions: []*commonpb.WorkflowExecution{
|
|
{WorkflowId: "test-wf-id", RunId: "test-run-id"},
|
|
},
|
|
Operation: &adminservice.StartAdminBatchOperationRequest_RefreshTasksOperation{
|
|
RefreshTasksOperation: &adminservice.BatchOperationRefreshTasks{},
|
|
},
|
|
})
|
|
s.Error(err)
|
|
s.Equal(codes.InvalidArgument, serviceerror.ToStatus(err).Code())
|
|
}
|
|
|
|
func (s *AdminBatchRefreshWorkflowTasksTestSuite) TestStartAdminBatchOperation_InvalidArgument_NoJobId() {
|
|
env := s.newTestEnv()
|
|
|
|
// Request without job_id should fail
|
|
_, err := env.AdminClient().StartAdminBatchOperation(s.Context(), &adminservice.StartAdminBatchOperationRequest{
|
|
Namespace: env.Namespace().String(),
|
|
Reason: "test",
|
|
Identity: "test-identity",
|
|
Executions: []*commonpb.WorkflowExecution{
|
|
{WorkflowId: "test-wf-id", RunId: "test-run-id"},
|
|
},
|
|
Operation: &adminservice.StartAdminBatchOperationRequest_RefreshTasksOperation{
|
|
RefreshTasksOperation: &adminservice.BatchOperationRefreshTasks{},
|
|
},
|
|
})
|
|
s.Error(err)
|
|
s.Equal(codes.InvalidArgument, serviceerror.ToStatus(err).Code())
|
|
}
|
|
|
|
func (s *AdminBatchRefreshWorkflowTasksTestSuite) TestStartAdminBatchOperation_InvalidArgument_NoExecutionsOrQuery() {
|
|
env := s.newTestEnv()
|
|
|
|
// Request without executions or visibility_query should fail
|
|
_, err := env.AdminClient().StartAdminBatchOperation(s.Context(), &adminservice.StartAdminBatchOperationRequest{
|
|
Namespace: env.Namespace().String(),
|
|
JobId: uuid.NewString(),
|
|
Reason: "test",
|
|
Identity: "test-identity",
|
|
Operation: &adminservice.StartAdminBatchOperationRequest_RefreshTasksOperation{
|
|
RefreshTasksOperation: &adminservice.BatchOperationRefreshTasks{},
|
|
},
|
|
})
|
|
s.Error(err)
|
|
s.Equal(codes.InvalidArgument, serviceerror.ToStatus(err).Code())
|
|
}
|
|
|
|
func (s *AdminBatchRefreshWorkflowTasksTestSuite) TestStartAdminBatchOperation_0_SeparateLimitFromFrontendBatchOperation() {
|
|
env := s.newTestEnv(
|
|
testcore.WithDynamicConfig(dynamicconfig.FrontendMaxConcurrentBatchOperationPerNamespace, 1),
|
|
testcore.WithDynamicConfig(dynamicconfig.FrontendMaxConcurrentAdminBatchOperationPerNamespace, 1),
|
|
)
|
|
|
|
env.SdkWorker().RegisterWorkflow(s.simpleWorkflow)
|
|
|
|
// Create workflows
|
|
workflowRun1 := s.createWorkflow(env, s.simpleWorkflow)
|
|
workflowRun2 := s.createWorkflow(env, s.simpleWorkflow)
|
|
|
|
// Wait for workflows to complete
|
|
var out string
|
|
err := workflowRun1.Get(s.Context(), &out)
|
|
s.NoError(err)
|
|
err = workflowRun2.Get(s.Context(), &out)
|
|
s.NoError(err)
|
|
|
|
_, err = env.FrontendClient().StartBatchOperation(s.Context(), &workflowservice.StartBatchOperationRequest{
|
|
Namespace: env.Namespace().String(),
|
|
Executions: []*commonpb.WorkflowExecution{
|
|
{WorkflowId: workflowRun1.GetID(), RunId: workflowRun1.GetRunID()},
|
|
},
|
|
JobId: uuid.NewString(),
|
|
Reason: "test frontend batch",
|
|
Operation: &workflowservice.StartBatchOperationRequest_SignalOperation{
|
|
SignalOperation: &batchpb.BatchOperationSignal{
|
|
Signal: "test-signal",
|
|
Input: payloads.EncodeString("test-input"),
|
|
Identity: "test-identity",
|
|
},
|
|
},
|
|
})
|
|
s.NoError(err, "frontend batch operation should succeed")
|
|
|
|
_, err = env.AdminClient().StartAdminBatchOperation(s.Context(), &adminservice.StartAdminBatchOperationRequest{
|
|
Namespace: env.Namespace().String(),
|
|
Executions: []*commonpb.WorkflowExecution{
|
|
{WorkflowId: workflowRun2.GetID(), RunId: workflowRun2.GetRunID()},
|
|
},
|
|
JobId: uuid.NewString(),
|
|
Reason: "test admin batch",
|
|
Identity: "test-identity",
|
|
Operation: &adminservice.StartAdminBatchOperationRequest_RefreshTasksOperation{
|
|
RefreshTasksOperation: &adminservice.BatchOperationRefreshTasks{},
|
|
},
|
|
})
|
|
s.NoError(err, "admin batch operation should succeed because it uses a separate limit")
|
|
}
|