RclexTesting

A Phoenix.LiveViewTest-style testing library for ROS 2 applications built with rclex.

Describe robot behaviour as executable scenario specifications — without manually creating nodes, publishers, subscribers, timers, clients, or assertion loops.

defmodule RobotTest do
  use ExUnit.Case
  use RclexTesting

  scenario "robot starts cleaning" do
    mock_service "/safety/check", SafetyCheck,
      fn _req -> %SafetyCheck.Response{allowed: true} end

    watch "/robot/state", RobotState
    publish "/mission/start", %MissionStart{id: 42}

    expect "/robot/state",
      matches(%{mode: :cleaning, speed: gt(0)})
  end
end

Features

Installation

Add rclex_testing to your test dependencies in mix.exs:

def deps do
  [
    {:rclex, "~> 0.12"},
    {:rclex_testing, path: "path/to/rclex_testing", only: :test}
  ]
end

Add the ROS 2 message/service/action types you need to config/config.exs:

import Config

config :rclex,
  ros2_message_types: [
    "std_msgs/msg/String",
    "geometry_msgs/msg/Twist"
    # add your types here
  ],
  ros2_service_types: [
    "std_srvs/srv/SetBool"
  ],
  ros2_action_types: []

Then run:

mix rclex.gen.msgs
mix rclex.gen.srvs
mix rclex.gen.action

Usage

Add use RclexTesting to any ExUnit.Case module. Each scenario block automatically starts a unique ROS node, cleans up all publishers, subscriptions, services, and action servers when the test exits.

defmodule MyApp.RobotTest do
  use ExUnit.Case, async: false
  use RclexTesting

  alias Rclex.Pkgs.StdMsgs

  scenario "publishes and receives a message" do
    watch "/chatter", StdMsgs.Msg.String
    publish "/chatter", %StdMsgs.Msg.String{data: "hello"}
    msg = expect("/chatter", &(&1.data == "hello"))
    assert msg.data == "hello"
  end
end

Topics

watch/2

Subscribe to a topic and buffer all arriving messages.

watch "/robot/state", RobotState

publish/2

Publish a message. The type is inferred from the struct. A publisher is started lazily and reused within the scenario.

publish "/cmd_vel", %Twist{linear: %Vector3{x: 1.0}}

expect/2,3

Assert that a message matching the predicate arrives. Returns the matched message. Cursor-based: consecutive expect calls on the same topic advance past already-matched messages.

expect "/robot/state", &(&1.mode == :running)
expect "/robot/state", RobotState, &(&1.mode == :running)
expect "/robot/state", &(&1.mode == :running), timeout: 10_000

expect_sequence/2,3

Assert a sequence of matchers in order on the same topic.

expect_sequence "/robot/state", [
  matches(%{mode: :idle}),
  matches(%{mode: :starting}),
  matches(%{mode: :running})
]

refute_message/1,2,3,4

Assert that no (matching) message arrives within a timeout window.

refute_message "/fault_topic", timeout: 500
refute_message "/fault_topic", FaultMsg, &(&1.severity == :critical)

expect_count/2,3

Assert that exactly N messages arrive.

msgs = expect_count "/sensor/data", 5

messages/1

Return all buffered messages for a topic in arrival order.

all = messages("/robot/state")

eventually/1,2

Poll a zero-arity predicate until truthy or timeout.

eventually(fn -> length(messages("/data")) >= 10 end, timeout: 10_000)

Services

mock_service/3

Register a test service on the scenario node. Requests are recorded in the collector and forwarded to the handler function.

mock_service "/safety/check", SafetyCheck,
  fn _req -> %SafetyCheck.Response{allowed: true} end

call_service/3,4

Call a service synchronously and return the response.

response = call_service("/safety/check", SafetyCheck, %SafetyCheck.Request{})

expect_service_call/2,3

Assert that the mocked service received a (matching) request.

expect_service_call "/safety/check", SafetyCheck
expect_service_call "/safety/check", SafetyCheck, fn req -> req.robot_id == "r1" end

refute_service_call/2,3,4

Assert that no (matching) request arrives.

refute_service_call "/delete_map", DeleteMap

Actions

mock_action/3

Register a test action server. Goals, feedback, and results are all recorded. The handler receives (goal, publish_feedback) and must return a result struct.

mock_action "/navigate", Navigate, fn goal, publish_feedback ->
  publish_feedback.(%Navigate.Feedback{progress: 50})
  publish_feedback.(%Navigate.Feedback{progress: 100})
  %Navigate.Result{success: true}
end

send_goal/3,4

Send a goal to an action server and return the goal UUID.

uuid = send_goal("/navigate", Navigate, %Navigate.Goal{x: 10, y: 20})

expect_goal/2,3, refute_goal/2,3,4

Assert that a (matching) goal was or was not submitted.

goal = expect_goal "/navigate", Navigate, fn g -> g.x == 10 end
refute_goal "/navigate", Navigate

expect_feedback/2,3, expect_result/2,3

Assert that feedback or a result was produced.

expect_feedback "/navigate", Navigate, fn fb -> fb.progress == 100 end
expect_result "/navigate", Navigate, fn r -> r.success end

Matchers

Matchers return predicate functions and compose with all assertion DSL.

expect "/robot/state",
  matches(%{
    mode: :running,
    speed: gt(0),
    battery: between(20, 100),
    pose: matches(%{x: approx(10.0, delta: 0.1)})
  })

Simulated Time

Test-Controlled Time (use_sim_time: true)

The test controls time deterministically. Use this for unit tests and CI pipelines.

scenario "timer fires after 1 second", use_sim_time: true do
  advance_time(1_000)  # advance 1000 ms
  expect "/tick", MyPkg.Msg.Tick
end

The scenario:

1. Starts the node with use_sim_time:=true so it subscribes to /clock
2. Creates a :ros_time clock with manual override enabled
3. Publishes an initial Clock message at t=0
4. On each advance_time(ms), updates the override and publishes the new time

Use set_time(ms) to jump directly to a specific time:

scenario "multiple phases", use_sim_time: true do
  # Phase 1: around t=0
  expect "/phase", &(&1 == :startup)
  
  # Jump to t=5s for phase 2
  set_time(5_000)
  expect "/phase", &(&1 == :running)
  
  # Advance gradually during phase 3
  advance_time(2_000)
  expect "/phase", &(&1 == :complete)
end

External Time (use_sim_time: :external)

An external simulator (Gazebo, motion capture, etc.) publishes on /clock. Use this when testing against live simulators.

scenario "robot moves in Gazebo", use_sim_time: :external do
  # Gazebo publishes on /clock; don't call advance_time
  watch "/robot/pose", GeometryMsgs.Msg.Pose
  expect "/robot/pose", &(&1.position.z < 10)
end

The scenario:

1. Starts the node with use_sim_time:=true so it subscribes to /clock
2. Creates a :ros_time clock (without override) — it syncs automatically from /clock
3. Publishes nothing; the external source drives all time

Call sim_time() to read the current clock value (driven by external source).

Ensuring Your Application Uses Sim Time

Any ROS node—in your application under test or spawned in a scenario—must start with the ROS arg to recognize sim time:

Rclex.start_node("my_node", ros_args: ["--ros-args", "-p", "use_sim_time:=true"])

The rclex_testing scenario automatically passes this to the test node when use_sim_time: true or use_sim_time: :external is set. For other nodes you spawn or test against, ensure they also receive this argument or they will ignore /clock and use wall time.

Multi-Stream Event Assertions

Beyond single-stream expect, rclex_testing provides a set of cross-stream workflow assertions that verify ordering and causality across topics, services, and actions.

Event Descriptors

Build blocks for multi-stream assertions. Each descriptor references a stream and an optional predicate:

message("/robot/state")                           # any message
message("/robot/state", &(&1.mode == :running))   # with predicate
message("/robot/state", matches(%{mode: :running})) # with matcher

service_call("/safety/check")
service_call("/safety/check", fn req -> req.allowed end)

goal("/navigate")
goal("/navigate", fn g -> g.x > 0 end)

feedback("/navigate", fn fb -> fb.progress >= 50 end)
result("/navigate", fn r -> r.success end)

The stream (watch/2, mock_service/3, mock_action/3) must be set up before the assertion runs.

expect_sequence/1

Assert an ordered sequence of events across any mix of topics, services, and actions. Each event must arrive after the previous one (based on the global event log).

watch "/mission/start", MissionStart
watch "/robot/state", RobotState
mock_service "/safety/check", SafetyCheck, fn _ -> ... end
mock_action "/navigate", Navigate, fn _, _ -> ... end

expect_sequence [
  message("/mission/start"),
  service_call("/safety/check"),
  goal("/navigate"),
  feedback("/navigate", fn fb -> fb.progress >= 100 end),
  result("/navigate", fn r -> r.success end),
  message("/robot/state", matches(%{mode: :completed}))
]

Returns the list of matched payloads in order.

For single-topic ordered assertions, the shorter form in Topics DSL is still available:

expect_sequence "/robot/state", [
  matches(%{mode: :starting}),
  matches(%{mode: :running}),
  matches(%{mode: :completed})
]

expect_before/2 and expect_after/2

Assert a causal relationship between two events.

# Safety check must happen before robot enters running state
{req, state} = expect_before(
  service_call("/safety/check"),
  message("/robot/state", matches(%{mode: :running}))
)

# Completed must happen after running
{earlier, later} = expect_after(
  message("/robot/state", matches(%{mode: :completed})),
  message("/robot/state", matches(%{mode: :running}))
)

expect_before(a, b) returns {a_payload, b_payload}. expect_after(later, earlier) returns {earlier_payload, later_payload}.

expect_together/1

Assert all conditions are eventually satisfied in any order. Each event in the log can satisfy at most one descriptor (greedy left-to-right matching).

[state_msg, battery_msg] = expect_together [
  message("/robot/state", matches(%{mode: :running})),
  message("/battery/state", matches(%{charging: false}))
]

expect_workflow/1

Readable alias for expect_sequence/1 using :given / :then keyword steps. Ideal for describing cause-and-effect workflows.

expect_workflow [
  given: message("/mission/start"),
  then:  service_call("/safety/check"),
  then:  message("/robot/state", matches(%{mode: :running})),
  then:  message("/robot/state", matches(%{mode: :completed}))
]

Failure Messages

When a sequence assertion fails, the error includes which event was expected and what was actually observed:

expect_sequence: timed out waiting for event.

Expected:
  message on "/robot/state"

Events observed after seq 2:
  [seq 3] message on /robot/state: %RobotState{mode: :completed}

Design

• Scenario node — each scenario block starts a uniquely-named ROS node owned by the test; all entities are stopped in on_exit.
• Agent buffer — subscription and action callbacks write to a per-scenario Agent; the test process retains the buffer for messages/1 and structural inspection.
• Unified event log — every incoming message, service call, goal, feedback, and result is also appended (with a monotonic sequence number) to a global log. The multi-stream assertions (expect_sequence, expect_before, etc.) query this log to verify ordering across streams.
• Message-passing wakeup — callbacks send a tagged notification to the test process; expect uses receive with a deadline instead of polling, giving sub-millisecond response latency. Cross-stream assertions additionally receive an :any notification after each event.
• Cursor semantics — each expect/expect_service_call/expect_goal etc. advances a per-stream cursor so consecutive assertions on the same stream always observe distinct events.