Durable.Testing (Durable v1.1.0)

View Source

Helpers for testing workflows.

Runs execute inline in the test process, with no queue or pollers, so they share the test's Ecto sandbox. drain/2 keeps running a workflow and every run it spawns until nothing is left to do. Sleeps end immediately and retries skip their backoff.

Setup

Turn queue processing off in config/test.exs, so pollers don't race the test for the same runs:

config :durable, disable_queue_processing: true

Then import the helpers where you test workflows, for example in your DataCase:

import Durable.Testing

Example

test "charges the order" do
  run = run_workflow(OrderWorkflow, %{"order_id" => 42})

  assert_status(run, :completed)
  assert_step(run, :charge)
  assert %{"charged" => true} = run.context
end

A waiting run is resumed with the regular API, then drained:

run = run_workflow(ApprovalWorkflow, %{})
assert_waiting_for_input(run, "manager_approval")

:ok = Durable.provide_input(run.id, "manager_approval", %{"approved" => true})

run
|> drain()
|> assert_status(:completed)

Every function takes a :durable option that names the Durable instance (default Durable). Runs can be passed as a %WorkflowExecution{}, a map with an :id, or the id itself.

Summary

Functions

assert_status(run, expected, opts \\ [])

Asserts the run's status, such as :completed, :failed or :waiting. On a mismatch the failure shows the run's current step and error.

assert_step(run, step, status \\ :completed, opts \\ [])

Asserts that the step's latest attempt ended with status (default :completed). Steps inside parallel and branch blocks are matched by their own name, so it also checks which branch ran.

assert_waiting_for_event(run, event_name, opts \\ [])

Asserts that the run is waiting for the event named event_name. Returns the %PendingEvent{}.

assert_waiting_for_input(run, input_name, opts \\ [])

Asserts that the run is waiting for the input named input_name. Returns the %PendingInput{}.

drain(run, opts \\ [])

Runs every pending execution in the run's tree (the run, its parallel branches and its child workflows) until none is left, then returns the reloaded run.

run_workflow(module, input \\ %{}, opts \\ [])

Starts the workflow in module with input and drains it. Returns the run as a %WorkflowExecution{}, finished or waiting.

steps(run, opts \\ [])

Returns the run's step executions, oldest first, including the steps its parallel blocks ran. There is one row per attempt, and a step that waited has one more row for the run after it resumed.

Functions

assert_status(run, expected, opts \\ [])

@spec assert_status(
  Durable.Storage.Schemas.WorkflowExecution.t() | map() | String.t(),
  atom(),
  keyword()
) :: Durable.Storage.Schemas.WorkflowExecution.t()

Asserts the run's status, such as :completed, :failed or :waiting. On a mismatch the failure shows the run's current step and error.

Returns the reloaded run.

assert_step(run, step, status \\ :completed, opts \\ [])

@spec assert_step(
  Durable.Storage.Schemas.WorkflowExecution.t() | map() | String.t(),
  atom(),
  atom(),
  keyword()
) :: Durable.Storage.Schemas.StepExecution.t()

Asserts that the step's latest attempt ended with status (default :completed). Steps inside parallel and branch blocks are matched by their own name, so it also checks which branch ran.

Returns that %StepExecution{}, so its output, error and attempt can be checked as well:

assert assert_step(run, :charge).attempt == 3

assert_waiting_for_event(run, event_name, opts \\ [])

@spec assert_waiting_for_event(
  Durable.Storage.Schemas.WorkflowExecution.t() | map() | String.t(),
  String.t(),
  keyword()
) :: Durable.Storage.Schemas.PendingEvent.t()

Asserts that the run is waiting for the event named event_name. Returns the %PendingEvent{}.

assert_waiting_for_input(run, input_name, opts \\ [])

@spec assert_waiting_for_input(
  Durable.Storage.Schemas.WorkflowExecution.t() | map() | String.t(),
  String.t(),
  keyword()
) :: Durable.Storage.Schemas.PendingInput.t()

Asserts that the run is waiting for the input named input_name. Returns the %PendingInput{}.

drain(run, opts \\ [])

@spec drain(
  Durable.Storage.Schemas.WorkflowExecution.t() | map() | String.t(),
  keyword()
) :: Durable.Storage.Schemas.WorkflowExecution.t()

Runs every pending execution in the run's tree (the run, its parallel branches and its child workflows) until none is left, then returns the reloaded run.

Sleeps end immediately and retries skip their backoff. Use it after Durable.provide_input/3 or Durable.send_event/3.

run_workflow(module, input \\ %{}, opts \\ [])

@spec run_workflow(module(), map(), keyword()) ::
  Durable.Storage.Schemas.WorkflowExecution.t()

Starts the workflow in module with input and drains it. Returns the run as a %WorkflowExecution{}, finished or waiting.

Takes the same options as Durable.start/3.

steps(run, opts \\ [])

@spec steps(
  Durable.Storage.Schemas.WorkflowExecution.t() | map() | String.t(),
  keyword()
) :: [Durable.Storage.Schemas.StepExecution.t()]

Returns the run's step executions, oldest first, including the steps its parallel blocks ran. There is one row per attempt, and a step that waited has one more row for the run after it resumed.