ROS2 easy-test

A Python test framework for ROS2 allowing for:

• simple and expressive assertions based on message and service interactions (black box testing)
• easy integration of existing nodes and launch files
• testing of nodes implemented in any programming language (C++, Python, etc.)
• works with and without tools like colcon test and pytest
• is minimalistic and has very few dependencies
• is typed and documented
• is tested, used in practice, indexed, and maintained

Installation

Note: The former PyPI package ros2-easy-test is now deprecated and will not be updated anymore. Thus, only use pip install ros2-easy-test for ROS versions prior to Kilted.

Just run (ROS2 Rolling and Kilted+):

rosdep install ros2_easy_test

Or add it to your package.xml (see the ROS2 docs):

<test_depend>ros2_easy_test</test_depend>

Examples

The following two examples demonstrate the usage of the Python decorators @with_single_node and @with_launch_file, which provide this package’s core functionality. To get a better understanding of their inner workings, please have a look at their implementation here. Besides the simple examples here, you can embed everything in unittest.TestCase as well. To check out how, have a look at the provided test in tests/demo/ for some advanced examples.

Testing a Node

In simple settings, where a single node is to be tested, the decorator @with_single_node can be used:

from ros2_easy_test import ROS2TestEnvironment, with_launch_file, with_single_node
from my_nodes import Talker
from std_msgs.msg import String

@with_single_node(Talker, watch_topics={"/chatter": String})
def test_simple_publisher(env: ROS2TestEnvironment) -> None:
    response: String = env.assert_message_published("/chatter", timeout=5)
    assert response.data == "Hello World: 0"

It is also possible to specify custom QoSProfiles for the ROS2TestEnvironment to use when subscribing or publishing on the specified topics. See tests/demo/latching_topic_test.py for an example. If no profile is specified, the default QoSProfile(history=QoSHistoryPolicy.KEEP_ALL) is used.

You can optionally provide more parameters to the test setting, i.e., additionally pass parameters={"some.thing": 30.2} to the decorator. The argument of the test function receiving the ROS2TestEnvironment must be named env.

Testing a Launch File

For more complex scenarios involving multiple nodes using a launch file (both nodes and launch file being implemented in any language supported by ROS2), the @with_launch_file decorator can be used.

@with_launch_file(
    "example_launch_file.yaml",
    watch_topics={"/some/interesting/response": ColorRGBA},
)
def test_simple_update_launch_file(env: ROS2TestEnvironment) -> None:
    env.publish("/topic/for/node_input", ColorRGBA(r=0.5, g=0.2, b=0.9, a=1.0))
    response_color = env.assert_message_published("/some/interesting/response")
    assert response_color.r == 0.5

You can also pass the literal launch file contents as a str instead of a path like "example_launch_file.yaml". The argument of the test function receiving the ROS2TestEnvironment must be named env.

Note that, however, this method is much slower than the one above. One reason for this is the requirement of a fixed warm-up time for the nodes to be started. This is because the test environment has to wait for the nodes to be ready before it can start listening for messages.

Usage

How you can interact with the node(s)

Using ROS2TestEnvironment, you can call:

• publish(topic: str, message: RosMessage) -> None
• listen_for_messages(topic: str, time_span: float) -> List[RosMessage]
• clear_messages(topic: str) -> None to forget all messages that have been received so far.
• call_service(name: str, request: Request, ...) -> Response
• send_action_goal(name: str, goal: Any, ...) -> Tuple[ClientGoalHandle, List[FeedbackMsg]]
• send_action_goal_and_wait_for_result(name: str, goal: Any, ...) -> Tuple[List[FeedbackMsg], ResultMsg]

Note that a ROS2TestEnvironment is a normal rclpy.node.Node and thus has all the methods of any other ROS2 node. So feel free to offer a service with env.create_service() and cover more specific use cases. Extend as you please!

File truncated at 100 lines see the full file