niklas/ros2_tracing
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add assertNumEvents method and use assertFieldEquals

Browse source
This commit is contained in:
Christophe Bedard 2019-06-23 12:56:50 +02:00
parent 4b1545ccb4
commit 0abddffeff
7 changed files with 52 additions and 41 deletions
Download patch file Download diff file Expand all files Collapse all files

28
tracetools_test/test/test_intra.py
View file

@ -46,12 +46,12 @@ class TestIntra(TraceTestCase):
'topic_name', 'topic_name',
'/the_topic/_intra', '/the_topic/_intra',
sub_init_events) sub_init_events)
self.assertEqual( self.assertNumEvents(
len(sub_init_normal_events), sub_init_normal_events,
1, 1,
'none or more than 1 sub init event for normal sub') 'none or more than 1 sub init event for normal sub')
self.assertEqual( self.assertNumEvents(
len(sub_init_intra_events), sub_init_intra_events,
1, 1,
'none or more than 1 sub init event for intra sub') 'none or more than 1 sub init event for intra sub')
@ -68,8 +68,8 @@ class TestIntra(TraceTestCase):
sub_handle_intra, sub_handle_intra,
self.get_events_with_name( self.get_events_with_name(
'ros2:rclcpp_subscription_callback_added')) 'ros2:rclcpp_subscription_callback_added'))
self.assertEqual( self.assertNumEvents(
len(callback_added_events), callback_added_events,
1, 1,
'none or more than 1 callback added event') 'none or more than 1 callback added event')
callback_added_event = callback_added_events[0] callback_added_event = callback_added_events[0]
@ -79,8 +79,8 @@ class TestIntra(TraceTestCase):
start_events = self.get_events_with_name('ros2:callback_start') start_events = self.get_events_with_name('ros2:callback_start')
end_events = self.get_events_with_name('ros2:callback_end') end_events = self.get_events_with_name('ros2:callback_end')
# Should still have two start:end pairs (1 normal + 1 intra) # Should still have two start:end pairs (1 normal + 1 intra)
self.assertEqual(len(start_events), 2, 'does not have 2 callback start events') self.assertNumEvents(start_events, 2, 'does not have 2 callback start events')
self.assertEqual(len(end_events), 2, 'does not have 2 callback end events') self.assertNumEvents(end_events, 2, 'does not have 2 callback end events')
start_events_intra = self.get_events_with_field_value( start_events_intra = self.get_events_with_field_value(
'callback', 'callback',
callback_handle_intra, callback_handle_intra,
@ -89,12 +89,12 @@ class TestIntra(TraceTestCase):
'callback', 'callback',
callback_handle_intra, callback_handle_intra,
end_events) end_events)
self.assertEqual( self.assertNumEvents(
len(start_events_intra), start_events_intra,
1, 1,
'none or more than one intra start event') 'none or more than one intra start event')
self.assertEqual( self.assertNumEvents(
len(end_events_intra), end_events_intra,
1, 1,
'none or more than one intra end event') 'none or more than one intra end event')
@ -111,8 +111,8 @@ class TestIntra(TraceTestCase):
'callback', 'callback',
callback_handle_intra, callback_handle_intra,
start_events) start_events)
self.assertEqual( self.assertNumEvents(
len(start_events_not_intra), start_events_not_intra,
1, 1,
'none or more than one normal start event') 'none or more than one normal start event')
start_event_not_intra = start_events_not_intra[0] start_event_not_intra = start_events_not_intra[0]

12
tracetools_test/test/test_publisher.py
View file

@ -49,16 +49,16 @@ class TestPublisher(TraceTestCase):
'topic_name', 'topic_name',
'/the_topic', '/the_topic',
test_pub_init_events) test_pub_init_events)
self.assertEqual( self.assertNumEvents(
len(test_pub_init_topic_events), test_pub_init_topic_events,
1, 1,
'none or more than 1 pub_init even for test topic') 'none or more than 1 pub_init even for test topic')
# Check queue_depth value # Check queue_depth value
test_pub_init_topic_event = test_pub_init_topic_events[0] test_pub_init_topic_event = test_pub_init_topic_events[0]
test_queue_depth = self.get_field(test_pub_init_topic_event, 'queue_depth') self.assertFieldEquals(
self.assertEqual( test_pub_init_topic_event,
test_queue_depth, 'queue_depth',
10, 10,
'pub_init event does not have expected queue depth value') 'pub_init event does not have expected queue depth value')
@ -67,7 +67,7 @@ class TestPublisher(TraceTestCase):
test_pub_node_init_events = self.get_events_with_procname( test_pub_node_init_events = self.get_events_with_procname(
'test_publisher', 'test_publisher',
node_init_events) node_init_events)
self.assertEqual(len(test_pub_node_init_events), 1, 'none or more than 1 node_init event') self.assertNumEvents(test_pub_node_init_events, 1, 'none or more than 1 node_init event')
test_pub_node_init_event = test_pub_node_init_events[0] test_pub_node_init_event = test_pub_node_init_events[0]
self.assertMatchingField( self.assertMatchingField(
test_pub_node_init_event, test_pub_node_init_event,

2
tracetools_test/test/test_service.py
View file

@ -61,7 +61,7 @@ class TestService(TraceTestCase):
test_srv_node_init_events = self.get_events_with_procname( test_srv_node_init_events = self.get_events_with_procname(
'test_service', 'test_service',
node_init_events) node_init_events)
self.assertEqual(len(test_srv_node_init_events), 1, 'none or more than 1 node_init event') self.assertNumEvents(test_srv_node_init_events, 1, 'none or more than 1 node_init event')
test_srv_node_init_event = test_srv_node_init_events[0] test_srv_node_init_event = test_srv_node_init_events[0]
self.assertMatchingField( self.assertMatchingField(
test_srv_node_init_event, test_srv_node_init_event,

9
tracetools_test/test/test_service_callback.py
View file

@ -38,13 +38,12 @@ class TestServiceCallback(TraceTestCase):
start_events = self.get_events_with_name('ros2:callback_start') start_events = self.get_events_with_name('ros2:callback_start')
for event in start_events: for event in start_events:
self.assertValidHandle(event, 'callback') self.assertValidHandle(event, 'callback')
is_intra_process_value = self.get_field(event, 'is_intra_process')
self.assertIsInstance(is_intra_process_value, int, 'is_intra_process not int')
# Should not be 1 for services (yet) # Should not be 1 for services (yet)
self.assertEqual( self.assertFieldEquals(
is_intra_process_value, event,
'is_intra_process',
0, 0,
f'invalid value for is_intra_process: {is_intra_process_value}') 'invalid value for is_intra_process')
end_events = self.get_events_with_name('ros2:callback_end') end_events = self.get_events_with_name('ros2:callback_end')
for event in end_events: for event in end_events:

12
tracetools_test/test/test_subscription.py
View file

@ -54,13 +54,13 @@ class TestSubscription(TraceTestCase):
'topic_name', 'topic_name',
'/the_topic', '/the_topic',
sub_init_events) sub_init_events)
self.assertEqual(len(test_sub_init_events), 1, 'cannot find test topic name') self.assertNumEvents(test_sub_init_events, 1, 'cannot find test topic name')
test_sub_init_event = test_sub_init_events[0] test_sub_init_event = test_sub_init_events[0]
# Check queue_depth value # Check queue_depth value
test_queue_depth = self.get_field(test_sub_init_event, 'queue_depth') self.assertFieldEquals(
self.assertEqual( test_sub_init_event,
test_queue_depth, 'queue_depth',
10, 10,
'sub_init event does not have expected queue depth value') 'sub_init event does not have expected queue depth value')
@ -69,8 +69,8 @@ class TestSubscription(TraceTestCase):
test_sub_node_init_events = self.get_events_with_procname( test_sub_node_init_events = self.get_events_with_procname(
'test_subscription', 'test_subscription',
node_init_events) node_init_events)
self.assertEqual( self.assertNumEvents(
len(test_sub_node_init_events), test_sub_node_init_events,
1, 1,
'none or more than 1 node_init event') 'none or more than 1 node_init event')
test_sub_node_init_event = test_sub_node_init_events[0] test_sub_node_init_event = test_sub_node_init_events[0]

11
tracetools_test/test/test_timer.py
View file

@ -51,13 +51,12 @@ class TestTimer(TraceTestCase):
start_events = self.get_events_with_name('ros2:callback_start') start_events = self.get_events_with_name('ros2:callback_start')
for event in start_events: for event in start_events:
self.assertValidHandle(event, 'callback') self.assertValidHandle(event, 'callback')
is_intra_process_value = self.get_field(event, 'is_intra_process')
self.assertIsInstance(is_intra_process_value, int, 'is_intra_process not int')
# Should not be 1 for timer # Should not be 1 for timer
self.assertEqual( self.assertEqual(
is_intra_process_value, event,
'is_intra_process',
0, 0,
f'invalid value for is_intra_process: {is_intra_process_value}') 'invalid value for is_intra_process')
end_events = self.get_events_with_name('ros2:callback_end') end_events = self.get_events_with_name('ros2:callback_end')
for event in end_events: for event in end_events:
@ -67,9 +66,7 @@ class TestTimer(TraceTestCase):
test_timer_init_event = self.get_events_with_procname('test_timer', init_events) test_timer_init_event = self.get_events_with_procname('test_timer', init_events)
self.assertEqual(len(test_timer_init_event), 1) self.assertEqual(len(test_timer_init_event), 1)
test_init_event = test_timer_init_event[0] test_init_event = test_timer_init_event[0]
test_period = self.get_field(test_init_event, 'period') self.assertFieldEquals(test_init_event, 'period', 1000000, 'invalid period')
self.assertIsInstance(test_period, int)
self.assertEqual(test_period, 1000000, f'invalid period: {test_period}')
# Check that the timer_init:callback_added pair exists and has a common timer handle # Check that the timer_init:callback_added pair exists and has a common timer handle
self.assertMatchingField( self.assertMatchingField(

19
tracetools_test/tracetools_test/case.py
View file

@ -160,6 +160,14 @@ class TraceTestCase(unittest.TestCase):
""" """
self.assertTrue(self.are_events_ordered(first_event, second_event)) self.assertTrue(self.are_events_ordered(first_event, second_event))
def assertNumEvents(
self,
events: List[DictEvent],
expected_number: int,
msg: str = 'wrong number of events'
):
self.assertEqual(len(events), expected_number, msg)
def assertMatchingField( def assertMatchingField(
self, self,
initial_event: DictEvent, initial_event: DictEvent,
@ -200,16 +208,23 @@ class TraceTestCase(unittest.TestCase):
1, 1,
'matching field event not after initial event') 'matching field event not after initial event')
def assertFieldEquals(self, event: DictEvent, field_name: str, value: Any): def assertFieldEquals(
self,
event: DictEvent,
field_name: str,
value: Any,
msg: str = 'wrong field value'
):
""" """
Check the value of a field. Check the value of a field.
:param event: the event :param event: the event
:param field_name: the name of the field to check :param field_name: the name of the field to check
:param value: to value to compare the field value to :param value: to value to compare the field value to
:param msg: the message to display on failure
""" """
actual_value = self.get_field(event, field_name) actual_value = self.get_field(event, field_name)
self.assertEqual(actual_value, value, 'invalid field value') self.assertEqual(actual_value, value, msg)
def get_field(self, event: DictEvent, field_name: str) -> Any: def get_field(self, event: DictEvent, field_name: str) -> Any:
""" """