Testing

Tests are first-class citizens in Emmett. Once your business logic is a set of functions returning events, testing turns into a repeatable pattern you can apply from a single decision up to the whole running API. This guide shows you how, at three levels: the business logic on its own, the HTTP API against an in-memory store, and the whole slice end-to-end against a real database. Read models get their own section too.

No matter which level you pick, the shape stays the same:

TIP

• GIVEN the events already recorded,
• WHEN you run a command or a request,
• THEN you assert the new events, or the error.

The helpers change; the pattern doesn't. For the thinking behind it, read Behaviour-Driven Design is more than tests.

Test Business Logic

Business logic is where your rules live, and in Emmett it's a plain function: decide takes the current state and a command, and returns events. No I/O, no framework, nothing to mock. That makes it the cheapest thing to test, so this is where most of your tests belong. Set up a DeciderSpecification once with your decide, evolve, and initialState:

import { DeciderSpecification } from '@event-driven-io/emmett';

const given = DeciderSpecification.for({
  decide,
  evolve,
  initialState: initialState,
});

For each command, three questions cover it: does it produce the right events when the rule allows the action, does it say no when the rule forbids it, and does it stay quiet when there's nothing to do? That's three tests, each named after the rule it guards.

Assert the events it produces

Give it a state through the given events, run the command, and check what comes back. Assert the events, not the state the decision saw. The events are what the rest of your system reacts to; the state is just how the decision got there.

void it('should add product item', () => {
  given([])
    .when({
      type: 'AddProductItemToShoppingCart',
      data: {
        shoppingCartId,
        productItem,
      },
      metadata: { now },
    })
    .then([
      {
        type: 'ProductItemAddedToShoppingCart',
        data: {
          shoppingCartId,
          productItem,
          addedAt: now,
        },
      },
    ]);
});

Emmett matches each expected event against the produced events, comparing the fields you write and ignoring any you leave out. The two lists must also hold the same number of events. We recommend writing every field, as the test above does. This is why Emmett favours deterministic decisions and projections: inject the time and other inputs instead of reading them inside the logic, and each field takes a known value you can assert, the way the injected now fixes addedAt here (see Make Time and External Data Injectable). Leave a field out only when its value is deliberately non-deterministic, such as a timestamp the logic assigns that you chose not to pass in.

Assert with a custom check

Some checks don't fit a fixed list: a computed total, a value in a range, a rule across several events. For those, pass then a callback. It receives the events, and you can assert them with Emmett's built-in assertions or your preferred library, such as node:assert, Vitest, or Jest. The test fails if the callback throws or returns an Error:

void it('asserts events with a custom check', () => {
  givenCart([
    {
      type: 'ProductItemAdded',
      data: { shoppingCartId, productItem: shoes },
    },
  ])
    .when({
      type: 'RemoveProductItem',
      data: { shoppingCartId, productItem: shoes, removedBy: 'user-456' },
    })
    // the callback receives the produced events; assert them however you
    // need, here with node:assert, including checks an exact list can't state
    .then((events) => {
      equal(events.length, 1);
      const [event] = events;
      ok(event?.type === 'ProductItemRemoved');
      ok(event.data.productItem.price > 0);
    });
});

Assert the rule it enforces

A rule isn't proven until you show it saying no. Set up a state that should reject the command, then assert the exact error, not merely that something threw. That same error becomes the caller's 403 later, so it's worth pinning down. thenThrows takes the error type, a check on the message, or both:

void it('should not add products', () => {
  given([
    {
      type: 'ProductItemAddedToShoppingCart',
      data: {
        shoppingCartId,
        productItem,
        addedAt: oldTime,
      },
    },
    {
      type: 'ShoppingCartConfirmed',
      data: { shoppingCartId, confirmedAt: oldTime },
    },
  ])
    .when({
      type: 'AddProductItemToShoppingCart',
      data: {
        shoppingCartId,
        productItem,
      },
      metadata: { now },
    })
    .thenThrows(
      (error: Error) => error.message === 'Shopping Cart already closed',
    );
});

Assert it stays quiet

Some commands are valid but have nothing to add, like removing an item that's already gone. Returning no events for those is what keeps a command safe to send twice. Assert that path with thenNothingHappened().

Test the API In-Memory

A lot happens between the request and the decision. The request gets mapped to a command, validated, run through middleware, and its result turned into a status code. None of that is visible to a unit test, so let's cover it, still in memory, so the tests stay fast enough to run continuously.

ApiSpecification gives you the same given/when/then, this time over HTTP. Point it at the seams the unit tests can't reach, and leave the rule-by-rule coverage to them.

Set up the specification

Inject the in-memory store and stub what the slice reaches for (the price lookup and the clock), so the results are deterministic:

import { getInMemoryEventStore } from '@event-driven-io/emmett';
import {
  ApiSpecification,
  getApplication,
} from '@event-driven-io/emmett-expressjs';

const unitPrice = 100;
const now = new Date();

const given = ApiSpecification.for<ShoppingCartEvent>(
  () => getInMemoryEventStore(),
  (eventStore) =>
    getApplication({
      apis: [
        shoppingCartApi(
          eventStore,
          () => Promise.resolve(unitPrice),
          () => now,
        ),
      ],
    }),
);

Assert the response and the new events

existingStream seeds the stream, when sends the request, and then checks both sides of the outcome at once: the status the caller gets, and the events that land in the store.

import {
  existingStream,
  expectNewEvents,
  expectResponse,
} from '@event-driven-io/emmett-expressjs';

void describe('When opened with product item', () => {
  void it('should confirm', () => {
    return given(
      existingStream(shoppingCartId, [
        {
          type: 'ProductItemAddedToShoppingCart',
          data: {
            shoppingCartId,
            productItem,
            addedAt: oldTime,
          },
        },
      ]),
    )
      .when((request) =>
        request.post(`/clients/${clientId}/shopping-carts/current/confirm`),
      )
      .then([
        expectResponse(204),
        expectNewEvents(shoppingCartId, [
          {
            type: 'ShoppingCartConfirmed',
            data: {
              shoppingCartId,
              confirmedAt: now,
            },
          },
        ]),
      ]);
  });
});

Assert the failure the caller sees

The unit test already showed the rule throws. Here you show the throw turning into the right HTTP contract: getApplication maps IllegalStateError to a 403 with a Problem Details body. Assert it with expectError, from @event-driven-io/emmett-expressjs alongside the helpers above:

void it('should not add products', () => {
  return given(
    existingStream(shoppingCartId, [
      {
        type: 'ProductItemAddedToShoppingCart',
        data: {
          shoppingCartId,
          productItem,
          addedAt: oldTime,
        },
      },
      {
        type: 'ShoppingCartConfirmed',
        data: { shoppingCartId, confirmedAt: oldTime },
      },
    ]),
  )
    .when((request) =>
      request
        .post(`/clients/${clientId}/shopping-carts/current/product-items`)
        .send(productItem),
    )
    .then(
      expectError(403, {
        detail: 'Shopping Cart already closed',
        status: 403,
        title: 'Forbidden',
        type: 'about:blank',
      }),
    );
});

Test End-to-End Against PostgreSQL

In-memory is great for a fast loop, but it never touches a real database. Serialisation and queries only run for real against the actual store, and that's exactly where surprises hide. So keep a small end-to-end set for the flows that matter most, and treat the API as a black box.

Start a database container

Spin up PostgreSQL in a throwaway container with TestContainers, which randomises the port and cleans up after itself. Start one for the whole suite and reuse it, since the store keeps a connection pool, then close both in afterAll:

import {
  getPostgreSQLEventStore,
  type PostgresEventStore,
} from '@event-driven-io/emmett-postgresql';
import type { StartedPostgreSqlContainer } from '@testcontainers/postgresql';
import { PostgreSqlContainer } from '@testcontainers/postgresql';

void describe('ShoppingCart E2E', () => {
  let postgreSQLContainer: StartedPostgreSqlContainer;
  let eventStore: PostgresEventStore;

  // Set up a container and event store before all tests
  beforeAll(async () => {
    postgreSQLContainer = await new PostgreSqlContainer(
      'postgres:18.1',
    ).start();
    eventStore = getPostgreSQLEventStore(
      postgreSQLContainer.getConnectionUri(),
    );
  });

  // Close PostgreSQL connection and stop container once we finished testing
  afterAll(async () => {
    await eventStore.close();
    return postgreSQLContainer.stop();
  });
  // (...) Tests will go here
});

Point the specification at it

It's the same getApplication as before, now backed by the container's store. That's the nice part: you're re-running a known slice against real infrastructure, not writing it twice.

import {
  ApiE2ESpecification,
  getApplication,
} from '@event-driven-io/emmett-expressjs';

const given = ApiE2ESpecification.for({
  getEventStore: () => eventStore,
  getApplication: (eventStore) =>
    getApplication({
      apis: [
        shoppingCartApi(
          eventStore,
          () => Promise.resolve(unitPrice),
          () => now,
        ),
      ],
    }),
});

Drive it through HTTP

Setup runs through requests too, so you assert only the responses. Here we open a cart with a product, then confirm it:

import { expectResponse } from '@event-driven-io/emmett-expressjs';

void describe('When opened with product item', () => {
  const openedShoppingCartWithProduct: TestRequest = (request) =>
    request
      .post(`/clients/${clientId}/shopping-carts/current/product-items`)
      .send(productItem);

  void it('should confirm', () => {
    return given(openedShoppingCartWithProduct)
      .when((request) =>
        request.post(`/clients/${clientId}/shopping-carts/current/confirm`),
      )
      .then([expectResponse(204)]);
  });
});

Test Projections

Projections earn their keep in a real database, so that's the only honest place to test them. Serialisation and querying are the whole point, and an in-memory fake would paper over both. Assert the stored document as the read model evolves: the first event creates it, later events add to it, and a terminal event clears it. PostgreSQLProjectionSpec keeps the same given/when/then, this time over Pongo documents:

import {
  eventInStream,
  eventsInStream,
  expectPongoDocuments,
  newEventsInStream,
  PostgreSQLProjectionSpec,
} from '@event-driven-io/emmett-postgresql';

void describe('Shopping Cart Short Info Projection', () => {
  let postgres: StartedPostgreSqlContainer;
  let given: PostgreSQLProjectionSpec<ProductItemAdded | DiscountApplied>;
  let shoppingCartId: string;

  beforeAll(async () => {
    postgres = await getPostgreSQLStartedContainer();

    given = PostgreSQLProjectionSpec.for({
      projection: shoppingCartShortInfoProjection,
      connectionString: postgres.getConnectionUri(),
    });
  });

  beforeEach(() => (shoppingCartId = `shoppingCart:${uuid()}:${uuid()}`));

  afterAll(async () => {
    await postgres.stop();
  });

  void it('creates summary from first event', () =>
    given([])
      .when([
        eventInStream(shoppingCartId, {
          type: 'ProductItemAdded',
          data: {
            productItem: { price: 100, productId: 'shoes', quantity: 100 },
          },
        }),
      ])
      .then(
        expectPongoDocuments
          .fromCollection<ShoppingCartShortInfo>(
            shoppingCartShortInfoCollectionName,
          )
          .withId(shoppingCartId)
          .toBeEqual({
            productItemsCount: 100,
            totalAmount: 10000,
            appliedDiscounts: [],
          }),
      ));

  void it('accumulates across events', () => {
    const couponId = uuid();

    return given(
      eventsInStream<ProductItemAdded>(shoppingCartId, [
        {
          type: 'ProductItemAdded',
          data: {
            productItem: { price: 100, productId: 'shoes', quantity: 100 },
          },
        },
      ]),
    )
      .when(
        newEventsInStream(shoppingCartId, [
          {
            type: 'DiscountApplied',
            data: { percent: 10, couponId },
          },
        ]),
      )
      .then(
        expectPongoDocuments
          .fromCollection<ShoppingCartShortInfo>(
            shoppingCartShortInfoCollectionName,
          )
          .withId(shoppingCartId)
          .toBeEqual({
            productItemsCount: 100,
            totalAmount: 9000,
            appliedDiscounts: [couponId],
          }),
      );
  });
});

The same spec ships for every store: swap PostgreSQLProjectionSpec for SQLiteProjectionSpec, MongoDBInlineProjectionSpec, or InMemoryProjectionSpec (no container, so it runs at unit speed). The given/when/then doesn't change.

Assert it handles duplicates

Emmett processes each event exactly once today: inline projections run inside the append transaction, async ones through transactional checkpointing. Even so, keeping a projection idempotent is worth it. Emmett may later add consumers, such as ones backed by Kafka, RabbitMQ, or SQS, that deliver an event more than once, and a projection that double-counts on the second pass would be a latent bug. Replay the same events with { numberOfTimes } to prove yours holds. Here a discount guarded by its coupon id is handled twice and applied once:

void it('ignores a redelivered event', () => {
  const couponId = uuid();

  return given(
    eventsInStream<ProductItemAdded>(shoppingCartId, [
      {
        type: 'ProductItemAdded',
        data: {
          productItem: { price: 100, productId: 'shoes', quantity: 100 },
        },
      },
    ]),
  )
    .when(
      newEventsInStream(shoppingCartId, [
        {
          type: 'DiscountApplied',
          data: { percent: 10, couponId },
        },
      ]),
      // 👇 deliver the same event twice
      { numberOfTimes: 2 },
    )
    .then(
      expectPongoDocuments
        .fromCollection<ShoppingCartShortInfo>(
          shoppingCartShortInfoCollectionName,
        )
        .withId(shoppingCartId)
        .toBeEqual({
          productItemsCount: 100,
          totalAmount: 9000,
          appliedDiscounts: [couponId],
        }),
    );
});

For raw SQL projections, deletion, and multi-stream projections, see Test a Projection in the Read Models guide.

Test Async Consumers

An async consumer runs in the background, so a test has to wait for it to process what the test appended before asserting. The unreliable ways to wait are a fixed sleep, which is either flaky or slow, and a stop condition that closes over a position assigned after the append, which races the poller and hangs when the poller wins. Wait on the consumer's own progress instead.

Start the consumer, append, and let whenCaughtUp resolve once every processor has reached the store's tail as of the call. It observes committed checkpoints, so it resolves the moment processing catches up and never hangs on an append it has already passed:

const processed: GuestStayEvent[] = [];

const consumer = sqliteEventStoreConsumer({
  driver: sqlite3EventStoreDriver,
  fileName,
});
consumer.reactor<GuestStayEvent>({
  processorId: uuid(),
  eachMessage: (event) => {
    processed.push(event);
  },
});

const guestId = uuid();
const events: GuestStayEvent[] = [
  { type: 'GuestCheckedIn', data: { guestId } },
  { type: 'GuestCheckedOut', data: { guestId } },
];

let consumerPromise: Promise<void> | undefined;
try {
  consumerPromise = consumer.start();
  await consumer.whenStarted();

  await eventStore.appendToStream(`guestStay-${guestId}`, events);

  // resolves once every processor has reached the store's tail
  await consumer.whenCaughtUp();

  assertThatArray(processed).containsElementsMatching(events);
} finally {
  await consumer.close();
  await consumerPromise;
}

When you care about one precise point rather than the whole tail, wait for the position the append returned:

const appendResult = await eventStore.appendToStream(
  `guestStay-${guestId}`,
  [
    { type: 'GuestCheckedIn', data: { guestId } },
    { type: 'GuestCheckedOut', data: { guestId } },
  ],
);

await consumer.whenProcessed(appendResult.lastEventGlobalPosition);

Both waits observe checkpoints rather than a message callback, so they work for projectors too, which have no eachMessage to hook. Bound them with a timeout so a consumer that never catches up fails fast with a descriptive error instead of hanging until the test runner kills it:

// rejects with a descriptive error after 5s rather than hanging
await consumer.whenCaughtUp({ timeout: 5000 });

Choose the Right Level

The proportion between the levels is up to you, but one rule keeps it honest: put each test at the lowest level that can fail for the reason you care about.

• A business rule breaks in the decision, so test it as a unit. Most of your tests live here.
• Wiring (mapping, validation, status codes, concurrency) breaks above the decision, so test it in-memory.
• Serialisation and queries break against the database, so keep a lean end-to-end and projection set for those.

Cover each concern once. Running the same scenario through all three levels buys the same confidence three times over, and you'll pay for it again on every future change. Once the in-memory tests are this cheap, you can lean on them; Martin Thwaites makes that case well.

Best Practices

Assert Behaviour, Not State

Check the events a decision returns and the documents a projection writes, never the intermediate state the code built along the way. Events and documents are the contract other code depends on; internal state is free to change under a refactor, and your tests shouldn't break when it does.

Make Time and External Data Injectable

Route the current time through command metadata and inject dependencies like the price lookup, so a test can fix them. The specifications above pass () => now and () => Promise.resolve(unitPrice), which is why the expected addedAt and totals come out as exact values.

Give Every Test Its Own Stream

Share the container and store across a suite, but give each test a fresh stream id from randomUUID in beforeEach, so no test can see another's events. The integration spec shows the pattern.

Wait for a Point, Don't Race a Stop

When a test drives an async consumer, wait on whenCaughtUp or whenProcessed, not on a stop condition that closes over a position assigned after appendToStream. The awaiters observe committed progress, so they resolve as soon as processing catches up and can't miss an append they already passed. A stop that races a late position is the classic source of a consumer test that passes locally and times out on CI.

Troubleshooting

A Timestamp Assertion Fails By Milliseconds

The decision is reading the wall clock instead of an injected time. Route time through command metadata and inject a fixed clock into the API setup, as Make Time and External Data Injectable shows, and the expected value lands exactly.

The Suite Is Slow

A container is starting per test. Start one in beforeAll, reuse it, and give each test a fresh stream id instead. Close the store and stop the container in afterAll so connections get released.

A Consumer Test Hangs or Times Out

The test is waiting on something the consumer never reaches. Replace any sleep or position-racing stop with whenCaughtUp, and pass a timeout so the wait rejects with a message naming the processor and the checkpoint it stopped at instead of hanging until the runner kills it. If a processor deliberately does not persist checkpoints, for example one resuming from an explicit in-memory position, its progress is invisible to whenCaughtUp; wait on a message from its handler instead.

Further Readings

• Getting Started - Unit Testing
• Command Handling - the handler these tests exercise
• Read Models - testing projections in detail
• API Reference: Decider - the specification helpers
• Behaviour-Driven Design is more than tests
• Testing Event Sourcing, Emmett edition
• Building Operable Software with TDD