mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-04 02:04:03 +00:00
Alex ShtinandGitHub
8e8f1b0f7c
## What changed?
<!-- Describe what has changed in this PR -->
Remove license header from every file. Because it is really hard to
follow in this PR here is the summary:
1. License header is removed from all `.go` and `.proto` files
:fireworks::fireworks:🎆.
2. `LICENSE` file in the root directory has only Temporal and Uber
copyrights.
3. 5 other `LICENSE` files added to the packages which have copyrights
different from Temporal and Uber: Datadog, Xargin, "Mat Ryer, Tyler
Bunnell and contributors".
4. `license_file` flag is removed from all code generation tools.
5. `copyright_file` flag is removed from `go:generate mockgen`
directive.
6. All copyright related targets are removed from `Makefile`.
7. Updated Temporal copyright year to 2025 everywhere.
## Why?
<!-- Tell your future self why have you made these changes -->
I double checked with legal department that it is not needed to have
license header in every file. One file per repo is enough. I put all
copyrights to the root `LICENSE` file and removed header from all other
files. Also updated tools and `Makefile`.
291 lines
9.4 KiB
Go
291 lines
9.4 KiB
Go
package matching
|
|
|
|
import (
|
|
"context"
|
|
"math/rand"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/google/uuid"
|
|
"github.com/stretchr/testify/require"
|
|
"github.com/stretchr/testify/suite"
|
|
commonpb "go.temporal.io/api/common/v1"
|
|
enumspb "go.temporal.io/api/enums/v1"
|
|
"go.temporal.io/api/serviceerror"
|
|
"go.temporal.io/server/api/historyservice/v1"
|
|
"go.temporal.io/server/api/historyservicemock/v1"
|
|
persistencespb "go.temporal.io/server/api/persistence/v1"
|
|
"go.temporal.io/server/common/cluster"
|
|
"go.temporal.io/server/common/namespace"
|
|
"go.temporal.io/server/common/primitives/timestamp"
|
|
"go.uber.org/mock/gomock"
|
|
"google.golang.org/protobuf/types/known/timestamppb"
|
|
)
|
|
|
|
type (
|
|
taskValidatorSuite struct {
|
|
suite.Suite
|
|
*require.Assertions
|
|
|
|
controller *gomock.Controller
|
|
clusterMetadata *cluster.MockMetadata
|
|
historyClient *historyservicemock.MockHistoryServiceClient
|
|
namespaceCache *namespace.MockRegistry
|
|
|
|
namespaceID string
|
|
workflowID string
|
|
runID string
|
|
scheduleEventID int64
|
|
task *persistencespb.AllocatedTaskInfo
|
|
|
|
taskValidator *taskValidatorImpl
|
|
}
|
|
)
|
|
|
|
func TestTaskValidatorSuite(t *testing.T) {
|
|
s := new(taskValidatorSuite)
|
|
suite.Run(t, s)
|
|
}
|
|
|
|
func (s *taskValidatorSuite) SetupTest() {
|
|
s.Assertions = require.New(s.T())
|
|
|
|
s.controller = gomock.NewController(s.T())
|
|
s.clusterMetadata = cluster.NewMockMetadata(s.controller)
|
|
s.historyClient = historyservicemock.NewMockHistoryServiceClient(s.controller)
|
|
s.namespaceCache = namespace.NewMockRegistry(s.controller)
|
|
|
|
s.namespaceID = uuid.New().String()
|
|
s.workflowID = uuid.New().String()
|
|
s.runID = uuid.New().String()
|
|
s.scheduleEventID = rand.Int63()
|
|
s.task = &persistencespb.AllocatedTaskInfo{
|
|
Data: &persistencespb.TaskInfo{
|
|
NamespaceId: s.namespaceID,
|
|
WorkflowId: s.workflowID,
|
|
RunId: s.runID,
|
|
ScheduledEventId: s.scheduleEventID,
|
|
CreateTime: timestamp.TimeNowPtrUtc(),
|
|
},
|
|
}
|
|
|
|
s.taskValidator = newTaskValidator(context.Background(), s.clusterMetadata, s.namespaceCache, s.historyClient)
|
|
}
|
|
|
|
func (s *taskValidatorSuite) TestPreValidateActive_NewTask_Skip_WithCreationTime() {
|
|
s.taskValidator.lastValidatedTaskInfo = taskValidationInfo{
|
|
taskID: s.task.TaskId - 1,
|
|
validationTime: time.Unix(0, rand.Int63()).UTC(),
|
|
}
|
|
s.task.Data.CreateTime = timestamppb.New(time.Unix(0, rand.Int63()))
|
|
|
|
shouldValidate := s.taskValidator.preValidateActive(s.task)
|
|
s.False(shouldValidate)
|
|
s.Equal(taskValidationInfo{
|
|
taskID: s.task.TaskId,
|
|
validationTime: s.task.Data.CreateTime.AsTime(),
|
|
}, s.taskValidator.lastValidatedTaskInfo)
|
|
}
|
|
|
|
func (s *taskValidatorSuite) TestPreValidateActive_NewTask_Skip_WithoutCreationTime() {
|
|
s.taskValidator.lastValidatedTaskInfo = taskValidationInfo{
|
|
taskID: s.task.TaskId - 1,
|
|
validationTime: time.Unix(0, rand.Int63()).UTC(),
|
|
}
|
|
s.task.Data.CreateTime = nil
|
|
|
|
shouldValidate := s.taskValidator.preValidateActive(s.task)
|
|
s.False(shouldValidate)
|
|
s.Equal(s.task.TaskId, s.taskValidator.lastValidatedTaskInfo.taskID)
|
|
s.True(time.Now().Sub(s.taskValidator.lastValidatedTaskInfo.validationTime) < time.Second)
|
|
}
|
|
|
|
func (s *taskValidatorSuite) TestPreValidateActive_ExistingTask_Validate() {
|
|
s.taskValidator.lastValidatedTaskInfo = taskValidationInfo{
|
|
taskID: s.task.TaskId,
|
|
validationTime: time.Now().Add(-taskReaderValidationThreshold * 2),
|
|
}
|
|
|
|
shouldValidate := s.taskValidator.preValidateActive(s.task)
|
|
s.True(shouldValidate)
|
|
}
|
|
|
|
func (s *taskValidatorSuite) TestPreValidateActive_ExistingTask_Skip() {
|
|
s.taskValidator.lastValidatedTaskInfo = taskValidationInfo{
|
|
taskID: s.task.TaskId,
|
|
validationTime: time.Now().Add(taskReaderValidationThreshold * 2),
|
|
}
|
|
|
|
shouldValidate := s.taskValidator.preValidateActive(s.task)
|
|
s.False(shouldValidate)
|
|
}
|
|
|
|
func (s *taskValidatorSuite) TestPreValidatePassive_NewTask_Skip_WithCreationTime() {
|
|
s.taskValidator.lastValidatedTaskInfo = taskValidationInfo{
|
|
taskID: s.task.TaskId - 1,
|
|
validationTime: time.Unix(0, rand.Int63()).UTC(),
|
|
}
|
|
s.task.Data.CreateTime = timestamppb.New(time.Now().Add(-taskReaderValidationThreshold / 2))
|
|
|
|
shouldValidate := s.taskValidator.preValidatePassive(s.task)
|
|
s.False(shouldValidate)
|
|
s.Equal(taskValidationInfo{
|
|
taskID: s.task.TaskId,
|
|
validationTime: s.task.Data.CreateTime.AsTime(),
|
|
}, s.taskValidator.lastValidatedTaskInfo)
|
|
}
|
|
|
|
func (s *taskValidatorSuite) TestPreValidatePassive_NewTask_Validate_WithCreationTime() {
|
|
s.taskValidator.lastValidatedTaskInfo = taskValidationInfo{
|
|
taskID: s.task.TaskId - 1,
|
|
validationTime: time.Unix(0, rand.Int63()).UTC(),
|
|
}
|
|
s.task.Data.CreateTime = timestamppb.New(time.Now().Add(-taskReaderValidationThreshold * 2))
|
|
|
|
shouldValidate := s.taskValidator.preValidatePassive(s.task)
|
|
s.True(shouldValidate)
|
|
s.Equal(taskValidationInfo{
|
|
taskID: s.task.TaskId,
|
|
validationTime: s.task.Data.CreateTime.AsTime(),
|
|
}, s.taskValidator.lastValidatedTaskInfo)
|
|
}
|
|
|
|
func (s *taskValidatorSuite) TestPreValidatePassive_NewTask_Skip_WithoutCreationTime() {
|
|
s.taskValidator.lastValidatedTaskInfo = taskValidationInfo{
|
|
taskID: s.task.TaskId - 1,
|
|
validationTime: time.Unix(0, rand.Int63()).UTC(),
|
|
}
|
|
s.task.Data.CreateTime = nil
|
|
|
|
shouldValidate := s.taskValidator.preValidatePassive(s.task)
|
|
s.False(shouldValidate)
|
|
s.Equal(s.task.TaskId, s.taskValidator.lastValidatedTaskInfo.taskID)
|
|
s.True(time.Now().Sub(s.taskValidator.lastValidatedTaskInfo.validationTime) < time.Second)
|
|
}
|
|
|
|
func (s *taskValidatorSuite) TestPreValidatePassive_ExistingTask_Validate() {
|
|
s.taskValidator.lastValidatedTaskInfo = taskValidationInfo{
|
|
taskID: s.task.TaskId,
|
|
validationTime: time.Now().Add(-taskReaderValidationThreshold * 2),
|
|
}
|
|
|
|
shouldValidate := s.taskValidator.preValidatePassive(s.task)
|
|
s.True(shouldValidate)
|
|
}
|
|
|
|
func (s *taskValidatorSuite) TestPreValidatePassive_ExistingTask_Skip() {
|
|
s.taskValidator.lastValidatedTaskInfo = taskValidationInfo{
|
|
taskID: s.task.TaskId,
|
|
validationTime: time.Now().Add(taskReaderValidationThreshold * 2),
|
|
}
|
|
|
|
shouldValidate := s.taskValidator.preValidatePassive(s.task)
|
|
s.False(shouldValidate)
|
|
}
|
|
|
|
func (s *taskValidatorSuite) TestIsTaskValid_ActivityTask_Valid() {
|
|
taskType := enumspb.TASK_QUEUE_TYPE_ACTIVITY
|
|
|
|
s.historyClient.EXPECT().IsActivityTaskValid(gomock.Any(), &historyservice.IsActivityTaskValidRequest{
|
|
NamespaceId: s.namespaceID,
|
|
Execution: &commonpb.WorkflowExecution{
|
|
WorkflowId: s.workflowID,
|
|
RunId: s.runID,
|
|
},
|
|
Clock: s.task.Data.Clock,
|
|
ScheduledEventId: s.task.Data.ScheduledEventId,
|
|
}).Return(&historyservice.IsActivityTaskValidResponse{IsValid: true}, nil)
|
|
|
|
valid, err := s.taskValidator.isTaskValid(s.task, taskType)
|
|
s.NoError(err)
|
|
s.True(valid)
|
|
}
|
|
|
|
func (s *taskValidatorSuite) TestIsTaskValid_ActivityTask_NotFound() {
|
|
taskType := enumspb.TASK_QUEUE_TYPE_ACTIVITY
|
|
|
|
s.historyClient.EXPECT().IsActivityTaskValid(gomock.Any(), &historyservice.IsActivityTaskValidRequest{
|
|
NamespaceId: s.namespaceID,
|
|
Execution: &commonpb.WorkflowExecution{
|
|
WorkflowId: s.workflowID,
|
|
RunId: s.runID,
|
|
},
|
|
Clock: s.task.Data.Clock,
|
|
ScheduledEventId: s.task.Data.ScheduledEventId,
|
|
}).Return(nil, &serviceerror.NotFound{})
|
|
|
|
valid, err := s.taskValidator.isTaskValid(s.task, taskType)
|
|
s.NoError(err)
|
|
s.False(valid)
|
|
}
|
|
|
|
func (s *taskValidatorSuite) TestIsTaskValid_ActivityTask_Error() {
|
|
taskType := enumspb.TASK_QUEUE_TYPE_ACTIVITY
|
|
|
|
s.historyClient.EXPECT().IsActivityTaskValid(gomock.Any(), &historyservice.IsActivityTaskValidRequest{
|
|
NamespaceId: s.namespaceID,
|
|
Execution: &commonpb.WorkflowExecution{
|
|
WorkflowId: s.workflowID,
|
|
RunId: s.runID,
|
|
},
|
|
Clock: s.task.Data.Clock,
|
|
ScheduledEventId: s.task.Data.ScheduledEventId,
|
|
}).Return(nil, &serviceerror.Unavailable{})
|
|
|
|
_, err := s.taskValidator.isTaskValid(s.task, taskType)
|
|
s.Error(err)
|
|
}
|
|
|
|
func (s *taskValidatorSuite) TestIsTaskValid_WorkflowTask_Valid() {
|
|
taskType := enumspb.TASK_QUEUE_TYPE_WORKFLOW
|
|
|
|
s.historyClient.EXPECT().IsWorkflowTaskValid(gomock.Any(), &historyservice.IsWorkflowTaskValidRequest{
|
|
NamespaceId: s.namespaceID,
|
|
Execution: &commonpb.WorkflowExecution{
|
|
WorkflowId: s.workflowID,
|
|
RunId: s.runID,
|
|
},
|
|
Clock: s.task.Data.Clock,
|
|
ScheduledEventId: s.task.Data.ScheduledEventId,
|
|
}).Return(&historyservice.IsWorkflowTaskValidResponse{IsValid: true}, nil)
|
|
|
|
valid, err := s.taskValidator.isTaskValid(s.task, taskType)
|
|
s.NoError(err)
|
|
s.True(valid)
|
|
}
|
|
|
|
func (s *taskValidatorSuite) TestIsTaskValid_WorkflowTask_NotFound() {
|
|
taskType := enumspb.TASK_QUEUE_TYPE_WORKFLOW
|
|
|
|
s.historyClient.EXPECT().IsWorkflowTaskValid(gomock.Any(), &historyservice.IsWorkflowTaskValidRequest{
|
|
NamespaceId: s.namespaceID,
|
|
Execution: &commonpb.WorkflowExecution{
|
|
WorkflowId: s.workflowID,
|
|
RunId: s.runID,
|
|
},
|
|
Clock: s.task.Data.Clock,
|
|
ScheduledEventId: s.task.Data.ScheduledEventId,
|
|
}).Return(nil, &serviceerror.NotFound{})
|
|
|
|
valid, err := s.taskValidator.isTaskValid(s.task, taskType)
|
|
s.NoError(err)
|
|
s.False(valid)
|
|
}
|
|
|
|
func (s *taskValidatorSuite) TestIsTaskValid_WorkflowTask_Error() {
|
|
taskType := enumspb.TASK_QUEUE_TYPE_WORKFLOW
|
|
|
|
s.historyClient.EXPECT().IsWorkflowTaskValid(gomock.Any(), &historyservice.IsWorkflowTaskValidRequest{
|
|
NamespaceId: s.namespaceID,
|
|
Execution: &commonpb.WorkflowExecution{
|
|
WorkflowId: s.workflowID,
|
|
RunId: s.runID,
|
|
},
|
|
Clock: s.task.Data.Clock,
|
|
ScheduledEventId: s.task.Data.ScheduledEventId,
|
|
}).Return(nil, &serviceerror.Unavailable{})
|
|
|
|
_, err := s.taskValidator.isTaskValid(s.task, taskType)
|
|
s.Error(err)
|
|
}
|