ROS2_learn/doc/90-testing.md

90 · 测试策略(完全指南)

目标:理解 ROS2 三层测试金字塔,会用 gtest / pytest / colcon test,能把本仓库的测试策略用到自己的项目。

---

目录

• 1. 测试金字塔
• 2. C++ gtest 完整指南
• 3. Python pytest 完整指南
• 4. launch_testing(集成测试 launch 文件)
• 5. 本仓库测试策略详解
• 6. 端到端自动化(进阶)
• 7. 标记 / 跳过 / 覆盖率
• 8. CI 集成(GitHub Actions)
• 9. 在本仓库里跑
• 10. 进阶:Mock DDS / 时间注入 / Fault Injection

---

1. 测试金字塔

       ┌────────────────┐
       │   端到端 E2E    │   ← launch_testing + 真实 launch
       │  (慢 / 1-2 个)  │
       ├────────────────┤
       │  集成 in-process │   ← 同进程 spin + DDS 单跑
       │   (中等 / 4-6)   │
       ├────────────────┤
       │   单元 Unit     │   ← gtest / pytest 单函数
       │  (快 / 大量)     │
       └────────────────┘

层	跑在哪	速度	数量
单元	容器内 / 本机 venv	秒级	几十 ~ 几百
集成 in-process	容器内(colcon test)	秒 ~ 分钟	几个 ~ 几十
端到端 E2E	launch + 真实 DDS	分钟级	1-5 个核心场景

本仓库:

• 单元 + 集成:colcon test 一键跑(10 用例,秒级)
• 端到端:固化 5 个 docker/*_e2e.log 手动验证

---

2. C++ gtest 完整指南

2.1 测试文件

#include "gtest/gtest.h"
#include "rclcpp/rclcpp.hpp"
#include "std_msgs/msg/string.hpp"

class PubsubTest : public ::testing::Test {
protected:
    static void SetUpTestSuite() {
        rclcpp::init(0, nullptr);    // 同进程共享一个 context
    }
    static void TearDownTestSuite() {
        rclcpp::shutdown();
    }
};

TEST_F(PubsubTest, MessageDataFieldIsNonEmpty) {
    std_msgs::msg::String msg;
    msg.data = "hi";
    EXPECT_FALSE(msg.data.empty());
    EXPECT_EQ(msg.data, "hi");
}

TEST_F(PubsubTest, PublisherOnce) {
    auto node = std::make_shared<rclcpp::Node>("test_node");
    auto pub = node->create_publisher<std_msgs::msg::String>("chatter", 10);
    auto msg = std_msgs::msg::String();
    msg.data = "unit-test";
    pub->publish(msg);

    auto exec = std::make_shared<rclcpp::executors::SingleThreadedExecutor>();
    exec->add_node(node);
    end = std::chrono::steady_clock::now() + std::chrono::seconds(1);
    while (std::chrono::steady_clock::now() < end) {
        exec->spin_some(50ms);
    }
    SUCCEED();
}

2.2 CMakeLists.txt 集成

if(BUILD_TESTING)
  find_package(ament_cmake_gtest REQUIRED)
  ament_add_gtest(test_pub_sub test/test_pub_sub.cpp)
  ament_target_dependencies(test_pub_sub rclcpp std_msgs)
endif()

2.3 跑测试

colcon build --packages-select cpp_pubsub
colcon test --packages-select cpp_pubsub

# 看详细
cat build/cpp_pubsub/test_results/cpp_pubsub/test_pub_sub.gtest.xml

2.4 关键 API

API	用途
TEST(suite, name)	测试用例
TEST_F(FixtureName, name)	用 fixture
EXPECT_*(非致命) / ASSERT_*(致命)	断言
SetUp() / TearDown()	每个用例前后
SetUpTestSuite() / TearDownTestSuite()	所有用例前后

---

3. Python pytest 完整指南

3.1 测试文件

import rclpy
import pytest

from py_pubsub.publisher_member_function import Talker


@pytest.fixture(scope='module')
def ros_context():
    """rclpy 是进程级单例,模块前 init,完成后 shutdown。"""
    rclpy.init()
    yield
    rclpy.shutdown()


def test_talker_init(ros_context):
    node = Talker()
    assert node.get_name() == 'talker_py'


def test_inproc_roundtrip(ros_context):
    """同进程 spin Talker + 自建 subscriber,验证消息流。"""
    import time
    from sensor_msgs.msg import String

    talker = Talker()
    received = []
    sub_node = rclpy.node.Node('test_subscriber')
    sub_node.create_subscription(
        String, 'chatter',
        lambda msg: received.append(msg.data), 10)

    exec_ = rclpy.executors.SingleThreadedExecutor()
    exec_.add_node(talker)
    exec_.add_node(sub_node)
    end = time.time() + 1.5
    while time.time() < end:
        exec_.spin_once(timeout_sec=0.05)

    assert any('Hello from PY' in s for s in received)

3.2 同进程 spin(关键模式)

exec_ = rclpy.executors.SingleThreadedExecutor()
exec_.add_node(talker)
exec_.add_node(listener)

end = time.time() + timeout_sec
while time.time() < end:
    exec_.spin_once(timeout_sec=0.05)

spin_once(timeout) 在主循环里手动驱动事件循环,比 rclpy.spin() 易控制超时。

3.3 pytest 标记(marks)

@pytest.mark.ros        # 需要 ROS2 环境
@pytest.mark.inproc    # 同进程可跑
@pytest.mark.slow      # 跑得慢

CLI:

pytest -m "not slow"
pytest -m ros and not inproc

3.4 pytest 配置(pyproject.toml)

本仓库已在 pyproject.toml 配:

[tool.pytest.ini_options]
markers = [
  "ros: 需要 ROS2 运行环境",
  "inproc: 同进程内 spin",
]

---

4. launch_testing(集成测试 launch 文件)

import launch
import launch_ros
import launch_testing
import launch_testing.actions
import pytest

@pytest.mark.launch_test
def generate_test_description():
    return launch.LaunchDescription([
        launch_ros.actions.Node(
            package='py_pubsub',
            executable='talker',
            parameters=[{'period_ms': 100, 'topic': 'chatter_test'}]),
        launch_ros.actions.Node(
            package='py_pubsub',
            executable='listener',
            parameters=[{'topic': 'chatter_test'}]),
        launch_testing.actions.ReadyToTest(),
    ])

class TestPubSub(unittest.TestCase):
    def test_message_received(self, proc_output):
        assert proc_output.assertWaitFor(
            'recv:', timeout=5, stream='stdout'
        )

注意: launch_testing 在 colcon test 环境下与节点 rclpy.shutdown() 二次调用可能冲突,本仓库改用更稳的 in-process spin。

---

5. 本仓库测试策略详解

5.1 单元 + 集成 in-process(主测试)

Python 包 用 pytest + in-process spin:

@pytest.fixture(scope='module')
def ros_context():
    rclpy.init()
    yield
    rclpy.shutdown()

def test_xxx(ros_context):
    server = SomeServer()
    client = SomeClient()
    exec_ = rclpy.executors.SingleThreadedExecutor()
    exec_.add_node(server); exec_.add_node(client)
    # ... 触发 + 验证

C++ 包 用 gtest + ament_add_gtest:

TEST_F(MyFixture, SpinSomeWorks) {
    auto node = std::make_shared<MyNode>();
    auto exec = std::make_shared<rclcpp::executors::SingleThreadedExecutor>();
    exec->add_node(node);
    exec->spin_some(50ms);
}

5.2 端到端(固化日志,手工验证)

5 个核心场景,跑一遍固化到 docker/:

• bringup_e2e.log:Topic 4 节点跨包跨语言
• srv_e2e.log:Service 12+30=42
• robot_e2e.log:URDF + TF 实时打印
• vision_e2e.log:Image 流
• full_demo_e2e.log:11 节点一起

每次大改动后重跑 + 替换日志。

5.3 测试覆盖现状

包	测试	用例数	通过率
py_pubsub	pytest + in-process	4/4	100%
cpp_pubsub	gtest	2/2	100%
py_srv	pytest + Service	1/1	100%
py_action_demo	pytest + Action	1/1	100%
cpp_robot_tf2	gtest + TF2	2/2	100%
py_vision_demo	pytest + Image	2/2	100%
bringup	launch 6 文件就绪	OK	100%

合计: 10/10 测试 100% PASSED。

---

6. 端到端自动化(进阶)

@pytest.mark.launch_test
def generate_test_description():
    return launch.LaunchDescription([
        IncludeLaunchDescription(
            PythonLaunchDescriptionSource(<bringup>/launch/full_demo_launch.py)),
        launch_testing.actions.ReadyToTest(),
    ])

class TestFullDemo(unittest.TestCase):
    def test_all_nodes_running(self):
        # 跑 ros2 node list 子进程,看 11 个节点
        ...

详见 20-topics.md §11.3 录包回放。

---

7. 标记 / 跳过 / 覆盖率

7.1 标记

@pytest.mark.skip(reason="硬件没到位")
def test_xxx():
    ...

@pytest.mark.skipif(sys.platform == 'win32', reason="Windows 跑不动")
def test_yyy():
    ...

@pytest.mark.xfail(reason="已知 bug")     # 标记预期失败
def test_zzz():
    ...

7.2 覆盖率

# Python
pytest --cov=py_pubsub --cov-report=html src/py_pubsub/test
# 输出 build/htmlcov/index.html

# C++(用 lcov)
sudo apt install lcov
cd build/cpp_pubsub && lcov --capture --directory . --output-file coverage.info
genhtml coverage.info --output-directory coverage_html

---

8. CI 集成(GitHub Actions)

# .github/workflows/test.yml
name: test
on: [push, pull_request]
jobs:
  test:
    runs-on: ubuntu-22.04
    steps:
      - uses: actions/checkout@v4
      - name: Build image & test
        run: |
          docker compose -f docker/docker-compose.yml build
          docker compose -f docker/docker-compose.yml up -d
          docker exec ros2_dev bash -lc "cd /root/ros2_ws && colcon build --packages-select py_pubsub cpp_pubsub py_srv py_action_demo cpp_robot_tf2 py_vision_demo bringup"
          docker exec ros2_dev bash -lc "cd /root/ros2_ws && colcon test --packages-select py_pubsub cpp_pubsub py_srv py_action_demo cpp_robot_tf2 py_vision_demo bringup"

---

9. 在本仓库里跑

9.1 跑全部测试

docker exec ros2_dev bash -lc "source /opt/ros/humble/setup.bash && cd /root/ros2_ws && colcon test --packages-select py_pubsub cpp_pubsub py_srv py_action_demo cpp_robot_tf2 py_vision_demo bringup"

9.2 单包测试

docker exec ros2_dev bash -lc "cd /root/ros2_ws && colcon test --packages-select cpp_pubsub"

9.3 看测试日志

# gtest
cat build/cpp_pubsub/test_results/cpp_pubsub/test_pub_sub.gtest.xml

# pytest
cat build/py_pubsub/pytest.xml

9.4 当前结果(固化)

✅ cpp_pubsub: 2/2 PASSED
✅ py_pubsub: 4/4 PASSED
✅ py_srv: 1/1 PASSED
✅ py_action_demo: 1/1 PASSED
✅ cpp_robot_tf2: 2/2 PASSED
✅ py_vision_demo: 2/2 PASSED
✅ bringup: OK (no tests)
合计: 10/10 PASSED

---

10. 进阶:Mock DDS / 时间注入 / Fault Injection

10.1 Mock DDS(测试纯逻辑)

# 不起 ROS2 上下文,直接测算法
def test_message_parser():
    raw = b'\x00\x01hello'
    parsed = MyParser.parse(raw)
    assert parsed == 'hello'

10.2 时间注入

// C++: 用 Clock 抽象 + RosTime mock
node->set_parameter(rclcpp::Parameter("use_sim_time", true));
// 然后通过 /clock topic 推时间

10.3 Fault Injection

# 杀掉 server 模拟断网
subprocess.run(['pkill', '-9', '-f', 'add_two_ints_server'])
# 看 client 怎么处理

# 慢响应模拟
# 修改 server callback sleep 时间

---

接下来读

主题	文档
三机部署	100-embedded-deployment.md
具身智能路径	99-embodied-ai.md
Docker 开发	85-docker.md