Add tests for ros2trace with --dual-session option

Signed-off-by: Shravan Deva <[email protected]>

Author: ShravanDeva5327

test_ros2trace/test/test_ros2trace/test_trace.py

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import asyncio
 from collections.abc import Mapping
 import os
 import shutil
@@ -28,6 +29,8 @@
 from launch import LaunchService
 from launch_ros.actions import Node
 from lttngpy import impl as lttngpy
+import osrf_pycommon.process_utils
+from tracetools_launch.action import Trace
 from tracetools_read import get_event_name
 from tracetools_test.mark_process import get_corresponding_trace_test_events
 from tracetools_test.mark_process import get_trace_test_id
@@ -36,6 +39,7 @@
 from tracetools_trace.tools import tracepoints
 from tracetools_trace.tools.lttng import is_lttng_installed
 from tracetools_trace.tools.names import DEFAULT_EVENTS_ROS
+from tracetools_trace.tools.names import DEFAULT_INIT_EVENTS_ROS
 
 
 def are_tracepoints_included() -> bool:
@@ -295,6 +299,52 @@ def run_nodes(self, env: Optional[Mapping[str, str]] = None) -> None:
         exit_code = ls.run()
         self.assertEqual(0, exit_code)
 
+    def pre_configure_dual_session_and_run_nodes(
+        self,
+        base_path: str,
+        session_name: str,
+        env: Optional[Mapping[str, str]] = None,
+    ) -> Tuple[LaunchService, asyncio.Task]:
+        # Set trace test ID env var for spawned processes
+        additional_env: Dict[str, str] = {}
+        if env is not None:
+            additional_env.update(env)
+        assert self.trace_test_id
+        additional_env[TRACE_TEST_ID_ENV_VAR] = self.trace_test_id
+        actions = [
+            Trace(
+                session_name=session_name,
+                dual_session=True,
+                base_path=base_path,
+                events_ust=DEFAULT_INIT_EVENTS_ROS + [TRACE_TEST_ID_TP_NAME],
+            ),
+            Node(
+                package='test_tracetools',
+                executable='test_ping',
+                output='screen',
+                arguments=['do_more'],  # run indefinitely
+                additional_env=additional_env,
+            ),
+            Node(
+                package='test_tracetools',
+                executable='test_pong',
+                output='screen',
+                arguments=['do_more'],  # run indefinitely
+                additional_env=additional_env,
+            ),
+        ]
+        ld = LaunchDescription(actions)
+        ls = LaunchService()
+        ls.include_launch_description(ld)
+
+        loop = osrf_pycommon.process_utils.get_loop()
+        launch_task = loop.create_task(ls.run_async())
+
+        # Give the launch service time to start actions
+        loop.run_until_complete(asyncio.sleep(1.0))
+
+        return ls, launch_task
+
     def test_default(self) -> None:
         tmpdir = self.create_test_tmpdir('test_default')
 
@@ -671,7 +721,7 @@ def test_snapshot_start_pause_resume_stop(self) -> None:
         ret = self.run_trace_subcommand(
             [
                 'start', session_name,
-                '--ust', tracepoints.rcl_subscription_init, TRACE_TEST_ID_TP_NAME,
+                '--ust', 'ros2:*', TRACE_TEST_ID_TP_NAME,
                 '--path', tmpdir,
                 '--snapshot-mode',
             ]
@@ -736,6 +786,97 @@ def test_snapshot_start_pause_resume_stop(self) -> None:
 
         shutil.rmtree(tmpdir)
 
+    @unittest.skipIf(not are_tracepoints_included(), 'tracepoints are required')
+    def test_dual_session_start_pause_resume_stop(self) -> None:
+        tmpdir = self.create_test_tmpdir('test_dual_session_start_pause_resume_stop')
+        session_name = 'test_dual_session_start_pause_resume_stop'
+
+        # Pre-configure dual session, start nodes and check that the snapshot session exists
+        ls, launch_task = self.pre_configure_dual_session_and_run_nodes(tmpdir, session_name)
+        self.assertTracingSessionExist(session_name+'-snapshot', snapshot_mode=True)
+
+        # Start tracing
+        ret = self.run_trace_subcommand(
+            [
+                'start', session_name,
+                '--ust', 'ros2:*',
+                '--path', tmpdir,
+                '--dual-session',
+            ]
+        )
+        self.assertEqual(0, ret)
+        trace_dir = os.path.join(tmpdir, session_name)
+        snapshot_trace_dir = os.path.join(tmpdir, session_name, 'snapshot')
+        runtime_trace_dir = os.path.join(tmpdir, session_name, 'runtime')
+        self.assertTraceExist(snapshot_trace_dir)
+        self.assertTraceExist(runtime_trace_dir)
+        self.assertTracingSessionExist(session_name+'-snapshot', snapshot_mode=True)
+        self.assertTracingSessionExist(session_name+'-runtime')
+
+        # Pause tracing and check trace
+        ret = self.run_trace_subcommand(['pause', session_name, '--dual-session'])
+        self.assertEqual(0, ret)
+        self.assertTracingSessionExist(session_name+'-snapshot', snapshot_mode=True)
+        self.assertTracingSessionExist(session_name+'-runtime')
+        expected_trace_data = [
+            ('topic_name', '/ping'),
+            ('topic_name', '/pong'),
+        ]
+        num_events = self.assertTraceContains(trace_dir, expected_field_value=expected_trace_data)
+
+        # Pausing again should give an error but not affect anything
+        ret = self.run_trace_subcommand(['pause', session_name, '--dual-session'])
+        self.assertEqual(1, ret)
+        self.assertTracingSessionExist(session_name+'-snapshot', snapshot_mode=True)
+        self.assertTracingSessionExist(session_name+'-runtime')
+        new_num_events = self.assertTraceContains(
+            trace_dir=tmpdir,
+            expected_field_value=expected_trace_data,
+        )
+        self.assertEqual(num_events, new_num_events, 'unexpected new events in trace')
+
+        # Resume tracing
+        ret = self.run_trace_subcommand(['resume', session_name, '--dual-session'])
+        self.assertEqual(0, ret)
+        self.assertTracingSessionExist(session_name+'-snapshot', snapshot_mode=True)
+        self.assertTracingSessionExist(session_name+'-runtime')
+
+        # Resuming tracing again should give an error but not affect anything
+        ret = self.run_trace_subcommand(['resume', session_name, '--dual-session'])
+        self.assertEqual(1, ret)
+        self.assertTracingSessionExist(session_name+'-snapshot', snapshot_mode=True)
+        self.assertTracingSessionExist(session_name+'-runtime')
+
+        # Stop tracing and check that trace changed
+        ret = self.run_trace_subcommand(['stop', session_name, '--dual-session'])
+        self.assertEqual(0, ret)
+        # Snapshot session should still exist
+        self.assertTracingSessionExist(session_name+'-snapshot', snapshot_mode=True)
+        self.assertTracingSessionNotExist(session_name+'-runtime')
+        new_num_events = self.assertTraceContains(
+            trace_dir,
+            expected_field_value=expected_trace_data,
+        )
+        self.assertGreater(new_num_events, num_events, 'no new events in trace')
+
+        # Stopping tracing again should give an error but not affect anything
+        ret = self.run_trace_subcommand(['stop', session_name, '--dual-session'])
+        self.assertEqual(1, ret)
+        self.assertTracingSessionExist(session_name+'-snapshot', snapshot_mode=True)
+        self.assertTracingSessionNotExist(session_name+'-runtime')
+
+        # Shutdown launch service
+        loop = osrf_pycommon.process_utils.get_loop()
+        if not launch_task.done():
+            loop.create_task(ls.shutdown())
+            loop.run_until_complete(launch_task)
+        assert 0 == launch_task.result()
+
+        # Shutting down the launch service should destroy the snapshot session
+        self.assertTracingSessionNotExist(session_name+'-snapshot', snapshot_mode=True)
+
+        shutil.rmtree(tmpdir)
+
     @unittest.skipIf(not are_tracepoints_included(), 'tracepoints are required')
     def test_runtime_disable(self) -> None:
         tmpdir = self.create_test_tmpdir('test_runtime_disable')