testkit

Documentation

Index

• func RequireStreamErrorMessage(t testing.TB, msg engram.StreamMessage)
• func RequireStreamMessage(t testing.TB, msg engram.StreamMessage)
• func RequireStructuredError(t testing.TB, errObj runsv1alpha1.StructuredError)
• func ValidateStreamErrorMessage(msg engram.StreamMessage) error
• func ValidateStreamMessage(msg engram.StreamMessage) error
• func ValidateStructuredError(errObj runsv1alpha1.StructuredError) error
• type BatchHarness
  • func (h BatchHarness[C, I]) Run(ctx context.Context) (*engram.Result, error)
• type StreamHarness
  • func (h StreamHarness[C]) Run(ctx context.Context) ([]engram.StreamMessage, error)

Functions

func RequireStreamErrorMessage

func RequireStreamErrorMessage(t testing.TB, msg engram.StreamMessage)

RequireStreamErrorMessage fails the test when the error envelope is invalid.

func RequireStreamMessage

func RequireStreamMessage(t testing.TB, msg engram.StreamMessage)

RequireStreamMessage fails the test when the message is invalid.

func RequireStructuredError

func RequireStructuredError(t testing.TB, errObj runsv1alpha1.StructuredError)

RequireStructuredError fails the test when the StructuredError payload is invalid.

func ValidateStreamErrorMessage

func ValidateStreamErrorMessage(msg engram.StreamMessage) error

ValidateStreamErrorMessage verifies the StructuredError envelope for error StreamMessages.

func ValidateStreamMessage

func ValidateStreamMessage(msg engram.StreamMessage) error

ValidateStreamMessage verifies the minimum validity requirements for streaming messages.

func ValidateStructuredError

func ValidateStructuredError(errObj runsv1alpha1.StructuredError) error

ValidateStructuredError checks that a StructuredError payload matches the v1 contract.

Types

type BatchHarness

type BatchHarness[C any, I any] struct {
	// Engram is the implementation under test.
	Engram engram.BatchEngram[C, I]
	// Config is passed to Engram.Init.
	Config C
	// Inputs is passed to Engram.Process.
	Inputs I
	// Secrets are expanded the same way as in SDK runtime before Init is called.
	Secrets map[string]string
	// StoryInfo seeds the synthetic execution context passed to Process.
	StoryInfo engram.StoryInfo
	// Logger overrides the default logger used in the synthetic execution context.
	Logger *slog.Logger
	// Tracer overrides the default no-op tracer used in the synthetic execution context.
	Tracer trace.Tracer
	// CELContext is injected into the synthetic execution context for template/CEL consumers.
	CELContext map[string]any
}

BatchHarness runs a BatchEngram with explicit config/inputs and a synthetic ExecutionContext. It is intended for unit tests that want to exercise Init + Process without a controller.

func (BatchHarness[C, I]) Run

func (h BatchHarness[C, I]) Run(ctx context.Context) (*engram.Result, error)

Run executes Init + Process and returns the resulting output.

type StreamHarness

type StreamHarness[C any] struct {
	// Engram is the implementation under test.
	Engram engram.StreamingEngram[C]
	// Config is passed to Engram.Init.
	Config C
	// Secrets are expanded the same way as in SDK runtime before Init is called.
	Secrets map[string]string
	// StoryInfo seeds the synthetic execution context used by Stream.
	StoryInfo engram.StoryInfo
	// Logger overrides the default logger used in the synthetic execution context.
	Logger *slog.Logger
	// Tracer overrides the default no-op tracer used in the synthetic execution context.
	Tracer trace.Tracer
	// CELContext is injected into the synthetic execution context for template/CEL consumers.
	CELContext map[string]any
	// Inputs are converted to InboundMessages and delivered to Stream in order.
	Inputs []engram.StreamMessage
	// OnInputProcessed runs when an input message calls Done through the SDK
	// receipt hook. Tests can use this to verify acknowledgement behavior.
	OnInputProcessed func(engram.StreamMessage)
}

StreamHarness runs a StreamingEngram with provided input messages and returns collected outputs. It mimics the SDK runtime by closing the output channel after Stream returns.

func (StreamHarness[C]) Run

func (h StreamHarness[C]) Run(ctx context.Context) ([]engram.StreamMessage, error)

Run executes Init + Stream and returns the collected output messages.