streamtest

package
v0.0.0-...-e968d04 Latest Latest
Warning

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

 Go to latest Published: Aug 9, 2026 License: Apache-2.0 Imports: 12 Imported by: 0

Documentation

Overview

Package streamtest provides utilities for testing sync event publishing. It allows declarative specification of expected events and automatic verification.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func GetReceived 

func GetReceived[T any](exp *Expectation[T]) []T

GetReceived returns the received events for a specific expectation. Useful for additional assertions after verification.

func MatchAny 

func MatchAny[T any]() func(T) bool

MatchAny returns a matcher that accepts any event.

Types

type CountConstraint 

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

CountConstraint specifies how many events are expected.

func Any 

func Any() CountConstraint

Any expects any number of events (including zero).

func AtLeast 

func AtLeast(n int) CountConstraint

AtLeast expects at least n events.

func AtMost 

func AtMost(n int) CountConstraint

AtMost expects at most n events.

func Between 

func Between(min, max int) CountConstraint

Between expects between min and max events (inclusive).

func Exactly 

func Exactly(n int) CountConstraint

Exactly expects exactly n events.

func (CountConstraint) String 

func (c CountConstraint) String() string

type EventType 

type EventType string

EventType represents the type of sync event (insert, update, delete). 

const (
	Insert EventType = "insert"
	Update EventType = "update"
	Delete EventType = "delete"
)

type Expectation 

type Expectation[T any] struct {
	// contains filtered or unexported fields
}

Expectation represents an expected event with optional matching criteria.

type ExpectationResult 

type ExpectationResult struct {
	Name     string
	Expected string
	Actual   int
	Passed   bool
	Details  string
}

ExpectationResult contains the result of checking an expectation.

type StreamSubscription 

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

StreamSubscription holds a subscription that will be collected.

type Verifier 

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

Verifier collects and verifies sync events against expectations.

func Expect 

func Expect[Topic any, Payload any](
	v *Verifier,
	name string,
	stream eventstream.SyncStreamer[Topic, Payload],
	eventType EventType,
	count CountConstraint,
	getType func(Payload) string,
	matcher func(Payload) bool,
) *Verifier

Expect adds an expectation for events from a specific stream. The getType function extracts the event type string (e.g., "insert", "update", "delete") from the payload. The matcher function determines if a specific event matches this expectation.

func New 

func New(t *testing.T) *Verifier

New creates a new Verifier for testing sync events.

func (*Verifier) ExpectCredential 

func (v *Verifier) ExpectCredential(
	stream eventstream.SyncStreamer[rcredential.CredentialTopic, rcredential.CredentialEvent],
	eventType EventType,
	count CountConstraint,
	matcher func(rcredential.CredentialEvent) bool,
) *Verifier

ExpectCredential adds an expectation for credential events.

func (*Verifier) ExpectCredentialAnthropic 

func (v *Verifier) ExpectCredentialAnthropic(
	stream eventstream.SyncStreamer[rcredential.CredentialAnthropicTopic, rcredential.CredentialAnthropicEvent],
	eventType EventType,
	count CountConstraint,
	matcher func(rcredential.CredentialAnthropicEvent) bool,
) *Verifier

ExpectCredentialAnthropic adds an expectation for credential Anthropic events.

func (*Verifier) ExpectCredentialAnthropicDelete 

func (v *Verifier) ExpectCredentialAnthropicDelete(
	stream eventstream.SyncStreamer[rcredential.CredentialAnthropicTopic, rcredential.CredentialAnthropicEvent],
	count CountConstraint,
	matcher func(rcredential.CredentialAnthropicEvent) bool,
) *Verifier

ExpectCredentialAnthropicDelete is a shorthand for ExpectCredentialAnthropic with Delete type.

func (*Verifier) ExpectCredentialAnthropicInsert 

func (v *Verifier) ExpectCredentialAnthropicInsert(
	stream eventstream.SyncStreamer[rcredential.CredentialAnthropicTopic, rcredential.CredentialAnthropicEvent],
	count CountConstraint,
	matcher func(rcredential.CredentialAnthropicEvent) bool,
) *Verifier

ExpectCredentialAnthropicInsert is a shorthand for ExpectCredentialAnthropic with Insert type.

func (*Verifier) ExpectCredentialAnthropicUpdate 

func (v *Verifier) ExpectCredentialAnthropicUpdate(
	stream eventstream.SyncStreamer[rcredential.CredentialAnthropicTopic, rcredential.CredentialAnthropicEvent],
	count CountConstraint,
	matcher func(rcredential.CredentialAnthropicEvent) bool,
) *Verifier

ExpectCredentialAnthropicUpdate is a shorthand for ExpectCredentialAnthropic with Update type.

func (*Verifier) ExpectCredentialDelete 

func (v *Verifier) ExpectCredentialDelete(
	stream eventstream.SyncStreamer[rcredential.CredentialTopic, rcredential.CredentialEvent],
	count CountConstraint,
	matcher func(rcredential.CredentialEvent) bool,
) *Verifier

ExpectCredentialDelete is a shorthand for ExpectCredential with Delete type.

func (*Verifier) ExpectCredentialGemini 

func (v *Verifier) ExpectCredentialGemini(
	stream eventstream.SyncStreamer[rcredential.CredentialGeminiTopic, rcredential.CredentialGeminiEvent],
	eventType EventType,
	count CountConstraint,
	matcher func(rcredential.CredentialGeminiEvent) bool,
) *Verifier

ExpectCredentialGemini adds an expectation for credential Gemini events.

func (*Verifier) ExpectCredentialGeminiDelete 

func (v *Verifier) ExpectCredentialGeminiDelete(
	stream eventstream.SyncStreamer[rcredential.CredentialGeminiTopic, rcredential.CredentialGeminiEvent],
	count CountConstraint,
	matcher func(rcredential.CredentialGeminiEvent) bool,
) *Verifier

ExpectCredentialGeminiDelete is a shorthand for ExpectCredentialGemini with Delete type.

func (*Verifier) ExpectCredentialGeminiInsert 

func (v *Verifier) ExpectCredentialGeminiInsert(
	stream eventstream.SyncStreamer[rcredential.CredentialGeminiTopic, rcredential.CredentialGeminiEvent],
	count CountConstraint,
	matcher func(rcredential.CredentialGeminiEvent) bool,
) *Verifier

ExpectCredentialGeminiInsert is a shorthand for ExpectCredentialGemini with Insert type.

func (*Verifier) ExpectCredentialGeminiUpdate 

func (v *Verifier) ExpectCredentialGeminiUpdate(
	stream eventstream.SyncStreamer[rcredential.CredentialGeminiTopic, rcredential.CredentialGeminiEvent],
	count CountConstraint,
	matcher func(rcredential.CredentialGeminiEvent) bool,
) *Verifier

ExpectCredentialGeminiUpdate is a shorthand for ExpectCredentialGemini with Update type.

func (*Verifier) ExpectCredentialInsert 

func (v *Verifier) ExpectCredentialInsert(
	stream eventstream.SyncStreamer[rcredential.CredentialTopic, rcredential.CredentialEvent],
	count CountConstraint,
	matcher func(rcredential.CredentialEvent) bool,
) *Verifier

ExpectCredentialInsert is a shorthand for ExpectCredential with Insert type.

func (*Verifier) ExpectCredentialOpenAi 

func (v *Verifier) ExpectCredentialOpenAi(
	stream eventstream.SyncStreamer[rcredential.CredentialOpenAiTopic, rcredential.CredentialOpenAiEvent],
	eventType EventType,
	count CountConstraint,
	matcher func(rcredential.CredentialOpenAiEvent) bool,
) *Verifier

ExpectCredentialOpenAi adds an expectation for credential OpenAI events.

func (*Verifier) ExpectCredentialOpenAiDelete 

func (v *Verifier) ExpectCredentialOpenAiDelete(
	stream eventstream.SyncStreamer[rcredential.CredentialOpenAiTopic, rcredential.CredentialOpenAiEvent],
	count CountConstraint,
	matcher func(rcredential.CredentialOpenAiEvent) bool,
) *Verifier

ExpectCredentialOpenAiDelete is a shorthand for ExpectCredentialOpenAi with Delete type.

func (*Verifier) ExpectCredentialOpenAiInsert 

func (v *Verifier) ExpectCredentialOpenAiInsert(
	stream eventstream.SyncStreamer[rcredential.CredentialOpenAiTopic, rcredential.CredentialOpenAiEvent],
	count CountConstraint,
	matcher func(rcredential.CredentialOpenAiEvent) bool,
) *Verifier

ExpectCredentialOpenAiInsert is a shorthand for ExpectCredentialOpenAi with Insert type.

func (*Verifier) ExpectCredentialOpenAiUpdate 

func (v *Verifier) ExpectCredentialOpenAiUpdate(
	stream eventstream.SyncStreamer[rcredential.CredentialOpenAiTopic, rcredential.CredentialOpenAiEvent],
	count CountConstraint,
	matcher func(rcredential.CredentialOpenAiEvent) bool,
) *Verifier

ExpectCredentialOpenAiUpdate is a shorthand for ExpectCredentialOpenAi with Update type.

func (*Verifier) ExpectCredentialUpdate 

func (v *Verifier) ExpectCredentialUpdate(
	stream eventstream.SyncStreamer[rcredential.CredentialTopic, rcredential.CredentialEvent],
	count CountConstraint,
	matcher func(rcredential.CredentialEvent) bool,
) *Verifier

ExpectCredentialUpdate is a shorthand for ExpectCredential with Update type.

func (*Verifier) ExpectEdge 

func (v *Verifier) ExpectEdge(
	stream eventstream.SyncStreamer[rflowv2.EdgeTopic, rflowv2.EdgeEvent],
	eventType EventType,
	count CountConstraint,
	matcher func(rflowv2.EdgeEvent) bool,
) *Verifier

ExpectEdge adds an expectation for edge events.

func (*Verifier) ExpectEnv 

func (v *Verifier) ExpectEnv(
	stream eventstream.SyncStreamer[renv.EnvironmentTopic, renv.EnvironmentEvent],
	eventType EventType,
	count CountConstraint,
	matcher func(renv.EnvironmentEvent) bool,
) *Verifier

ExpectEnv adds an expectation for environment events.

func (*Verifier) ExpectEnvInsert 

func (v *Verifier) ExpectEnvInsert(
	stream eventstream.SyncStreamer[renv.EnvironmentTopic, renv.EnvironmentEvent],
	matcher func(renv.EnvironmentEvent) bool,
) *Verifier

ExpectEnvInsert is a shorthand for ExpectEnv with Insert type and Exactly(1).

func (*Verifier) ExpectEnvVar 

func (v *Verifier) ExpectEnvVar(
	stream eventstream.SyncStreamer[renv.EnvironmentVariableTopic, renv.EnvironmentVariableEvent],
	eventType EventType,
	count CountConstraint,
	matcher func(renv.EnvironmentVariableEvent) bool,
) *Verifier

ExpectEnvVar adds an expectation for environment variable events.

func (*Verifier) ExpectEnvVarInsert 

func (v *Verifier) ExpectEnvVarInsert(
	stream eventstream.SyncStreamer[renv.EnvironmentVariableTopic, renv.EnvironmentVariableEvent],
	count CountConstraint,
	matcher func(renv.EnvironmentVariableEvent) bool,
) *Verifier

ExpectEnvVarInsert is a shorthand for ExpectEnvVar with Insert type.

func (*Verifier) ExpectEnvVarUpdate 

func (v *Verifier) ExpectEnvVarUpdate(
	stream eventstream.SyncStreamer[renv.EnvironmentVariableTopic, renv.EnvironmentVariableEvent],
	count CountConstraint,
	matcher func(renv.EnvironmentVariableEvent) bool,
) *Verifier

ExpectEnvVarUpdate is a shorthand for ExpectEnvVar with Update type.

func (*Verifier) ExpectFile 

func (v *Verifier) ExpectFile(
	stream eventstream.SyncStreamer[rfile.FileTopic, rfile.FileEvent],
	eventType EventType,
	count CountConstraint,
	matcher func(rfile.FileEvent) bool,
) *Verifier

ExpectFile adds an expectation for file events.

func (*Verifier) ExpectFlow 

func (v *Verifier) ExpectFlow(
	stream eventstream.SyncStreamer[rflowv2.FlowTopic, rflowv2.FlowEvent],
	eventType EventType,
	count CountConstraint,
	matcher func(rflowv2.FlowEvent) bool,
) *Verifier

ExpectFlow adds an expectation for flow events.

func (*Verifier) ExpectFlowInsert 

func (v *Verifier) ExpectFlowInsert(
	stream eventstream.SyncStreamer[rflowv2.FlowTopic, rflowv2.FlowEvent],
	matcher func(rflowv2.FlowEvent) bool,
) *Verifier

ExpectFlowInsert is a shorthand for ExpectFlow with Insert type and Exactly(1).

func (*Verifier) ExpectHttp 

func (v *Verifier) ExpectHttp(
	stream eventstream.SyncStreamer[rhttp.HttpTopic, rhttp.HttpEvent],
	eventType EventType,
	count CountConstraint,
	matcher func(rhttp.HttpEvent) bool,
) *Verifier

ExpectHttp adds an expectation for HTTP events.

func (*Verifier) ExpectHttpAssert 

func (v *Verifier) ExpectHttpAssert(
	stream eventstream.SyncStreamer[rhttp.HttpAssertTopic, rhttp.HttpAssertEvent],
	eventType EventType,
	count CountConstraint,
	matcher func(rhttp.HttpAssertEvent) bool,
) *Verifier

ExpectHttpAssert adds an expectation for HTTP assert events.

func (*Verifier) ExpectHttpBodyForm 

func (v *Verifier) ExpectHttpBodyForm(
	stream eventstream.SyncStreamer[rhttp.HttpBodyFormTopic, rhttp.HttpBodyFormEvent],
	eventType EventType,
	count CountConstraint,
	matcher func(rhttp.HttpBodyFormEvent) bool,
) *Verifier

ExpectHttpBodyForm adds an expectation for HTTP body form events.

func (*Verifier) ExpectHttpBodyRaw 

func (v *Verifier) ExpectHttpBodyRaw(
	stream eventstream.SyncStreamer[rhttp.HttpBodyRawTopic, rhttp.HttpBodyRawEvent],
	eventType EventType,
	count CountConstraint,
	matcher func(rhttp.HttpBodyRawEvent) bool,
) *Verifier

ExpectHttpBodyRaw adds an expectation for HTTP body raw events.

func (*Verifier) ExpectHttpBodyUrlEncoded 

func (v *Verifier) ExpectHttpBodyUrlEncoded(
	stream eventstream.SyncStreamer[rhttp.HttpBodyUrlEncodedTopic, rhttp.HttpBodyUrlEncodedEvent],
	eventType EventType,
	count CountConstraint,
	matcher func(rhttp.HttpBodyUrlEncodedEvent) bool,
) *Verifier

ExpectHttpBodyUrlEncoded adds an expectation for HTTP body URL-encoded events.

func (*Verifier) ExpectHttpHeader 

func (v *Verifier) ExpectHttpHeader(
	stream eventstream.SyncStreamer[rhttp.HttpHeaderTopic, rhttp.HttpHeaderEvent],
	eventType EventType,
	count CountConstraint,
	matcher func(rhttp.HttpHeaderEvent) bool,
) *Verifier

ExpectHttpHeader adds an expectation for HTTP header events.

func (*Verifier) ExpectHttpInsert 

func (v *Verifier) ExpectHttpInsert(
	stream eventstream.SyncStreamer[rhttp.HttpTopic, rhttp.HttpEvent],
	count CountConstraint,
	matcher func(rhttp.HttpEvent) bool,
) *Verifier

ExpectHttpInsert is a shorthand for ExpectHttp with Insert type.

func (*Verifier) ExpectHttpSearchParam 

func (v *Verifier) ExpectHttpSearchParam(
	stream eventstream.SyncStreamer[rhttp.HttpSearchParamTopic, rhttp.HttpSearchParamEvent],
	eventType EventType,
	count CountConstraint,
	matcher func(rhttp.HttpSearchParamEvent) bool,
) *Verifier

ExpectHttpSearchParam adds an expectation for HTTP search param events.

func (*Verifier) ExpectNode 

func (v *Verifier) ExpectNode(
	stream eventstream.SyncStreamer[rflowv2.NodeTopic, rflowv2.NodeEvent],
	eventType EventType,
	count CountConstraint,
	matcher func(rflowv2.NodeEvent) bool,
) *Verifier

ExpectNode adds an expectation for node events.

func (*Verifier) ExpectNodeInsert 

func (v *Verifier) ExpectNodeInsert(
	stream eventstream.SyncStreamer[rflowv2.NodeTopic, rflowv2.NodeEvent],
	count CountConstraint,
	matcher func(rflowv2.NodeEvent) bool,
) *Verifier

ExpectNodeInsert is a shorthand for ExpectNode with Insert type.

func (*Verifier) Verify 

func (v *Verifier) Verify() bool

Verify immediately checks all expectations without waiting.

func (*Verifier) WaitAndVerify 

func (v *Verifier) WaitAndVerify(timeout time.Duration) bool

WaitAndVerify waits for the specified duration, then verifies all expectations. Returns true if all expectations passed.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.