ROS2►核心

ros_testing 源码详细分析

ros_testing 源码详细分析

工作区路径：/home/cp/work2/ros2Learn/ros2_humble/src/ros2/ros_testing
版本：0.4.0（Humble），子包 2 个，许可证 Apache 2.0。

ros_testing 仓库体量很小（源码文件约 16 个），本身几乎不包含测试框架逻辑。它的定位是 ROS 2 集成测试的 统一入口（metapackage）：把 launch_testing 生态、ros2 test CLI 与 CMake add_ros_test() 串成一条链路，供涉及 Node / DDS / launch 的包做 colcon test 集成测试。

重要区分：真正的 launch 集成测试框架在 launch 与 launch_ros 仓库的 launch_testing / launch_testing_ros 包中。ros_testing 是对 ROS 场景的薄封装与依赖聚合。

1. 总体认识

1.1 核心职责

能力 实现位置 说明
CMake 注册 launch 测试 ros_testing/cmake/add_ros_test.cmake 包装 ament_add_test，命令改为 ros2 test
依赖聚合 ros_testing/package.xml 导出 launch_testing* + ros2test，下游 test_depend 一处搞定
CLI 入口 ros2test/command/test.py ros2 test 子命令，带 ROS 域隔离
ROS 专用 runner 委托 launch_testing_ros.LaunchTestRunner 与 launch_test 相同，预留 ROS 扩展点

1.2 在 ROS 2 测试栈中的位置

被测包 CMakeLists.txtros_testing 仓库launch 生态基础设施ros2 test ...add_ros_test\ntest/foo.pyros_testing\nCMake extrasros2test\nros2 test CLIlaunch_testing_ament_cmake\nparse_launch_test_argumentslaunch_testing\nLaunchTestRunnerlaunch_testing_ros\nWaitForTopics 等colcon test / ctestdomain_coordinator\nROS_DOMAIN_ID
对比项 ros_testing launch_testing_ament_cmake launch_testing
角色 ROS 集成测试 入口包 通用 CMake 集成 Python 测试 引擎
CMake 函数 add_ros_test() add_launch_test() —
运行命令 ros2 test python -m launch_testing.launch_test 被上述两者调用
域隔离 有（domain_coordinator） 无 无
适用场景 涉及 ROS Node 的包 任意 launch 测试（含非 ROS） 框架本体

1.3 与 ROS 1 rostest 的关系

README 仍写 “rostest”，CMake 宏里保留 rostest__strip_prefix 命名——这是从 ROS 1 rostest 移植 CMake 辅助逻辑时的历史痕迹。ROS 2 中：

  • 不再有 rostest 包与 test_rostest 宏；
  • 等价能力 = launch_testing + launch_testing_ros + ros_testing；
  • 测试文件是 Python launch test（*_launch_test.py 或 test_*.py），不是 .test XML。

2. 仓库结构

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
ros_testing/
├── README.md                          # 一行说明：Node 测试统一入口
├── LICENSE
├── ros_testing/                       # ★ CMake 聚合包（无编译产物）
│   ├── package.xml
│   ├── CMakeLists.txt                 # 仅 install cmake/ + ament_package
│   ├── ros_testing-extras.cmake       # include add_ros_test.cmake
│   ├── cmake/add_ros_test.cmake       # 唯一核心 CMake 逻辑（~82 行）
│   └── CHANGELOG.rst
└── ros2test/                          # ★ ros2cli 扩展
    ├── package.xml
    ├── setup.py                       # entry_point: ros2cli.command
    ├── ros2test/command/test.py       # TestCommand（~51 行）
    ├── pytest.ini
    └── test/                          # ament  lint 自检
包 构建类型 源码规模 职责
ros_testing ament_cmake（project(... NONE)） ~82 行 CMake 导出 add_ros_test、传递依赖
ros2test ament_python ~51 行业务逻辑 ros2 test 命令

3. 包 ros_testing：CMake 入口

3.1 CMakeLists.txt

1
2
3
4
project(ros_testing NONE)
find_package(ament_cmake_core REQUIRED)
ament_package(CONFIG_EXTRAS "${PROJECT_NAME}-extras.cmake")
install(DIRECTORY cmake DESTINATION share/${PROJECT_NAME})

特点：project(... NONE) 表示 不编译任何目标，纯配置/安装包。安装后下游通过 find_package(ros_testing REQUIRED) 获得 add_ros_test()。

3.2 ros_testing-extras.cmake

1
2
find_package(launch_testing_ament_cmake REQUIRED)
include("${ros_testing_DIR}/add_ros_test.cmake")

依赖链：ros_testing → launch_testing_ament_cmake（复用其 parse_launch_test_arguments 宏）。

3.3 add_ros_test() — 核心逻辑

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
function(add_ros_test filename)
  parse_launch_test_arguments(_ros_test ${filename} ${ARGN})

  set(cmd
    "ros2"
    "test"
    "${_ros_test_FILE_NAME}"
    "${_ros_test_ARGS}"
    "--junit-xml=${_ros_test_RESULT_FILE}"
    "--package-name=${PROJECT_NAME}"
  )

  ament_add_test(
    "${_ros_test_TARGET}"
    COMMAND ${cmd}
    OUTPUT_FILE "${CMAKE_BINARY_DIR}/ros_test/${_ros_test_TARGET}.txt"
    RESULT_FILE "${_ros_test_RESULT_FILE}"
    TIMEOUT "${_ros_test_TIMEOUT}"
    ${_ros_test_UNPARSED_ARGUMENTS}
  )
endfunction()

与 add_launch_test() 的 唯一实质差异 是 COMMAND：

函数 执行命令
add_launch_test ${PYTHON_EXECUTABLE} -m launch_testing.launch_test ...
add_ros_test ros2 test ...

其余（TARGET 推导、TIMEOUT 默认 60s、xUnit 路径、ARGS 传递）均来自共享宏 parse_launch_test_arguments。

parse_launch_test_arguments 行为摘要

（定义于 launch_testing_ament_cmake/cmake/add_launch_test.cmake）

参数 默认 说明
TIMEOUT 60 ctest 超时秒数
TARGET 由文件路径生成 去掉 PROJECT_SOURCE_DIR 前缀，/ → _
ARGS — 传给 launch 的 name:=value 参数
LABELS launch_test ctest 标签（add_ros_test 未显式设置，继承 ament_add_test 默认）
RESULT_FILE $AMENT_TEST_RESULTS_DIR/$PROJECT_NAME/$TARGET.xunit.xml JUnit 报告

3.4 package.xml 依赖设计

1
2
3
4
5
6
7
<buildtool_export_depend>launch_testing_ament_cmake</buildtool_export_depend>
<buildtool_export_depend>ros2test</buildtool_export_depend>
<build_export_depend>launch_testing</build_export_depend>
<build_export_depend>launch_testing_ros</build_export_depend>
<exec_depend>launch_testing</exec_depend>
<exec_depend>launch_testing_ros</exec_depend>
<exec_depend>ros2test</exec_depend>

下游包只需：

1
<test_depend>ros_testing</test_depend>
1
2
find_package(ros_testing REQUIRED)
add_ros_test(test/test_lifecycle.py TIMEOUT 60)

即可在 colcon test 时拉起完整 ROS launch 测试链。不必单独声明 launch_testing_ament_cmake / ros2test（但直接依赖 launch_testing_ament_cmake 也常见，见下文）。

4. 包 ros2test：ros2 test CLI

4.1 注册方式

setup.py：

1
2
3
4
5
entry_points={
    'ros2cli.command': [
        'test = ros2test.command.test:TestCommand',
    ],
}

安装后可用：

1
2
3
4
ros2 test /path/to/test_lifecycle.py
ros2 test --show-args test/foo.py
ros2 test test/foo.py arg1:=value --verbose
ros2 test test/foo.py --disable-isolation

4.2 TestCommand 完整流程

1
2
3
4
5
6
7
8
9
10
11
12
13
class TestCommand(CommandExtension):
    def add_arguments(self, parser, cli_name):
        launch_testing.launch_test.add_arguments(parser)
        parser.add_argument('--disable-isolation', ...)

    def main(self, *, parser, args):
        with contextlib.ExitStack() as stack:
            if 'ROS_DOMAIN_ID' not in os.environ and not args.disable_isolation:
                domain_id = stack.enter_context(domain_coordinator.domain_id())
                os.environ['ROS_DOMAIN_ID'] = str(domain_id)
            return launch_testing.launch_test.run(
                parser, args, test_runner_cls=launch_testing_ros.LaunchTestRunner
            )

两步增强（相对 python -m launch_testing.launch_test）：

  1. ROS_DOMAIN_ID 自动隔离

    • 若环境未设置 ROS_DOMAIN_ID 且未 --disable-isolation，通过 domain_coordinator.domain_id() 上下文管理器选取 未被占用 的 domain id（基于端口协调，避免并行 colcon test 互相发现）。
    • 若用户已设置 ROS_DOMAIN_ID，则 尊重 用户值。
    • --disable-isolation 关闭自动选取（与 CI 固定 domain 场景兼容）。

  2. 指定 LaunchTestRunner

    • Humble 上 launch_testing_ros.LaunchTestRunner 继承基类且 无额外 override——占位以便将来注入 ROS 专用行为。
    • 实际 ROS 辅助工具（WaitForTopics 等）在测试文件中 显式 import 使用。

4.3 依赖

依赖 用途
ros2cli CommandExtension 基类
launch_testing add_arguments / run
launch_testing_ros LaunchTestRunner
launch / launch_ros 间接（launch 文件常用）
domain_coordinator 并行测试 domain 协调

4.4 历史变更（CHANGELOG）

  • 0.1.0：引入 ros_testing metapackage + ros2test
  • --disable-isolation 从 launch_testing 迁至 ros2test（避免非 ROS 场景误用）
  • domain_coordinator API 改为 context manager（with domain_coordinator.domain_id()）

5. 端到端执行流程

5.1 从 colcon test 到断言

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
colcon test --packages-select lifecycle
  └── ctest → add_ros_test 注册的目标
        └── ros2 test /path/to/test_lifecycle.py --junit-xml=... --package-name=lifecycle
              ├── [可选] domain_coordinator 分配 ROS_DOMAIN_ID
              └── launch_testing.launch_test.run(..., LaunchTestRunner)
                    ├── importlib 加载测试 Python 文件
                    ├── LoadTestsFromPythonModule → 解析 generate_test_description / TestCase
                    ├── LaunchTestRunner.validate()
                    └── LaunchTestRunner.run()
                          ├── LaunchService 启动 LaunchDescription 中的 Node/Process
                          ├── ReadyToTest → 启动 pre-shutdown unittest（后台线程）
                          ├── 测试可通过 proc_info / proc_output / rclpy 与运行中进程交互
                          ├── 关闭 launch
                          └── post_shutdown_test 类运行（检查 exit code 等）
                    └── 写 xUnit XML，返回 exit code

5.2 Launch 测试文件契约

测试文件必须提供：

元素 要求
generate_test_description() 返回 LaunchDescription 或 (LaunchDescription, dict)
launch_testing.actions.ReadyToTest() 必须出现在 LD 中，通知框架可开始 active 测试
unittest.TestCase 子类 pre-shutdown 测试（与 launch 并发）
@launch_testing.post_shutdown_test() 可选，shutdown 后测试

可选第二返回值（context dict）将 action 对象注入测试方法参数，例如：

1
2
return launch.LaunchDescription([...]), {'talker_node': talker_node}
# 测试方法签名：def test_foo(self, proc_output, talker_node): ...

5.3 测试阶段划分

Post-shutdown TestsPre-shutdown TestsLaunchServicePost-shutdown TestsPre-shutdown TestsLaunchServicepar[并发]启动 Node / ExecuteProcessReadyToTest 触发assertWaitFor / rclpy 订阅进程运行关闭所有进程assertExitCodes 等

注入对象（launch_testing 自动绑定）：

  • proc_info — 进程 exit code、运行状态
  • proc_output — stdout/stderr 捕获
  • launch_service — 运行中可动态发 Event（仅 pre-shutdown）
  • test_args — CLI 传入的 key:=value launch 参数

6. 下游使用示例（lifecycle demo）

demos/lifecycle/CMakeLists.txt：

1
2
3
4
5
find_package(ros_testing REQUIRED)
add_ros_test(
  test/test_lifecycle.py
  TIMEOUT 60
)

test/test_lifecycle.py 要点：

  1. generate_test_description()

    • 启动 LifecycleNode + 普通 Node
    • 用 RegisterEventHandler + OnStateTransition 驱动 lifecycle 状态机
    • 末尾 ReadyToTest()

  2. TestLifecyclePubSub（pre-shutdown）

    • proc_output.assertWaitFor('on_configure() is called', ...)
    • 正则匹配 listener 收到的消息

  3. TestLifecyclePubSubAfterShutdown（@post_shutdown_test）

    • launch_testing.asserts.assertExitCodes(proc_info, process=talker_node)

手动运行等价命令：

1
2
3
4
ros2 test src/ros2/demos/lifecycle/test/test_lifecycle.py
# 或
launch_test src/ros2/demos/lifecycle/test/test_lifecycle.py
# （后者无自动 ROS_DOMAIN_ID 隔离）

7. 关联包详解（框架本体，非本仓库）

理解 ros_testing 必须熟悉以下包：

7.1 launch_testing（launch 仓库）

模块 职责
launch_test.py CLI：run() 加载模块、写 JUnit
test_runner.py LaunchTestRunner：线程模型、LaunchService 生命周期
loader.py 从 Python 模块加载 TestRun
asserts.py assertExitCodes、assertInStdout 等
proc_info_handler.py / io_handler.py 进程信息与 I/O 捕获

线程约束：LaunchService.run() 必须在主线程；pre-shutdown 测试在 后台线程 与 launch 并发（见 launch/issues/126）。

7.2 launch_testing_ros（launch_ros 仓库）

工具 用途
WaitForTopics 阻塞直到指定 topic 收到消息
DataRepublisher 订阅→修改→再发布（fuzz 测试）
MessagePump 异步消息泵
launch_testing_ros/tools 输出解析等
pytest/hooks.py @pytest.mark.rostest 与 pytest 集成

Pytest 集成：launch_testing_ros_pytest_entrypoint.py 注册 rostest marker，使 pytest 收集 launch test 模块。

7.3 launch_testing_ament_cmake

提供 add_launch_test() 与 parse_launch_test_arguments。大量系统测试包 直接依赖 此包而非 ros_testing，例如：

  • system_tests/test_communication
  • demos/demo_nodes_cpp
  • rcl/rcl/test

选择建议：

场景 推荐
测试涉及 ROS Node、需并行 domain 隔离 test_depend ros_testing + add_ros_test
纯 launch/process 测试、CI 固定 domain launch_testing_ament_cmake + add_launch_test
本地快速调试 ros2 test 或 launch_test

7.4 domain_coordinator（ament_cmake_ros 仓库）

1
2
with domain_coordinator.domain_id() as domain_id:
    os.environ['ROS_DOMAIN_ID'] = str(domain_id)

通过端口锁协调，保证多个测试进程不共用同一 domain。仅 ros2 test 默认启用。

8. add_ros_test vs add_launch_test 对照

1
2
3
4
5
6
7
# ros_testing — ROS 场景推荐
find_package(ros_testing REQUIRED)
add_ros_test(test/my_test.py TIMEOUT 120 ARGS "param:=value")

# launch_testing_ament_cmake — 通用
find_package(launch_testing_ament_cmake REQUIRED)
add_launch_test(test/my_test.py TIMEOUT 120)
维度 add_ros_test add_launch_test
命令 ros2 test python -m launch_testing.launch_test
Domain 隔离 默认有 无
输出目录 build/ros_test/ build/launch_test/
xUnit 相同路径规则 相同
依赖包 ros_testing（更重） launch_testing_ament_cmake（更轻）

9. 常见断言与 ROS 交互模式

9.1 仅看进程输出（无 rclpy）

1
2
proc_output.assertWaitFor('expected string', process=node_action, timeout=5)
launch_testing.asserts.assertExitCodes(proc_info, process=node_action)

9.2 测试中启动 rclpy 节点

1
2
3
4
5
6
7
8
9
10
11
class TestFoo(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        rclpy.init()
    @classmethod
    def tearDownClass(cls):
        rclpy.shutdown()

    def test_sub(self):
        node = rclpy.create_node('test')
        # create_subscription / spin_until_future_complete

见 launch_testing_ros/test/examples/talker_listener_launch_test.py。

9.3 WaitForTopics

1
2
3
from launch_testing_ros import WaitForTopics
with WaitForTopics([('/chatter', String)], timeout=5.0):
    ...

10. 与 ros2cli 的循环依赖说明

CHANGELOG 记录：ros2topic 等包曾依赖 ros_testing，后改为 直接依赖 launch 包 以避免与 ros2cli ↔ ros_testing ↔ ros2cli 的循环。

结论：

  • ros2test 依赖 ros2cli，但 ros2cli 各插件不应 test_depend ros_testing；
  • 需要 launch 测试的 库/节点包 才 test_depend ros_testing。

11. 限制与注意事项

topic 说明
LaunchService 线程 pre-shutdown 测试中勿阻塞主线程式调用 LaunchService.run
Domain 并行测试务必用 ros2 test 或自行协调 ROS_DOMAIN_ID
超时 默认 60s，复杂 lifecycle/导航测试需显式 TIMEOUT
Keyed topic discovery 等内部 topic 与 RMW 能力相关；测试文件需自洽
Windows parse_launch_test_arguments 含 Debug Python 可执行文件分支
框架代码位置 调试断言/ runner 问题应查 launch_testing，不是 ros_testing

12. 调试建议

  1. 单测文件手动跑：
    ros2 test path/to/test.py --verbose

  2. 查看 launch 参数：
    ros2 test --show-args path/to/test.py

  3. 隔离 domain 问题：
    对比 ros2 test 与 launch_test（后者不自动分配 domain）

  4. 看 stdout 日志：
    build/ros_test/<target>.txt（add_ros_test）或 build/launch_test/（add_launch_test）

  5. JUnit：
    build/test_results/<package>/<target>.xunit.xml

  6. 并行 ctest 失败且表现为 discovery 串扰：
    确认未 --disable-isolation 且未多个测试强制同一 ROS_DOMAIN_ID

13. 推荐阅读顺序

  1. 本仓库 add_ros_test.cmake — 理解入口仅一行命令差异
  2. ros2test/command/test.py — domain 隔离 + runner 选择
  3. launch_testing/README.md — launch test 文件契约与断言 API
  4. launch_testing/launch_test.py — run() 加载与 JUnit
  5. launch_testing/test_runner.py — 线程与 LaunchService 生命周期
  6. demos/lifecycle/test/test_lifecycle.py — 真实 ROS 集成测试样例
  7. launch_testing_ros/test/examples/ — talker/listener、WaitForTopics
  8. launch_testing_ament_cmake/cmake/add_launch_test.cmake — CMake 宏细节

14. 小结

ros_testing 仓库是 ROS 2 集成测试的薄入口层，价值在于 聚合依赖 与 统一 CLI：

  • ros_testing 包：导出 add_ros_test()，把 ctest 接到 ros2 test；
  • ros2test 包：在 launch_testing.launch_test.run 外包一层 ROS_DOMAIN_ID 隔离；
  • 测试框架本体在 launch_testing / launch_testing_ros，Launch 文件 + unittest 两阶段（running / post-shutdown）模型与 ROS 1 rostest 哲学相似，但实现完全基于 launch 系统。

编写涉及 Node 的包测试时，推荐 test_depend ros_testing + add_ros_test()；若仅需 launch 进程测试或要避免引入 ros2cli 链，可直接使用 launch_testing_ament_cmake 的 add_launch_test()。两种方式的 测试 Python 文件格式相同，差异几乎只在 谁执行命令 与 是否自动隔离 DDS domain。

文章互动

阅读 --

留言

0 条留言

正在加载留言…