xtesting

func AppendToEventStream ¶

func AppendToEventStream(
	t testing.TB,
	x xsql.Executor,
	streamID *uuidpb.UUID,
	events ...dogma.Event,
) []*envelopepb.Envelope

AppendToEventStream appends events to an existing event stream.

The returned envelopes have the envelopepb.EventStreamPosition extension set, as they would when read back from the stream.

func AppendToNewEventStream ¶ added in v0.2.0

func AppendToNewEventStream(
	t testing.TB,
	x xsql.Executor,
	events ...dogma.Event,
) (*uuidpb.UUID, []*envelopepb.Envelope)

AppendToNewEventStream creates a new event stream and appends the given events to it.

func ContextWithManualDeadline ¶ added in v0.2.0

func ContextWithManualDeadline(parent context.Context) (context.Context, func())

ContextWithManualDeadline returns a context that has a deadline, and a function that makes the context behave as though that deadline had just passed. The deadline is never reached otherwise.

It allows a test to verify the behavior of an operation when its deadline passes, without depending on the wall clock.

func CreateEventStream ¶ added in v0.2.0

func CreateEventStream(
	t testing.TB,
	x xsql.Executor,
) *uuidpb.UUID

CreateEventStream creates a new event stream and returns its ID.

func CreateEventStreams ¶

func CreateEventStreams(
	t testing.TB,
	x xsql.Executor,
	count int,
) []*uuidpb.UUID

CreateEventStreams creates the given number of event streams.

func DialGRPC ¶ added in v0.2.0

func DialGRPC(t testing.TB, engines ...*runkit.Engine) *grpc.ClientConn

DialGRPC returns an insecure gRPC client connection to the given engines, which must be listening for gRPC requests.

The connection sends requests to the first of the engines, and moves to the next, in the order given, if that engine stops. It is closed when the test ends.

func DialGRPCAddr ¶ added in v0.2.0

func DialGRPCAddr(
	t testing.TB,
	target string,
	options ...grpc.DialOption,
) *grpc.ClientConn

DialGRPCAddr returns an insecure gRPC client connection to the given target, which is usually the address of a gRPC server. The options are applied after the defaults, so they may override them.

It is closed when the test ends.

func EnqueuePostponedCommand ¶

func EnqueuePostponedCommand(
	t testing.TB,
	x xsql.Executor,
	command dogma.Command,
) *envelopepb.Envelope

EnqueuePostponedCommand inserts a command directly into the command queue with deliver_at set far in the future, so it will not be picked up by the dequeue loop. It returns the envelope of the enqueued command.

func ExecOne ¶

func ExecOne(
	t testing.TB,
	x xsql.Executor,
	query string,
	args ...any,
)

ExecOne executes the given query and fails the test if it returns an error or does not affect exactly one row.

func ExecuteCommand ¶

func ExecuteCommand(
	t testing.TB,
	executor dogma.CommandExecutor,
	command dogma.Command,
	options ...dogma.ExecuteCommandOption,
) *envelopepb.Envelope

ExecuteCommand executes the given command using the executor, and fails the test if it returns an error.

func ExecuteCommandAndWait ¶

func ExecuteCommandAndWait(
	t testing.TB,
	executor dogma.CommandExecutor,
	db *sql.DB,
	command dogma.Command,
	options ...dogma.ExecuteCommandOption,
) *envelopepb.Envelope

ExecuteCommandAndWait executes the given command using the executor and waits for it to be removed from the command queue.

func ExecuteCommandWithHook ¶

func ExecuteCommandWithHook(
	t testing.TB,
	executor dogma.CommandExecutor,
	command dogma.Command,
	hook func(contexthook.ExecuteCommand),
	options ...dogma.ExecuteCommandOption,
) *envelopepb.Envelope

ExecuteCommandWithHook executes the given command using the executor, and fails the test if it returns an error.

func ExecuteCommandsSequentially ¶

func ExecuteCommandsSequentially(
	t testing.TB,
	executor dogma.CommandExecutor,
	db *sql.DB,
	commands ...dogma.Command,
) []*envelopepb.Envelope

ExecuteCommandsSequentially executes the given commands using the executor sequentially, and fails the test if any of them returns an error.

func Expect ¶ added in v0.2.0

func Expect[T any](
	t testing.TB,
	description string,
	got, want T,
)

Expect asserts that got and want represent the same value.

func ExpectCommandIDToBeQueued ¶

func ExpectCommandIDToBeQueued(
	t testing.TB,
	q xsql.Querier,
	messageID *uuidpb.UUID,
)

ExpectCommandIDToBeQueued asserts that the command with the given ID is present in the queue.

func ExpectCommandToBeQueued ¶

func ExpectCommandToBeQueued(
	t testing.TB,
	q xsql.Querier,
	want dogma.Command,
)

ExpectCommandToBeQueued asserts that the command queue contains the given command.

func ExpectCommandToBeUnattempted ¶

func ExpectCommandToBeUnattempted(
	t testing.TB,
	q xsql.Querier,
	messageID *uuidpb.UUID,
)

ExpectCommandToBeUnattempted asserts that the command with the given ID is present in the queue and has never been attempted (failures = 0).

func ExpectContiguousEvents ¶

func ExpectContiguousEvents(
	t testing.TB,
	q xsql.Querier,
	streamID *uuidpb.UUID,
	offset uint64,
	want ...dogma.Event,
)

ExpectContiguousEvents asserts that an event stream contains the given events, starting at the given offset, with no gaps.

func ExpectEventCount ¶

func ExpectEventCount(
	t testing.TB,
	q xsql.Querier,
	want int,
)

ExpectEventCount asserts that the total number of events in the database matches the given value.

func ExpectN ¶ added in v0.2.0

func ExpectN[T any](
	t testing.TB,
	description string,
	got []T,
	want ...T,
)

ExpectN asserts that got and want represent the same values.

func ExpectNoReceive ¶ added in v0.2.0

func ExpectNoReceive[T any](
	t testing.TB,
	description string,
	ch <-chan T,
)

ExpectNoReceive asserts that nothing is received on the channel within a short period.

func ExpectQueryResult ¶

func ExpectQueryResult[T comparable](
	t testing.TB,
	description string,
	want T,
	q xsql.Querier,
	query string,
	args ...any,
)

ExpectQueryResult executes a database query and fails the test if it does not produce a row with the wanted value in the first column.

func ExpectToReceiveN ¶ added in v0.2.0

func ExpectToReceiveN[T any](
	t testing.TB,
	description string,
	got <-chan T,
	want ...T,
)

ExpectToReceiveN asserts that a channel receives the values in the want slice, in order.

func Go ¶ added in v0.2.0

func Go(
	t testing.TB,
	fn func(ctx context.Context),
) func()

Go runs the given function in a goroutine until the test ends.

It returns a function that can be called to stop the goroutine early, which blocks until fn returns.

func GoE ¶ added in v0.2.0

func GoE(
	t testing.TB,
	fn func(ctx context.Context) error,
) func()

GoE runs the given function in a goroutine until the test ends.

If fn returns an error before the goroutine is stopped or the test ends, the test fails. An error returned while the test is ending is ignored, as fn may fail because something it depends on has already stopped.

It returns a function that can be called to stop the goroutine early, which blocks until fn returns.

func NewDatabase ¶

func NewDatabase(t testing.TB) *sql.DB

NewDatabase returns a database pool that connects to an isolated test database that has the engine's schema DDL applied.

The database is removed when the test ends.

func NewDatabaseWithFailingQueries ¶ added in v0.2.0

func NewDatabaseWithFailingQueries(
	t testing.TB,
	fail func(query string, args []any) bool,
) *sql.DB

NewDatabaseWithFailingQueries returns a database pool like NewDatabase, except that each query for which the given function returns true fails without being sent to the database.

The function is called with the SQL and arguments of every query, including those that apply the schema.

func NewDatabaseWithoutSchema ¶

func NewDatabaseWithoutSchema(t testing.TB) *sql.DB

NewDatabaseWithoutSchema returns a database pool that connects to an isolated test database that does not have the engine's schema DDL applied.

The database is removed when the test ends.

func PopulateEventStreams ¶

func PopulateEventStreams(
	t testing.TB,
	x xsql.Executor,
	fn func(streamID *uuidpb.UUID, offset uint64) dogma.Event,
	counts ...uint64,
) []*uuidpb.UUID

PopulateEventStreams creates event streams and populates them with events generated by the given function.

Each element of counts specifies the number of events to append to one stream, so the number of streams created equals len(counts). Events are distributed by cycling through the streams, appending one event at a time to each in turn.

It returns the stream IDs that were created.

func PostponeStreamConsumption ¶

func PostponeStreamConsumption(
	t testing.TB,
	x xsql.Executor,
	handlerKey string,
	streamID *uuidpb.UUID,
)

PostponeStreamConsumption inserts a handler_checkpoints row with resume_at set 24 hours in the future, preventing the handler from consuming the stream.

func RunEngines ¶

func RunEngines(
	t *testing.T,
	fn func(
		testing.TB,
		*Cluster,
		*sql.DB,
	),
	routes ...dogma.HandlerRoute,
)

RunEngines runs the given Dogma application on a cluster of nodes, and executes the given function while the cluster is running.

Multiple nodes are run to ensure that behavior is consistent when multiple engines run against the same database. fn receives the Cluster, which executes each command on a node chosen at random.

If any of the nodes stop before fn returns, the context passed to fn is canceled, and the test fails.

func RunEnginesWithDB ¶

func RunEnginesWithDB(
	t *testing.T,
	db *sql.DB,
	fn func(testing.TB, *Cluster, *sql.DB),
	routes ...dogma.HandlerRoute,
)

RunEnginesWithDB is like RunEngines but uses the given database instead of creating one.

Unlike RunEngines and SetupThenRunEngines, it does not call t.Parallel().

func SetupThenRunEngines ¶

func SetupThenRunEngines(
	t *testing.T,
	setup func(testing.TB, *sql.DB),
	fn func(testing.TB, *Cluster, *sql.DB),
	routes ...dogma.HandlerRoute,
)

SetupThenRunEngines is like RunEngines but accepts a setup function that receives the test database before any engines start.

func Transact ¶

func Transact(t testing.TB, db *sql.DB, fn func(*sql.Tx))

Transact executes the given function within a database transaction, and fails the test if it returns an error.

func WaitFor ¶ added in v0.2.0

func WaitFor(
	t testing.TB,
	description string,
	cond func() bool,
)

WaitFor polls cond until it returns true. It fails the test if that does not occur within WaitTimeout.

func WaitForCommandToBePostponed ¶

func WaitForCommandToBePostponed(
	t testing.TB,
	q xsql.Querier,
	messageID *uuidpb.UUID,
)

WaitForCommandToBePostponed waits until the command with the given ID has been postponed. If this does not occur within WaitTimeout, the test fails.

func WaitForCommandToBeRemovedFromQueue ¶

func WaitForCommandToBeRemovedFromQueue(
	t testing.TB,
	q xsql.Querier,
	messageID *uuidpb.UUID,
)

WaitForCommandToBeRemovedFromQueue waits until the command with the given ID is removed from the queue. If this does not occur within WaitTimeout, the test fails.

func WaitForEmptyCommandQueue ¶

func WaitForEmptyCommandQueue(
	t testing.TB,
	q xsql.Querier,
)

WaitForEmptyCommandQueue waits until the command queue is empty. If this does not occur within WaitTimeout, the test fails.

func WaitForHandlerToConsumeAllEvents ¶

func WaitForHandlerToConsumeAllEvents(
	t testing.TB,
	q xsql.Querier,
	handlerKey string,
)

WaitForHandlerToConsumeAllEvents waits until the handler with the given key has caught up to the tail of all non-empty event streams. If this does not occur within WaitTimeout, the test fails.

It assumes that all events on all streams are of types that the handler consumes. If that is not the case this assertion will hang until it times out.

func WaitForHandlerToPostponeConsumingStream ¶

func WaitForHandlerToPostponeConsumingStream(
	t testing.TB,
	q xsql.Querier,
	handlerKey string,
	streamID *uuidpb.UUID,
)

WaitForHandlerToPostponeConsumingStream waits until the handler with the given key has postponed consuming events from the given stream.

func WaitForLatches ¶ added in v0.2.0

func WaitForLatches(
	t testing.TB,
	latches ...*cosyne.Latch,
)

WaitForLatches waits for all of the given latches to be set. It fails the test if all latches are not set within WaitTimeout.

func WaitForNoPendingDeadlines ¶

func WaitForNoPendingDeadlines(
	t testing.TB,
	q xsql.Querier,
)

WaitForNoPendingDeadlines waits until there are no pending deadlines. If this does not occur within WaitTimeout, the test fails.

func WaitForObservable ¶ added in v0.2.0

func WaitForObservable[T any](
	t testing.TB,
	description string,
	o cosyne.ObservableR[T],
	ok func(T) bool,
)

WaitForObservable waits until the value of o satisfies ok. It fails the test if that does not occur within WaitTimeout, and reports the most recent value.

func WaitForProcessHandlerInitialization ¶

func WaitForProcessHandlerInitialization(
	t testing.TB,
	db *sql.DB,
	handlerKey string,
)

WaitForProcessHandlerInitialization waits until the given process message handler has been fully initialized.

func WaitForQueryResult ¶

func WaitForQueryResult[T comparable](
	t testing.TB,
	description string,
	want T,
	q xsql.Querier,
	query string,
	args ...any,
)

WaitForQueryResult executes a database query repeatedly until it produces a row with the wanted value in the first column. If this does not occur within WaitTimeout, the test fails.

An error from the query does not fail the test until the timeout, as it may be transient. If the timeout is reached, the most recent error is reported.

func WaitForStreamFailureCounter ¶

func WaitForStreamFailureCounter(
	t testing.TB,
	q xsql.Querier,
	handlerKey string,
	min uint64,
	streamIDs ...*uuidpb.UUID,
)

WaitForStreamFailureCounter waits until the handler with the given key has its failures counter reach (or exceed) the given value on each of the given streams.

func WaitForStreamFailureCounterToReset ¶

func WaitForStreamFailureCounterToReset(
	t testing.TB,
	q xsql.Querier,
	handlerKey string,
	streamIDs ...*uuidpb.UUID,
)

WaitForStreamFailureCounterToReset waits until the handler with the given key has its failures counter reset to zero for each of the given streams.