conformance

Documentation

Overview

Package conformance provides an end-to-end conformance suite that tests the full event sourcing pipeline:

Command → CommandBus → Decider → Store → Subscription → Projection → Query

This catches bugs that live at the seams between components — a store can pass its own conformance suite but break when the subscription's global sequence doesn't align, or when snapshot versions cause off-by-one errors.

Usage:

func TestE2E_MemoryStore(t *testing.T) {
    conformance.RunSuite(t, conformance.MemoryFactory())
}

Index

• func RunSuite(t *testing.T, factory Factory)
• type AccountCommand
• type AccountEvent
• type AccountOpened
• type AccountState
• type BalanceProjection
  • func NewBalanceProjection() *BalanceProjection
  • func (p *BalanceProjection) Balance(id string) (int, bool)
  • func (p *BalanceProjection) EventCount() int64
  • func (p *BalanceProjection) Handle(_ context.Context, evt subscription.GlobalEvent[AccountEvent]) error
• type Deposit
  • func (c Deposit) StreamID() string
• type Deposited
• type Factory
  • func MemoryFactory() Factory
• type Infra
• type NotifyingStore
  • func NewNotifyingStore[E any](store eskit.EventStore[E], notifier *subscription.ChannelNotifier, ...) *NotifyingStore[E]
  • func (s *NotifyingStore[E]) Append(ctx context.Context, streamID string, expectedVersion int, events []E, ...) ([]eskit.Event[E], error)
• type OpenAccount
  • func (c OpenAccount) StreamID() string
• type Withdraw
  • func (c Withdraw) StreamID() string
• type Withdrawn

Functions

func RunSuite

func RunSuite(t *testing.T, factory Factory)

RunSuite runs the full end-to-end conformance suite.

Types

type AccountCommand

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

type AccountEvent

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

type AccountOpened

type AccountOpened struct{ Balance int }

type AccountState

type AccountState struct {
	Exists  bool
	Balance int
}

type BalanceProjection

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

func NewBalanceProjection

func NewBalanceProjection() *BalanceProjection

func (*BalanceProjection) Balance

func (p *BalanceProjection) Balance(id string) (int, bool)

func (*BalanceProjection) EventCount

func (p *BalanceProjection) EventCount() int64

func (*BalanceProjection) Handle

func (p *BalanceProjection) Handle(_ context.Context, evt subscription.GlobalEvent[AccountEvent]) error

type Deposit

type Deposit struct {
	ID     string
	Amount int
}

func (Deposit) StreamID

func (c Deposit) StreamID() string

type Deposited

type Deposited struct{ Amount int }

type Factory

type Factory func(t *testing.T) *Infra

Factory creates fresh infrastructure for each test.

func MemoryFactory

func MemoryFactory() Factory

type Infra

type Infra struct {
	Store         eskit.EventStore[AccountEvent]
	SnapshotStore eskit.SnapshotStore[AccountState]
	GlobalReader  subscription.GlobalReader[AccountEvent]
	Notifier      *subscription.ChannelNotifier
}

Infra holds the wired-up infrastructure for a single test run.

type NotifyingStore

type NotifyingStore[E any] struct {
	eskit.EventStore[E]
	// contains filtered or unexported fields
}

NotifyingStore wraps an EventStore to signal a notifier on append.

func NewNotifyingStore

func NewNotifyingStore[E any](
	store eskit.EventStore[E],
	notifier *subscription.ChannelNotifier,
	seqFn func(ctx context.Context) (uint64, error),
) *NotifyingStore[E]

func (*NotifyingStore[E]) Append

func (s *NotifyingStore[E]) Append(ctx context.Context, streamID string, expectedVersion int, events []E, metadata ...eskit.Metadata) ([]eskit.Event[E], error)

type OpenAccount

type OpenAccount struct {
	ID      string
	Balance int
}

func (OpenAccount) StreamID

func (c OpenAccount) StreamID() string

type Withdraw

type Withdraw struct {
	ID     string
	Amount int
}

func (Withdraw) StreamID

func (c Withdraw) StreamID() string

type Withdrawn

type Withdrawn struct{ Amount int }