Forja.Testing (Forja v0.4.0)

View Source

Helpers for testing applications that use Forja.

Provides functions to verify that events were emitted and to process pending events synchronously in tests.

Inline mode

In tests, Forja can be configured to process events synchronously without Oban:

start_supervised!(
  {Forja,
   name: :test,
   repo: MyApp.Repo,
   pubsub: MyApp.PubSub,
   handlers: [MyHandler]}
)

Usage example

defmodule MyApp.OrderTest do
  use MyApp.DataCase

  import Forja.Testing

  test "emitting order event" do
    Forja.emit(:test, MyApp.Events.OrderCreated, payload: %{id: 1})

    assert_event_emitted(:test, MyApp.Events.OrderCreated)
    assert_event_emitted(:test, MyApp.Events.OrderCreated, %{"id" => 1})
  end
end

Summary

Functions

assert_event_caused_by(name, parent_id, child_type)

Asserts that an event of child_type was caused by the event parent_id.

assert_event_deduplicated(name, idempotency_key)

Asserts that an event with the given idempotency key was deduplicated.

assert_event_emitted(name, type, payload_match \\ %{})

Asserts that an event of the specified type was emitted in the name instance.

assert_same_correlation(events)

Asserts that all given events share the same non-nil correlation_id.

invoke_handler(handler_module, type, payload \\ %{}, meta \\ %{})

Invokes a handler directly with a fabricated event.

process_all_pending(name)

Synchronously processes all pending events (without processed_at).

refute_event_emitted(name, type)

Asserts that NO event of the specified type was emitted in the name instance.

Functions

assert_event_caused_by(name, parent_id, child_type)

@spec assert_event_caused_by(atom(), String.t(), String.t() | module()) ::
  Forja.Event.t()

Asserts that an event of child_type was caused by the event parent_id.

assert_event_deduplicated(name, idempotency_key)

@spec assert_event_deduplicated(atom(), String.t()) :: Forja.Event.t()

Asserts that an event with the given idempotency key was deduplicated.

Verifies that exactly one event exists with the specified key, confirming deduplication worked correctly.

assert_event_emitted(name, type, payload_match \\ %{})

@spec assert_event_emitted(atom(), String.t() | module(), map()) :: Forja.Event.t()

Asserts that an event of the specified type was emitted in the name instance.

Optionally verifies that the payload contains the specified fields.

assert_same_correlation(events)

@spec assert_same_correlation([Forja.Event.t()]) :: String.t()

Asserts that all given events share the same non-nil correlation_id.

invoke_handler(handler_module, type, payload \\ %{}, meta \\ %{})

@spec invoke_handler(module(), String.t() | module(), map(), map()) ::
  :ok | {:error, term()}

Invokes a handler directly with a fabricated event.

Useful for testing handlers in isolation without emitting real events.

process_all_pending(name)

@spec process_all_pending(atom()) :: :ok

Synchronously processes all pending events (without processed_at).

Useful for tests that need to ensure all handlers have been executed before making assertions.

refute_event_emitted(name, type)

@spec refute_event_emitted(atom(), String.t() | module()) :: :ok

Asserts that NO event of the specified type was emitted in the name instance.