Add ros_launch_test bazel rule

Author: Francisco Rossi

bazel_ros2_rules/lib/extensions.bzl

@@ -34,6 +34,8 @@ def _local_ros2_implementation(module_ctx):
         include_packages += [
             "action_msgs",
             "builtin_interfaces",
+            "launch_testing",
+            "launch_testing_ros",
             "ros2cli_common_extensions",
             "ros2cli",
             "rosidl_default_generators",

bazel_ros2_rules/lib/private/ros_py.bzl

@@ -246,6 +246,110 @@ def ros_launch(
         **kwargs
     )
 
+_LAUNCH_TEST_MAIN_TEMPLATE = """\
+import os
+import sys
+
+from python.runfiles import runfiles as runfiles_api
+
+assert __name__ == "__main__"
+runfiles = runfiles_api.Create()
+launch_test_file = runfiles.Rlocation({launch_test_respath})
+
+# This shim only ever runs inside a Bazel test sandbox. The sandbox sets
+# HOME=/does_not_exist (read-only) and the project .bazelrc sets ROS_HOME to
+# a similar sentinel. RCL log-dir priority: ROS_LOG_DIR > $ROS_HOME/log >
+# $HOME/.ros/log. Redirect all three unconditionally to the per-test tmpdir.
+_tmpdir = os.environ.get("TEST_TMPDIR", "/tmp")
+os.environ["ROS_LOG_DIR"] = _tmpdir
+os.environ["ROS_HOME"] = _tmpdir
+os.environ["HOME"] = _tmpdir
+
+sys.argv = ["launch_test", launch_test_file] + sys.argv[1:]
+
+from launch_testing.launch_test import main
+sys.exit(main())
+"""
+
+def ros_launch_test(
+        name,
+        launch_test_file,
+        workspace_name = None,
+        executables = [],
+        share = [],
+        data = [],
+        deps = [],
+        rmw_implementation = None,
+        isolate = True,
+        **kwargs):
+    """
+    Builds a launch_testing test and wraps it with a shim that will inject the
+    minimal runtime environment necessary for execution when depending on
+    targets from this ROS 2 local repository.
+
+    The test file must export generate_test_description() and any number of
+    unittest.TestCase subclasses following the launch_testing convention.
+
+    Args:
+        name: test target name
+        launch_test_file: label of the launch_testing test file
+        workspace_name: optional ament package name override (defaults to
+            the Bzlmod module name)
+        executables: executable targets to register in the ament index,
+            enabling Node(package=<pkg>, executable=...) lookups
+        share: data file targets to register under share/<pkg>/, enabling
+            FindPackageShare(<pkg>) and get_package_share_directory(<pkg>)
+        data: additional runtime data deps
+        deps: additional Python deps beyond launch_testing and launch
+        rmw_implementation: optional RMW implementation to run against
+        isolate: whether to use network namespace isolation (default True)
+
+    Additional keyword arguments are forwarded to ros_py_test.
+    """
+    main = "{}_launch_test_main.py".format(name)
+    launch_test_respath = _make_respath(launch_test_file, workspace_name)
+
+    generate_file(
+        name = main,
+        content = _LAUNCH_TEST_MAIN_TEMPLATE.format(
+            launch_test_respath = repr(launch_test_respath),
+        ),
+        visibility = ["//visibility:private"],
+        testonly = True,
+    )
+
+    if executables or share:
+        package_name = workspace_name if workspace_name != None else native.module_name()
+        index_target = "_{}_ament_index".format(name)
+        ament_index_share_files(
+            name = index_target,
+            package_name = package_name,
+            executables = executables,
+            srcs = share,
+            visibility = ["//visibility:private"],
+        )
+        data = data + executables + share + [":" + index_target]
+
+    tags = list(kwargs.pop("tags", []))
+    if "nolint" not in tags:
+        tags.append("nolint")
+
+    ros_py_test(
+        name = name,
+        srcs = [main],
+        main = main,
+        data = data + [launch_test_file],
+        deps = deps + [
+            "@bazel_ros2_rules//deps/python/runfiles",
+            "@ros2//:launch_testing_py",
+            "@ros2//:launch_py",
+        ],
+        rmw_implementation = rmw_implementation,
+        isolate = isolate,
+        tags = tags,
+        **kwargs
+    )
+
 def ros_py_test(
         name,
         rmw_implementation = None,

ros2_example_bazel_installed/ros2_example_apps/BUILD.bazel

@@ -2,7 +2,7 @@
 # vi: set ft=python :
 
 load("@ros2//:ros_cc.bzl", "ros_cc_binary", "ros_cc_test")
-load("@ros2//:ros_py.bzl", "ros_launch", "ros_py_binary", "ros_py_test")
+load("@ros2//:ros_py.bzl", "ros_launch", "ros_launch_test", "ros_py_binary", "ros_py_test")
 load("@ros2//:rosidl.bzl", "rosidl_interfaces_group")
 load("@rules_cc//cc:cc_library.bzl", "cc_library")
 load("//tools:cmd_test.bzl", "cmd_test")
@@ -259,6 +259,35 @@ ros_launch(
     workspace_name = workspace_name,
 )
 
+ros_launch_test(
+    name = "roslaunch_eg_launch_test",
+    size = "medium",
+    # roslaunch_eg_py carries eg_talker and eg_listener in its ament index
+    # (via executables= on its ros_launch declaration). Listing it in data=
+    # propagates those indexes into AMENT_PREFIX_PATH at test runtime.
+    data = [
+        ":roslaunch_eg_py",
+    ],
+    launch_test_file = "test/roslaunch_eg_launch_test.py",
+    rmw_implementation = "rmw_cyclonedds_cpp",
+)
+
+ros_launch_test(
+    name = "roslaunch_eg_node_launch_test",
+    size = "medium",
+    # roslaunch_eg_py registers eg_talker and eg_listener in the ament index
+    # under package "ros2_example_bazel_installed", making them discoverable
+    # by launch_ros.actions.Node without an explicit Rlocation call.
+    data = [
+        ":roslaunch_eg_py",
+    ],
+    launch_test_file = "test/roslaunch_eg_node_launch_test.py",
+    rmw_implementation = "rmw_cyclonedds_cpp",
+    deps = [
+        "@ros2//:launch_ros_py",
+    ],
+)
+
 ros_py_test(
     name = "roslaunch_eg_test",
     srcs = ["test/roslaunch_eg_test.py"],

ros2_example_bazel_installed/ros2_example_apps/test/roslaunch_eg_launch_test.py

@@ -0,0 +1,64 @@
+# -*- python -*-
+# Copyright 2024 Open Source Robotics Foundation
+
+"""launch_testing example: verify talker/listener exchange via ros_launch_test.
+
+This test spins up the eg_talker and eg_listener from the roslaunch_eg_py
+launch binary, then checks exit codes in a post-shutdown test.
+
+To run via Bazel:
+    bazel test //ros2_example_apps:roslaunch_eg_launch_test
+"""
+
+import unittest
+
+import launch
+import launch.actions
+import launch.event_handlers
+import launch_testing
+import launch_testing.actions
+import launch_testing.asserts
+import launch_testing.markers
+from python.runfiles import runfiles as runfiles_api
+
+
+@launch_testing.markers.keep_alive
+def generate_test_description():
+    r = runfiles_api.Create()
+
+    # Locate the ros_launch binary via Bazel runfiles.
+    # The binary already carries eg_talker and eg_listener in its ament index
+    # because roslaunch_eg_py was declared with executables=[":eg_listener",
+    # ":eg_talker"]. No additional ament registration is needed here.
+    launch_bin = r.Rlocation(
+        "ros2_example_bazel_installed/ros2_example_apps/roslaunch_eg_py"
+    )
+
+    launch_action = launch.actions.ExecuteProcess(
+        cmd=[launch_bin],
+        output="screen",
+    )
+
+    # Fire ReadyToTest() only after roslaunch_eg_py exits on its own.
+    # This prevents launch_testing from sending SIGINT before the talker and
+    # listener have finished exchanging their 10 messages.
+    return (
+        launch.LaunchDescription(
+            [
+                launch_action,
+                launch.actions.RegisterEventHandler(
+                    launch.event_handlers.OnProcessExit(
+                        target_action=launch_action,
+                        on_exit=[launch_testing.actions.ReadyToTest()],
+                    )
+                ),
+            ]
+        ),
+        {"launch_proc": launch_action},
+    )
+
+
+@launch_testing.post_shutdown_test()
+class TestExitCode(unittest.TestCase):
+    def test_launch_exit_code(self, launch_proc, proc_info):
+        launch_testing.asserts.assertExitCodes(proc_info, [0], launch_proc)

ros2_example_bazel_installed/ros2_example_apps/test/roslaunch_eg_node_launch_test.py

@@ -0,0 +1,69 @@
+# -*- python -*-
+# Copyright 2024 Open Source Robotics Foundation
+
+"""launch_testing Pattern B example: launch nodes via launch_ros.actions.Node.
+
+Unlike roslaunch_eg_launch_test.py (which locates the binary via Rlocation),
+this test refers to nodes by their ament package/executable name.  The ament
+index entries for eg_talker and eg_listener are made visible at test runtime
+because roslaunch_eg_py (which registered them via executables=) is listed in
+the data= of the ros_launch_test target.
+
+To run via Bazel:
+    bazel test //ros2_example_apps:roslaunch_eg_node_launch_test
+"""
+
+import unittest
+
+import launch
+import launch.actions
+import launch.event_handlers
+from launch_ros.actions import Node
+import launch_testing
+import launch_testing.actions
+import launch_testing.asserts
+import launch_testing.markers
+
+
+@launch_testing.markers.keep_alive
+def generate_test_description():
+    talker = Node(
+        package="ros2_example_bazel_installed",
+        executable="eg_talker",
+        output="screen",
+    )
+    listener = Node(
+        package="ros2_example_bazel_installed",
+        executable="eg_listener",
+        output="screen",
+    )
+
+    # The listener exits first (transient-local QoS lets it buffer all 10
+    # messages quickly).  Fire ReadyToTest() from the talker's OnProcessExit
+    # so the transition to post-shutdown tests happens only after both nodes
+    # have exited cleanly.  keep_alive prevents LaunchService from
+    # auto-shutting down when the first node exits.
+    return (
+        launch.LaunchDescription(
+            [
+                talker,
+                listener,
+                launch.actions.RegisterEventHandler(
+                    launch.event_handlers.OnProcessExit(
+                        target_action=talker,
+                        on_exit=[launch_testing.actions.ReadyToTest()],
+                    )
+                ),
+            ]
+        ),
+        {"talker": talker, "listener": listener},
+    )
+
+
+@launch_testing.post_shutdown_test()
+class TestExitCodes(unittest.TestCase):
+    def test_talker_exit_code(self, talker, proc_info):
+        launch_testing.asserts.assertExitCodes(proc_info, [0], talker)
+
+    def test_listener_exit_code(self, listener, proc_info):
+        launch_testing.asserts.assertExitCodes(proc_info, [0], listener)