Continuum.Test (continuum v0.8.1)

Copy Markdown View Source

Public helpers for testing Continuum workflows.

The helpers in this module are deliberately small and stable. They cover the v0.1 testing loop:

  • run workflows against the in-memory journal
  • load or persist event histories
  • replay a committed golden history (through Continuum.Replay)
  • inject signals and timers in deterministic tests
  • check out an Ecto SQL Sandbox connection for Postgres-backed tests

The in-memory journal is process-local and not durable. Use unique run IDs or call reset_in_memory!/0 between tests that need a clean journal.

Summary

Functions

Assert that a workflow replays from history without drift.

Assert that a workflow replays from history to expected.

Kill a run's engine process the way a node loss would.

Drive the durable runtime until run_id reaches a terminal state.

Drive the durable runtime until run_id is in one of states.

Persist a run history to path as an Erlang external term.

Make every unfired timer on a run due now.

Expire a run's lease so recovery may rescue it.

Inject a fired timer event for the latest pending timer in a run's history.

Load a run's event history from a journal.

Inject a signal into a run and wake its local engine when one exists.

Load a history previously written by dump_history!/3.

Replay a workflow from an existing history.

Reset the in-memory journal.

Start a workflow run against the Postgres journal.

Start a workflow run synchronously against the in-memory journal.

Types

replay_result()

@type replay_result() ::
  {:ok, term()}
  | {:suspended, term()}
  | {:continued, binary()}
  | {:error, term()}

Functions

assert_replays(workflow_module, input, history)

@spec assert_replays(module(), term(), [map()]) :: term()

Assert that a workflow replays from history without drift.

Returns the replayed result.

assert_replays(workflow_module, input, history, expected)

@spec assert_replays(module(), term(), [map()], term()) :: term()

Assert that a workflow replays from history to expected.

checkout_sandbox(repo \\ Application.get_env(:continuum, :repo), opts \\ [])

@spec checkout_sandbox(
  module() | nil,
  keyword()
) :: :ok

Check out an Ecto SQL Sandbox connection.

Pass shared: true when workflow engines or workers need to use the test process' checked-out connection.

crash!(run_id, opts \\ [])

(since 0.8.0)
@spec crash!(
  binary(),
  keyword()
) :: :ok | {:error, :no_engine}

Kill a run's engine process the way a node loss would.

Returns once the engine is dead and has left the instance registry, so the next drive/2 starts a genuinely fresh engine rather than racing the old one. Pair with expire_lease!/2: a killed engine's lease is still valid until it expires, and rescuing it earlier would be lease theft.

Drive the run to a resting state first (drive_until_state/3). Killing an engine that is mid-statement is fine in production but takes the SQL Sandbox shared connection down with it, which fails the rest of the test for a reason that has nothing to do with the workflow.

drive(run_id, opts \\ [])

(since 0.8.0)
@spec drive(
  binary(),
  keyword()
) :: {:ok, map()} | {:error, term()}

Drive the durable runtime until run_id reaches a terminal state.

Test suites normally start Continuum with its pollers disabled so nothing moves behind a test's back. This turns the crank instead: on each tick it rescues expired leases, dispatches runnable runs, runs due activity tasks, and fires due timers, until the run completes, fails, is cancelled, or the deadline passes.

Returns whatever Continuum.await/3 returns for the finished run.

{:ok, run_id} = Continuum.Test.start_postgres(Checkout, order)
assert {:ok, %{state: :completed}} = Continuum.Test.drive(run_id)

Options: :instance, :timeout (default 5_000 ms), :batch_size (default 10).

drive_until_state(run_id, states, opts \\ [])

(since 0.8.0)
@spec drive_until_state(binary(), [atom()], keyword()) :: :ok | {:error, :timeout}

Drive the durable runtime until run_id is in one of states.

The same crank as drive/2, stopped earlier. Useful for getting a run to the point you want to crash it at.

Continuum.Test.drive_until_state(run_id, [:suspended])

dump_history!(run_id, path, opts \\ [])

@spec dump_history!(binary(), Path.t(), keyword()) :: :ok

Persist a run history to path as an Erlang external term.

The resulting file is intended for golden-history tests committed to the repository.

elapse_timers!(run_id, opts \\ [])

(since 0.8.0)
@spec elapse_timers!(
  binary(),
  keyword()
) :: :ok

Make every unfired timer on a run due now.

A multi-day timer/1 is the thing you least want to wait for in a test; this moves its fires_at into the past so the next drive/2 fires it.

expire_lease!(run_id, opts \\ [])

(since 0.8.0)
@spec expire_lease!(
  binary(),
  keyword()
) :: :ok

Expire a run's lease so recovery may rescue it.

Recovery deliberately refuses to touch a run whose lease is still live, so a crash-resume test has to move the clock rather than wait out a real TTL.

fire_timer(run_id, opts \\ [])

@spec fire_timer(
  binary(),
  keyword()
) :: :ok | {:error, term()}

Inject a fired timer event for the latest pending timer in a run's history.

history(run_id, opts \\ [])

@spec history(
  binary(),
  keyword()
) :: [map()]

Load a run's event history from a journal.

:journal accepts the shorthand :postgres or :in_memory as well as an adapter module, so a test never has to name a Continuum.Runtime.* module. Defaults to the in-memory journal.

inject_signal(run_id, name, payload, opts \\ [])

@spec inject_signal(binary(), atom(), term(), keyword()) :: :ok | {:error, term()}

Inject a signal into a run and wake its local engine when one exists.

Delivery goes through the same Continuum.Runtime.SignalRouter path as Continuum.signal/4: in-memory signals are buffered in the run's mailbox and consumed by the matching await signal, journaling signal_received with the await's command identity — injected signals exercise the same command-identity drift detection as production deliveries.

load_history!(path)

@spec load_history!(Path.t()) :: [map()]

Load a history previously written by dump_history!/3.

replay(workflow_module, input, history, opts \\ [])

@spec replay(module(), term(), [map()], keyword()) :: replay_result()

Replay a workflow from an existing history.

Delegates to Continuum.Replay.run/4, which is read-only by default: stepping past the journaled tail reports {:suspended, {:history_exhausted, _}} instead of executing the next activity for real. Pass journal: explicitly to replay through a live adapter.

Returns {:ok, result} when the workflow completes from history, or {:suspended, reason} if the history ends at a pending effect.

reset_in_memory!()

@spec reset_in_memory!() :: :ok

Reset the in-memory journal.

start_postgres(workflow_module, input, opts \\ [])

@spec start_postgres(module(), term(), keyword()) ::
  {:ok, binary()} | {:error, term()}

Start a workflow run against the Postgres journal.

start_synchronous(workflow_module, input, opts \\ [])

@spec start_synchronous(module(), term(), keyword()) ::
  {:ok, binary()} | {:error, term()}

Start a workflow run synchronously against the in-memory journal.

Activities run inline in the engine process — no worker pool, no retry, no timeout. Child workflows run inline too, as their own in-memory engines.

Stubbing activities

Pass :activities to stand in for activity bodies, so a unit test can drive a workflow's branches without the activity's real dependencies:

Continuum.Test.start_synchronous(Checkout, order,
  activities: %{
    {Payments, :charge} => fn _order -> {:ok, "ch_test"} end,
    {Shipping, :book} => {:error, :out_of_stock}
  }
)

Keys are {Module, :function}, or {Module, :function, arity} when one module exports the same activity name at several arities; the more specific key wins. A value that is a function of the activity's arity is called with the activity's arguments, and any other value is returned as-is.

Stub returns are validated with Continuum.DurableTerm, because in-memory writes otherwise skip that check — a stub returning a PID would pass the unit test and be rejected in production.

Stubs are refused on the Postgres journal: a durable activity runs in a worker process out of a claimed task row, which a stub cannot reach. They also cannot influence command identity, which is computed at macro expansion from the call site, so a stubbed run journals the same command ids as a real one.