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

Add subscription callback test with ping+pong nodes

Browse source
This commit is contained in:
Christophe Bedard 2019-06-03 11:16:05 +02:00
parent 3b85b3cbd8
commit 8f09ce339d
7 changed files with 158 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

2
tracetools_test/test/test_node.py
View file

@ -17,7 +17,7 @@ class TestNode(unittest.TestCase):
def test_creation(self):
session_name_prefix = 'session-test-node-creation'
base_path = '/tmp'
test_node = 'test_publisher'
test_node = ['test_publisher']
exit_code, full_path = run_and_trace(base_path, session_name_prefix, node_creation_events, None, PKG, test_node)
self.assertEqual(exit_code, 0)

2
tracetools_test/test/test_publisher.py
View file

@ -16,7 +16,7 @@ class TestPublisher(unittest.TestCase):
def test_creation(self):
session_name_prefix = 'session-test-publisher-creation'
base_path = '/tmp'
test_node = 'test_publisher'
test_node = ['test_publisher']
exit_code, full_path = run_and_trace(base_path, session_name_prefix, publisher_creation_events, None, PKG, test_node)
self.assertEqual(exit_code, 0)

21
tracetools_test/test/test_subscription.py
View file

@ -12,12 +12,17 @@ subscription_creation_events = [
'ros2:rclcpp_subscription_callback_added',
]
subscription_callback_events = [
'ros2:rclcpp_subscription_callback_start',
'ros2:rclcpp_subscription_callback_end',
]
class TestSubscription(unittest.TestCase):
def test_creation(self):
session_name_prefix = 'session-test-subscription-creation'
base_path = '/tmp'
test_node = 'test_subscription'
test_node = ['test_subscription']
exit_code, full_path = run_and_trace(base_path, session_name_prefix, subscription_creation_events, None, PKG, test_node)
self.assertEqual(exit_code, 0)
@ -28,6 +33,20 @@ class TestSubscription(unittest.TestCase):
cleanup_trace(full_path)
def test_callback(self):
session_name_prefix = 'session-test-subscription-callback'
base_path = '/tmp'
test_nodes = ['test_ping', 'test_pong']
exit_code, full_path = run_and_trace(base_path, session_name_prefix, subscription_callback_events, None, PKG, test_nodes)
self.assertEqual(exit_code, 0)
trace_events = get_trace_event_names(full_path)
print(f'trace_events: {trace_events}')
self.assertSetEqual(set(subscription_callback_events), trace_events)
cleanup_trace(full_path)
if __name__ == '__main__':
unittest.main()