Assertions

roboticks.assertions ships five rclpy-aware assertion helpers. Each one spins a node briefly, evaluates a predicate, and raises AssertionError (or TimeoutError) with an informative message on failure.

from roboticks.assertions import (
    assert_topic_published,
    assert_service_response,
    assert_action_result,
    assert_param_equals,
    assert_tf_transform,
)

rclpy-guarded. Importing roboticks.assertions on a host without rclpy installed raises a clear RuntimeError pointing you at the ROS2 install docs. Decorators in roboticks itself are not guarded — you can import them anywhere.

`assert_topic_published`

Wait until a message matching the predicate is published on topic.

Signature

def assert_topic_published(
    topic: str,
    msg_type: type,
    *,
    within: float = 5.0,
    predicate: Callable[[Any], bool] | None = None,
    qos: rclpy.qos.QoSProfile | None = None,
    node: rclpy.node.Node | None = None,
) -> Any: ...

Parameter	Type	Default	Description
`topic`	`str`	—	Fully-qualified topic (e.g. `/cmd_vel`).
`msg_type`	message class	—	The ROS2 message class to subscribe with.
`within`	`float`	`5.0`	Wall-clock seconds to wait.
`predicate`	`Callable[[msg], bool]`	accept-first	Filter — wait for the first message that returns True.
`qos`	`QoSProfile`	rclpy default	Subscription QoS.
`node`	`Node`	ephemeral	Use an existing node instead of spinning a one-shot subscriber.

Raises

TimeoutError — no matching message in within seconds.
RuntimeError — rclpy not initialised, topic type mismatch.

Returns

The first message matching the predicate, as a deserialised Python object of msg_type.

Example

from geometry_msgs.msg import Twist
from roboticks.assertions import assert_topic_published

msg = assert_topic_published(
    "/cmd_vel", Twist,
    within=3.0,
    predicate=lambda m: m.linear.x > 0.1,
)
assert msg.linear.x > 0.1

`assert_service_response`

Send a request, wait for the response.

Signature

def assert_service_response(
    service: str,
    srv_type: type,
    request: Any,
    *,
    within: float = 5.0,
    node: rclpy.node.Node | None = None,
) -> Any: ...

Raises

TimeoutError — service unavailable or response not received in within seconds.
RuntimeError — rclpy not initialised, service type mismatch.

Returns

The deserialised response object.

Example

from example_interfaces.srv import AddTwoInts
from roboticks.assertions import assert_service_response

response = assert_service_response(
    "/add_two_ints", AddTwoInts,
    AddTwoInts.Request(a=2, b=3),
    within=2.0,
)
assert response.sum == 5

`assert_action_result`

Send a goal, wait for the terminal result (success or failure).

Signature

def assert_action_result(
    action: str,
    action_type: type,
    goal: Any,
    *,
    within: float = 30.0,
    feedback_cb: Callable[[Any], None] | None = None,
    node: rclpy.node.Node | None = None,
) -> ClientGoalHandle: ...

Parameter	Description
`action`	Action name (e.g. `/navigate_to_pose`).
`action_type`	The ROS2 action class.
`goal`	A `<ActionType>.Goal()` instance.
`within`	Seconds to wait for a terminal result.
`feedback_cb`	Optional callback fired on each feedback message.

Raises

TimeoutError — no terminal result in within seconds.
AssertionError — goal rejected by action server.
RuntimeError — rclpy not initialised.

Returns

The completed ClientGoalHandle. Access .result() for the action-specific result.

Example

from nav2_msgs.action import NavigateToPose
from roboticks.assertions import assert_action_result

goal = NavigateToPose.Goal()
goal.pose.pose.position.x = 2.0

handle = assert_action_result(
    "/navigate_to_pose", NavigateToPose,
    goal, within=45.0,
)
assert handle.result().result.error_code == 0

`assert_param_equals`

Read a parameter from a node and assert its value.

Signature

def assert_param_equals(
    node_name: str,
    param: str,
    expected: Any,
    *,
    within: float = 2.0,
    node: rclpy.node.Node | None = None,
) -> None: ...

Parameter	Description
`node_name`	Fully-qualified node name (e.g. `/nav2_planner`).
`param`	Parameter name.
`expected`	Expected value. Comparison is `==`.
`within`	Seconds to wait for the param service to be available.

Raises

TimeoutError — parameter service unavailable.
AssertionError — parameter value doesn’t match expected.
RuntimeError — node doesn’t exist, parameter not declared.

Example

from roboticks.assertions import assert_param_equals

assert_param_equals("/nav2_planner", "use_sim_time", True)
assert_param_equals("/nav2_planner", "plugin_lib_names", ["GridBased"])

`assert_tf_transform`

Wait until a TF transform from source_frame to target_frame is available, then assert on its components.

Signature

def assert_tf_transform(
    source_frame: str,
    target_frame: str,
    *,
    within: float = 5.0,
    translation: tuple[float, float, float] | None = None,
    rotation: tuple[float, float, float, float] | None = None,
    tol: float = 0.01,
    node: rclpy.node.Node | None = None,
) -> TransformStamped: ...

Parameter	Description
`source_frame`	Source TF frame ID.
`target_frame`	Target TF frame ID.
`within`	Seconds to wait for the transform.
`translation`	Optional expected (x, y, z) — asserted within `tol`.
`rotation`	Optional expected quaternion (x, y, z, w) — asserted within `tol`.
`tol`	Tolerance for translation/rotation comparison.

Raises

TimeoutError — TF transform not available.
AssertionError — translation/rotation deviates beyond tol.

Returns

The geometry_msgs/msg/TransformStamped object.

Example

from roboticks.assertions import assert_tf_transform

tf = assert_tf_transform(
    "map", "base_link",
    within=5.0,
    translation=(0.0, 0.0, 0.0),
    tol=0.05,
)
print(tf.header.stamp)

Shared semantics

Concern	Behaviour
Node reuse	All helpers accept `node=` to share a long-lived node. Defaults to an ephemeral node created per-call.
Threading	Each helper internally spins on a `SingleThreadedExecutor` for the duration of the wait. Safe to call from pytest’s default sync runner.
rclpy.init	Caller must have called `rclpy.init()` (typical in a session-scoped fixture).
Cancellation	Test framework SIGINT (Ctrl-C) cleanly aborts the wait.
Logging	Each helper logs to the `roboticks.assertions` logger at `DEBUG`.

Fault injection

Pair assertions with fault injection for negative tests.

Writing tests in pytest

Worked examples that put these helpers in context.

MCAP capture

Record bag files when assertions fail.

Launch testing

System-test patterns that use these helpers across processes.

​Assertions

​assert_topic_published

​Signature

​Raises

​Returns

​Example

​assert_service_response

​Signature

​Raises

​Returns

​Example

​assert_action_result

​Signature

​Raises

​Returns

​Example

​assert_param_equals

​Signature

​Raises

​Example

​assert_tf_transform

​Signature

​Raises

​Returns

​Example

​Shared semantics

​Next

Fault injection

Writing tests in pytest

MCAP capture

Launch testing

Assertions

`assert_topic_published`

Signature

Raises

Returns

Example

`assert_service_response`

Signature

Raises

Returns

Example

`assert_action_result`

Signature

Raises

Returns

Example

`assert_param_equals`

Signature

Raises

Example

`assert_tf_transform`

Signature

Raises

Returns

Example

Shared semantics

Next