Drops more legacy launch API use cases.

Signed-off-by: Michel Hidalgo <michel@ekumenlabs.com>

test_cli_remapping/package.xml

@@ -17,7 +17,7 @@
   <test_depend>ament_cmake_pytest</test_depend>
   <test_depend>ament_lint_auto</test_depend>
   <test_depend>ament_lint_common</test_depend>
-  <test_depend>launch</test_depend>
+  <test_depend>launch_testing</test_depend>
   <test_depend>rclcpp</test_depend>
   <test_depend>rclpy</test_depend>
   <test_depend>test_msgs</test_depend>

test_cli_remapping/test/test_cli_remapping.py

@@ -87,7 +87,7 @@ def test_func(node_fixture):
             launch_test.add_test_action(ld, OpaqueCoroutine(
                 coroutine=coroutine_test, args=[node_fixture], ignore_context=True
             ))
-            launch_service = LaunchService(debug=True)
+            launch_service = LaunchService()
             launch_service.include_launch_description(ld)
             return_code = launch_test.run(launch_service)
             assert return_code == 0, 'Launch failed with exit code %r' % (return_code,)

test_communication/package.xml

@@ -23,7 +23,7 @@
   <test_depend>ament_cmake_pytest</test_depend>
   <test_depend>ament_lint_auto</test_depend>
   <test_depend>ament_lint_common</test_depend>
-  <test_depend>launch</test_depend>
+  <test_depend>launch_testing</test_depend>
   <test_depend>osrf_testing_tools_cpp</test_depend>
   <test_depend>rcl</test_depend>
   <test_depend>rclcpp</test_depend>

test_communication/test/test_action_client_server.py.in

@@ -4,15 +4,18 @@ import os
 import sys
 import time
 
-from launch.legacy import LaunchDescriptor
-from launch.legacy.exit_handler import primary_exit_handler
-from launch.legacy.launcher import DefaultLauncher
+from launch import LaunchDescription
+from launch import LaunchService
+from launch.actions import ExecuteProcess
+from launch_testing import LaunchTestService
 
 
 def test_action_client_server():
     namespace = '/test_time_%s' % time.strftime('%H_%M_%S', time.gmtime())
 
-    ld = LaunchDescriptor()
+    launch_test = LaunchTestService()
+    launch_description = LaunchDescription()
+
     action_client_cmd = ['@TEST_ACTION_CLIENT_EXECUTABLE@', '@TEST_ACTION_TYPE@', namespace]
     action_server_cmd = ['@TEST_ACTION_SERVER_EXECUTABLE@', '@TEST_ACTION_TYPE@', namespace]
 
@@ -28,24 +31,29 @@ def test_action_client_server():
 
     action_server_env['RCL_ASSERT_RMW_ID_MATCHES'] = '@ACTION_SERVER_RMW@'
     action_server_env['RMW_IMPLEMENTATION'] = '@ACTION_SERVER_RMW@'
-    ld.add_process(
-        cmd=action_server_cmd,
-        name='test_action_server',
-        env=action_server_env,
+
+    launch_test.add_fixture_action(
+        launch_description, ExecuteProcess(
+            cmd=action_server_cmd,
+            name='test_publisher',
+            env=action_server_env,
+        )
     )
 
     action_client_env['RCL_ASSERT_RMW_ID_MATCHES'] = '@ACTION_CLIENT_RMW@'
     action_client_env['RMW_IMPLEMENTATION'] = '@ACTION_CLIENT_RMW@'
-    ld.add_process(
-        cmd=action_client_cmd,
-        name='test_action_client',
-        env=action_client_env,
-        exit_handler=primary_exit_handler,
+
+    launch_test.add_test_action(
+        launch_description, ExecuteProcess(
+            cmd=action_client_cmd,
+            name='test_action_client',
+            env=action_client_env,
+        )
     )
 
-    launcher = DefaultLauncher()
-    launcher.add_launch_descriptor(ld)
-    rc = launcher.launch()
+    launch_service = LaunchService()
+    launch_service.include_launch_description(launch_description)
+    rc = launch_test.run(launch_service)
 
     assert rc == 0, \
         "The launch file failed with exit code '" + str(rc) + "'. " \

test_communication/test/test_publisher_subscriber.py.in

@@ -4,14 +4,18 @@ import os
 import sys
 import time
 
-from launch.legacy import LaunchDescriptor
-from launch.legacy.exit_handler import primary_exit_handler
-from launch.legacy.launcher import DefaultLauncher
+from launch import LaunchDescription
+from launch import LaunchService
+from launch.actions import ExecuteProcess
+from launch_testing import LaunchTestService
 
 
 def test_publisher_subscriber():
     namespace = '/test_time_%s' % time.strftime('%H_%M_%S', time.gmtime())
-    ld = LaunchDescriptor()
+
+    launch_test = LaunchTestService()
+    launch_description = LaunchDescription()
+
     publisher_cmd = ['@TEST_PUBLISHER_EXECUTABLE@', '@TEST_MESSAGE_TYPE@', namespace]
     subscriber_cmd = ['@TEST_SUBSCRIBER_EXECUTABLE@', '@TEST_MESSAGE_TYPE@', namespace]
 
@@ -27,24 +31,29 @@ def test_publisher_subscriber():
 
     publisher_env['RCL_ASSERT_RMW_ID_MATCHES'] = '@PUBLISHER_RMW@'
     publisher_env['RMW_IMPLEMENTATION'] = '@PUBLISHER_RMW@'
-    ld.add_process(
-        cmd=publisher_cmd,
-        name='test_publisher',
-        env=publisher_env,
+
+    launch_test.add_fixture_action(
+        launch_description, ExecuteProcess(
+            cmd=publisher_cmd,
+            name='test_publisher',
+            env=publisher_env,
+        )
     )
 
     subscriber_env['RCL_ASSERT_RMW_ID_MATCHES'] = '@SUBSCRIBER_RMW@'
     subscriber_env['RMW_IMPLEMENTATION'] = '@SUBSCRIBER_RMW@'
-    ld.add_process(
-        cmd=subscriber_cmd,
-        name='test_subscriber',
-        env=subscriber_env,
-        exit_handler=primary_exit_handler,
+
+    launch_test.add_test_action(
+        launch_description, ExecuteProcess(
+            cmd=subscriber_cmd,
+            name='test_subscriber',
+            env=subscriber_env,
+        )
     )
 
-    launcher = DefaultLauncher()
-    launcher.add_launch_descriptor(ld)
-    rc = launcher.launch()
+    launch_service = LaunchService()
+    launch_service.include_launch_description(launch_description)
+    rc = launch_test.run(launch_service)
 
     assert rc == 0, \
         "The launch file failed with exit code '" + str(rc) + "'. " \

test_communication/test/test_requester_replier.py.in

@@ -4,15 +4,18 @@ import os
 import sys
 import time
 
-from launch.legacy import LaunchDescriptor
-from launch.legacy.exit_handler import primary_exit_handler
-from launch.legacy.launcher import DefaultLauncher
+from launch import LaunchDescription
+from launch import LaunchService
+from launch.actions import ExecuteProcess
+from launch_testing import LaunchTestService
 
 
 def test_requester_replier():
     namespace = '/test_time_%s' % time.strftime('%H_%M_%S', time.gmtime())
 
-    ld = LaunchDescriptor()
+    launch_test = LaunchTestService()
+    launch_description = LaunchDescription()
+
     requester_cmd = ['@TEST_REQUESTER_EXECUTABLE@', '@TEST_SERVICE_TYPE@', namespace]
     replier_cmd = ['@TEST_REPLIER_EXECUTABLE@', '@TEST_SERVICE_TYPE@', namespace]
 
@@ -28,24 +31,29 @@ def test_requester_replier():
 
     replier_env['RCL_ASSERT_RMW_ID_MATCHES'] = '@REPLIER_RMW@'
     replier_env['RMW_IMPLEMENTATION'] = '@REPLIER_RMW@'
-    ld.add_process(
-        cmd=replier_cmd,
-        name='test_replier',
-        env=replier_env,
+
+    launch_test.add_fixture_action(
+        launch_description, ExecuteProcess(
+            cmd=replier_cmd,
+            name='test_replier',
+            env=replier_env,
+        )
     )
 
     requester_env['RCL_ASSERT_RMW_ID_MATCHES'] = '@REQUESTER_RMW@'
     requester_env['RMW_IMPLEMENTATION'] = '@REQUESTER_RMW@'
-    ld.add_process(
-        cmd=requester_cmd,
-        name='test_requester',
-        env=requester_env,
-        exit_handler=primary_exit_handler,
+
+    launch_test.add_test_action(
+        launch_description, ExecuteProcess(
+            cmd=requester_cmd,
+            name='test_requester',
+            env=requester_env,
+        )
     )
 
-    launcher = DefaultLauncher()
-    launcher.add_launch_descriptor(ld)
-    rc = launcher.launch()
+    launch_service = LaunchService()
+    launch_service.include_launch_description(launch_description)
+    rc = launch_test.run(launch_service)
 
     assert rc == 0, \
         "The launch file failed with exit code '" + str(rc) + "'. " \

test_rclcpp/package.xml

@@ -16,7 +16,6 @@
   <test_depend>ament_cmake_pytest</test_depend>
   <test_depend>ament_lint_auto</test_depend>
   <test_depend>ament_lint_common</test_depend>
-  <test_depend>launch</test_depend>
   <test_depend>launch_testing</test_depend>
   <test_depend>rclcpp</test_depend>
   <test_depend>rmw_implementation</test_depend>

test_rclcpp/test/test_executable_output.py.in

@@ -1,52 +1,55 @@
 # generated from test_rclcpp/test/test_executable_output.py.in
 # generated code does not contain a copyright notice
 
-from launch.legacy import LaunchDescriptor
-from launch.legacy.exit_handler import primary_exit_handler
-from launch.legacy.launcher import DefaultLauncher
-from launch.legacy.output_handler import ConsoleOutput
-
-from launch_testing import create_handler
+from launch import LaunchDescription
+from launch import LaunchService
+from launch.actions import ExecuteProcess
+from launch_testing import LaunchTestService
+from launch_testing.output import create_output_lines_filter
+from launch_testing.output import create_output_test_from_file
 
 
 def @TEST_NAME@():
-    ld = LaunchDescriptor()
-
-    output_handlers = [ConsoleOutput()]
-
-    shutdown_trigger_handler = create_handler(
-        '@TEST_EXECUTABLE_NAME@',
-        ld,
-        '@TEST_EXECUTABLE_TRIGGER_SHUTDOWN_OUTPUT@',
-        exit_on_match=True,
-        filtered_rmw_implementation='@rmw_implementation@',
+    launch_test = LaunchTestService()
+    launch_description = LaunchDescription()
+
+    action = launch_test.add_fixture_action(
+        launch_description, ExecuteProcess(
+            cmd=['@TEST_EXECUTABLE@'],
+            name='@TEST_EXECUTABLE_NAME@',
+            sigterm_timeout='15',
+            output='screen'
+        ), exit_allowed=True
     )
-    output_handlers.append(shutdown_trigger_handler)
-
-    output_check_handler = create_handler(
-        '@TEST_EXECUTABLE_NAME@',
-        ld,
-        '@TEST_EXECUTABLE_EXPECTED_OUTPUT@',
-        exit_on_match=False,
-        filtered_rmw_implementation='@rmw_implementation@',
+
+    launch_test.add_output_test(
+        launch_description, action,
+        output_test=create_output_test_from_file(
+            output_file='@TEST_EXECUTABLE_EXPECTED_OUTPUT@'
+        ),
+        output_filter=create_output_lines_filter(
+            filtered_rmw_implementation='@rmw_implementation@'
+        )
     )
-    output_handlers.append(output_check_handler)
 
-    ld.add_process(
-        cmd=['@TEST_EXECUTABLE@'],
-        name='@TEST_EXECUTABLE_NAME@',
-        exit_handler=primary_exit_handler,
-        output_handlers=output_handlers,
+    launch_test.add_output_test(
+        launch_description, action,
+        output_test=create_output_test_from_file(
+            output_file='@TEST_EXECUTABLE_TRIGGER_SHUTDOWN_OUTPUT@'
+        ),
+        output_filter=create_output_lines_filter(
+            filtered_rmw_implementation='@rmw_implementation@'
+        ),
+        test_suffix='trigger_shutdown_output',
+        side_effect='shutdown'
     )
 
-    launcher = DefaultLauncher()
-    launcher.add_launch_descriptor(ld)
-    rc = launcher.launch()
+    launch_service = LaunchService()
+    launch_service.include_launch_description(launch_description)
+    rc = launch_test.run(launch_service)
 
     assert rc == 0, "The launch file failed with exit code '" + str(rc) + "'. "
 
-    output_check_handler.check()
-
 
 if __name__ == '__main__':
     @TEST_NAME@()

test_rclcpp/test/test_n_nodes.py.in

@@ -2,37 +2,42 @@
 
 import os
 
-from launch.legacy import LaunchDescriptor
-from launch.legacy.exit_handler import primary_exit_handler
-from launch.legacy.launcher import DefaultLauncher
+from launch import LaunchDescription
+from launch import LaunchService
+from launch.actions import ExecuteProcess
+from launch_testing import LaunchTestService
 
 
 def test_@TEST_NUM_NODES@_nodes():
-    ld = LaunchDescriptor()
+    launch_test = LaunchTestService()
+    launch_description = LaunchDescription()
 
     env = None
     if '@TEST_RMW_IMPLEMENTATION@':
         env = dict(os.environ)
         env['RMW_IMPLEMENTATION'] = '@TEST_RMW_IMPLEMENTATION@'
 
     for i in range(@TEST_NUM_NODES@):
-        ld.add_process(
-            cmd=['@TEST_EXECUTABLE1@'],
-            name='node_with_name_' + str(i),
-            env=env,
+        launch_test.add_fixture_action(
+            launch_description, ExecuteProcess(
+                cmd=['@TEST_EXECUTABLE1@'],
+                name='node_with_name_' + str(i),
+                env=env,
+            )
         )
 
     cmd = ['@TEST_EXECUTABLE2@'] + '@TEST_NUM_NODES@'.split(' ')
-    ld.add_process(
-        cmd=cmd,
-        name='node_check_names',
-        env=env,
-        exit_handler=primary_exit_handler,
+    launch_test.add_test_action(
+        launch_description, ExecuteProcess(
+            cmd=cmd,
+            name='node_check_names',
+            env=env
+        )
     )
 
-    launcher = DefaultLauncher()
-    launcher.add_launch_descriptor(ld)
-    rc = launcher.launch()
+    launch_service = LaunchService()
+    launch_service.include_launch_description(launch_description)
+    rc = launch_test.run(launch_service)
 
     assert rc == 0, "The launch file failed with exit code '" + str(rc) + "'. "