Documentation
¶
Index ¶
- Constants
- Variables
- func ApplyTestClusterOptions(options []TestClusterOption) testClusterParams
- func DecodeString(t require.TestingT, pls *commonpb.Payloads) string
- func EventBatchesToVersionHistory(versionHistory *historyspb.VersionHistory, eventBatches []*historypb.History) (*historyspb.VersionHistory, error)
- func ExtractReplicationMessages(msg proto.Message) *replicationspb.WorkflowReplicationMessages
- func GetPersistenceTestDefaults() persistencetests.TestBaseOptions
- func MustToPayload(t require.TestingT, v any) *commonpb.Payload
- func NewContext(parent ...context.Context) context.Context
- func NewTestDataConverter() converter.DataConverter
- func PickRolloutSplit(t *testing.T, namespace string, percent int) (accepted, rejected string)
- func RandomizeStr(id string) string
- func RandomizedNexusEndpoint(name string) string
- func UseCassandraPersistence() bool
- func UseSQLVisibility() bool
- func UseSuiteScopedCluster(t *testing.T)deprecated
- func WithDropTask(o *PollAndProcessWorkflowTaskOptions)
- func WithDumpHistory(o *PollAndProcessWorkflowTaskOptions)
- func WithForceNewWorkflowTask(o *PollAndProcessWorkflowTaskOptions)
- func WithNoDumpCommands(o *PollAndProcessWorkflowTaskOptions)
- func WithPollSticky(o *PollAndProcessWorkflowTaskOptions)
- func WithRespondSticky(o *PollAndProcessWorkflowTaskOptions)
- func WithoutRetries(o *PollAndProcessWorkflowTaskOptions)
- type ActivityTaskHandler
- type CapturedReplicationMessage
- type Env
- type FrontendConfig
- type FunctionalTestBase
- func (s *FunctionalTestBase) AdminClient() adminservice.AdminServiceClient
- func (s *FunctionalTestBase) CloseShard(namespaceID string, workflowID string)
- func (s *FunctionalTestBase) DecodePayloadsInt(ps *commonpb.Payloads) int
- func (s *FunctionalTestBase) ExternalNamespace() namespace.Name
- func (s *FunctionalTestBase) FrontendClient() workflowservice.WorkflowServiceClient
- func (s *FunctionalTestBase) FrontendGRPCAddress() string
- func (s *FunctionalTestBase) GetHistory(namespace string, execution *commonpb.WorkflowExecution) []*historypb.HistoryEvent
- func (s *FunctionalTestBase) GetHistoryFunc(namespace string, execution *commonpb.WorkflowExecution) func() []*historypb.HistoryEvent
- func (s *FunctionalTestBase) GetNamespaceID(namespace string) string
- func (s *FunctionalTestBase) GetTestCluster() *TestCluster
- func (s *FunctionalTestBase) GetTestClusterConfig() *TestClusterConfig
- func (s *FunctionalTestBase) HttpAPIAddress() string
- func (s *FunctionalTestBase) InjectHook(hook testhooks.Hook) (cleanup func())
- func (s *FunctionalTestBase) MarkNamespaceAsDeleted(nsName namespace.Name) error
- func (s *FunctionalTestBase) Namespace() namespace.Name
- func (s *FunctionalTestBase) NamespaceID() namespace.ID
- func (s *FunctionalTestBase) OperatorClient() operatorservice.OperatorServiceClient
- func (s *FunctionalTestBase) OverrideDynamicConfig(setting dynamicconfig.GenericSetting, value any) (cleanup func())
- func (s *FunctionalTestBase) Poisoned() bool
- func (s *FunctionalTestBase) RegisterNamespace(nsName namespace.Name, retentionDays int32, ...) (namespace.ID, error)
- func (s *FunctionalTestBase) RegisterTest(t testlogger.CleanupCapableT)
- func (s *FunctionalTestBase) RunTestWithMatchingBehavior(subtest func())
- func (s *FunctionalTestBase) SdkClient() sdkclient.Client
- func (s *FunctionalTestBase) SdkWorker() sdkworker.Worker
- func (s *FunctionalTestBase) SendSignal(nsName string, execution *commonpb.WorkflowExecution, signalName string, ...) error
- func (s *FunctionalTestBase) SendToChannel(ctx context.Context, ch chan struct{})deprecated
- func (s *FunctionalTestBase) SetupSubTest()
- func (s *FunctionalTestBase) SetupSuite()
- func (s *FunctionalTestBase) SetupSuiteWithCluster(options ...TestClusterOption)
- func (s *FunctionalTestBase) SetupTest()
- func (s *FunctionalTestBase) TaskPoller() *taskpoller.TaskPoller
- func (s *FunctionalTestBase) TaskQueue() string
- func (s *FunctionalTestBase) TearDownCluster()
- func (s *FunctionalTestBase) TearDownSubTest()
- func (s *FunctionalTestBase) TearDownSuite()
- func (s *FunctionalTestBase) TearDownTest()
- func (s *FunctionalTestBase) WaitForChannel(ctx context.Context, ch chan struct{})deprecated
- func (s *FunctionalTestBase) WorkerGRPCAddress() string
- type GlobalMetricCapture
- type HistoryConfig
- type HistoryTaskRecorder
- func (r *HistoryTaskRecorder) AddHistoryTasks(ctx context.Context, request *persistence.AddHistoryTasksRequest) error
- func (r *HistoryTaskRecorder) AppendHistoryNodes(ctx context.Context, request *persistence.AppendHistoryNodesRequest) (*persistence.AppendHistoryNodesResponse, error)
- func (r *HistoryTaskRecorder) AppendRawHistoryNodes(ctx context.Context, request *persistence.AppendRawHistoryNodesRequest) (*persistence.AppendHistoryNodesResponse, error)
- func (r *HistoryTaskRecorder) Close()
- func (r *HistoryTaskRecorder) CompleteHistoryTask(ctx context.Context, request *persistence.CompleteHistoryTaskRequest) error
- func (r *HistoryTaskRecorder) ConflictResolveWorkflowExecution(ctx context.Context, ...) (*persistence.ConflictResolveWorkflowExecutionResponse, error)
- func (r *HistoryTaskRecorder) CountMatchingTasks(category tasks.Category, matcher TaskMatcher) int
- func (r *HistoryTaskRecorder) CountTasksForNamespace(category tasks.Category, namespaceID string, matcher TaskMatcher) int
- func (r *HistoryTaskRecorder) CountTasksForWorkflow(category tasks.Category, namespaceID string, workflowID string, runID string, ...) int
- func (r *HistoryTaskRecorder) CreateWorkflowExecution(ctx context.Context, request *persistence.CreateWorkflowExecutionRequest) (*persistence.CreateWorkflowExecutionResponse, error)
- func (r *HistoryTaskRecorder) DeleteCurrentWorkflowExecution(ctx context.Context, ...) error
- func (r *HistoryTaskRecorder) DeleteHistoryBranch(ctx context.Context, request *persistence.DeleteHistoryBranchRequest) error
- func (r *HistoryTaskRecorder) DeleteReplicationTaskFromDLQ(ctx context.Context, request *persistence.DeleteReplicationTaskFromDLQRequest) error
- func (r *HistoryTaskRecorder) DeleteWorkflowExecution(ctx context.Context, request *persistence.DeleteWorkflowExecutionRequest) error
- func (r *HistoryTaskRecorder) ForkHistoryBranch(ctx context.Context, request *persistence.ForkHistoryBranchRequest) (*persistence.ForkHistoryBranchResponse, error)
- func (r *HistoryTaskRecorder) GetAllHistoryTreeBranches(ctx context.Context, request *persistence.GetAllHistoryTreeBranchesRequest) (*persistence.GetAllHistoryTreeBranchesResponse, error)
- func (r *HistoryTaskRecorder) GetAllRecordedTasks() map[tasks.Category][]RecordedTask
- func (r *HistoryTaskRecorder) GetAllTasks() map[tasks.Category][]tasks.Task
- func (r *HistoryTaskRecorder) GetCurrentExecution(ctx context.Context, request *persistence.GetCurrentExecutionRequest) (*persistence.GetCurrentExecutionResponse, error)
- func (r *HistoryTaskRecorder) GetHistoryBranchUtil() persistence.HistoryBranchUtil
- func (r *HistoryTaskRecorder) GetHistoryTasks(ctx context.Context, request *persistence.GetHistoryTasksRequest) (*persistence.GetHistoryTasksResponse, error)
- func (r *HistoryTaskRecorder) GetName() string
- func (r *HistoryTaskRecorder) GetRecordedTasksByCategoryFiltered(category tasks.Category, filter TaskFilter) []RecordedTask
- func (r *HistoryTaskRecorder) GetReplicationTasksFromDLQ(ctx context.Context, request *persistence.GetReplicationTasksFromDLQRequest) (*persistence.GetHistoryTasksResponse, error)
- func (r *HistoryTaskRecorder) GetWorkflowExecution(ctx context.Context, request *persistence.GetWorkflowExecutionRequest) (*persistence.GetWorkflowExecutionResponse, error)
- func (r *HistoryTaskRecorder) IsReplicationDLQEmpty(ctx context.Context, request *persistence.GetReplicationTasksFromDLQRequest) (bool, error)
- func (r *HistoryTaskRecorder) ListConcreteExecutions(ctx context.Context, request *persistence.ListConcreteExecutionsRequest) (*persistence.ListConcreteExecutionsResponse, error)
- func (r *HistoryTaskRecorder) MatchTasks(category tasks.Category, matcher TaskMatcher) []RecordedTask
- func (r *HistoryTaskRecorder) MatchTasksForNamespace(category tasks.Category, namespaceID string, matcher TaskMatcher) []RecordedTask
- func (r *HistoryTaskRecorder) MatchTasksForWorkflow(category tasks.Category, namespaceID string, workflowID string, runID string, ...) []RecordedTask
- func (r *HistoryTaskRecorder) PutReplicationTaskToDLQ(ctx context.Context, request *persistence.PutReplicationTaskToDLQRequest) error
- func (r *HistoryTaskRecorder) RangeCompleteHistoryTasks(ctx context.Context, request *persistence.RangeCompleteHistoryTasksRequest) error
- func (r *HistoryTaskRecorder) RangeDeleteReplicationTaskFromDLQ(ctx context.Context, ...) error
- func (r *HistoryTaskRecorder) ReadHistoryBranch(ctx context.Context, request *persistence.ReadHistoryBranchRequest) (*persistence.ReadHistoryBranchResponse, error)
- func (r *HistoryTaskRecorder) ReadHistoryBranchByBatch(ctx context.Context, request *persistence.ReadHistoryBranchRequest) (*persistence.ReadHistoryBranchByBatchResponse, error)
- func (r *HistoryTaskRecorder) ReadHistoryBranchReverse(ctx context.Context, request *persistence.ReadHistoryBranchReverseRequest) (*persistence.ReadHistoryBranchReverseResponse, error)
- func (r *HistoryTaskRecorder) ReadRawHistoryBranch(ctx context.Context, request *persistence.ReadHistoryBranchRequest) (*persistence.ReadRawHistoryBranchResponse, error)
- func (r *HistoryTaskRecorder) SetWorkflowExecution(ctx context.Context, request *persistence.SetWorkflowExecutionRequest) (*persistence.SetWorkflowExecutionResponse, error)
- func (r *HistoryTaskRecorder) TrimHistoryBranch(ctx context.Context, request *persistence.TrimHistoryBranchRequest) (*persistence.TrimHistoryBranchResponse, error)
- func (r *HistoryTaskRecorder) UpdateWorkflowExecution(ctx context.Context, request *persistence.UpdateWorkflowExecutionRequest) (*persistence.UpdateWorkflowExecutionResponse, error)
- func (r *HistoryTaskRecorder) WriteToLog(filePath string) error
- type MatchingBehavior
- type MatchingConfig
- type MessageHandler
- type NamespaceMetricCapture
- type PollAndProcessWorkflowTaskOptionFunc
- type PollAndProcessWorkflowTaskOptions
- type PollAndProcessWorkflowTaskResponse
- type QueryHandler
- type RecordedTask
- type ReplicationStreamRecorder
- func (r *ReplicationStreamRecorder) Clear()
- func (r *ReplicationStreamRecorder) GetMessages() []CapturedReplicationMessage
- func (r *ReplicationStreamRecorder) SetOutputFile(filePath string)
- func (r *ReplicationStreamRecorder) StreamInterceptor(clusterName string) grpc.StreamClientInterceptor
- func (r *ReplicationStreamRecorder) StreamServerInterceptor(clusterName string) grpc.StreamServerInterceptor
- func (r *ReplicationStreamRecorder) UnaryInterceptor(clusterName string) grpc.UnaryClientInterceptor
- func (r *ReplicationStreamRecorder) UnaryServerInterceptor(clusterName string) grpc.UnaryServerInterceptor
- func (r *ReplicationStreamRecorder) WriteToLog() error
- type TaskFilter
- type TaskMatcher
- type TaskPollerdeprecated
- func (p *TaskPoller) HandlePartialWorkflowTask(response *workflowservice.PollWorkflowTaskQueueResponse, ...) (*workflowservice.RespondWorkflowTaskCompletedResponse, error)
- func (p *TaskPoller) PollAndProcessActivityTask(dropTask bool) error
- func (p *TaskPoller) PollAndProcessActivityTaskWithID(dropTask bool) error
- func (p *TaskPoller) PollAndProcessWorkflowTask(funcs ...PollAndProcessWorkflowTaskOptionFunc) (res PollAndProcessWorkflowTaskResponse, err error)
- func (p *TaskPoller) PollAndProcessWorkflowTaskWithOptions(opts *PollAndProcessWorkflowTaskOptions) (res PollAndProcessWorkflowTaskResponse, err error)
- type TestCluster
- func (tc *TestCluster) AdminClient() adminservice.AdminServiceClient
- func (tc *TestCluster) ClusterName() string
- func (tc *TestCluster) ExecutionManager() persistence.ExecutionManager
- func (tc *TestCluster) FrontendClient() workflowservice.WorkflowServiceClient
- func (tc *TestCluster) GetHistoryTaskRecorder() *HistoryTaskRecorder
- func (tc *TestCluster) GetReplicationStreamRecorder() *ReplicationStreamRecorder
- func (tc *TestCluster) HistoryClient() historyservice.HistoryServiceClient
- func (tc *TestCluster) Host() *temporalImpl
- func (tc *TestCluster) InjectHook(t *testing.T, hook testhooks.Hook, scope any) func()
- func (tc *TestCluster) MatchingClient() matchingservice.MatchingServiceClient
- func (tc *TestCluster) OperatorClient() operatorservice.OperatorServiceClient
- func (tc *TestCluster) OverrideDynamicConfig(t *testing.T, key dynamicconfig.GenericSetting, value any) (cleanup func())
- func (tc *TestCluster) SchedulerClient() schedulerpb.SchedulerServiceClient
- func (tc *TestCluster) TearDownCluster() error
- func (tc *TestCluster) TestBase() *persistencetests.TestBase
- func (tc *TestCluster) WorkerGRPCAddress() string
- type TestClusterConfig
- type TestClusterFactory
- type TestClusterOption
- func WithClusterHistoryTaskRecorder() TestClusterOption
- func WithClusterLogger(logger log.Logger) TestClusterOption
- func WithDCRedirectionPolicy(policy config.DCRedirectionPolicy) TestClusterOption
- func WithDynamicConfigOverrides(overrides map[dynamicconfig.Key]any) TestClusterOption
- func WithFaultInjectionConfig(cfg *config.FaultInjection) TestClusterOption
- func WithNumHistoryShards(n int32) TestClusterOption
- func WithReplicationStreamRecorder() TestClusterOption
- func WithSharedCluster() TestClusterOption
- type TestDataConverter
- func (tdc *TestDataConverter) FromPayload(payload *commonpb.Payload, valuePtr any) error
- func (tdc *TestDataConverter) FromPayloads(payloads *commonpb.Payloads, valuePtrs ...any) error
- func (tdc *TestDataConverter) ToPayload(value any) (*commonpb.Payload, error)
- func (tdc *TestDataConverter) ToPayloads(values ...any) (*commonpb.Payloads, error)
- func (tdc *TestDataConverter) ToString(payload *commonpb.Payload) string
- func (tdc *TestDataConverter) ToStrings(payloads *commonpb.Payloads) []string
- type TestEnv
- func (e *TestEnv) CloseShard(namespaceID string, workflowID string)
- func (e *TestEnv) Context() context.Contextdeprecated
- func (e *TestEnv) Error(err error, msgAndArgs ...any)deprecated
- func (e *TestEnv) InjectHook(hook testhooks.Hook) (cleanup func())
- func (e *TestEnv) Namespace() namespace.Name
- func (e *TestEnv) NamespaceID() namespace.ID
- func (e *TestEnv) NoError(err error, msgAndArgs ...any)deprecated
- func (e *TestEnv) OverrideDynamicConfig(setting dynamicconfig.GenericSetting, value any) (cleanup func())
- func (e *TestEnv) Run(name string, subtest func()) booldeprecated
- func (e *TestEnv) SdkClient() sdkclient.Client
- func (e *TestEnv) SdkWorker() sdkworker.Worker
- func (e *TestEnv) SendToChannel(ch chan<- struct{})
- func (e *TestEnv) SetOnAuthorize(...)
- func (e *TestEnv) SetOnGetClaims(fn func(*authorization.AuthInfo) (*authorization.Claims, error))
- func (e *TestEnv) StartGlobalMetricCapture() *GlobalMetricCapture
- func (e *TestEnv) StartNamespaceMetricCapture() *NamespaceMetricCapture
- func (e *TestEnv) StartNamespaceMetricCaptureFor(namespaceName string) *NamespaceMetricCapture
- func (e *TestEnv) T() *testing.Tdeprecated
- func (e *TestEnv) TaskPoller() *taskpoller.TaskPoller
- func (e *TestEnv) Tv() *testvars.TestVars
- func (e *TestEnv) WaitForChannel(ch <-chan struct{})
- func (e *TestEnv) WorkerTaskQueue() string
- type TestOption
- func WithArchival() TestOption
- func WithCustomArchivers(historyFactory provider.CustomHistoryArchiverFactory, ...) TestOption
- func WithDedicatedCluster() TestOption
- func WithDisableTestloggerFailure() TestOption
- func WithDynamicConfig(setting dynamicconfig.GenericSetting, value any) TestOption
- func WithHistoryShardCount(n int32) TestOption
- func WithHistoryTaskRecorder() TestOption
- func WithLogger(logger log.Logger) TestOption
- func WithMTLS() TestOption
- func WithPersistenceFaultInjection(cfg *config.FaultInjection) TestOption
- func WithSdkWorker() TestOptiondeprecated
- func WithTestVars(fn func(*testvars.TestVars) *testvars.TestVars) TestOption
- func WithWorkerService(reason string) TestOption
- type WorkerConfig
- type WorkflowTaskHandler
Constants ¶
const ( DefaultPageSize = 5 PprofTestPort = 7000 TlsCertCommonName = "my-common-name" ClientSuiteLimit = 10 // TODO (alex): replace all sleeps with WaitForESToSettle with s.Eventually() WaitForESToSettle = 4 * time.Second // wait es shards for some time ensure data consistent )
const ( DirectionSend = "send" DirectionRecv = "recv" DirectionServerSend = "server_send" DirectionServerRecv = "server_recv" )
Message direction constants
const NamespaceCacheRefreshInterval = time.Second
Variables ¶
var ( ErrEncodingIsNotSet = errors.New("payload encoding metadata is not set") ErrEncodingIsNotSupported = errors.New("payload encoding is not supported") )
var (
ErrNoTasks = errors.New("no tasks")
)
Functions ¶
func ApplyTestClusterOptions ¶ added in v1.27.0
func ApplyTestClusterOptions(options []TestClusterOption) testClusterParams
func EventBatchesToVersionHistory ¶
func EventBatchesToVersionHistory( versionHistory *historyspb.VersionHistory, eventBatches []*historypb.History, ) (*historyspb.VersionHistory, error)
func ExtractReplicationMessages ¶ added in v1.30.0
func ExtractReplicationMessages(msg proto.Message) *replicationspb.WorkflowReplicationMessages
ExtractReplicationMessages extracts WorkflowReplicationMessages from a proto message. This is a helper for tests that need to inspect replication message contents.
func GetPersistenceTestDefaults ¶ added in v1.31.0
func GetPersistenceTestDefaults() persistencetests.TestBaseOptions
GetPersistenceTestDefaults returns the default persistence options based on CLI flags. Use this when creating TestClusterConfig to ensure proper database configuration.
func MustToPayload ¶ added in v1.31.0
MustToPayload converts a value to a Payload using the default data converter.
func NewContext ¶
NewContext creates a context with default timeout and RPC headers.
NOTE: If you're using testcore.NewEnv, you can use env.Context() directly - it already includes RPC headers. This function is primarily for legacy tests or creating standalone contexts outside of the TestEnv framework.
If a parent context is provided, the returned context will be canceled when either the timeout expires OR the parent is canceled.
func NewTestDataConverter ¶
func NewTestDataConverter() converter.DataConverter
TODO (alex): use it by default SdkCleint everywhere?
func PickRolloutSplit ¶
PickRolloutSplit returns two distinct business IDs in the given namespace such that the first is accepted by RolloutAccepts at percent and the second is rejected. Fails the test if no split is found.
func RandomizeStr ¶
func RandomizedNexusEndpoint ¶
func UseCassandraPersistence ¶ added in v1.29.0
func UseCassandraPersistence() bool
func UseSQLVisibility ¶ added in v1.27.0
func UseSQLVisibility() bool
func UseSuiteScopedCluster
deprecated
func WithDropTask ¶
func WithDropTask(o *PollAndProcessWorkflowTaskOptions)
func WithDumpHistory ¶
func WithDumpHistory(o *PollAndProcessWorkflowTaskOptions)
func WithForceNewWorkflowTask ¶
func WithForceNewWorkflowTask(o *PollAndProcessWorkflowTaskOptions)
func WithNoDumpCommands ¶
func WithNoDumpCommands(o *PollAndProcessWorkflowTaskOptions)
func WithPollSticky ¶
func WithPollSticky(o *PollAndProcessWorkflowTaskOptions)
func WithRespondSticky ¶
func WithRespondSticky(o *PollAndProcessWorkflowTaskOptions)
func WithoutRetries ¶
func WithoutRetries(o *PollAndProcessWorkflowTaskOptions)
Types ¶
type ActivityTaskHandler ¶
type ActivityTaskHandler func(task *workflowservice.PollActivityTaskQueueResponse) (*commonpb.Payloads, bool, error)
type CapturedReplicationMessage ¶ added in v1.30.0
type CapturedReplicationMessage struct {
Timestamp string `json:"timestamp"`
Method string `json:"method"`
Direction string `json:"direction"`
ClusterName string `json:"clusterName"`
TargetAddress string `json:"targetAddress"`
MessageType string `json:"messageType"`
IsStreamCall bool `json:"isStreamCall"`
Request proto.Message `json:"-"` // Don't marshal directly
Response proto.Message `json:"-"` // Don't marshal directly
Message json.RawMessage `json:"message,omitempty"`
}
CapturedReplicationMessage represents a captured replication message
type Env ¶ added in v1.31.0
type Env interface {
// T returns the *testing.T.
//
// Deprecated: use the suite's T() method instead.
T() *testing.T
Namespace() namespace.Name
NamespaceID() namespace.ID
FrontendClient() workflowservice.WorkflowServiceClient
AdminClient() adminservice.AdminServiceClient
GetTestCluster() *TestCluster
CloseShard(namespaceID string, workflowID string)
OverrideDynamicConfig(setting dynamicconfig.GenericSetting, value any) (cleanup func())
// Deprecated: use the suite's Context() method instead.
Context() context.Context
InjectHook(hook testhooks.Hook) (cleanup func())
}
type FrontendConfig ¶
type FrontendConfig struct {
NumFrontendHosts int
}
FrontendConfig is the config for the frontend service
type FunctionalTestBase ¶
type FunctionalTestBase struct {
suite.Suite
// `suite.Suite` embeds `*assert.Assertions` which, by default, makes all asserts (like `s.NoError(err)`)
// only log the error, continue test execution, and only then fail the test.
// This is not desired behavior in most cases. The idiomatic way to change this behavior
// is to replace `*assert.Assertions` with `*require.Assertions` by embedding it in every test suite
// (or base struct of every test suite).
*require.Assertions
protorequire.ProtoAssertions
historyrequire.HistoryRequire
updateutils.UpdateUtils
Logger log.Logger
// contains filtered or unexported fields
}
func (*FunctionalTestBase) AdminClient ¶
func (s *FunctionalTestBase) AdminClient() adminservice.AdminServiceClient
func (*FunctionalTestBase) CloseShard ¶ added in v1.31.0
func (s *FunctionalTestBase) CloseShard(namespaceID string, workflowID string)
CloseShard closes the shard that contains the given workflow. This is a cluster-global operation and cannot be called on shared clusters.
func (*FunctionalTestBase) DecodePayloadsInt ¶
func (s *FunctionalTestBase) DecodePayloadsInt(ps *commonpb.Payloads) int
func (*FunctionalTestBase) ExternalNamespace ¶ added in v1.29.0
func (s *FunctionalTestBase) ExternalNamespace() namespace.Name
func (*FunctionalTestBase) FrontendClient ¶
func (s *FunctionalTestBase) FrontendClient() workflowservice.WorkflowServiceClient
func (*FunctionalTestBase) FrontendGRPCAddress ¶
func (s *FunctionalTestBase) FrontendGRPCAddress() string
func (*FunctionalTestBase) GetHistory ¶
func (s *FunctionalTestBase) GetHistory(namespace string, execution *commonpb.WorkflowExecution) []*historypb.HistoryEvent
func (*FunctionalTestBase) GetHistoryFunc ¶
func (s *FunctionalTestBase) GetHistoryFunc(namespace string, execution *commonpb.WorkflowExecution) func() []*historypb.HistoryEvent
func (*FunctionalTestBase) GetNamespaceID ¶
func (s *FunctionalTestBase) GetNamespaceID(namespace string) string
func (*FunctionalTestBase) GetTestCluster ¶
func (s *FunctionalTestBase) GetTestCluster() *TestCluster
func (*FunctionalTestBase) GetTestClusterConfig ¶
func (s *FunctionalTestBase) GetTestClusterConfig() *TestClusterConfig
func (*FunctionalTestBase) HttpAPIAddress ¶
func (s *FunctionalTestBase) HttpAPIAddress() string
func (*FunctionalTestBase) InjectHook ¶ added in v1.27.0
func (s *FunctionalTestBase) InjectHook(hook testhooks.Hook) (cleanup func())
InjectHook sets a test hook inside the cluster.
func (*FunctionalTestBase) MarkNamespaceAsDeleted ¶ added in v1.27.0
func (s *FunctionalTestBase) MarkNamespaceAsDeleted( nsName namespace.Name, ) error
func (*FunctionalTestBase) Namespace ¶
func (s *FunctionalTestBase) Namespace() namespace.Name
func (*FunctionalTestBase) NamespaceID ¶ added in v1.27.0
func (s *FunctionalTestBase) NamespaceID() namespace.ID
func (*FunctionalTestBase) OperatorClient ¶
func (s *FunctionalTestBase) OperatorClient() operatorservice.OperatorServiceClient
func (*FunctionalTestBase) OverrideDynamicConfig ¶
func (s *FunctionalTestBase) OverrideDynamicConfig(setting dynamicconfig.GenericSetting, value any) (cleanup func())
func (*FunctionalTestBase) Poisoned ¶
func (s *FunctionalTestBase) Poisoned() bool
Poisoned reports whether the cluster's logger has recorded a failing log.
func (*FunctionalTestBase) RegisterNamespace ¶ added in v1.27.0
func (s *FunctionalTestBase) RegisterNamespace( nsName namespace.Name, retentionDays int32, archivalState enumspb.ArchivalState, historyArchivalURI string, visibilityArchivalURI string, ) (namespace.ID, error)
Register namespace using persistence API because:
- The Retention period is set to 0 for archival tests, and this can't be done through FE,
- Update search attributes would require an extra API call,
- One more extra API call would be necessary to get namespace.ID.
func (*FunctionalTestBase) RegisterTest ¶
func (s *FunctionalTestBase) RegisterTest(t testlogger.CleanupCapableT)
RegisterTest records t as currently using this cluster. At t's Cleanup it fails t if the cluster was poisoned during t's window. This fails all active tests currently running. The cluster will be torn down if t was the last active test on a poisoned cluster. The cluster pool's slot reference is replaced as soon as poison is observed.
func (*FunctionalTestBase) RunTestWithMatchingBehavior ¶
func (s *FunctionalTestBase) RunTestWithMatchingBehavior(subtest func())
func (*FunctionalTestBase) SdkClient ¶ added in v1.28.0
func (s *FunctionalTestBase) SdkClient() sdkclient.Client
func (*FunctionalTestBase) SdkWorker ¶ added in v1.31.0
func (s *FunctionalTestBase) SdkWorker() sdkworker.Worker
func (*FunctionalTestBase) SendSignal ¶ added in v1.27.0
func (s *FunctionalTestBase) SendSignal(nsName string, execution *commonpb.WorkflowExecution, signalName string, input *commonpb.Payloads, identity string) error
TODO (alex): change to nsName namespace.Name
func (*FunctionalTestBase) SendToChannel
deprecated
added in
v1.31.0
func (s *FunctionalTestBase) SendToChannel(ctx context.Context, ch chan struct{})
Deprecated: use (*TestEnv).SendToChannel instead.
func (*FunctionalTestBase) SetupSubTest ¶ added in v1.27.0
func (s *FunctionalTestBase) SetupSubTest()
func (*FunctionalTestBase) SetupSuite ¶
func (s *FunctionalTestBase) SetupSuite()
func (*FunctionalTestBase) SetupSuiteWithCluster ¶ added in v1.27.0
func (s *FunctionalTestBase) SetupSuiteWithCluster(options ...TestClusterOption)
func (*FunctionalTestBase) SetupTest ¶
func (s *FunctionalTestBase) SetupTest()
All test suites that inherit FunctionalTestBase and overwrite SetupTest must call this testcore FunctionalTestBase.SetupTest function to distribute the tests into partitions. Otherwise, the test suite will be executed multiple times in each partition.
func (*FunctionalTestBase) TaskPoller ¶ added in v1.28.0
func (s *FunctionalTestBase) TaskPoller() *taskpoller.TaskPoller
func (*FunctionalTestBase) TaskQueue ¶ added in v1.28.0
func (s *FunctionalTestBase) TaskQueue() string
func (*FunctionalTestBase) TearDownCluster ¶ added in v1.27.0
func (s *FunctionalTestBase) TearDownCluster()
func (*FunctionalTestBase) TearDownSubTest ¶ added in v1.28.0
func (s *FunctionalTestBase) TearDownSubTest()
**IMPORTANT**: When overridding this, make sure to invoke `s.FunctionalTestBase.TearDownSubTest()`.
func (*FunctionalTestBase) TearDownSuite ¶
func (s *FunctionalTestBase) TearDownSuite()
func (*FunctionalTestBase) TearDownTest ¶ added in v1.28.0
func (s *FunctionalTestBase) TearDownTest()
**IMPORTANT**: When overridding this, make sure to invoke `s.FunctionalTestBase.TearDownTest()`.
func (*FunctionalTestBase) WaitForChannel
deprecated
added in
v1.27.0
func (s *FunctionalTestBase) WaitForChannel(ctx context.Context, ch chan struct{})
Deprecated: use (*TestEnv).WaitForChannel instead.
func (*FunctionalTestBase) WorkerGRPCAddress ¶ added in v1.31.0
func (s *FunctionalTestBase) WorkerGRPCAddress() string
type GlobalMetricCapture ¶ added in v1.31.0
type GlobalMetricCapture struct {
// contains filtered or unexported fields
}
func (*GlobalMetricCapture) CollectMetric ¶ added in v1.31.0
func (c *GlobalMetricCapture) CollectMetric(name string, keep func(*metricstest.CapturedRecording) bool) []*metricstest.CapturedRecording
CollectMetric returns the recordings for the named metric that the caller chooses to keep.
func (*GlobalMetricCapture) Metric ¶ added in v1.31.0
func (c *GlobalMetricCapture) Metric(name string) []*metricstest.CapturedRecording
type HistoryConfig ¶
HistoryConfig contains configs for history service
type HistoryTaskRecorder ¶
type HistoryTaskRecorder struct {
// contains filtered or unexported fields
}
HistoryTaskRecorder wraps an ExecutionManager to record ALL task writes to the history task queues (transfer, timer, replication, visibility, archival, etc.). This is useful for integration tests where you want to assert on what tasks were generated and in what order. Tasks are stored flattened by category - all tasks of the same type are in a single list, with each task wrapped with metadata about when/where it was written.
func NewHistoryTaskRecorder ¶
func NewHistoryTaskRecorder(delegate persistence.ExecutionManager, logger log.Logger) *HistoryTaskRecorder
NewHistoryTaskRecorder creates a recorder that wraps the given ExecutionManager
func (*HistoryTaskRecorder) AddHistoryTasks ¶
func (r *HistoryTaskRecorder) AddHistoryTasks( ctx context.Context, request *persistence.AddHistoryTasksRequest, ) error
AddHistoryTasks records the task write and then delegates to the underlying manager
func (*HistoryTaskRecorder) AppendHistoryNodes ¶
func (r *HistoryTaskRecorder) AppendHistoryNodes( ctx context.Context, request *persistence.AppendHistoryNodesRequest, ) (*persistence.AppendHistoryNodesResponse, error)
func (*HistoryTaskRecorder) AppendRawHistoryNodes ¶
func (r *HistoryTaskRecorder) AppendRawHistoryNodes( ctx context.Context, request *persistence.AppendRawHistoryNodesRequest, ) (*persistence.AppendHistoryNodesResponse, error)
func (*HistoryTaskRecorder) Close ¶
func (r *HistoryTaskRecorder) Close()
func (*HistoryTaskRecorder) CompleteHistoryTask ¶
func (r *HistoryTaskRecorder) CompleteHistoryTask( ctx context.Context, request *persistence.CompleteHistoryTaskRequest, ) error
func (*HistoryTaskRecorder) ConflictResolveWorkflowExecution ¶
func (r *HistoryTaskRecorder) ConflictResolveWorkflowExecution( ctx context.Context, request *persistence.ConflictResolveWorkflowExecutionRequest, ) (*persistence.ConflictResolveWorkflowExecutionResponse, error)
func (*HistoryTaskRecorder) CountMatchingTasks ¶
func (r *HistoryTaskRecorder) CountMatchingTasks(category tasks.Category, matcher TaskMatcher) int
CountMatchingTasks returns the count of tasks in a category that match the given matcher
func (*HistoryTaskRecorder) CountTasksForNamespace ¶
func (r *HistoryTaskRecorder) CountTasksForNamespace( category tasks.Category, namespaceID string, matcher TaskMatcher, ) int
CountTasksForNamespace returns the count of tasks in a category for a specific namespace
func (*HistoryTaskRecorder) CountTasksForWorkflow ¶
func (r *HistoryTaskRecorder) CountTasksForWorkflow( category tasks.Category, namespaceID string, workflowID string, runID string, matcher TaskMatcher, ) int
CountTasksForWorkflow returns the count of tasks in a category for a specific workflow If namespaceID is empty, it matches any namespace If runID is empty, it matches any runID for the given workflowID
func (*HistoryTaskRecorder) CreateWorkflowExecution ¶
func (r *HistoryTaskRecorder) CreateWorkflowExecution( ctx context.Context, request *persistence.CreateWorkflowExecutionRequest, ) (*persistence.CreateWorkflowExecutionResponse, error)
func (*HistoryTaskRecorder) DeleteCurrentWorkflowExecution ¶
func (r *HistoryTaskRecorder) DeleteCurrentWorkflowExecution( ctx context.Context, request *persistence.DeleteCurrentWorkflowExecutionRequest, ) error
func (*HistoryTaskRecorder) DeleteHistoryBranch ¶
func (r *HistoryTaskRecorder) DeleteHistoryBranch( ctx context.Context, request *persistence.DeleteHistoryBranchRequest, ) error
func (*HistoryTaskRecorder) DeleteReplicationTaskFromDLQ ¶
func (r *HistoryTaskRecorder) DeleteReplicationTaskFromDLQ( ctx context.Context, request *persistence.DeleteReplicationTaskFromDLQRequest, ) error
func (*HistoryTaskRecorder) DeleteWorkflowExecution ¶
func (r *HistoryTaskRecorder) DeleteWorkflowExecution( ctx context.Context, request *persistence.DeleteWorkflowExecutionRequest, ) error
func (*HistoryTaskRecorder) ForkHistoryBranch ¶
func (r *HistoryTaskRecorder) ForkHistoryBranch( ctx context.Context, request *persistence.ForkHistoryBranchRequest, ) (*persistence.ForkHistoryBranchResponse, error)
func (*HistoryTaskRecorder) GetAllHistoryTreeBranches ¶
func (r *HistoryTaskRecorder) GetAllHistoryTreeBranches( ctx context.Context, request *persistence.GetAllHistoryTreeBranchesRequest, ) (*persistence.GetAllHistoryTreeBranchesResponse, error)
func (*HistoryTaskRecorder) GetAllRecordedTasks ¶
func (r *HistoryTaskRecorder) GetAllRecordedTasks() map[tasks.Category][]RecordedTask
GetAllRecordedTasks returns all recorded tasks WITH metadata, grouped by category
func (*HistoryTaskRecorder) GetAllTasks ¶
func (r *HistoryTaskRecorder) GetAllTasks() map[tasks.Category][]tasks.Task
GetAllTasks returns all tasks grouped by category (unwrapped, without metadata)
func (*HistoryTaskRecorder) GetCurrentExecution ¶
func (r *HistoryTaskRecorder) GetCurrentExecution( ctx context.Context, request *persistence.GetCurrentExecutionRequest, ) (*persistence.GetCurrentExecutionResponse, error)
func (*HistoryTaskRecorder) GetHistoryBranchUtil ¶
func (r *HistoryTaskRecorder) GetHistoryBranchUtil() persistence.HistoryBranchUtil
func (*HistoryTaskRecorder) GetHistoryTasks ¶
func (r *HistoryTaskRecorder) GetHistoryTasks( ctx context.Context, request *persistence.GetHistoryTasksRequest, ) (*persistence.GetHistoryTasksResponse, error)
func (*HistoryTaskRecorder) GetName ¶
func (r *HistoryTaskRecorder) GetName() string
func (*HistoryTaskRecorder) GetRecordedTasksByCategoryFiltered ¶
func (r *HistoryTaskRecorder) GetRecordedTasksByCategoryFiltered(category tasks.Category, filter TaskFilter) []RecordedTask
GetRecordedTasksByCategoryFiltered returns recorded tasks WITH metadata for a specific category, filtered by namespace (required) and optionally by workflow ID and run ID. This is the preferred API for tests to ensure tasks are properly scoped.
func (*HistoryTaskRecorder) GetReplicationTasksFromDLQ ¶
func (r *HistoryTaskRecorder) GetReplicationTasksFromDLQ( ctx context.Context, request *persistence.GetReplicationTasksFromDLQRequest, ) (*persistence.GetHistoryTasksResponse, error)
func (*HistoryTaskRecorder) GetWorkflowExecution ¶
func (r *HistoryTaskRecorder) GetWorkflowExecution( ctx context.Context, request *persistence.GetWorkflowExecutionRequest, ) (*persistence.GetWorkflowExecutionResponse, error)
func (*HistoryTaskRecorder) IsReplicationDLQEmpty ¶
func (r *HistoryTaskRecorder) IsReplicationDLQEmpty( ctx context.Context, request *persistence.GetReplicationTasksFromDLQRequest, ) (bool, error)
func (*HistoryTaskRecorder) ListConcreteExecutions ¶
func (r *HistoryTaskRecorder) ListConcreteExecutions( ctx context.Context, request *persistence.ListConcreteExecutionsRequest, ) (*persistence.ListConcreteExecutionsResponse, error)
func (*HistoryTaskRecorder) MatchTasks ¶
func (r *HistoryTaskRecorder) MatchTasks(category tasks.Category, matcher TaskMatcher) []RecordedTask
MatchTasks returns all tasks in a category that match the given matcher function
func (*HistoryTaskRecorder) MatchTasksForNamespace ¶
func (r *HistoryTaskRecorder) MatchTasksForNamespace( category tasks.Category, namespaceID string, matcher TaskMatcher, ) []RecordedTask
MatchTasksForNamespace returns all tasks in a category for a specific namespace
func (*HistoryTaskRecorder) MatchTasksForWorkflow ¶
func (r *HistoryTaskRecorder) MatchTasksForWorkflow( category tasks.Category, namespaceID string, workflowID string, runID string, matcher TaskMatcher, ) []RecordedTask
MatchTasksForWorkflow returns all tasks in a category for a specific workflow If namespaceID is empty, it matches any namespace If runID is empty, it matches any runID for the given workflowID
func (*HistoryTaskRecorder) PutReplicationTaskToDLQ ¶
func (r *HistoryTaskRecorder) PutReplicationTaskToDLQ( ctx context.Context, request *persistence.PutReplicationTaskToDLQRequest, ) error
func (*HistoryTaskRecorder) RangeCompleteHistoryTasks ¶
func (r *HistoryTaskRecorder) RangeCompleteHistoryTasks( ctx context.Context, request *persistence.RangeCompleteHistoryTasksRequest, ) error
func (*HistoryTaskRecorder) RangeDeleteReplicationTaskFromDLQ ¶
func (r *HistoryTaskRecorder) RangeDeleteReplicationTaskFromDLQ( ctx context.Context, request *persistence.RangeDeleteReplicationTaskFromDLQRequest, ) error
func (*HistoryTaskRecorder) ReadHistoryBranch ¶
func (r *HistoryTaskRecorder) ReadHistoryBranch( ctx context.Context, request *persistence.ReadHistoryBranchRequest, ) (*persistence.ReadHistoryBranchResponse, error)
func (*HistoryTaskRecorder) ReadHistoryBranchByBatch ¶
func (r *HistoryTaskRecorder) ReadHistoryBranchByBatch( ctx context.Context, request *persistence.ReadHistoryBranchRequest, ) (*persistence.ReadHistoryBranchByBatchResponse, error)
func (*HistoryTaskRecorder) ReadHistoryBranchReverse ¶
func (r *HistoryTaskRecorder) ReadHistoryBranchReverse( ctx context.Context, request *persistence.ReadHistoryBranchReverseRequest, ) (*persistence.ReadHistoryBranchReverseResponse, error)
func (*HistoryTaskRecorder) ReadRawHistoryBranch ¶
func (r *HistoryTaskRecorder) ReadRawHistoryBranch( ctx context.Context, request *persistence.ReadHistoryBranchRequest, ) (*persistence.ReadRawHistoryBranchResponse, error)
func (*HistoryTaskRecorder) SetWorkflowExecution ¶
func (r *HistoryTaskRecorder) SetWorkflowExecution( ctx context.Context, request *persistence.SetWorkflowExecutionRequest, ) (*persistence.SetWorkflowExecutionResponse, error)
func (*HistoryTaskRecorder) TrimHistoryBranch ¶
func (r *HistoryTaskRecorder) TrimHistoryBranch( ctx context.Context, request *persistence.TrimHistoryBranchRequest, ) (*persistence.TrimHistoryBranchResponse, error)
func (*HistoryTaskRecorder) UpdateWorkflowExecution ¶
func (r *HistoryTaskRecorder) UpdateWorkflowExecution( ctx context.Context, request *persistence.UpdateWorkflowExecutionRequest, ) (*persistence.UpdateWorkflowExecutionResponse, error)
func (*HistoryTaskRecorder) WriteToLog ¶
func (r *HistoryTaskRecorder) WriteToLog(filePath string) error
WriteToLog writes all captured tasks to a file in JSON format
type MatchingBehavior ¶ added in v1.31.0
MatchingBehavior describes a test scenario for matching service behavior.
func AllMatchingBehaviors ¶ added in v1.31.0
func AllMatchingBehaviors() []MatchingBehavior
AllMatchingBehaviors returns all 8 combinations of matching behaviors for testing.
func (MatchingBehavior) InjectHooks ¶ added in v1.31.0
func (b MatchingBehavior) InjectHooks(env hookInjector)
InjectHooks injects the test hooks for this matching behavior.
func (MatchingBehavior) Name ¶ added in v1.31.0
func (b MatchingBehavior) Name() string
Name returns a descriptive name for this behavior combination.
func (MatchingBehavior) Options ¶ added in v1.31.0
func (b MatchingBehavior) Options() []TestOption
Options returns the TestOptions to configure matching behavior.
type MatchingConfig ¶
type MatchingConfig struct {
NumMatchingHosts int
}
MatchingConfig is the config for the matching service
type MessageHandler ¶
type MessageHandler func(task *workflowservice.PollWorkflowTaskQueueResponse) ([]*protocolpb.Message, error)
type NamespaceMetricCapture ¶ added in v1.31.0
type NamespaceMetricCapture struct {
// contains filtered or unexported fields
}
func (*NamespaceMetricCapture) CollectMetric ¶ added in v1.31.0
func (c *NamespaceMetricCapture) CollectMetric(name string, keep func(*metricstest.CapturedRecording) bool) []*metricstest.CapturedRecording
CollectMetric returns the recordings for the named metric that belong to the test namespace and that the caller chooses to keep. It panics if the requested metric is not namespace-scoped.
func (*NamespaceMetricCapture) Metric ¶ added in v1.31.0
func (c *NamespaceMetricCapture) Metric(name string) []*metricstest.CapturedRecording
type PollAndProcessWorkflowTaskOptionFunc ¶
type PollAndProcessWorkflowTaskOptionFunc func(*PollAndProcessWorkflowTaskOptions)
func WithExpectedAttemptCount ¶
func WithExpectedAttemptCount(c int) PollAndProcessWorkflowTaskOptionFunc
func WithRetries ¶
func WithRetries(c int) PollAndProcessWorkflowTaskOptionFunc
type PollAndProcessWorkflowTaskResponse ¶
type PollAndProcessWorkflowTaskResponse struct {
IsQueryTask bool
NewTask *workflowservice.RespondWorkflowTaskCompletedResponse
}
type QueryHandler ¶
type QueryHandler func(task *workflowservice.PollWorkflowTaskQueueResponse) (*commonpb.Payloads, error)
type RecordedTask ¶ added in v1.30.0
type RecordedTask struct {
Timestamp time.Time `json:"timestamp"`
TaskType string `json:"taskType"` // The specific task type (e.g., "TASK_TYPE_ACTIVITY_RETRY_TIMER")
ShardID int32 `json:"shardId"`
RangeID int64 `json:"rangeId,omitempty"`
NamespaceID string `json:"namespaceId"`
WorkflowID string `json:"workflowId"`
RunID string `json:"runId"`
Task tasks.Task `json:"task"` // The actual task object
}
RecordedTask wraps a task with metadata about when and where it was written
type ReplicationStreamRecorder ¶ added in v1.30.0
type ReplicationStreamRecorder struct {
// contains filtered or unexported fields
}
ReplicationStreamRecorder captures replication stream messages for testing
func NewReplicationStreamRecorder ¶ added in v1.30.0
func NewReplicationStreamRecorder() *ReplicationStreamRecorder
func (*ReplicationStreamRecorder) Clear ¶ added in v1.30.0
func (r *ReplicationStreamRecorder) Clear()
func (*ReplicationStreamRecorder) GetMessages ¶ added in v1.30.0
func (r *ReplicationStreamRecorder) GetMessages() []CapturedReplicationMessage
func (*ReplicationStreamRecorder) SetOutputFile ¶ added in v1.30.0
func (r *ReplicationStreamRecorder) SetOutputFile(filePath string)
SetOutputFile sets the file path for writing captured messages on-demand
func (*ReplicationStreamRecorder) StreamInterceptor ¶ added in v1.30.0
func (r *ReplicationStreamRecorder) StreamInterceptor(clusterName string) grpc.StreamClientInterceptor
StreamInterceptor returns a gRPC stream client interceptor that captures stream messages
func (*ReplicationStreamRecorder) StreamServerInterceptor ¶ added in v1.30.0
func (r *ReplicationStreamRecorder) StreamServerInterceptor(clusterName string) grpc.StreamServerInterceptor
StreamServerInterceptor returns a gRPC stream server interceptor that captures stream messages
func (*ReplicationStreamRecorder) UnaryInterceptor ¶ added in v1.30.0
func (r *ReplicationStreamRecorder) UnaryInterceptor(clusterName string) grpc.UnaryClientInterceptor
UnaryInterceptor returns a gRPC unary client interceptor that captures messages
func (*ReplicationStreamRecorder) UnaryServerInterceptor ¶ added in v1.30.0
func (r *ReplicationStreamRecorder) UnaryServerInterceptor(clusterName string) grpc.UnaryServerInterceptor
UnaryServerInterceptor returns a gRPC unary server interceptor that captures messages
func (*ReplicationStreamRecorder) WriteToLog ¶ added in v1.30.0
func (r *ReplicationStreamRecorder) WriteToLog() error
WriteToLog writes all captured messages to the configured output file
type TaskFilter ¶ added in v1.30.0
type TaskFilter struct {
NamespaceID string // Required: namespace ID to filter by
WorkflowID string // Optional: workflow ID to filter by (empty string means no filter)
RunID string // Optional: run ID to filter by (empty string means no filter)
}
TaskFilter specifies criteria for filtering recorded tasks
type TaskMatcher ¶ added in v1.30.0
type TaskMatcher func(RecordedTask) bool
TaskMatcher is a function that tests whether a RecordedTask matches some criteria
type TaskPoller
deprecated
type TaskPoller struct {
Client workflowservice.WorkflowServiceClient
Namespace string
TaskQueue *taskqueuepb.TaskQueue
StickyTaskQueue *taskqueuepb.TaskQueue
StickyScheduleToStartTimeout time.Duration
Identity string
WorkflowTaskHandler WorkflowTaskHandler
ActivityTaskHandler ActivityTaskHandler
QueryHandler QueryHandler
MessageHandler MessageHandler
Logger log.Logger
T *testing.T
}
Deprecated: TaskPoller is deprecated. Use taskpoller.TaskPoller instead. TaskPoller is used in functional tests to poll workflow or activity task queues.
func (*TaskPoller) HandlePartialWorkflowTask ¶
func (p *TaskPoller) HandlePartialWorkflowTask(response *workflowservice.PollWorkflowTaskQueueResponse, forceCreateNewWorkflowTask bool) (*workflowservice.RespondWorkflowTaskCompletedResponse, error)
HandlePartialWorkflowTask for workflow task
func (*TaskPoller) PollAndProcessActivityTask ¶
func (p *TaskPoller) PollAndProcessActivityTask(dropTask bool) error
PollAndProcessActivityTask for activity tasks
func (*TaskPoller) PollAndProcessActivityTaskWithID ¶
func (p *TaskPoller) PollAndProcessActivityTaskWithID(dropTask bool) error
PollAndProcessActivityTaskWithID is similar to PollAndProcessActivityTask but using RespondActivityTask...ByID
func (*TaskPoller) PollAndProcessWorkflowTask ¶
func (p *TaskPoller) PollAndProcessWorkflowTask(funcs ...PollAndProcessWorkflowTaskOptionFunc) (res PollAndProcessWorkflowTaskResponse, err error)
func (*TaskPoller) PollAndProcessWorkflowTaskWithOptions ¶
func (p *TaskPoller) PollAndProcessWorkflowTaskWithOptions(opts *PollAndProcessWorkflowTaskOptions) (res PollAndProcessWorkflowTaskResponse, err error)
type TestCluster ¶
type TestCluster struct {
// contains filtered or unexported fields
}
TestCluster is a testcore struct for functional tests
func (*TestCluster) AdminClient ¶
func (tc *TestCluster) AdminClient() adminservice.AdminServiceClient
func (*TestCluster) ClusterName ¶ added in v1.27.0
func (tc *TestCluster) ClusterName() string
func (*TestCluster) ExecutionManager ¶
func (tc *TestCluster) ExecutionManager() persistence.ExecutionManager
ExecutionManager returns an execution manager factory from the test cluster
func (*TestCluster) FrontendClient ¶
func (tc *TestCluster) FrontendClient() workflowservice.WorkflowServiceClient
func (*TestCluster) GetHistoryTaskRecorder ¶
func (tc *TestCluster) GetHistoryTaskRecorder() *HistoryTaskRecorder
func (*TestCluster) GetReplicationStreamRecorder ¶ added in v1.30.0
func (tc *TestCluster) GetReplicationStreamRecorder() *ReplicationStreamRecorder
func (*TestCluster) HistoryClient ¶
func (tc *TestCluster) HistoryClient() historyservice.HistoryServiceClient
HistoryClient returns a history client from the test cluster
func (*TestCluster) Host ¶
func (tc *TestCluster) Host() *temporalImpl
TODO (alex): expose only needed objects from TemporalImpl.
func (*TestCluster) InjectHook ¶
func (*TestCluster) MatchingClient ¶
func (tc *TestCluster) MatchingClient() matchingservice.MatchingServiceClient
MatchingClient returns a matching client from the test cluster
func (*TestCluster) OperatorClient ¶
func (tc *TestCluster) OperatorClient() operatorservice.OperatorServiceClient
func (*TestCluster) OverrideDynamicConfig ¶
func (tc *TestCluster) OverrideDynamicConfig(t *testing.T, key dynamicconfig.GenericSetting, value any) (cleanup func())
func (*TestCluster) SchedulerClient ¶ added in v1.31.0
func (tc *TestCluster) SchedulerClient() schedulerpb.SchedulerServiceClient
SchedulerClient returns a scheduler client from the test cluster
func (*TestCluster) TearDownCluster ¶
func (tc *TestCluster) TearDownCluster() error
TearDownCluster tears down the test cluster
func (*TestCluster) TestBase ¶
func (tc *TestCluster) TestBase() *persistencetests.TestBase
TODO (alex): remove this method. Replace usages with concrete methods.
func (*TestCluster) WorkerGRPCAddress ¶ added in v1.31.0
func (tc *TestCluster) WorkerGRPCAddress() string
type TestClusterConfig ¶
type TestClusterConfig struct {
IsMasterCluster bool
ClusterMetadata cluster.Config
Persistence persistencetests.TestBaseOptions
FrontendConfig FrontendConfig
HistoryConfig HistoryConfig
MatchingConfig MatchingConfig
WorkerConfig WorkerConfig
ESConfig *esclient.Config
MockAdminClient map[string]adminservice.AdminServiceClient
FaultInjection *config.FaultInjection
DCRedirectionPolicy config.DCRedirectionPolicy
DynamicConfigOverrides map[dynamicconfig.Key]any
EnableMTLS bool
EnableMetricsCapture bool
EnableHistoryTaskRecorder bool
EnableReplicationRecorder bool
EnableArchival bool
SpanExporters map[telemetry.SpanExporterType]sdktrace.SpanExporter
TokenProvider auth.TokenProvider
TLSConfigProvider *encryption.FixedTLSConfigProvider
AdditionalServerOptions []temporal.ServerOption
}
TestClusterConfig are config for a test cluster
type TestClusterFactory ¶
type TestClusterFactory interface {
NewCluster(t *testing.T, clusterConfig *TestClusterConfig, logger log.Logger) (*TestCluster, error)
}
func NewTestClusterFactory ¶
func NewTestClusterFactory() TestClusterFactory
type TestClusterOption ¶ added in v1.27.0
type TestClusterOption func(params *testClusterParams)
func WithClusterHistoryTaskRecorder ¶
func WithClusterHistoryTaskRecorder() TestClusterOption
func WithClusterLogger ¶
func WithClusterLogger(logger log.Logger) TestClusterOption
WithClusterLogger sets a custom logger for the test cluster, used instead of the default test logger. Useful for intercepting server log output.
func WithDCRedirectionPolicy ¶
func WithDCRedirectionPolicy(policy config.DCRedirectionPolicy) TestClusterOption
func WithDynamicConfigOverrides ¶ added in v1.27.0
func WithDynamicConfigOverrides(overrides map[dynamicconfig.Key]any) TestClusterOption
func WithFaultInjectionConfig ¶ added in v1.28.0
func WithFaultInjectionConfig(cfg *config.FaultInjection) TestClusterOption
func WithNumHistoryShards ¶ added in v1.28.0
func WithNumHistoryShards(n int32) TestClusterOption
func WithReplicationStreamRecorder ¶
func WithReplicationStreamRecorder() TestClusterOption
func WithSharedCluster ¶ added in v1.31.0
func WithSharedCluster() TestClusterOption
type TestDataConverter ¶
type TestDataConverter struct {
NumOfCallToPayloads int // for testing to know testDataConverter is called as expected
NumOfCallFromPayloads int
}
TestDataConverter implements encoded.DataConverter using gob
func (*TestDataConverter) FromPayload ¶
func (tdc *TestDataConverter) FromPayload(payload *commonpb.Payload, valuePtr any) error
func (*TestDataConverter) FromPayloads ¶
func (tdc *TestDataConverter) FromPayloads(payloads *commonpb.Payloads, valuePtrs ...any) error
func (*TestDataConverter) ToPayload ¶
func (tdc *TestDataConverter) ToPayload(value any) (*commonpb.Payload, error)
func (*TestDataConverter) ToPayloads ¶
func (tdc *TestDataConverter) ToPayloads(values ...any) (*commonpb.Payloads, error)
type TestEnv ¶ added in v1.31.0
type TestEnv struct {
*FunctionalTestBase
// Shadows FunctionalTestBase.Assertions with a per-test instance bound to
// this TestEnv's own *testing.T, avoiding data races when parallel tests
// share the same *FunctionalTestBase cluster.
// TODO: remove once all tests are migrated to TestEnv (and no longer use FunctionalTestBase directly).
*require.Assertions
Logger log.Logger
// contains filtered or unexported fields
}
func NewEnv ¶ added in v1.31.0
func NewEnv(t *testing.T, opts ...TestOption) *TestEnv
NewEnv creates a new test environment with access to a Temporal cluster.
func (*TestEnv) CloseShard ¶
CloseShard closes the shard that contains the given workflow. This is a cluster-global operation and cannot be called on shared clusters.
func (*TestEnv) Context
deprecated
added in
v1.31.0
Context returns the test-level timeout context with RPC version headers already included. This context will be canceled when the test timeout occurs. Use this directly for all RPC operations - no need to wrap with NewContext or add headers manually.
For custom timeouts, use:
ctx, cancel := context.WithTimeout(env.Context(), 10*time.Second) defer cancel()
Deprecated: use the suite's Context() method instead.
func (*TestEnv) Error
deprecated
added in
v1.31.0
func (*TestEnv) InjectHook ¶ added in v1.31.0
InjectHook sets a test hook inside the cluster.
It auto-detects the scope from the hook: - For namespace-scoped hooks: scopes it to the test's namespace - For global hooks: requires a dedicated cluster, except for suite-scoped legacy clusters.
func (*TestEnv) Namespace ¶ added in v1.31.0
Use test env-specific namespace here for test isolation.
func (*TestEnv) NamespaceID ¶ added in v1.31.0
func (*TestEnv) NoError
deprecated
added in
v1.31.0
func (*TestEnv) OverrideDynamicConfig ¶ added in v1.31.0
func (e *TestEnv) OverrideDynamicConfig(setting dynamicconfig.GenericSetting, value any) (cleanup func())
OverrideDynamicConfig overrides a dynamic config setting for the duration of this test. For settings that can be namespace-scoped, a namespace constraint is applied. All others cannot be applied to a shared cluster and require `WithDedicatedCluster`.
func (*TestEnv) SdkClient ¶ added in v1.31.0
SdkClient returns the SDK client. It is lazily initialized on the first call.
func (*TestEnv) SdkWorker ¶ added in v1.31.0
SdkWorker returns the SDK worker. It is lazily initialized on the first call.
func (*TestEnv) SendToChannel ¶
func (e *TestEnv) SendToChannel(ch chan<- struct{})
SendToChannel sends to ch using the TestEnv context.
func (*TestEnv) SetOnAuthorize ¶
func (e *TestEnv) SetOnAuthorize( fn func(context.Context, *authorization.Claims, *authorization.CallTarget) (authorization.Result, error), )
func (*TestEnv) SetOnGetClaims ¶
func (e *TestEnv) SetOnGetClaims(fn func(*authorization.AuthInfo) (*authorization.Claims, error))
func (*TestEnv) StartGlobalMetricCapture ¶ added in v1.31.0
func (e *TestEnv) StartGlobalMetricCapture() *GlobalMetricCapture
StartGlobalMetricCapture starts a cluster-global metrics capture for this test and automatically stops it during cleanup. Metric capture is cluster-global, so it is only safe on dedicated clusters. Misuse detection is best-effort and only applies to queried metrics that produced recordings.
func (*TestEnv) StartNamespaceMetricCapture ¶ added in v1.31.0
func (e *TestEnv) StartNamespaceMetricCapture() *NamespaceMetricCapture
StartNamespaceMetricCapture starts a metrics capture scoped to this test's namespace. Namespace captures are safe on shared clusters because reads are restricted to per-metric namespace-filtered iteration and reject non-namespaced metrics.
func (*TestEnv) StartNamespaceMetricCaptureFor ¶
func (e *TestEnv) StartNamespaceMetricCaptureFor(namespaceName string) *NamespaceMetricCapture
StartNamespaceMetricCaptureFor starts a metrics capture scoped to the provided namespace.
func (*TestEnv) TaskPoller ¶ added in v1.31.0
func (e *TestEnv) TaskPoller() *taskpoller.TaskPoller
func (*TestEnv) WaitForChannel ¶
func (e *TestEnv) WaitForChannel(ch <-chan struct{})
WaitForChannel waits for ch to receive using the TestEnv context.
func (*TestEnv) WorkerTaskQueue ¶ added in v1.31.0
WorkerTaskQueue returns the task queue name used by the SDK Worker.
type TestOption ¶ added in v1.31.0
type TestOption func(*testOptions)
func WithArchival ¶
func WithArchival() TestOption
WithArchival enables archival on the test's cluster. This implies a dedicated cluster because archival is configured at the cluster level.
func WithCustomArchivers ¶
func WithCustomArchivers(historyFactory provider.CustomHistoryArchiverFactory, visibilityFactory provider.CustomVisibilityArchiverFactory) TestOption
WithCustomArchivers configures custom history and visibility archiver factories on the test's cluster. This implies a dedicated cluster because the factories are configured at the cluster level.
func WithDedicatedCluster ¶ added in v1.31.0
func WithDedicatedCluster() TestOption
WithDedicatedCluster requests a dedicated (non-shared) cluster for the test. Use this for tests that have cluster-global side effects.
func WithDisableTestloggerFailure ¶
func WithDisableTestloggerFailure() TestOption
WithDisableTestloggerFailure disables the test logger's behavior of failing the test when an error log matches a registered expectation (e.g. soft-assert errors tagged with tag.FailedAssertion). Use for tests that intentionally trigger and then verify soft-assert errors. Implies WithDedicatedCluster, because FailOnError is cluster-wide and disabling it on a shared cluster may hide failures in concurrent tests.
func WithDynamicConfig ¶ added in v1.31.0
func WithDynamicConfig(setting dynamicconfig.GenericSetting, value any) TestOption
WithDynamicConfig overrides a dynamic config setting for the test. For settings that can be namespace-scoped, a namespace constraint is applied. For all others that require a dedicated cluster, this implies `WithDedicatedCluster`.
func WithHistoryShardCount ¶
func WithHistoryShardCount(n int32) TestOption
WithHistoryShardCount sets the number of history shards for the test's cluster. This implies a dedicated cluster, since shard count cannot be changed on a shared cluster.
func WithHistoryTaskRecorder ¶
func WithHistoryTaskRecorder() TestOption
func WithLogger ¶
func WithLogger(logger log.Logger) TestOption
WithLogger sets a custom logger for the test's cluster, letting a test intercept server log output. This implies a dedicated cluster, since a custom logger cannot be shared across tests.
func WithMTLS ¶ added in v1.28.0
func WithMTLS() TestOption
WithMTLS enables mutual TLS on the test's cluster. This implies a dedicated cluster, since the TLS configuration cannot be shared across tests.
func WithPersistenceFaultInjection ¶
func WithPersistenceFaultInjection(cfg *config.FaultInjection) TestOption
WithPersistenceFaultInjection requests a dedicated cluster with the given persistence fault injection config.
func WithSdkWorker
deprecated
added in
v1.31.0
func WithSdkWorker() TestOption
Deprecated: this option is no longer required and will be removed once all callers have been updated.
func WithTestVars ¶
func WithTestVars(fn func(*testvars.TestVars) *testvars.TestVars) TestOption
WithTestVars customizes the default test variables for the environment.
func WithWorkerService ¶
func WithWorkerService(reason string) TestOption
WithWorkerService enables the system worker service. The service is off by default to avoid the worker overhead. This implies a dedicated cluster.
type WorkerConfig ¶
WorkerConfig is the config for the worker service
type WorkflowTaskHandler ¶
type WorkflowTaskHandler func(task *workflowservice.PollWorkflowTaskQueueResponse) ([]*commandpb.Command, error)