xtesting

func AppendToEventStream ¶

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

AppendToEventStream appends events to an existing event stream.

func CreateEventStreams ¶

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

CreateEventStreams creates the given number of event streams.

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,
	engine *dogmaengine.Engine,
	command dogma.Command,
	options ...dogma.ExecuteCommandOption,
) *envelopepb.Envelope

ExecuteCommand executes the given command on the engine, and fails the test if it returns an error.

func ExecuteCommandAndWait ¶

func ExecuteCommandAndWait(
	t testing.TB,
	engine *dogmaengine.Engine,
	command dogma.Command,
	options ...dogma.ExecuteCommandOption,
) *envelopepb.Envelope

ExecuteCommandAndWait executes the given command on the engine and waits for it to be removed from the command queue.

func ExecuteCommandWithHook ¶

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

ExecuteCommandWithHook executes the given command on the engine, and fails the test if it returns an error.

func ExecuteCommandsSequentially ¶

func ExecuteCommandsSequentially(
	t testing.TB,
	engine *dogmaengine.Engine,
	commands ...dogma.Command,
) []*envelopepb.Envelope

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

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 ExpectEqualEvents ¶

func ExpectEqualEvents(
	t testing.TB,
	description string,
	got []dogma.Event,
	want ...dogma.Event,
)

ExpectEqualEvents asserts that got and want contain the same events.

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 ExpectLatchesSetEventually ¶

func ExpectLatchesSetEventually(
	t testing.TB,
	latches ...*xsync.Latch,
)

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

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 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 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,
		*dogmaengine.Engine,
	),
	routes ...dogma.HandlerRoute,
)

RunEngines runs the given Dogma application in a test engine and executes the given function while the engine is running.

Multiple engines are run concurrently to ensure that behavior is consistent when multiple engines running against the same database.

If any of the engines 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, *dogmaengine.Engine),
	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, *dogmaengine.Engine),
	routes ...dogma.HandlerRoute,
)

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

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 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 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 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 and fails the test if it does not produce a row with the wanted value in the first column within WaitTimeout.

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.