Make tracing test assert messages more descriptive (#93)

Signed-off-by: Christophe Bedard <[email protected]>

test_tracetools/test/test_buffer.py

@@ -59,26 +59,20 @@ def test_all(self):
             target_buffer_to_ipb_event = self.get_events_with_field_value(
                 'buffer',
                 target_buffer,
-                buffer_to_ipb_events)
-            # Only 1 for our given buffer
-            self.assertNumEventsEqual(
-                target_buffer_to_ipb_event,
-                1,
-                'none or more than 1 buffer_to_ipb event for the buffer',
+                buffer_to_ipb_events,
             )
+            # Only 1 for our given buffer
+            self.assertNumEventsEqual(target_buffer_to_ipb_event, 1)
 
             target_ipb = self.get_field(target_buffer_to_ipb_event[0], 'ipb')
             target_ipb_to_subscription_event = self.get_events_with_field_value(
                 'ipb',
                 target_ipb,
-                ipb_to_subscription_events)
+                ipb_to_subscription_events,
+            )
 
             # Only 1 for our given ipb
-            self.assertNumEventsEqual(
-                target_ipb_to_subscription_event,
-                1,
-                'none or more than 1 ipb_to_subscription event for the ipb',
-            )
+            self.assertNumEventsEqual(target_ipb_to_subscription_event, 1)
 
             # Check subscription init order
             #   * rclcpp_construct_ring_buffer
@@ -87,7 +81,7 @@ def test_all(self):
             self.assertEventOrder([
                 construct_event,
                 target_buffer_to_ipb_event[0],
-                target_ipb_to_subscription_event[0]
+                target_ipb_to_subscription_event[0],
             ])
 
 

test_tracetools/test/test_intra.py

@@ -51,11 +51,7 @@ def test_all(self):
             rcl_sub_init_events,
         )
         # Only 1 for our given topic
-        self.assertNumEventsEqual(
-            rcl_sub_init_topic_events,
-            1,
-            'none or more than 1 rcl_sub_init event for the topic',
-        )
+        self.assertNumEventsEqual(rcl_sub_init_topic_events, 1)
 
         # Get subscription handle
         rcl_sub_init_topic_event = rcl_sub_init_topic_events[0]
@@ -70,11 +66,7 @@ def test_all(self):
         )
         # Should have 2 events for the given subscription handle
         # (Subscription and SubscriptionIntraProcess)
-        self.assertNumEventsEqual(
-            rclcpp_sub_init_topic_events,
-            2,
-            'number of rclcpp_sub_init events for the topic not equal to 2',
-        )
+        self.assertNumEventsEqual(rclcpp_sub_init_topic_events, 2)
 
         # Get the 2 subscription pointers
         events = rclcpp_sub_init_topic_events
@@ -96,30 +88,20 @@ def test_all(self):
         start_events = self.get_events_with_name(tp.callback_start)
         end_events = self.get_events_with_name(tp.callback_end)
         # Should have at least one start:end pair
-        self.assertNumEventsGreaterEqual(
-            start_events,
-            1,
-            'does not have at least 1 callback start event')
-        self.assertNumEventsGreaterEqual(
-            end_events,
-            1,
-            'does not have at least 1 callback end event')
+        self.assertNumEventsGreaterEqual(start_events, 1)
+        self.assertNumEventsGreaterEqual(end_events, 1)
         start_events_topic = self.get_events_with_field_value(
             'callback',
             callback_pointers,
-            start_events)
+            start_events,
+        )
         end_events_topic = self.get_events_with_field_value(
             'callback',
             callback_pointers,
-            end_events)
-        self.assertNumEventsGreaterEqual(
-            start_events_topic,
-            1,
-            'no callback_start event found for topic')
-        self.assertNumEventsGreaterEqual(
-            end_events_topic,
-            1,
-            'no callback_end found event for topic')
+            end_events,
+        )
+        self.assertNumEventsGreaterEqual(start_events_topic, 1)
+        self.assertNumEventsGreaterEqual(end_events_topic, 1)
 
         # Check for is_intra_process field value of 1/true
         start_events_intra = self.get_events_with_field_value(
@@ -128,23 +110,15 @@ def test_all(self):
             start_events_topic,
         )
         # Should have one event
-        self.assertNumEventsEqual(
-            start_events_intra,
-            1,
-            'none or more than 1 callback_start event for intra-process topic',
-        )
+        self.assertNumEventsEqual(start_events_intra, 1)
         # As a sanity check, make sure its callback pointer has a corresponding callback_end event
         callback_pointer_intra = self.get_field(start_events_intra[0], 'callback')
         end_events_intra = self.get_events_with_field_value(
             'callback',
             callback_pointer_intra,
             end_events_topic,
         )
-        self.assertNumEventsEqual(
-            end_events_intra,
-            1,
-            'none or more than 1 callback_end event for intra-process topic',
-        )
+        self.assertNumEventsEqual(end_events_intra, 1)
 
 
 if __name__ == '__main__':

test_tracetools/test/test_intra_pub_sub.py

@@ -67,22 +67,22 @@ def test_all(self):
             ipb_ = self.get_events_with_field_value(
                 'buffer',
                 buffer,
-                buffer_to_ipb_events
+                buffer_to_ipb_events,
             )
             self.assertNumEventsEqual(ipb_, 1)
 
             subscription_ = self.get_events_with_field_value(
                 'ipb',
                 self.get_field(ipb_[0], 'ipb'),
-                ipb_to_subscription_events
+                ipb_to_subscription_events,
             )
             self.assertNumEventsEqual(subscription_, 1)
 
             callback_ = self.get_events_with_field_value(
                 'subscription',
                 self.get_field(subscription_[0], 'subscription'),
-                rclcpp_subscription_callback_added_events
-                )
+                rclcpp_subscription_callback_added_events,
+            )
             self.assertNumEventsEqual(callback_, 1)
             buffer_to_callback[buffer] = self.get_field(callback_[0], 'callback')
 
@@ -92,43 +92,48 @@ def test_all(self):
             enqueue_event_cand = self.get_events_with_field_value(
                 'vtid',
                 self.get_field(intra_publish_event, 'vtid'),
-                ring_buffer_enqueue_events
+                ring_buffer_enqueue_events,
             )
             target_enqueue_event = self.get_corresponding_event(
                 self.get_field(intra_publish_event, '_timestamp'),
-                enqueue_event_cand)
+                enqueue_event_cand,
+            )
 
             # Find corresponding enqueue/dequeue event
             target_index = self.get_field(target_enqueue_event, 'index')
             target_buffer = self.get_field(target_enqueue_event, 'buffer')
             filtered_dequeue_events = self.get_filtered_event(
                 {'buffer': target_buffer, 'index': target_index},
-                ring_buffer_dequeue_events)
+                ring_buffer_dequeue_events,
+            )
             target_dequeue_event = self.get_corresponding_event(
                 self.get_field(target_enqueue_event, '_timestamp'),
-                filtered_dequeue_events)
+                filtered_dequeue_events,
+            )
 
             callback_ = buffer_to_callback[target_buffer]
 
             filterd_callback_start = self.get_events_with_field_value(
                 'callback',
                 callback_,
-                callback_start_events
+                callback_start_events,
             )
 
             # Find corresponding callback_start/callback_end event
             target_callback_start = self.get_corresponding_event(
                 self.get_field(target_dequeue_event, '_timestamp'),
-                filterd_callback_start)
+                filterd_callback_start,
+            )
 
             callback_end_cand = self.get_events_with_field_value(
                 'callback',
                 callback_,
-                callback_end_events
+                callback_end_events,
             )
             target_callback_end = self.get_corresponding_event(
                 self.get_field(target_callback_start, '_timestamp'),
-                callback_end_cand)
+                callback_end_cand,
+            )
 
             event_sequence = [
                 intra_publish_event,
@@ -142,7 +147,8 @@ def test_all(self):
             for e in event_sequence:
                 self.assertTrue(
                     e is not None,
-                    'cannot find corresponding event for intra_publish')
+                    'cannot find corresponding event for intra_publish',
+                )
 
             # Check the order of events.
             self.assertEventOrder(event_sequence)

test_tracetools/test/test_lifecycle_node.py

@@ -46,22 +46,14 @@ def test_all(self):
             'test_lifecycle_node',
             rcl_node_init_events,
         )
-        self.assertEqual(
-            len(lifecycle_node_matches),
-            1,
-            'no node init event for lifecycle node',
-        )
+        self.assertNumEventsEqual(lifecycle_node_matches, 1)
         lifecycle_node_handle = self.get_field(lifecycle_node_matches[0], 'node_handle')
 
         # Check the state machine init event
         rcl_lifecycle_state_machine_init_events = self.get_events_with_name(
             tp.rcl_lifecycle_state_machine_init,
         )
-        self.assertEqual(
-            len(rcl_lifecycle_state_machine_init_events),
-            1,
-            'more than one state machine init event',
-        )
+        self.assertNumEventsEqual(rcl_lifecycle_state_machine_init_events, 1)
         # Make sure the node handle matches the one from the node init event
         rcl_lifecycle_state_machine_init_event = rcl_lifecycle_state_machine_init_events[0]
         node_handle = self.get_field(rcl_lifecycle_state_machine_init_event, 'node_handle')
@@ -116,7 +108,10 @@ def test_all(self):
                 self.get_field(event, 'state_machine') == state_machine_handle
                 for event in rcl_lifecycle_transition_events
             ),
-            'state machine handle does not match',
+            (
+                f"state machine handles do not all match '{state_machine_handle}': "
+                f'{rcl_lifecycle_transition_events}'
+            ),
         )
 
 

test_tracetools/test/test_pub_sub.py

@@ -166,11 +166,11 @@ def test_all(self):
         ping_rcl_subscription_init_event = ping_rcl_subscription_init_events[0]
         pong_rcl_subscription_init_event = pong_rcl_subscription_init_events[0]
         ping_sub_handle = self.get_field(ping_rcl_subscription_init_event, 'subscription_handle')
-        ping_rmw_sub_handle = self.get_field(
-            ping_rcl_subscription_init_event, 'rmw_subscription_handle')
+        ping_rmw_sub_handle = \
+            self.get_field(ping_rcl_subscription_init_event, 'rmw_subscription_handle')
         pong_sub_handle = self.get_field(pong_rcl_subscription_init_event, 'subscription_handle')
-        pong_rmw_sub_handle = self.get_field(
-            pong_rcl_subscription_init_event, 'rmw_subscription_handle')
+        pong_rmw_sub_handle = \
+            self.get_field(pong_rcl_subscription_init_event, 'rmw_subscription_handle')
 
         # Find corresponding rmw_sub_init events
         ping_rmw_sub_init_events = self.get_events_with_field_value(

test_tracetools/test/test_publisher.py

@@ -86,32 +86,19 @@ def test_all(self):
             '/the_topic',
             test_pub_init_events,
         )
-        self.assertNumEventsEqual(
-            test_pub_init_topic_events,
-            1,
-            'none or more than 1 rcl_pub_init even for test topic',
-        )
+        self.assertNumEventsEqual(test_pub_init_topic_events, 1)
 
         # Check queue_depth value
         test_pub_init_topic_event = test_pub_init_topic_events[0]
-        self.assertFieldEquals(
-            test_pub_init_topic_event,
-            'queue_depth',
-            10,
-            'pub_init event does not have expected queue depth value',
-        )
+        self.assertFieldEquals(test_pub_init_topic_event, 'queue_depth', 10)
 
         # Check that the node handle matches with the node_init event
         node_init_events = self.get_events_with_name(tp.rcl_node_init)
         test_pub_node_init_events = self.get_events_with_procname(
             'test_publisher',
             node_init_events,
         )
-        self.assertNumEventsEqual(
-            test_pub_node_init_events,
-            1,
-            'none or more than 1 node_init event',
-        )
+        self.assertNumEventsEqual(test_pub_node_init_events, 1)
         test_pub_node_init_event = test_pub_node_init_events[0]
         self.assertMatchingField(
             test_pub_node_init_event,
@@ -127,11 +114,7 @@ def test_all(self):
             rmw_publisher_handle,
             rmw_pub_init_events,
         )
-        self.assertNumEventsEqual(
-            rmw_pub_init_events,
-            1,
-            'none or more than 1 rmw_pub_init event for test topic',
-        )
+        self.assertNumEventsEqual(rmw_pub_init_events, 1)
         rmw_pub_init_event = rmw_pub_init_events[0]
 
         # Check publisher creation events order (rmw then rcl)
@@ -145,11 +128,7 @@ def test_all(self):
             rmw_publisher_handle,
             rmw_publish_events,
         )
-        self.assertNumEventsEqual(
-            rmw_publish_topic_events,
-            1,
-            'none or more than 1 rmw_publish event for test topic',
-        )
+        self.assertNumEventsEqual(rmw_publish_topic_events, 1)
         rmw_publish_topic_event = rmw_publish_topic_events[0]
         pub_message = self.get_field(rmw_publish_topic_event, 'message')
         # Find corresponding rclcpp/rcl_publish event
@@ -163,24 +142,19 @@ def test_all(self):
             pub_message,
             rcl_publish_events,
         )
-        self.assertNumEventsEqual(
-            rclcpp_publish_topic_events,
-            1,
-            'none or more than 1 rclcpp_publish event for test topic',
-        )
-        self.assertNumEventsEqual(
-            rcl_publish_topic_events,
-            1,
-            'none or more than 1 rcl_publish event for test topic',
-        )
+        self.assertNumEventsEqual(rclcpp_publish_topic_events, 1)
+        self.assertNumEventsEqual(rcl_publish_topic_events, 1)
         rclcpp_publish_topic_event = rclcpp_publish_topic_events[0]
         rcl_publish_topic_event = rcl_publish_topic_events[0]
         # Get publisher handle from rcl_publisher_init event
         publisher_handle = self.get_field(test_pub_init_topic_event, 'publisher_handle')
         self.assertFieldEquals(rcl_publish_topic_event, 'publisher_handle', publisher_handle)
         # Check publication events order
-        self.assertEventOrder(
-            [rclcpp_publish_topic_event, rcl_publish_topic_event, rmw_publish_topic_event])
+        self.assertEventOrder([
+            rclcpp_publish_topic_event,
+            rcl_publish_topic_event,
+            rmw_publish_topic_event,
+        ])
 
 
 if __name__ == '__main__':