ROS node testing (publish/subscribe)

Question

I am struggling with an example integration test for a ROS node (in Python). I have a node that somehow processes input data and outputs some messages based on this:

#!/usr/bin/env python3

import rospy
from image_utils import ImageUtils
from sensor_msgs.msg import Image
from std_msgs.msg import String


class TemplateNode:
    def __init__(self):
        rospy.init_node("template_node", anonymous=True)

        self.input_topic = rospy.get_param("~input_topic", "/input_topic")
        self.output_topic = rospy.get_param("~output_topic", "/output_topic")
        self.image_utils = ImageUtils()

        self.input_sub = rospy.Subscriber(self.input_topic, Image, self.input_callback)
        self.output_pub = rospy.Publisher(self.output_topic, String, queue_size=10)

    def input_callback(self, data):
        try:
            img_shape = self.image_utils.get_imgmsg_shape(data)
            result = str(img_shape)

            rospy.loginfo(result)
            self.publish_result(result)

        except Exception as e:
            rospy.logerr("Error processing image: {}".format(e))

    def publish_result(self, data):
        self.output_pub.publish(data)

    def run(self):
        rospy.spin()


if __name__ == "__main__":
    try:
        node = TemplateNode()
        node.run()
    except rospy.ROSInterruptException:
        pass

I created the following test for it:

#!/usr/bin/env python3

import sys
import time
import unittest

import numpy as np
import rospy
from cv_bridge import CvBridge
from sensor_msgs.msg import Image
from std_msgs.msg import String

PKG = "template_pkg"
NAME = "template_node_test"
PUB_TOPIC = "/input_topic"
SUB_TOPIC = "/output_topic"
TEST_NODE_NAME = "template_node"


class TemplateNodeTest(unittest.TestCase):

    def setUp(self):
        self.success = False
        self.callback_data = None
        self.bridge = CvBridge()
        self.img_data = np.random.randint(0, 255, (200, 100, 3), dtype=np.uint8)
        self.pub = rospy.Publisher(PUB_TOPIC, Image)
        

        rospy.init_node(NAME, anonymous=True)

    def callback(self, data):
        self.callback_data = data
        if data == str(self.img_data.shape):
            self.success = True

    def test_image_response(self):

        timeout_t = time.time() + 5.0
        while (
            not rostest.is_subscriber(
                rospy.resolve_name(PUB_TOPIC), rospy.resolve_name(TEST_NODE_NAME)
            )
            and time.time() < timeout_t  # noqa: W503
        ):
            time.sleep(0.1)

        self.assert_(
            rostest.is_subscriber(
                rospy.resolve_name(PUB_TOPIC), rospy.resolve_name(TEST_NODE_NAME)
            ),
            "%s is not up" % TEST_NODE_NAME,
        )
        self.sub = rospy.Subscriber(SUB_TOPIC, String, self.callback)
        self.pub.publish(self.bridge.cv2_to_imgmsg(self.img_data, encoding="bgr8"))
        
        timeout_t = time.time() + 1.0
        while not rospy.is_shutdown() and not self.success and time.time() < timeout_t:
            time.sleep(0.1)

        self.assert_(self.success, "Wrong response: {}".format(self.callback_data))


if __name__ == "__main__":
    import rostest

    rostest.rosrun(PKG, NAME, TemplateNodeTest, sys.argv)

Launched using the following configuration:

<launch>
    <node name="template_node" pkg="template_pkg" type="template_node.py" output="screen" />
    <test test-name="template_node_test" pkg="template_pkg" type="template_node_test.py"/>
</launch>

It kinda works, but I the callback from the test is never called:

[INFO] [1710528976.836281]: (200, 100, 3)
[Testcase: testtemplate_node_test] ... ok

[ROSTEST]-----------------------------------------------------------------------

[template_pkg.rosunit-template_node_test/test_image_response][FAILURE]----------
False is not true : Wrong response: None
  File "/usr/lib/python3.8/unittest/case.py", line 60, in testPartExecutor
    yield
  File "/usr/lib/python3.8/unittest/case.py", line 676, in run
    self._callTestMethod(testMethod)
  File "/usr/lib/python3.8/unittest/case.py", line 633, in _callTestMethod
    method()
  File "/catkin_ws/src/template_pkg/test/template_node_test.py", line 65, in test_image_response
    self.assert_(self.success, "Wrong response: {}".format(self.callback_data))
  File "/usr/lib/python3.8/unittest/case.py", line 1410, in deprecated_func
    return original_func(*args, **kwargs)
  File "/usr/lib/python3.8/unittest/case.py", line 765, in assertTrue
    raise self.failureException(msg)
--------------------------------------------------------------------------------

I am confused, because I can see the node reporting the shape via logging:

[INFO] [1710528976.836281]: (200, 100, 3)

Apparently, the message published by the node is not received by the test callback, I just do not understand why.

Would be grateful for any hint, including the test design, maybe I shall do it differently?

Answer 1

It turned to be out that there was not enough time to set up and start receiving messages before the publisher in my node starts sending messages. I fixed it by adding a delay between the subscription and publishing:

self.sub = rospy.Subscriber(SUB_TOPIC, String, self.callback)
rospy.sleep(0.1)
self.pub.publish(self.bridge.cv2_to_imgmsg(self.img_data, encoding="bgr8"))