如何为依赖静态变换发布的ROS 2节点函数做单元测试？

问题核心 #

需要测试get_new_transform_coordinates函数在TF变换可用/不可用时的行为，但无法修改节点源码，且测试中节点通信异常。核心解决点在于正确初始化测试环境、模拟TF发布，并确保TF缓存能接收到变换数据。

完整测试代码实现 #

以下是test_transform_node.py的示例代码，覆盖两种测试场景：

import rclpy
from rclpy.node import Node
from tf2_ros import StaticTransformBroadcaster, TransformException
from tf2_ros.buffer import Buffer
from tf2_ros.transform_listener import TransformListener
from geometry_msgs.msg import TransformStamped, Time
from your_package.your_node import TargetNode  # 替换为你的节点类路径
import unittest
from unittest.mock import patch, MagicMock

class TestTransformFunction(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        rclpy.init()

    @classmethod
    def tearDownClass(cls):
        rclpy.shutdown()

    def setUp(self):
        # 创建测试节点与被测试节点
        self.test_node = Node("test_tf_node")
        self.target_node = TargetNode()

        # 初始化被测试节点的TF缓存与监听器（若原节点未在构造函数中初始化）
        self.target_node.tf_buffer = Buffer()
        self.target_node.tf_listener = TransformListener(self.target_node.tf_buffer, self.target_node)

        # 模拟base_link参数
        self.target_node.__base_link_frame_id = MagicMock()
        self.target_node.__base_link_frame_id.get_parameter_value().string_value = "base_link"

        # 创建静态TF发布者
        self.tf_broadcaster = StaticTransformBroadcaster(self.test_node)

    def tearDown(self):
        self.test_node.destroy_node()
        self.target_node.destroy_node()

    def test_transform_available(self):
        # 发布base_link到camera_frame的静态TF
        tf_msg = TransformStamped()
        tf_msg.header.stamp = self.test_node.get_clock().now().to_msg()
        tf_msg.header.frame_id = "base_link"
        tf_msg.child_frame_id = "camera_frame"
        tf_msg.transform.translation.x = 0.5
        tf_msg.transform.translation.y = 0.3
        tf_msg.transform.rotation.w = 1.0
        self.tf_broadcaster.sendTransform(tf_msg)

        # 等待节点处理TF消息
        rclpy.spin_once(self.test_node, timeout_sec=1.0)
        rclpy.spin_once(self.target_node, timeout_sec=1.0)

        # 构造测试用image_msg
        image_msg = {
            "timestamp": self.test_node.get_clock().now().to_msg(),
            "frame_id": "camera_frame"
        }

        # 模拟时间戳检查通过
        with patch.object(self.target_node, 'check_time_stamp_fresh', return_value=True):
            pos_x, pos_y = self.target_node.get_new_transform_coordinates(
                mission_x=1.0, image_msg=image_msg, camera_name="camera", image_type="rgb"
            )

        # 验证计算结果
        self.assertEqual(pos_x, 1500)  # (0.5+1.0)*1000
        self.assertEqual(pos_y, -300)  # 0.3*(-1000)

    def test_transform_unavailable(self):
        # 构造不存在的frame_id的image_msg
        image_msg = {
            "timestamp": self.test_node.get_clock().now().to_msg(),
            "frame_id": "non_existent_frame"
        }

        # 模拟时间戳检查通过
        with patch.object(self.target_node, 'check_time_stamp_fresh', return_value=True):
            pos_x, pos_y = self.target_node.get_new_transform_coordinates(
                mission_x=1.0, image_msg=image_msg, camera_name="camera", image_type="rgb"
            )

        # 验证返回值
        self.assertEqual(pos_x, -1)
        self.assertEqual(pos_y, -1)

        # 验证错误日志输出
        with patch.object(self.target_node.get_logger(), 'error') as mock_log:
            self.target_node.get_new_transform_coordinates(
                mission_x=1.0, image_msg=image_msg, camera_name="camera", image_type="rgb"
            )
            mock_log.assert_called()

关键细节说明 #

• 环境初始化：通过setUpClass/tearDownClass全局初始化/关闭rclpy，避免重复初始化。
• TF缓存初始化：若原节点未在构造函数中初始化tf_buffer和tf_listener，需在测试中手动初始化，确保能接收TF数据。
• 参数模拟：用MagicMock模拟私有参数__base_link_frame_id的返回值，无需修改原节点源码。
• 节点Spin：调用rclpy.spin_once让节点处理TF消息，否则tf_buffer无法缓存变换，导致查找失败。
• 时间戳模拟：通过patch模拟check_time_stamp_fresh返回True，跳过时间戳过期的分支，专注测试TF查找逻辑。
• 静态TF发布：使用StaticTransformBroadcaster发布持久化的TF变换，适合单元测试的稳定场景。

内容的提问来源于stack exchange，提问作者eduardovilasboas