Extract common test utils

tracetools_test/test/test_publisher.py

@@ -1,18 +1,8 @@
-import time
-import shutil
-from launch import LaunchDescription
-from launch import LaunchIntrospector
-from launch import LaunchService
-
-from launch_ros import get_default_launch_description
-import launch_ros.actions
 import unittest
-from tracetools_analysis.test.utils import get_trace_event_names
-from tracetools_trace.tools.lttng import (
-    lttng_setup,
-    lttng_start,
-    lttng_stop,
-    lttng_destroy,
+from tracetools_test.utils import (
+    get_trace_event_names,
+    run_and_trace,
+    cleanup_trace,
 )
 
 PKG = 'tracetools_test'
@@ -24,32 +14,18 @@ publisher_creation_events = [
 class TestPublisher(unittest.TestCase):
 
     def test_creation(self):
-        session_name = f'session-test-publisher-creation-{time.strftime("%Y%m%d%H%M%S")}'
-        path = '/tmp/' + session_name
-        print(f'trace directory: {path}')
+        session_name_prefix = 'session-test-publisher-creation'
+        base_path = '/tmp'
+        test_node = 'test_publisher'
 
-        lttng_setup(session_name, path, ros_events=publisher_creation_events, kernel_events=None)
-        lttng_start(session_name)
-
-        ld = LaunchDescription([
-            launch_ros.actions.Node(
-                package=PKG, node_executable='test_publisher', output='screen'),
-        ])
-        ls = LaunchService()
-        ls.include_launch_description(get_default_launch_description())
-        ls.include_launch_description(ld)
-
-        exit_code = ls.run()
+        exit_code, full_path = run_and_trace(session_name_prefix, base_path, publisher_creation_events, None, PKG, test_node)
         self.assertEqual(exit_code, 0)
 
-        lttng_stop(session_name)
-        lttng_destroy(session_name)
-
-        trace_events = get_trace_event_names(path)
+        trace_events = get_trace_event_names(full_path)
         print(f'trace_events: {trace_events}')
-        self.assertListEqual(publisher_creation_events, list(trace_events))
+        self.assertSetEqual(set(publisher_creation_events), trace_events)
 
-        shutil.rmtree(path)
+        cleanup_trace(full_path)
 
 
 if __name__ == '__main__':

tracetools_test/test/test_subscription.py

@@ -1,18 +1,8 @@
-import time
-import shutil
-from launch import LaunchDescription
-from launch import LaunchIntrospector
-from launch import LaunchService
-
-from launch_ros import get_default_launch_description
-import launch_ros.actions
 import unittest
-from tracetools_analysis.test.utils import get_trace_event_names
-from tracetools_trace.tools.lttng import (
-    lttng_setup,
-    lttng_start,
-    lttng_stop,
-    lttng_destroy,
+from tracetools_test.utils import (
+    get_trace_event_names,
+    run_and_trace,
+    cleanup_trace,
 )
 
 PKG = 'tracetools_test'
@@ -25,32 +15,18 @@ subscription_creation_events = [
 class TestSubscription(unittest.TestCase):
 
     def test_creation(self):
-        session_name = f'session-test-subscription-creation-{time.strftime("%Y%m%d%H%M%S")}'
-        path = '/tmp/' + session_name
-        print(f'trace directory: {path}')
+        session_name_prefix = 'session-test-subscription-creation'
+        base_path = '/tmp'
+        test_node = 'test_subscription'
 
-        lttng_setup(session_name, path, ros_events=subscription_creation_events, kernel_events=None)
-        lttng_start(session_name)
-
-        ld = LaunchDescription([
-            launch_ros.actions.Node(
-                package=PKG, node_executable='test_subscription', output='screen'),
-        ])
-        ls = LaunchService()
-        ls.include_launch_description(get_default_launch_description())
-        ls.include_launch_description(ld)
-
-        exit_code = ls.run()
+        exit_code, full_path = run_and_trace(session_name_prefix, base_path, subscription_creation_events, None, PKG, test_node)
         self.assertEqual(exit_code, 0)
 
-        lttng_stop(session_name)
-        lttng_destroy(session_name)
-
-        trace_events = get_trace_event_names(path)
+        trace_events = get_trace_event_names(full_path)
         print(f'trace_events: {trace_events}')
-        self.assertListEqual(subscription_creation_events, list(trace_events))
+        self.assertSetEqual(set(subscription_creation_events), trace_events)
 
-        shutil.rmtree(path)
+        cleanup_trace(full_path)
 
 
 if __name__ == '__main__':

tracetools_test/tracetools_test/utils.py

@@ -0,0 +1,60 @@
+# Utils for tracetools_test
+
+import time
+import shutil
+import subprocess
+import babeltrace
+from launch import LaunchDescription
+from launch import LaunchService
+from launch_ros import get_default_launch_description
+import launch_ros.actions
+from tracetools_trace.tools.lttng import (
+    lttng_setup,
+    lttng_start,
+    lttng_stop,
+    lttng_destroy,
+)
+
+def run_and_trace(session_name_prefix, base_path, ros_events, kernel_events, package_name, node_executable):
+    session_name = f'{session_name_prefix}-{time.strftime("%Y%m%d%H%M%S")}'
+    full_path = f'{base_path}/{session_name}'
+    print(f'trace directory: {full_path}')
+
+    lttng_setup(session_name, full_path, ros_events=ros_events, kernel_events=kernel_events)
+    lttng_start(session_name)
+
+    ld = LaunchDescription([
+        launch_ros.actions.Node(
+            package=package_name, node_executable=node_executable, output='screen'),
+    ])
+    ls = LaunchService()
+    ls.include_launch_description(get_default_launch_description())
+    ls.include_launch_description(ld)
+
+    exit_code = ls.run()
+
+    lttng_stop(session_name)
+    lttng_destroy(session_name)
+
+    return exit_code, full_path
+
+
+def cleanup_trace(full_path):
+    shutil.rmtree(full_path)
+
+
+def get_trace_event_names(trace_directory):
+    """
+    Get a set of event names in a trace
+    :param trace_directory (str): the path to the main/top trace directory
+    :return: event names (set(str))
+    """
+    tc = babeltrace.TraceCollection()
+    tc.add_traces_recursive(trace_directory, 'ctf')
+
+    event_names = set()
+
+    for event in tc.events:
+        event_names.add(event.name)
+
+    return event_names