Adapt tests to Zenoh (#988)

Signed-off-by: Alejandro Hernández Cordero <[email protected]>

Author: ahcorde

ros2topic/test/test_bw_delay_hz.py

@@ -33,6 +33,8 @@
 import rclpy
 from rclpy.executors import SingleThreadedExecutor
 from rclpy.qos import DurabilityPolicy
+from rclpy.qos import qos_check_compatible
+from rclpy.qos import QoSCompatibility
 from rclpy.qos import QoSProfile
 from rclpy.qos import ReliabilityPolicy
 from rclpy.utilities import get_rmw_implementation_identifier
@@ -115,10 +117,22 @@ def helper_verb_basic(self, launch_service, proc_info, proc_output, verb, succes
                         verb_extra_options = [
                             '--qos-reliability', 'reliable',
                             '--qos-durability', 'transient_local']
+                        # This QoS profile matched with the extra options defined above
+                        rostopic_qos_profile = QoSProfile(
+                            depth=10,
+                            reliability=ReliabilityPolicy.RELIABLE,
+                            durability=DurabilityPolicy.TRANSIENT_LOCAL)
                         publisher_qos_profile = QoSProfile(
                             depth=10,
                             reliability=ReliabilityPolicy.BEST_EFFORT,
                             durability=DurabilityPolicy.VOLATILE)
+                        # Skip this test if the QoS between the publisher and subscription
+                        # are compatible according to the underlying middleware.
+                        comp, reason = qos_check_compatible(
+                            rostopic_qos_profile, publisher_qos_profile)
+                        if comp == QoSCompatibility.OK or comp == QoSCompatibility.WARNING:
+                            raise unittest.SkipTest()
+
                 publisher = self.node.create_publisher(PointStamped, topic, publisher_qos_profile)
                 assert publisher
 

ros2topic/test/test_echo_pub.py

@@ -14,6 +14,7 @@
 
 import functools
 import pathlib
+import re
 import sys
 import unittest
 
@@ -32,6 +33,8 @@
 import rclpy
 from rclpy.executors import SingleThreadedExecutor
 from rclpy.qos import DurabilityPolicy
+from rclpy.qos import qos_check_compatible
+from rclpy.qos import QoSCompatibility
 from rclpy.qos import QoSProfile
 from rclpy.qos import ReliabilityPolicy
 from rclpy.utilities import get_rmw_implementation_identifier
@@ -124,12 +127,23 @@ def test_pub_basic(self, launch_service, proc_info, proc_output):
                         pub_extra_options = [
                             '--qos-reliability', 'best_effort',
                             '--qos-durability', 'volatile']
+                        # This QoS profile matched with the extra options defined above
+                        rostopic_qos_profile = QoSProfile(
+                            depth=10,
+                            reliability=ReliabilityPolicy.BEST_EFFORT,
+                            durability=DurabilityPolicy.VOLATILE)
                         subscription_qos_profile = QoSProfile(
                             depth=10,
                             reliability=ReliabilityPolicy.RELIABLE,
                             durability=DurabilityPolicy.TRANSIENT_LOCAL)
                         expected_maximum_message_count = 0
                         expected_minimum_message_count = 0
+                        # Skip this test if the QoS between the publisher and subscription
+                        # are compatible according to the underlying middleware.
+                        comp, reason = qos_check_compatible(
+                            rostopic_qos_profile, subscription_qos_profile)
+                        if comp == QoSCompatibility.OK:
+                            raise unittest.SkipTest()
 
                 future = rclpy.task.Future()
 
@@ -363,6 +377,17 @@ def test_echo_basic(self, launch_service, proc_info, proc_output):
                             depth=10,
                             reliability=ReliabilityPolicy.BEST_EFFORT,
                             durability=DurabilityPolicy.VOLATILE)
+                        # This QoS profile matched with the extra options defined above
+                        rostopic_qos_profile = QoSProfile(
+                            depth=10,
+                            reliability=ReliabilityPolicy.RELIABLE,
+                            durability=DurabilityPolicy.TRANSIENT_LOCAL)
+                        # Skip this test if the QoS between the publisher and subscription
+                        # are compatible according to the underlying middleware.
+                        comp, reason = qos_check_compatible(
+                            rostopic_qos_profile, publisher_qos_profile)
+                        if comp == QoSCompatibility.OK or comp == QoSCompatibility.WARNING:
+                            raise unittest.SkipTest()
                 if not message_lost:
                     echo_extra_options.append('--no-lost-messages')
                 publisher = self.node.create_publisher(String, topic, publisher_qos_profile)
@@ -494,9 +519,9 @@ def publish_message():
                 )
                 assert command.wait_for_output(functools.partial(
                     launch_testing.tools.expect_output, expected_lines=[
-                        "b'\\x00\\x01\\x00\\x00\\x06\\x00\\x00\\x00hello\\x00\\x00\\x00'",
-                        '---',
-                    ], strict=True
+                        re.compile(r"^b'\\x00\\x01\\x00\\x00\\x06\\x00\\x00\\x00hello(\\x00)*'$"),
+                        re.compile(r'^---$'),
+                    ], strict=False
                 ), timeout=10), 'Echo CLI did not print expected message'
             assert command.wait_for_shutdown(timeout=10)
 

ros2topic/test/test_use_sim_time.py

@@ -264,7 +264,8 @@ def publish_message():
                     launch_testing.tools.expect_output, expected_lines=[
                         # without speed up, the average band width should be 16 B
                         re.compile(r'8 B/s from \d+ messages'),
-                        re.compile(r'\s*Message size mean: 16 B min: 16 B max: 16 B')
+                        re.compile(
+                            r'\s*Message size mean: (14|16) B min: (14|16) B max: (14|16) B')
                     ], strict=False
                 ), timeout=10), 'Echo CLI did not print expected message'
             assert command.wait_for_shutdown(timeout=5)