testcore

package
v1.32.0 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Sep 11, 2026 License: MIT Imports: 114 Imported by: 1

Documentation

Index

Constants

View Source
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
)
View Source
const (
	DirectionSend       = "send"
	DirectionRecv       = "recv"
	DirectionServerSend = "server_send"
	DirectionServerRecv = "server_recv"
)

Message direction constants

View Source
const NamespaceCacheRefreshInterval = time.Second

Variables

View Source
var (
	ErrEncodingIsNotSet       = errors.New("payload encoding metadata is not set")
	ErrEncodingIsNotSupported = errors.New("payload encoding is not supported")
)
View Source
var (
	ErrNoTasks = errors.New("no tasks")
)

Functions

func ApplyTestClusterOptions added in v1.27.0

func ApplyTestClusterOptions(options []TestClusterOption) testClusterParams

func DecodeString

func DecodeString(t require.TestingT, pls *commonpb.Payloads) string

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

func MustToPayload(t require.TestingT, v any) *commonpb.Payload

MustToPayload converts a value to a Payload using the default data converter.

func NewContext

func NewContext(parent ...context.Context) context.Context

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

func PickRolloutSplit(t *testing.T, namespace string, percent int) (accepted, rejected string)

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 RandomizeStr(id string) string

func RandomizedNexusEndpoint

func RandomizedNexusEndpoint(name string) string

func UseCassandraPersistence added in v1.29.0

func UseCassandraPersistence() bool

func UseSQLVisibility added in v1.27.0

func UseSQLVisibility() bool

func UseSuiteScopedCluster deprecated

func UseSuiteScopedCluster(t *testing.T)

UseSuiteScopedCluster makes NewEnv use one cluster for all tests under `t`. The cluster is created on first use and torn down when `t` completes.

Deprecated: this only exists for backwards-compatibility with legacy sequential suite execution.

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 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 (*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 (*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 (*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:

  1. The Retention period is set to 0 for archival tests, and this can't be done through FE,
  2. Update search attributes would require an extra API call,
  3. 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

CollectMetric returns the recordings for the named metric that the caller chooses to keep.

func (*GlobalMetricCapture) Metric added in v1.31.0

type HistoryConfig

type HistoryConfig struct {
	NumHistoryShards int32
	NumHistoryHosts  int
}

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 (*HistoryTaskRecorder) AppendRawHistoryNodes

func (*HistoryTaskRecorder) Close

func (r *HistoryTaskRecorder) Close()

func (*HistoryTaskRecorder) CompleteHistoryTask

func (r *HistoryTaskRecorder) CompleteHistoryTask(
	ctx context.Context,
	request *persistence.CompleteHistoryTaskRequest,
) 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) 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 (*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 (*HistoryTaskRecorder) GetHistoryBranchUtil

func (r *HistoryTaskRecorder) GetHistoryBranchUtil() persistence.HistoryBranchUtil

func (*HistoryTaskRecorder) GetHistoryTasks

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 (*HistoryTaskRecorder) GetWorkflowExecution

func (*HistoryTaskRecorder) IsReplicationDLQEmpty

func (r *HistoryTaskRecorder) IsReplicationDLQEmpty(
	ctx context.Context,
	request *persistence.GetReplicationTasksFromDLQRequest,
) (bool, 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 (*HistoryTaskRecorder) ReadHistoryBranchByBatch

func (*HistoryTaskRecorder) ReadRawHistoryBranch

func (*HistoryTaskRecorder) SetWorkflowExecution

func (*HistoryTaskRecorder) TrimHistoryBranch

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

type MatchingBehavior struct {
	ForceTaskForward bool
	ForcePollForward bool
	ForceAsync       bool
}

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 NamespaceMetricCapture added in v1.31.0

type NamespaceMetricCapture struct {
	// contains filtered or unexported fields
}

func (*NamespaceMetricCapture) CollectMetric added in v1.31.0

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

type PollAndProcessWorkflowTaskOptionFunc

type PollAndProcessWorkflowTaskOptionFunc func(*PollAndProcessWorkflowTaskOptions)

type PollAndProcessWorkflowTaskOptions

type PollAndProcessWorkflowTaskOptions struct {
	DumpHistory          bool
	DumpCommands         bool
	DropTask             bool
	PollSticky           bool
	RespondSticky        bool
	ExpectedAttemptCount int
	Retries              int
	ForceNewWorkflowTask bool
	QueryResult          *querypb.WorkflowQueryResult
}

type PollAndProcessWorkflowTaskResponse

type PollAndProcessWorkflowTaskResponse struct {
	IsQueryTask bool
	NewTask     *workflowservice.RespondWorkflowTaskCompletedResponse
}

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 (*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 (tc *TestCluster) InjectHook(t *testing.T, hook testhooks.Hook, scope any) func()

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)

func (*TestDataConverter) ToString

func (tdc *TestDataConverter) ToString(payload *commonpb.Payload) string

func (*TestDataConverter) ToStrings

func (tdc *TestDataConverter) ToStrings(payloads *commonpb.Payloads) []string

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

func (e *TestEnv) 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 (*TestEnv) Context deprecated added in v1.31.0

func (e *TestEnv) Context() context.Context

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 (e *TestEnv) Error(err error, msgAndArgs ...any)

Error asserts that err is not nil.

Deprecated: use require.Error with the parent test or suite instead. TODO: remove once all tests are migrated to TestEnv (and no longer use FunctionalTestBase directly).

func (*TestEnv) InjectHook added in v1.31.0

func (e *TestEnv) InjectHook(hook testhooks.Hook) (cleanup func())

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

func (e *TestEnv) Namespace() namespace.Name

Use test env-specific namespace here for test isolation.

func (*TestEnv) NamespaceID added in v1.31.0

func (e *TestEnv) NamespaceID() namespace.ID

func (*TestEnv) NoError deprecated added in v1.31.0

func (e *TestEnv) NoError(err error, msgAndArgs ...any)

NoError asserts that err is nil.

Deprecated: use require.NoError with the parent test or suite instead. TODO: remove once all tests are migrated to TestEnv (and no longer use FunctionalTestBase directly).

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) Run deprecated added in v1.31.0

func (e *TestEnv) Run(name string, subtest func()) bool

Run executes a subtest.

Deprecated: use the suite's Run method instead. TODO: remove once all tests are migrated to TestEnv (and no longer use FunctionalTestBase directly).

func (*TestEnv) SdkClient added in v1.31.0

func (e *TestEnv) SdkClient() sdkclient.Client

SdkClient returns the SDK client. It is lazily initialized on the first call.

func (*TestEnv) SdkWorker added in v1.31.0

func (e *TestEnv) SdkWorker() sdkworker.Worker

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 (*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) T deprecated added in v1.31.0

func (e *TestEnv) T() *testing.T

T returns the *testing.T.

Deprecated: use the suite's T() method instead.

func (*TestEnv) TaskPoller added in v1.31.0

func (e *TestEnv) TaskPoller() *taskpoller.TaskPoller

func (*TestEnv) Tv added in v1.31.0

func (e *TestEnv) Tv() *testvars.TestVars

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

func (e *TestEnv) WorkerTaskQueue() string

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

type WorkerConfig struct {
	NumWorkers    int
	DisableWorker bool // overrides NumWorkers
}

WorkerConfig is the config for the worker service

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL