flowtest

Documentation

Overview

Package flowtest provides database-free helpers backed by Flow's production codecs and deterministic engine primitives.

Index

• func RunCommit[A, R any](ctx context.Context, registration flow.Registration, tx flow.Tx, args A, ...) error
• type CanonicalValue
  • func AssertCanonicalStable(t TestingT, value any) CanonicalValue
  • func Canonical(value any) (CanonicalValue, error)
• type DirectResult
  • func RunDirect[A, R any](ctx context.Context, root flow.Registration, args A, maxCommands int, ...) (DirectResult[R], error)
• type EventWait
• type RetryClass
• type RetryDecision
  • func DecideRetry(policy flow.RetryPolicy, input RetryInput) (RetryDecision, error)
• type RetryInput
• type StagedCommand
• type StagedEvent
• type TestingT
• type WorkerOption
  • func WithCommandInfo(info flow.CommandInfo) WorkerOption
  • func WithEvent[T any](event flow.Event[T], key string, payload T) WorkerOption
• type WorkerResult
  • func RunWorker[A, R any](ctx context.Context, registration flow.Registration, args A, ...) (WorkerResult[R], error)

Functions

func RunCommit

func RunCommit[A, R any](ctx context.Context, registration flow.Registration, tx flow.Tx,
	args A, result R, info flow.CommandInfo) error

RunCommit invokes the production declared commit function with a caller's transaction double. A registration without a commit function is a no-op.

Types

type CanonicalValue

type CanonicalValue struct {
	Bytes     []byte
	DigestHex string
}

func AssertCanonicalStable

func AssertCanonicalStable(t TestingT, value any) CanonicalValue

func Canonical

func Canonical(value any) (CanonicalValue, error)

type DirectResult

type DirectResult[R any] struct {
	Result   R
	Commands map[string]json.RawMessage
	Events   []StagedEvent
}

func RunDirect

func RunDirect[A, R any](ctx context.Context, root flow.Registration, args A, maxCommands int,
	resolver func(name string, version int) (flow.Registration, bool)) (DirectResult[R], error)

RunDirect recursively executes a closed tree of successful worker decisions. Resolver supplies registrations for staged sub-command name/version.

type EventWait

type EventWait struct {
	Name string
	Key  string
}

type RetryClass

type RetryClass string

const (
	Retryable   RetryClass = "retryable"
	RetryAt     RetryClass = "retry_after"
	Permanent   RetryClass = "permanent"
	Panic       RetryClass = "panic"
	Timeout     RetryClass = "timeout"
	Interrupted RetryClass = "interrupted"
	LeaseLost   RetryClass = "lease_lost"
)

type RetryDecision

type RetryDecision struct {
	Retry            bool
	ConsumesAttempt  bool
	ConsumedAttempts int
	NextAttemptAt    time.Time
	StopReason       string
}

func DecideRetry

func DecideRetry(policy flow.RetryPolicy, input RetryInput) (RetryDecision, error)

type RetryInput

type RetryInput struct {
	DBNow             time.Time
	BudgetStartedAt   time.Time
	ConsumedAttempts  int
	AttemptID         flow.AttemptID
	Class             RetryClass
	ExplicitDelay     time.Duration
	ExecutionDeadline *time.Time
}

type StagedCommand

type StagedCommand struct {
	Key        string
	Name       string
	Version    int
	Args       json.RawMessage
	Required   bool
	StartAfter time.Duration
	Waits      []EventWait
	Within     time.Duration
}

type StagedEvent

type StagedEvent struct {
	Name    string
	Key     string
	Payload json.RawMessage
}

type TestingT

type TestingT interface {
	Helper()
	Fatalf(string, ...any)
}

type WorkerOption

type WorkerOption interface {
	// contains filtered or unexported methods
}

func WithCommandInfo

func WithCommandInfo(info flow.CommandInfo) WorkerOption

func WithEvent

func WithEvent[T any](event flow.Event[T], key string, payload T) WorkerOption

WithEvent supplies one exact declared event input to a database-free worker decision. It uses Flow's production canonical payload encoding.

type WorkerResult

type WorkerResult[R any] struct {
	Result   R
	Err      error
	Panicked bool
	Commands []StagedCommand
	Events   []StagedEvent
}

func RunWorker

func RunWorker[A, R any](ctx context.Context, registration flow.Registration, args A, opts ...WorkerOption) (WorkerResult[R], error)

RunWorker invokes the registered production worker and production staged decision recorder without PostgreSQL.