Testing guidelines

Unit testing ROS 2 code is no different from unit testing any other software, but some care must be exercised.

Considerations for unit testing

• Data transport in ROS 2 is non-deterministic. So is callback scheduling when multi-threaded executors (as the one instantiated by process-wide APIs by default) are in place. As such, time sensitive and execution order dependent tests are bound to fail even if only sporadically. A multi-threaded executor spinning in a background thread, as provided by bdai_ros_wrappers.scope functionality, enables synchronization primitives as a mechanism to avoid these issues.
• ROS 2 middlewares perform peer discovery by default. This allows distributed architectures in production but leads to cross-talk during parallelized testing. domain_coordinator functionality simplifies ROS domain ID assignment enforcing host-wide uniqueness and with it, middleware isolation.

Therefore, as rules of thumb consider:

1. Using bdai_ros2_wrappers.scope.top to setup ROS 2 in your test fixtures.
   • Isolate it by passing a unique domain ID, as provided by domain_coordinator.domain_id.
2. Using synchronization primitives to wait with timeouts.
   • Note timeouts make the test time sensitive. Pick timeouts an order of magnitude above the expected test timing.

Writing unit tests using pytest (recommended)

pytest is a testing framework for Python software, the most common in ROS 2 Python codebases.

import bdai_ros2_wrappers.scope as ros_scope
from bdai_ros2_wrappers.scope import ROSAwareScope
from bdai_ros2_wrappers.subscription import wait_for_message

import domain_coordinator
import pytest

import std_msgs.msg

@pytest.fixture
def ros() -> Iterator[ROSAwareScope]:
    """
    A pytest fixture that will set up and yield a ROS 2 aware global scope to each test that requests it.

    See https://docs.pytest.org/en/7.4.x/fixture.html for a primer on pytest fixtures.
    """
    with domain_coordinator.domain_id() as domain_id:  # to ensure node isolation 
        with ros_scope.top(global_=True, namespace="fixture", domain_id=domain_id) as top:
            yield top

def test_pub_sub(ros: ROSAwareScope) -> None:
    """Asserts that a published message can be received on the other end."""
    pub = ros.node.create_publisher(std_msgs.msg.String, "test", 1)
    pub.publish(std_msgs.msg.String(data="test"))
    # Message will arrive at an unspecified point in the future, thus
    assert wait_for_message(std_msgs.msg.String, "test", timeout_sec=5.0)

Writing unit tests using unittest

unittest is the testing framework in Python's standard library.

import bdai_ros2_wrappers.scope as ros_scope
from bdai_ros2_wrappers.subscription import wait_for_message

import contextlib
import domain_coordinator
import unittest

import std_msgs.msg

class TestCase(unittest.TestCase):

    def setUp(self) -> None:
        """Sets up an isolated ROS 2 aware scope for all tests in the test case."""
        self.fixture = contextlib.ExitStack()
        domain_id = self.fixture.enter_context(domain_coordinator.domain_id())
        self.ros = self.fixture.enter_context(ros_scope.top(
            global_=True, namespace="fixture", domain_id=domain_id
        ))

    def tearDown(self) -> None:
        self.fixture.close()  # exits all contexts

    def test_pub_sub(self) -> None:
        """Asserts that a published message can be received on the other end."""
        pub = self.ros.node.create_publisher(std_msgs.msg.String, "test", 1)
        pub.publish(std_msgs.msg.String(data="test"))
        self.assertIsNotNone(wait_for_message(std_msgs.msg.String, "test", timeout_sec=5.0))

Adding unit tests to a package

A package's type and build system dictate how unit tests are to be added. Unit tests for ROS 2 packages are typically hosted under the test subdirectory, so the following assumes this convention is observed.

For ament_cmake packages, the CMakeLists.txt file should have:

if(BUILD_TESTING)
    find_package(ament_cmake_pytest REQUIRED)
    # Define an arbitrary target for your tests such as:
    ament_add_pytest_test(unit_tests test)
endif()

For ament_python packages, the setup.py file should have:

setup(
    # ...
    tests_require=['pytest'],
)