> ## Documentation Index
> Fetch the complete documentation index at: https://docs.roboticks.io/llms.txt
> Use this file to discover all available pages before exploring further.
# Assertions
> Full reference for the rclpy-aware assertion helpers - assert_topic_published, assert_service_response, assert_action_result, assert_param_equals, assert_tf_transform.
# 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.
```python theme={null}
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
```python theme={null}
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
```python theme={null}
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
```python theme={null}
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
```python theme={null}
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
```python theme={null}
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 `.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
```python theme={null}
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
```python theme={null}
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
```python theme={null}
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
```python theme={null}
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
```python theme={null}
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`. |
## Next
Pair assertions with fault injection for negative tests.
Worked examples that put these helpers in context.
Record bag files when assertions fail.
System-test patterns that use these helpers across processes.