Use deque instead of list and added unit tests

Signed-off-by: Pintaudi Giorgio <[email protected]>

Author: LastStarDust

launch_testing_ros/launch_testing_ros/wait_for_topics.py

@@ -16,6 +16,7 @@
 import string
 from threading import Event
 from threading import Thread
+from collections import deque
 
 import rclpy
 from rclpy.executors import SingleThreadedExecutor
@@ -50,9 +51,10 @@ def method_2():
         wait_for_topics.shutdown()
     """
 
-    def __init__(self, topic_tuples, timeout=5.0):
+    def __init__(self, topic_tuples, timeout=5.0, max_number_of_messages=100):
         self.topic_tuples = topic_tuples
         self.timeout = timeout
+        self.max_number_of_messages = max_number_of_messages
         self.__ros_context = rclpy.Context()
         rclpy.init(context=self.__ros_context)
         self.__ros_executor = SingleThreadedExecutor(context=self.__ros_context)
@@ -64,9 +66,13 @@ def __init__(self, topic_tuples, timeout=5.0):
         self.__ros_spin_thread.start()
 
     def _prepare_ros_node(self):
-        node_name = '_test_node_' +\
-            ''.join(random.choices(string.ascii_uppercase + string.digits, k=10))
-        self.__ros_node = _WaitForTopicsNode(name=node_name, node_context=self.__ros_context)
+        node_name = '_test_node_' + ''.join(
+            random.choices(string.ascii_uppercase + string.digits, k=10)
+        )
+        self.__ros_node = _WaitForTopicsNode(
+            name=node_name, node_context=self.__ros_context,
+            max_number_of_messages=self.max_number_of_messages
+        )
         self.__ros_executor.add_node(self.__ros_node)
 
     def wait(self):
@@ -108,9 +114,11 @@ def __exit__(self, exep_type, exep_value, trace):
 class _WaitForTopicsNode(Node):
     """Internal node used for subscribing to a set of topics."""
 
-    def __init__(self, name='test_node', node_context=None):
+    def __init__(self, name='test_node', node_context=None,
+                 max_number_of_messages=None):
         super().__init__(node_name=name, context=node_context)
         self.msg_event_object = Event()
+        self.max_number_of_messages = max_number_of_messages
 
     def start_subscribers(self, topic_tuples):
         self.subscriber_list = []
@@ -128,14 +136,16 @@ def start_subscribers(self, topic_tuples):
                     10
                 )
             )
+            # Initialize ring buffer of received messages
+            self.received_messages[topic_name] = deque(maxlen=self.max_number_of_messages)
 
     def callback_template(self, topic_name):
 
         def topic_callback(data):
             if topic_name not in self.received_topics:
                 self.get_logger().debug('Message received for ' + topic_name)
                 self.received_topics.add(topic_name)
-                self.received_messages.setdefault(topic_name, []).append(data)
+                self.received_messages[topic_name].append(data)
             if self.received_topics == self.expected_topics:
                 self.msg_event_object.set()
 

launch_testing_ros/test/examples/wait_for_topic_launch_test.py

@@ -15,6 +15,7 @@
 import os
 import sys
 import unittest
+import re
 
 import launch
 import launch.actions
@@ -57,11 +58,15 @@ def test_topics_successful(self, count):
             """All the supplied topics should be read successfully."""
             topic_list = [('chatter_' + str(i), String) for i in range(count)]
             expected_topics = {'chatter_' + str(i) for i in range(count)}
+            message_pattern = re.compile('Hello World: \d+')
 
             # Method 1 : Using the magic methods and 'with' keyword
-            with WaitForTopics(topic_list, timeout=2.0) as wait_for_node_object_1:
+            with WaitForTopics(topic_list, timeout=2.0, max_number_of_messages=10) as wait_for_node_object_1:
                 assert wait_for_node_object_1.topics_received() == expected_topics
                 assert wait_for_node_object_1.topics_not_received() == set()
+                for topic_name, _ in topic_list:
+                    message = wait_for_node_object_1.messages_received(topic_name).pop().data
+                    assert message_pattern.match(message)
 
             # Multiple instances of WaitForNode() can be created safely as
             # their internal nodes spin in separate contexts
@@ -70,6 +75,9 @@ def test_topics_successful(self, count):
             assert wait_for_node_object_2.wait()
             assert wait_for_node_object_2.topics_received() == expected_topics
             assert wait_for_node_object_2.topics_not_received() == set()
+            for topic_name, _ in topic_list:
+                message = wait_for_node_object_2.messages_received(topic_name).pop().data
+                assert message_pattern.match(message)
             wait_for_node_object_2.shutdown()
 
         def test_topics_unsuccessful(self, count):