diff --git a/composition/CMakeLists.txt b/composition/CMakeLists.txt index c0a861ac7..b88054803 100644 --- a/composition/CMakeLists.txt +++ b/composition/CMakeLists.txt @@ -130,6 +130,7 @@ if(BUILD_TESTING) ament_lint_auto_find_test_dependencies() find_package(ament_cmake_pytest REQUIRED) + find_package(launch_testing_ament_cmake REQUIRED) find_package(rmw_implementation_cmake REQUIRED) file(GENERATE @@ -165,12 +166,14 @@ if(BUILD_TESTING) OUTPUT "${CMAKE_CURRENT_BINARY_DIR}/${test_name}${target_suffix}_$.py" INPUT "${CMAKE_CURRENT_BINARY_DIR}/${test_name}${target_suffix}.py.genexp" ) - ament_add_pytest_test(${test_name}${target_suffix} + add_launch_test( "${CMAKE_CURRENT_BINARY_DIR}/${test_name}${target_suffix}_$.py" + TARGET ${test_name}${target_suffix} ENV RMW_IMPLEMENTATION=${rmw_implementation} APPEND_ENV AMENT_PREFIX_PATH=${CMAKE_CURRENT_BINARY_DIR}/test_ament_index/$ APPEND_LIBRARY_DIRS "${append_library_dirs}" - TIMEOUT 60) + TIMEOUT 60 + ) list( APPEND generated_python_files "${CMAKE_CURRENT_BINARY_DIR}/${test_name}${target_suffix}_$.py") diff --git a/composition/package.xml b/composition/package.xml index 7b2eb4fb5..fe0dc6ef7 100644 --- a/composition/package.xml +++ b/composition/package.xml @@ -27,6 +27,8 @@ ament_lint_common launch launch_testing + launch_testing_ros + launch_testing_ament_cmake rmw_implementation_cmake diff --git a/composition/test/test_api_pubsub_composition.py.in b/composition/test/test_api_pubsub_composition.py.in index d12ead8ec..442192538 100644 --- a/composition/test/test_api_pubsub_composition.py.in +++ b/composition/test/test_api_pubsub_composition.py.in @@ -12,67 +12,58 @@ # See the License for the specific language governing permissions and # limitations under the License. +import unittest + 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 - +from launch.actions import OpaqueFunction -def test_api_pubsub_composition(): - name = 'test_api_composition' - executable = '@API_COMPOSITION_EXECUTABLE@' - additional_processes = [ - { - 'cmd': [ - '@API_COMPOSITION_CLI_EXECUTABLE@', - 'composition', 'composition::Talker'], - 'name': 'load_talker_component', - }, - { - 'cmd': [ - '@API_COMPOSITION_CLI_EXECUTABLE@', - 'composition', 'composition::Listener'], - 'name': 'load_listener_component', - } - ] - output_file = '@EXPECTED_OUTPUT_PUBSUB@' - launch(name, [executable], output_file, additional_processes=additional_processes) +import launch_testing +import launch_testing.asserts +import launch_testing_ros -def launch(name, cmd, output_file, additional_processes=None): - launch_test = LaunchTestService() +def generate_test_description(ready_fn): launch_description = LaunchDescription() - - action = launch_test.add_fixture_action( - launch_description, ExecuteProcess( - cmd=cmd, - name=name, - output='screen' - ) + process_under_test = ExecuteProcess( + cmd=['@API_COMPOSITION_EXECUTABLE@'], + name='test_api_composition', + output='screen' ) - rmw_implementation = '@rmw_implementation@' - launch_test.add_output_test( - launch_description, action, - output_test=create_output_test_from_file(output_file), - output_filter=create_output_lines_filter( - filtered_rmw_implementation=rmw_implementation + launch_description.add_action(process_under_test) + launch_description.add_action( + ExecuteProcess( + cmd=[ + '@API_COMPOSITION_CLI_EXECUTABLE@', + 'composition', 'composition::Talker' + ], + name='load_talker_component' ) ) - for additional_process in (additional_processes or []): - launch_test.add_fixture_action( - launch_description, ExecuteProcess(**additional_process) + launch_description.add_action( + ExecuteProcess( + cmd=[ + '@API_COMPOSITION_CLI_EXECUTABLE@', + 'composition', 'composition::Listener' + ], + name='load_listener_component' ) + ) + launch_description.add_action( + OpaqueFunction(function=lambda context: ready_fn()) + ) + return launch_description, locals() - 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) + "'. " \ - 'Maybe the client did not receive any messages?' +class TestComposition(unittest.TestCase): -if __name__ == '__main__': - test_api_pubsub_composition() + def test_api_pubsub_composition(self, proc_output, process_under_test): + """Test process' output against expectations.""" + output_filter = launch_testing_ros.tools.basic_output_filter( + filtered_rmw_implementation='@rmw_implementation@' + ) + proc_output.assertWaitFor( + expected_output=launch_testing.tools.expected_output_from_file( + path='@EXPECTED_OUTPUT_PUBSUB@' + ), process=process_under_test, output_filter=output_filter, timeout=10 + ) diff --git a/composition/test/test_api_srv_composition.py.in b/composition/test/test_api_srv_composition.py.in index 35699373f..a61bb5811 100644 --- a/composition/test/test_api_srv_composition.py.in +++ b/composition/test/test_api_srv_composition.py.in @@ -12,67 +12,58 @@ # See the License for the specific language governing permissions and # limitations under the License. +import unittest + 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 - +from launch.actions import OpaqueFunction -def test_api_srv_composition(): - name = 'test_api_composition' - executable = '@API_COMPOSITION_EXECUTABLE@' - additional_processes = [ - { - 'cmd': [ - '@API_COMPOSITION_CLI_EXECUTABLE@', - 'composition', 'composition::Server'], - 'name': 'load_server_component', - }, - { - 'cmd': [ - '@API_COMPOSITION_CLI_EXECUTABLE@', - 'composition', 'composition::Client'], - 'name': 'load_client_component', - }, - ] - output_file = '@EXPECTED_OUTPUT_SRV@' - launch(name, [executable], output_file, additional_processes=additional_processes) +import launch_testing +import launch_testing.asserts +import launch_testing_ros -def launch(name, cmd, output_file, additional_processes=None): - launch_test = LaunchTestService() +def generate_test_description(ready_fn): launch_description = LaunchDescription() - - action = launch_test.add_fixture_action( - launch_description, ExecuteProcess( - cmd=cmd, - name=name, - output='screen' - ) + process_under_test = ExecuteProcess( + cmd=['@API_COMPOSITION_EXECUTABLE@'], + name='test_api_composition', + output='screen' ) - rmw_implementation = '@rmw_implementation@' - launch_test.add_output_test( - launch_description, action, - output_test=create_output_test_from_file(output_file), - output_filter=create_output_lines_filter( - filtered_rmw_implementation=rmw_implementation + launch_description.add_action(process_under_test) + launch_description.add_action( + ExecuteProcess( + cmd=[ + '@API_COMPOSITION_CLI_EXECUTABLE@', + 'composition', 'composition::Server' + ], + name='load_server_component' ) ) - for additional_process in (additional_processes or []): - launch_test.add_fixture_action( - launch_description, ExecuteProcess(**additional_process) + launch_description.add_action( + ExecuteProcess( + cmd=[ + '@API_COMPOSITION_CLI_EXECUTABLE@', + 'composition', 'composition::Client' + ], + name='load_client_component' ) + ) + launch_description.add_action( + OpaqueFunction(function=lambda context: ready_fn()) + ) + return launch_description, locals() - 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) + "'. " \ - 'Maybe the client did not receive any messages?' +class TestComposition(unittest.TestCase): -if __name__ == '__main__': - test_api_srv_composition() + def test_api_srv_composition(self, proc_output, process_under_test): + """Test process' output against expectations.""" + output_filter = launch_testing_ros.tools.basic_output_filter( + filtered_rmw_implementation='@rmw_implementation@' + ) + proc_output.assertWaitFor( + expected_output=launch_testing.tools.expected_output_from_file( + path='@EXPECTED_OUTPUT_SRV@' + ), process=process_under_test, output_filter=output_filter, timeout=10 + ) diff --git a/composition/test/test_api_srv_composition_client_first.py.in b/composition/test/test_api_srv_composition_client_first.py.in index f42726c0c..904186d6d 100644 --- a/composition/test/test_api_srv_composition_client_first.py.in +++ b/composition/test/test_api_srv_composition_client_first.py.in @@ -12,69 +12,59 @@ # See the License for the specific language governing permissions and # limitations under the License. +import unittest + 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 - +from launch.actions import OpaqueFunction -def test_api_srv_composition_client_first(): - # This is a regression test to ensure that the client starting first and waiting for a service - # does not block the executor from processing requests to load other components. - name = 'test_api_composition' - executable = '@API_COMPOSITION_EXECUTABLE@' - additional_processes = [ - { - 'cmd': [ - '@API_COMPOSITION_CLI_EXECUTABLE@', - 'composition', 'composition::Client'], - 'name': 'load_client_component', - }, - { - 'cmd': [ - '@API_COMPOSITION_CLI_EXECUTABLE@', - 'composition', 'composition::Server', '--delay', '5000'], - 'name': 'load_server_component', - }, - ] - output_file = '@EXPECTED_OUTPUT_SRV@' - launch(name, [executable], output_file, additional_processes=additional_processes) +import launch_testing +import launch_testing.asserts +import launch_testing_ros -def launch(name, cmd, output_file, additional_processes=None): - launch_test = LaunchTestService() +def generate_test_description(ready_fn): launch_description = LaunchDescription() - - action = launch_test.add_fixture_action( - launch_description, ExecuteProcess( - cmd=cmd, - name=name, - output='screen' - ) + process_under_test = ExecuteProcess( + cmd=['@API_COMPOSITION_EXECUTABLE@'], + name='test_api_composition', + output='screen' ) - rmw_implementation = '@rmw_implementation@' - launch_test.add_output_test( - launch_description, action, - output_test=create_output_test_from_file(output_file), - output_filter=create_output_lines_filter( - filtered_rmw_implementation=rmw_implementation + launch_description.add_action(process_under_test) + launch_description.add_action( + ExecuteProcess( + cmd=[ + '@API_COMPOSITION_CLI_EXECUTABLE@', + 'composition', 'composition::Client' + ], + name='load_client_component' ) ) - for additional_process in (additional_processes or []): - launch_test.add_fixture_action( - launch_description, ExecuteProcess(**additional_process) + launch_description.add_action( + ExecuteProcess( + cmd=[ + '@API_COMPOSITION_CLI_EXECUTABLE@', + 'composition', 'composition::Server', + '--delay', '5000' + ], + name='load_server_component' ) + ) + launch_description.add_action( + OpaqueFunction(function=lambda context: ready_fn()) + ) + return launch_description, locals() - 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) + "'. " \ - 'Maybe the client did not receive any messages?' +class TestComposition(unittest.TestCase): -if __name__ == '__main__': - test_api_srv_composition_client_first() + def test_api_srv_composition_client_first(self, proc_output, process_under_test): + """Test process' output against expectations.""" + output_filter = launch_testing_ros.tools.basic_output_filter( + filtered_rmw_implementation='@rmw_implementation@' + ) + proc_output.assertWaitFor( + expected_output=launch_testing.tools.expected_output_from_file( + path='@EXPECTED_OUTPUT_SRV@' + ), process=process_under_test, output_filter=output_filter, timeout=15 + ) diff --git a/composition/test/test_dlopen_composition.py.in b/composition/test/test_dlopen_composition.py.in index c0a95cb7b..e9d3034df 100644 --- a/composition/test/test_dlopen_composition.py.in +++ b/composition/test/test_dlopen_composition.py.in @@ -12,60 +12,46 @@ # See the License for the specific language governing permissions and # limitations under the License. +import unittest + 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 - +from launch.actions import OpaqueFunction -def test_dlopen_composition(): - name = 'test_dlopen_composition' - executable = '@DLOPEN_COMPOSITION_EXECUTABLE@' - talker_library = '@TALKER_LIBRARY@' - listener_library = '@LISTENER_LIBRARY@' - server_library = '@SERVER_LIBRARY@' - client_library = '@CLIENT_LIBRARY@' - output_file = '@EXPECTED_OUTPUT_ALL@' - launch( - name, - [executable, talker_library, listener_library, server_library, client_library], - output_file) +import launch_testing +import launch_testing.asserts +import launch_testing_ros -def launch(name, cmd, output_file, additional_processes=None): - launch_test = LaunchTestService() +def generate_test_description(ready_fn): launch_description = LaunchDescription() - - action = launch_test.add_fixture_action( - launch_description, ExecuteProcess( - cmd=cmd, - name=name, - output='screen' - ) + process_under_test = ExecuteProcess( + cmd=[ + '@DLOPEN_COMPOSITION_EXECUTABLE@', + '@TALKER_LIBRARY@', + '@LISTENER_LIBRARY@', + '@SERVER_LIBRARY@', + '@CLIENT_LIBRARY@' + ], + name='test_dlopen_composition', + output='screen' ) - rmw_implementation = '@rmw_implementation@' - launch_test.add_output_test( - launch_description, action, - output_test=create_output_test_from_file(output_file), - output_filter=create_output_lines_filter( - filtered_rmw_implementation=rmw_implementation - ) + launch_description.add_action(process_under_test) + launch_description.add_action( + OpaqueFunction(function=lambda context: ready_fn()) ) - for additional_process in (additional_processes or []): - launch_test.add_fixture_action( - launch_description, ExecuteProcess(**additional_process) - ) - - launch_service = LaunchService() - launch_service.include_launch_description(launch_description) - rc = launch_test.run(launch_service) + return launch_description, locals() - assert rc == 0, \ - "The launch file failed with exit code '" + str(rc) + "'. " \ - 'Maybe the client did not receive any messages?' +class TestComposition(unittest.TestCase): -if __name__ == '__main__': - test_dlopen_composition() + def test_dlopen_composition(self, proc_output, process_under_test): + """Test process' output against expectations.""" + output_filter = launch_testing_ros.tools.basic_output_filter( + filtered_rmw_implementation='@rmw_implementation@' + ) + proc_output.assertWaitFor( + expected_output=launch_testing.tools.expected_output_from_file( + path='@EXPECTED_OUTPUT_ALL@' + ), process=process_under_test, output_filter=output_filter, timeout=10 + ) diff --git a/composition/test/test_linktime_composition.py.in b/composition/test/test_linktime_composition.py.in index bac73a0e2..8cf30a294 100644 --- a/composition/test/test_linktime_composition.py.in +++ b/composition/test/test_linktime_composition.py.in @@ -13,58 +13,45 @@ # limitations under the License. import os + +import unittest from unittest.case import SkipTest 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 - +from launch.actions import OpaqueFunction -def test_linktime_composition(): - if os.name == 'nt': - print('The link time registration of classes does not work on Windows') - raise SkipTest - name = 'test_linktime_composition' - executable = '@LINKTIME_COMPOSITION_EXECUTABLE@' - output_file = '@EXPECTED_OUTPUT_ALL@' - launch(name, [executable], output_file) +import launch_testing +import launch_testing.asserts +import launch_testing_ros -def launch(name, cmd, output_file, additional_processes=None): - launch_test = LaunchTestService() +def generate_test_description(ready_fn): launch_description = LaunchDescription() - - action = launch_test.add_fixture_action( - launch_description, ExecuteProcess( - cmd=cmd, - name=name, - output='screen' - ) + process_under_test = ExecuteProcess( + cmd=['@LINKTIME_COMPOSITION_EXECUTABLE@'], + name='test_linktime_composition', + output='screen' ) - rmw_implementation = '@rmw_implementation@' - launch_test.add_output_test( - launch_description, action, - output_test=create_output_test_from_file(output_file), - output_filter=create_output_lines_filter( - filtered_rmw_implementation=rmw_implementation - ) + launch_description.add_action(process_under_test) + launch_description.add_action( + OpaqueFunction(function=lambda context: ready_fn()) ) - for additional_process in (additional_processes or []): - launch_test.add_fixture_action( - launch_description, ExecuteProcess(**additional_process) - ) - - launch_service = LaunchService() - launch_service.include_launch_description(launch_description) - rc = launch_test.run(launch_service) + return launch_description, locals() - assert rc == 0, \ - "The launch file failed with exit code '" + str(rc) + "'. " \ - 'Maybe the client did not receive any messages?' +class TestComposition(unittest.TestCase): -if __name__ == '__main__': - test_linktime_composition() + def test_linktime_composition(self, proc_output, process_under_test): + """Test process' output against expectations.""" + if os.name == 'nt': + print('The link time registration of classes does not work on Windows') + raise SkipTest + output_filter = launch_testing_ros.tools.basic_output_filter( + filtered_rmw_implementation='@rmw_implementation@' + ) + proc_output.assertWaitFor( + expected_output=launch_testing.tools.expected_output_from_file( + path='@EXPECTED_OUTPUT_ALL@' + ), process=process_under_test, output_filter=output_filter, timeout=10 + ) diff --git a/composition/test/test_manual_composition.py.in b/composition/test/test_manual_composition.py.in index d47893086..8c34c94e9 100644 --- a/composition/test/test_manual_composition.py.in +++ b/composition/test/test_manual_composition.py.in @@ -12,53 +12,40 @@ # See the License for the specific language governing permissions and # limitations under the License. +import unittest + 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 - +from launch.actions import OpaqueFunction -def test_manual_composition(): - name = 'test_manual_composition' - executable = '@MANUAL_COMPOSITION_EXECUTABLE@' - output_file = '@EXPECTED_OUTPUT_ALL@' - launch(name, [executable], output_file) +import launch_testing +import launch_testing.asserts +import launch_testing_ros -def launch(name, cmd, output_file, additional_processes=None): - launch_test = LaunchTestService() +def generate_test_description(ready_fn): launch_description = LaunchDescription() - - action = launch_test.add_fixture_action( - launch_description, ExecuteProcess( - cmd=cmd, - name=name, - output='screen' - ) + process_under_test = ExecuteProcess( + cmd=['@MANUAL_COMPOSITION_EXECUTABLE@'], + name='test_manual_composition', + output='screen' ) - rmw_implementation = '@rmw_implementation@' - launch_test.add_output_test( - launch_description, action, - output_test=create_output_test_from_file(output_file), - output_filter=create_output_lines_filter( - filtered_rmw_implementation=rmw_implementation - ) + launch_description.add_action(process_under_test) + launch_description.add_action( + OpaqueFunction(function=lambda context: ready_fn()) ) - for additional_process in (additional_processes or []): - launch_test.add_fixture_action( - launch_description, ExecuteProcess(**additional_process) - ) - - launch_service = LaunchService() - launch_service.include_launch_description(launch_description) - rc = launch_test.run(launch_service) + return launch_description, locals() - assert rc == 0, \ - "The launch file failed with exit code '" + str(rc) + "'. " \ - 'Maybe the client did not receive any messages?' +class TestComposition(unittest.TestCase): -if __name__ == '__main__': - test_manual_composition() + def test_manual_composition(self, proc_output, process_under_test): + """Test process' output against expectations.""" + output_filter = launch_testing_ros.tools.basic_output_filter( + filtered_rmw_implementation='@rmw_implementation@' + ) + proc_output.assertWaitFor( + expected_output=launch_testing.tools.expected_output_from_file( + path='@EXPECTED_OUTPUT_ALL@' + ), process=process_under_test, output_filter=output_filter, timeout=10 + ) diff --git a/demo_nodes_cpp/CMakeLists.txt b/demo_nodes_cpp/CMakeLists.txt index be2458306..bbaeb99f6 100644 --- a/demo_nodes_cpp/CMakeLists.txt +++ b/demo_nodes_cpp/CMakeLists.txt @@ -60,8 +60,8 @@ if(BUILD_TESTING) ament_lint_auto_find_test_dependencies() find_package(ament_cmake_pytest REQUIRED) + find_package(launch_testing_ament_cmake REQUIRED) find_package(rmw_implementation_cmake REQUIRED) - # Add each test case. Multi-executable tests can be specified in # semicolon-separated strings, like exe1;exe2. set(tutorial_tests @@ -104,12 +104,14 @@ if(BUILD_TESTING) INPUT "${CMAKE_CURRENT_BINARY_DIR}/test_${exe_list_underscore}${target_suffix}.py.configured" ) - ament_add_pytest_test(test_tutorial_${exe_list_underscore}${target_suffix} + add_launch_test( "${CMAKE_CURRENT_BINARY_DIR}/test_${exe_list_underscore}${target_suffix}_$.py" - TIMEOUT 30 + TARGET test_tutorial_${exe_list_underscore}${target_suffix} + TIMEOUT 60 ENV RCL_ASSERT_RMW_ID_MATCHES=${rmw_implementation} - RMW_IMPLEMENTATION=${rmw_implementation}) + RMW_IMPLEMENTATION=${rmw_implementation} + ) foreach(executable ${tutorial_executables}) set_property( TEST test_tutorial_${exe_list_underscore}${target_suffix} diff --git a/demo_nodes_cpp/package.xml b/demo_nodes_cpp/package.xml index 2dec7f712..3cca3f597 100644 --- a/demo_nodes_cpp/package.xml +++ b/demo_nodes_cpp/package.xml @@ -30,6 +30,8 @@ ament_lint_common launch launch_testing + launch_testing_ament_cmake + launch_testing_ros ament_cmake diff --git a/demo_nodes_cpp/src/parameters/parameter_events.cpp b/demo_nodes_cpp/src/parameters/parameter_events.cpp index 971705c20..3981f1e1b 100644 --- a/demo_nodes_cpp/src/parameters/parameter_events.cpp +++ b/demo_nodes_cpp/src/parameters/parameter_events.cpp @@ -50,12 +50,6 @@ int main(int argc, char ** argv) auto node = rclcpp::Node::make_shared("parameter_events"); - // Declare parameters that may be set on this node - node->declare_parameter("foo"); - node->declare_parameter("bar"); - node->declare_parameter("baz"); - node->declare_parameter("foobar"); - auto parameters_client = std::make_shared(node); while (!parameters_client->wait_for_service(1s)) { if (!rclcpp::ok()) { @@ -72,6 +66,12 @@ int main(int argc, char ** argv) on_parameter_event(event, node->get_logger()); }); + // Declare parameters that may be set on this node + node->declare_parameter("foo"); + node->declare_parameter("bar"); + node->declare_parameter("baz"); + node->declare_parameter("foobar"); + // Set several different types of parameters. auto set_parameters_results = parameters_client->set_parameters({ rclcpp::Parameter("foo", 2), @@ -88,7 +88,10 @@ int main(int argc, char ** argv) // TODO(wjwwood): Create and use delete_parameter - rclcpp::sleep_for(100ms); + // TODO(hidmic): Fast-RTPS takes a significant amount of time to deliver + // requests and response, thus the rather long sleep. Reduce + // once that's resolved. + rclcpp::sleep_for(3s); rclcpp::spin_some(node); diff --git a/demo_nodes_cpp/src/parameters/parameter_events_async.cpp b/demo_nodes_cpp/src/parameters/parameter_events_async.cpp index f7982dafa..5b71fb5ed 100644 --- a/demo_nodes_cpp/src/parameters/parameter_events_async.cpp +++ b/demo_nodes_cpp/src/parameters/parameter_events_async.cpp @@ -29,12 +29,6 @@ class ParameterEventsAsyncNode : public rclcpp::Node ParameterEventsAsyncNode() : Node("parameter_events") { - // Declare parameters that may be set on this node - this->declare_parameter("foo"); - this->declare_parameter("bar"); - this->declare_parameter("baz"); - this->declare_parameter("foobar"); - // Typically a parameter client is created for a remote node by passing the name of the remote // node in the constructor; in this example we create a parameter client for this node itself. parameters_client_ = std::make_shared(this); @@ -63,9 +57,15 @@ class ParameterEventsAsyncNode : public rclcpp::Node // Setup callback for changes to parameters. parameter_event_sub_ = parameters_client_->on_parameter_event(on_parameter_event_callback); + // Declare parameters that may be set on this node + this->declare_parameter("foo"); + this->declare_parameter("bar"); + this->declare_parameter("baz"); + this->declare_parameter("foobar"); + // Queue a `set_parameters` request as soon as `spin` is called on this node. // TODO(dhood): consider adding a "call soon" notion to Node to not require a timer for this. - timer_ = create_wall_timer(0s, + timer_ = create_wall_timer(200ms, [this]() { this->queue_first_set_parameter_request(); }); diff --git a/demo_nodes_cpp/test/parameter_events.txt b/demo_nodes_cpp/test/parameter_events.txt index 16d280f9d..543cfd9a9 100644 --- a/demo_nodes_cpp/test/parameter_events.txt +++ b/demo_nodes_cpp/test/parameter_events.txt @@ -40,3 +40,31 @@ Parameter event: bar deleted parameters: + +Parameter event: + new parameters: + changed parameters: + baz + deleted parameters: + + +Parameter event: + new parameters: + changed parameters: + foobar + deleted parameters: + + +Parameter event: + new parameters: + changed parameters: + foo + deleted parameters: + + +Parameter event: + new parameters: + changed parameters: + bar + deleted parameters: + diff --git a/demo_nodes_cpp/test/parameter_events_async.txt b/demo_nodes_cpp/test/parameter_events_async.txt index 16d280f9d..543cfd9a9 100644 --- a/demo_nodes_cpp/test/parameter_events_async.txt +++ b/demo_nodes_cpp/test/parameter_events_async.txt @@ -40,3 +40,31 @@ Parameter event: bar deleted parameters: + +Parameter event: + new parameters: + changed parameters: + baz + deleted parameters: + + +Parameter event: + new parameters: + changed parameters: + foobar + deleted parameters: + + +Parameter event: + new parameters: + changed parameters: + foo + deleted parameters: + + +Parameter event: + new parameters: + changed parameters: + bar + deleted parameters: + diff --git a/demo_nodes_cpp/test/test_executables_tutorial.py.in b/demo_nodes_cpp/test/test_executables_tutorial.py.in index f3f6ac57b..7285be0c3 100644 --- a/demo_nodes_cpp/test/test_executables_tutorial.py.in +++ b/demo_nodes_cpp/test/test_executables_tutorial.py.in @@ -3,62 +3,67 @@ import os +import unittest + 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 -from launch_testing.output import get_default_filtered_prefixes +from launch.actions import OpaqueFunction + +import launch_testing +import launch_testing.asserts +import launch_testing.util +import launch_testing_ros -def setup(): +def generate_test_description(ready_fn): os.environ['OSPL_VERBOSITY'] = '8' # 8 = OS_NONE # bare minimum formatting for console output matching os.environ['RCUTILS_CONSOLE_OUTPUT_FORMAT'] = '{message}' - -def test_executable(): - launch_test = LaunchTestService() launch_description = LaunchDescription() - - rmw_implementation = '@rmw_implementation@' - executables = '@DEMO_NODES_CPP_EXECUTABLE@'.split(';') - output_files = '@DEMO_NODES_CPP_EXPECTED_OUTPUT@'.split(';') - output_filter = create_output_lines_filter( - filtered_prefixes=get_default_filtered_prefixes() + [ - b'service not available, waiting again...' - ], - filtered_rmw_implementation=rmw_implementation + processes_under_test = [ + ExecuteProcess(cmd=[executable], name='test_executable_' + str(i), output='screen') + for i, executable in enumerate('@DEMO_NODES_CPP_EXECUTABLE@'.split(';')) + ] + for process in processes_under_test: + launch_description.add_action(process) + launch_description.add_action(launch_testing.util.KeepAliveProc()) + launch_description.add_action( + OpaqueFunction(function=lambda context: ready_fn()) ) - for i, (exe, output_file) in enumerate(zip(executables, output_files)): - name = 'test_executable_' + str(i) - # The last executable is taken to be the test program (the one whose - # output check can make the decision to shut everything down) - is_last_executable = (i == (len(executables) - 1)) + return launch_description, locals() - action = launch_test.add_fixture_action( - launch_description, ExecuteProcess( - cmd=[exe], - name=name, - output='screen' - ), exit_allowed=[0] if is_last_executable else True - ) - launch_test.add_output_test( - launch_description, action, - output_test=create_output_test_from_file(output_file), - output_filter=output_filter, - side_effect='shutdown' if is_last_executable else None - ) - launch_service = LaunchService() - launch_service.include_launch_description(launch_description) - rc = launch_test.run(launch_service) +class TestExecutablesTutorial(unittest.TestCase): - assert rc == 0, \ - "The launch file failed with exit code '" + str(rc) + "'. " \ - 'Maybe the client did not receive any messages?' + def test_processes_output(self, proc_output, processes_under_test): + """Test all processes output against expectations.""" + from launch_testing.tools.output import get_default_filtered_prefixes + output_filter = launch_testing_ros.tools.basic_output_filter( + filtered_prefixes=get_default_filtered_prefixes() + [ + 'service not available, waiting again...' + ], + filtered_rmw_implementation='@rmw_implementation@' + ) + output_files = '@DEMO_NODES_CPP_EXPECTED_OUTPUT@'.split(';') + for process, output_file in zip(processes_under_test, output_files): + proc_output.assertWaitFor( + expected_output=launch_testing.tools.expected_output_from_file( + path=output_file + ), process=process, output_filter=output_filter, timeout=30 + ) + # TODO(hidmic): either make the underlying executables resilient to + # interruptions close/during shutdown OR adapt the testing suite to + # better cope with it. + import time + time.sleep(5) +@launch_testing.post_shutdown_test() +class TestExecutablesTutorialAfterShutdown(unittest.TestCase): -if __name__ == '__main__': - test_executable() + def test_last_process_exit_code(self, proc_info, processes_under_test): + """Test last process exit code.""" + launch_testing.asserts.assertExitCodes( + proc_info, + process=processes_under_test[-1] + ) diff --git a/demo_nodes_cpp_native/CMakeLists.txt b/demo_nodes_cpp_native/CMakeLists.txt index 76b67d155..3bba84844 100644 --- a/demo_nodes_cpp_native/CMakeLists.txt +++ b/demo_nodes_cpp_native/CMakeLists.txt @@ -36,6 +36,7 @@ if(rmw_fastrtps_cpp_FOUND) ament_lint_auto_find_test_dependencies() find_package(ament_cmake_pytest REQUIRED) + find_package(launch_testing_ament_cmake REQUIRED) set(tutorial_executables "talker") @@ -60,12 +61,14 @@ if(rmw_fastrtps_cpp_FOUND) INPUT "${CMAKE_CURRENT_BINARY_DIR}/test_${exe_list_underscore}.py.configured" ) - ament_add_pytest_test(test_tutorial_${exe_list_underscore} + add_launch_test( "${CMAKE_CURRENT_BINARY_DIR}/test_${exe_list_underscore}_$.py" + TARGET test_tutorial_${exe_list_underscore} TIMEOUT 30 ENV RCL_ASSERT_RMW_ID_MATCHES=rmw_fastrtps_cpp - RMW_IMPLEMENTATION=rmw_fastrtps_cpp) + RMW_IMPLEMENTATION=rmw_fastrtps_cpp + ) foreach(executable ${tutorial_executables}) set_property( TEST test_tutorial_${exe_list_underscore} diff --git a/demo_nodes_cpp_native/package.xml b/demo_nodes_cpp_native/package.xml index 6d3b75b23..b5769b904 100644 --- a/demo_nodes_cpp_native/package.xml +++ b/demo_nodes_cpp_native/package.xml @@ -20,6 +20,8 @@ ament_lint_common launch launch_testing + launch_testing_ament_cmake + launch_testing_ros ament_cmake diff --git a/demo_nodes_cpp_native/test/test_executables_tutorial.py.in b/demo_nodes_cpp_native/test/test_executables_tutorial.py.in index 177979aa5..2fba13243 100644 --- a/demo_nodes_cpp_native/test/test_executables_tutorial.py.in +++ b/demo_nodes_cpp_native/test/test_executables_tutorial.py.in @@ -1,56 +1,56 @@ # generated from demo_nodes_cpp/test/test_executables_tutorial.py.in # generated code does not contain a copyright notice +import unittest + 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 -from launch_testing.output import get_default_filtered_prefixes +from launch.actions import OpaqueFunction +import launch_testing +import launch_testing.asserts +import launch_testing_ros -def test_executable(): - launch_test = LaunchTestService() - launch_description = LaunchDescription() - rmw_implementation = 'rmw_fastrtps_cpp' - executables = '@DEMO_NODES_CPP_EXECUTABLE@'.split(';') - output_files = '@DEMO_NODES_CPP_EXPECTED_OUTPUT@'.split(';') - output_filter = create_output_lines_filter( - filtered_prefixes=get_default_filtered_prefixes() + [ - b'service not available, waiting again...' - ], - filtered_rmw_implementation=rmw_implementation +def generate_test_description(ready_fn): + launch_description = LaunchDescription() + processes_under_test = [ + ExecuteProcess(cmd=[executable], name='test_executable_' + str(i), output='screen') + for i, executable in enumerate('@DEMO_NODES_CPP_EXECUTABLE@'.split(';')) + ] + for process in processes_under_test: + launch_description.add_action(process) + launch_description.add_action( + OpaqueFunction(function=lambda context: ready_fn()) ) - for i, (exe, output_file) in enumerate(zip(executables, output_files)): - name = 'test_executable_' + str(i) - # The last executable is taken to be the test program (the one whose - # output check can make the decision to shut everything down) - is_last_executable = (i == (len(executables) - 1)) - - action = launch_test.add_fixture_action( - launch_description, ExecuteProcess( - cmd=[exe], - name=name, - output='screen' - ), exit_allowed=[0] if is_last_executable else True - ) - launch_test.add_output_test( - launch_description, action, - output_test=create_output_test_from_file(output_file), - output_filter=output_filter, - side_effect='shutdown' if is_last_executable else None - ) - - launch_service = LaunchService() - launch_service.include_launch_description(launch_description) - rc = launch_test.run(launch_service) + return launch_description, locals() - assert rc == 0, \ - "The launch file failed with exit code '" + str(rc) + "'. " \ - 'Maybe the client did not receive any messages?' +class TestExecutablesTutorial(unittest.TestCase): -if __name__ == '__main__': - test_executable() + def test_executables_output(self, proc_output, processes_under_test): + """Test all processes output against expectations.""" + from launch_testing.tools.output import get_default_filtered_prefixes + output_filter = launch_testing_ros.tools.basic_output_filter( + filtered_prefixes=get_default_filtered_prefixes() + [ + 'service not available, waiting again...' + ], + filtered_rmw_implementation='rmw_fastrtps_cpp' + ) + output_files = '@DEMO_NODES_CPP_EXPECTED_OUTPUT@'.split(';') + for process, output_file in zip(processes_under_test, output_files): + proc_output.assertWaitFor( + expected_output=launch_testing.tools.expected_output_from_file( + path=output_file + ), process=process, output_filter=output_filter, timeout=10 + ) + + +@launch_testing.post_shutdown_test() +class TestExecutablesTutorialAfterShutdown(unittest.TestCase): + + def test_last_executable_exit_code(self, proc_info, processes_under_test): + """Test last process exit code.""" + launch_testing.asserts.assertExitCodes( + proc_info, process=processes_under_test[-1] + ) diff --git a/image_tools/CMakeLists.txt b/image_tools/CMakeLists.txt index 266b5ffcc..f7fa33667 100644 --- a/image_tools/CMakeLists.txt +++ b/image_tools/CMakeLists.txt @@ -51,6 +51,7 @@ if(BUILD_TESTING) ament_lint_auto_find_test_dependencies() find_package(ament_cmake_pytest REQUIRED) + find_package(launch_testing_ament_cmake REQUIRED) find_package(rmw_implementation_cmake REQUIRED) # These are the regex's for validating the output of the executables. @@ -72,8 +73,9 @@ if(BUILD_TESTING) INPUT "${CMAKE_CURRENT_BINARY_DIR}/test_showimage_cam2image${target_suffix}.py.genexp" ) - ament_add_pytest_test(test_showimage_cam2image${target_suffix} + add_launch_test( "${CMAKE_CURRENT_BINARY_DIR}/test_showimage_cam2image${target_suffix}_$.py" + TARGET test_showimage_cam2image${target_suffix} ENV RCL_ASSERT_RMW_ID_MATCHES=${rmw_implementation} RMW_IMPLEMENTATION=${rmw_implementation} diff --git a/image_tools/package.xml b/image_tools/package.xml index 39232f63e..e3c72a434 100644 --- a/image_tools/package.xml +++ b/image_tools/package.xml @@ -23,6 +23,8 @@ ament_lint_common launch launch_testing + launch_testing_ament_cmake + launch_testing_ros rmw_implementation_cmake diff --git a/image_tools/test/test_executables_demo.py.in b/image_tools/test/test_executables_demo.py.in index 8a18141a8..513e52689 100644 --- a/image_tools/test/test_executables_demo.py.in +++ b/image_tools/test/test_executables_demo.py.in @@ -13,92 +13,80 @@ # limitations under the License. import os +import unittest 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 +from launch.actions import OpaqueFunction - -def setup(): - os.environ['OSPL_VERBOSITY'] = '8' # 8 = OS_NONE - # bare minimum formatting for console output matching - os.environ['RCUTILS_CONSOLE_OUTPUT_FORMAT'] = '{message}' - - -def test_reliable_qos(): - pub_executable_args = ['-r', '1', '-s', '0', '-b', '-f', '5'] - sub_executable_args = ['-r', '1', '-s', '0', '-b'] - _test_executables(pub_executable_args, sub_executable_args) +import launch_testing +import launch_testing.asserts +import launch_testing_ros -def _test_executables(publisher_executable_args, subscriber_executable_args): - launch_test = LaunchTestService() +def generate_test_description(ready_fn): launch_description = LaunchDescription() - - rmw_implementation = '@rmw_implementation@' - output_filter = create_output_lines_filter( - filtered_rmw_implementation=rmw_implementation - ) + publisher_executable_args = ['-r', '1', '-s', '0', '-b', '-f', '5'] + subscriber_executable_args = ['-r', '1', '-s', '0', '-b'] env = dict(os.environ) - env['RMW_IMPLEMENTATION'] = rmw_implementation + env['OSPL_VERBOSITY'] = '8' # 8 = OS_NONE + # bare minimum formatting for console output matching + env['RCUTILS_CONSOLE_OUTPUT_FORMAT'] = '{message}' + env['RMW_IMPLEMENTATION'] = '@rmw_implementation@' # Launch the process that will receive the images. # This is the process that gets to decide when to tear the launcher down. # It will exit when the regex for receiving images is matched. showimage_executable = '@RCLCPP_DEMO_SHOWIMAGE_EXECUTABLE@' - showimage_output_file = '@RCLCPP_DEMO_SHOWIMAGE_EXPECTED_OUTPUT@' showimage_name = 'test_showimage' command = [showimage_executable] command.extend(subscriber_executable_args) - action = launch_test.add_fixture_action( - launch_description, ExecuteProcess( - cmd=command, - name=showimage_name, - env=env, - output='screen' - ) - ) - launch_test.add_output_test( - launch_description, action, - output_test=create_output_test_from_file(showimage_output_file), - output_filter=output_filter, - side_effect='shutdown' + showimage_process = ExecuteProcess( + cmd=command, + name=showimage_name, + env=env, + output='screen' ) + launch_description.add_action(showimage_process) # Launch the process that will publish the images. # This process will be exited when the launcher is torn down. cam2image_executable = '@RCLCPP_DEMO_CAM2IMAGE_EXECUTABLE@' - cam2image_output_file = '@RCLCPP_DEMO_CAM2IMAGE_EXPECTED_OUTPUT@' cam2image_name = 'test_cam2image' command = [cam2image_executable] command.extend(publisher_executable_args) - action = launch_test.add_fixture_action( - launch_description, ExecuteProcess( - cmd=command, - name=cam2image_name, - env=env, - output='screen' - ), exit_allowed=True + cam2image_process = ExecuteProcess( + cmd=command, + name=cam2image_name, + env=env, + output='screen' ) - launch_test.add_output_test( - launch_description, action, - output_test=create_output_test_from_file(cam2image_output_file), - output_filter=output_filter, + launch_description.add_action(cam2image_process) + + launch_description.add_action( + OpaqueFunction(function=lambda context: ready_fn()) ) - launch_service = LaunchService() - launch_service.include_launch_description(launch_description) - rc = launch_test.run(launch_service) + return launch_description, locals() - assert rc == 0, \ - "The launch file failed with exit code '" + str(rc) + "'. " +class TestExecutablesDemo(unittest.TestCase): -if __name__ == '__main__': - test_reliable_qos() + def test_reliable_qos(self, proc_output, showimage_process, cam2image_process): + """Test QoS settings for both processes work as expected.""" + output_filter = launch_testing_ros.tools.basic_output_filter( + filtered_rmw_implementation='@rmw_implementation@' + ) + proc_output.assertWaitFor( + expected_output=launch_testing.tools.expected_output_from_file( + path='@RCLCPP_DEMO_SHOWIMAGE_EXPECTED_OUTPUT@' + ), process=showimage_process, output_filter=output_filter, timeout=10 + ) + proc_output.assertWaitFor( + expected_output=launch_testing.tools.expected_output_from_file( + path='@RCLCPP_DEMO_CAM2IMAGE_EXPECTED_OUTPUT@' + ), process=cam2image_process, output_filter=output_filter, timeout=10 + ) diff --git a/intra_process_demo/CMakeLists.txt b/intra_process_demo/CMakeLists.txt index 4ad85fee8..192ae40c0 100644 --- a/intra_process_demo/CMakeLists.txt +++ b/intra_process_demo/CMakeLists.txt @@ -100,6 +100,7 @@ if(BUILD_TESTING) ament_lint_auto_find_test_dependencies() find_package(ament_cmake_pytest REQUIRED) + find_package(launch_testing_ament_cmake REQUIRED) find_package(rmw_implementation_cmake REQUIRED) # Add each test case. Multi-executable tests can be specified in @@ -132,12 +133,14 @@ if(BUILD_TESTING) INPUT "${CMAKE_CURRENT_BINARY_DIR}/test_${exe_list_underscore}${target_suffix}.py.configured" ) - ament_add_pytest_test(test_demo_${exe_list_underscore}${target_suffix} + add_launch_test( "${CMAKE_CURRENT_BINARY_DIR}/test_${exe_list_underscore}${target_suffix}_$.py" + TARGET test_demo_${exe_list_underscore}${target_suffix} TIMEOUT 30 ENV RCL_ASSERT_RMW_ID_MATCHES=${rmw_implementation} - RMW_IMPLEMENTATION=${rmw_implementation}) + RMW_IMPLEMENTATION=${rmw_implementation} + ) if(TEST test_demo_${exe_list_underscore}${target_suffix}) set_tests_properties(test_demo_${exe_list_underscore}${target_suffix} PROPERTIES DEPENDS "test_demo_${exe_list_underscore}${target_suffix} test_demo_${exe_list_underscore}${target_suffix}") diff --git a/intra_process_demo/package.xml b/intra_process_demo/package.xml index 9d809e996..4085a1fe9 100644 --- a/intra_process_demo/package.xml +++ b/intra_process_demo/package.xml @@ -23,6 +23,7 @@ ament_lint_common launch launch_testing + launch_testing_ament_cmake rmw_implementation_cmake ament_cmake diff --git a/intra_process_demo/test/test_executables_demo.py.in b/intra_process_demo/test/test_executables_demo.py.in index e3b414021..bcd553b53 100644 --- a/intra_process_demo/test/test_executables_demo.py.in +++ b/intra_process_demo/test/test_executables_demo.py.in @@ -1,41 +1,41 @@ # generated from intra_process_demo/test/test_executables_demo.py.in # generated code does not contain a copyright notice +import unittest + 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_test_from_file +from launch.actions import OpaqueFunction +import launch_testing +import launch_testing.asserts -def test_executable(): - launch_test = LaunchTestService() - launch_description = LaunchDescription() - executables = '@RCLCPP_DEMOS_EXECUTABLE@'.split(';') - output_files = '@RCLCPP_DEMOS_EXPECTED_OUTPUT@'.split(';') - for i, (exe, output_file) in enumerate(zip(executables, output_files)): - name = 'test_executable_' + str(i) - action = launch_test.add_fixture_action( - launch_description, ExecuteProcess( - cmd=[exe, 'test_executable'], - name=name, - output='screen' - ), exit_allowed=True - ) - launch_test.add_output_test( - launch_description, action, - output_test=create_output_test_from_file(output_file), - ) - - 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) + "'. " \ - 'Maybe the client did not receive any messages?' - - -if __name__ == '__main__': - test_executable() +def generate_test_description(ready_fn): + launch_description = LaunchDescription() + processes_under_test = [ + ExecuteProcess( + cmd=[executable, 'test_executable'], + name='test_executable_' + str(i), + output='screen' + ) for i, executable in enumerate('@RCLCPP_DEMOS_EXECUTABLE@'.split(';')) + ] + for process in processes_under_test: + launch_description.add_action(process) + launch_description.add_action( + OpaqueFunction(function=lambda context: ready_fn()) + ) + return launch_description, locals() + + +class TestExecutablesDemo(unittest.TestCase): + + def test_executables_output(self, proc_output, processes_under_test): + """Test all processes output against expectations.""" + output_files = '@RCLCPP_DEMOS_EXPECTED_OUTPUT@'.split(';') + for process, output_file in zip(processes_under_test, output_files): + proc_output.assertWaitFor( + expected_output=launch_testing.tools.expected_output_from_file( + path=output_file + ), process=process + ) diff --git a/logging_demo/CMakeLists.txt b/logging_demo/CMakeLists.txt index 52a152620..5a99a7888 100644 --- a/logging_demo/CMakeLists.txt +++ b/logging_demo/CMakeLists.txt @@ -69,6 +69,7 @@ if(BUILD_TESTING) ament_lint_auto_find_test_dependencies() find_package(ament_cmake_pytest REQUIRED) + find_package(launch_testing_ament_cmake REQUIRED) find_package(rmw_implementation_cmake REQUIRED) set(generated_python_files) @@ -90,11 +91,13 @@ if(BUILD_TESTING) OUTPUT "${CMAKE_CURRENT_BINARY_DIR}/test_logging_demo${target_suffix}_$.py" INPUT "${CMAKE_CURRENT_BINARY_DIR}/test_logging_demo${target_suffix}.py.genexp" ) - ament_add_pytest_test(test_logging_demo${target_suffix} + add_launch_test( "${CMAKE_CURRENT_BINARY_DIR}/test_logging_demo${target_suffix}_$.py" + TARGET test_logging_demo${target_suffix} ENV RMW_IMPLEMENTATION=${rmw_implementation} APPEND_LIBRARY_DIRS "${append_library_dirs}" - TIMEOUT 30) + TIMEOUT 30 + ) list( APPEND generated_python_files "${CMAKE_CURRENT_BINARY_DIR}/test_logging_demo${target_suffix}_$.py") diff --git a/logging_demo/package.xml b/logging_demo/package.xml index cb5951d3e..951614e47 100644 --- a/logging_demo/package.xml +++ b/logging_demo/package.xml @@ -29,6 +29,8 @@ ament_lint_common launch launch_testing + launch_testing_ament_cmake + launch_testing_ros rmw_implementation_cmake rosidl_interface_packages diff --git a/logging_demo/test/test_logging_demo.py.in b/logging_demo/test/test_logging_demo.py.in index ac1e35506..a71d3b7d9 100644 --- a/logging_demo/test/test_logging_demo.py.in +++ b/logging_demo/test/test_logging_demo.py.in @@ -12,61 +12,51 @@ # See the License for the specific language governing permissions and # limitations under the License. +import unittest + 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_default_severity(): - name = 'test_logging_demo_default_severity' - executable = '@LOGGING_DEMO_MAIN_EXECUTABLE@' - output_file = '@EXPECTED_OUTPUT_LOGGING_DEMO_MAIN_DEFAULT_SEVERITY@' - launch(name, [executable], output_file) +from launch.actions import OpaqueFunction +import launch_testing +import launch_testing.asserts +import launch_testing_ros -def test_debug_severity(): - name = 'test_logging_demo_debug_severity' - executable = '@LOGGING_DEMO_MAIN_EXECUTABLE@' - output_file = '@EXPECTED_OUTPUT_LOGGING_DEMO_MAIN_DEBUG_SEVERITY@' - launch(name, [executable], output_file) - -def launch(name, cmd, output_file, additional_processes=None): - launch_test = LaunchTestService() +def generate_test_description(ready_fn): launch_description = LaunchDescription() - - action = launch_test.add_fixture_action( - launch_description, ExecuteProcess( - cmd=cmd, - name=name, - output='screen' - ) + process_under_test = ExecuteProcess( + cmd=['@LOGGING_DEMO_MAIN_EXECUTABLE@'], + name='test_logging_demo', + output='screen' ) - rmw_implementation = '@rmw_implementation@' - launch_test.add_output_test( - launch_description, action, - output_test=create_output_test_from_file(output_file), - output_filter=create_output_lines_filter( - filtered_rmw_implementation=rmw_implementation - ) + launch_description.add_action(process_under_test) + launch_description.add_action( + OpaqueFunction(function=lambda context: ready_fn()) ) - for additional_process in (additional_processes or []): - launch_test.add_fixture_action( - launch_description, ExecuteProcess(**additional_process) - ) + return launch_description, locals() + - launch_service = LaunchService() - launch_service.include_launch_description(launch_description) - rc = launch_test.run(launch_service) +class TestLoggingDemo(unittest.TestCase): - assert rc == 0, \ - "The launch file failed with exit code '" + str(rc) + "'. " \ - 'Maybe the client did not receive any messages?' + @classmethod + def setUpClass(cls): + cls.output_filter = launch_testing_ros.tools.basic_output_filter( + filtered_rmw_implementation='@rmw_implementation@' + ) + def test_default_severity(self, proc_output, process_under_test): + """Test process' logs at default severity.""" + proc_output.assertWaitFor( + expected_output=launch_testing.tools.expected_output_from_file( + path='@EXPECTED_OUTPUT_LOGGING_DEMO_MAIN_DEFAULT_SEVERITY@' + ), process=process_under_test, output_filter=TestLoggingDemo.output_filter, timeout=30 + ) -if __name__ == '__main__': - test_default_severity() - test_debug_severity() + def test_debug_severity(self, proc_output, process_under_test): + """Test process' output at debug severity.""" + proc_output.assertWaitFor( + expected_output=launch_testing.tools.expected_output_from_file( + path='@EXPECTED_OUTPUT_LOGGING_DEMO_MAIN_DEBUG_SEVERITY@' + ), process=process_under_test, output_filter=TestLoggingDemo.output_filter, timeout=30 + ) diff --git a/pendulum_control/CMakeLists.txt b/pendulum_control/CMakeLists.txt index b19a8f4f0..8ef16e014 100644 --- a/pendulum_control/CMakeLists.txt +++ b/pendulum_control/CMakeLists.txt @@ -68,6 +68,7 @@ if(BUILD_TESTING) ament_lint_auto_find_test_dependencies() find_package(ament_cmake_pytest REQUIRED) + find_package(launch_testing_ament_cmake REQUIRED) find_package(rmw_implementation_cmake REQUIRED) set(RCLCPP_DEMO_PENDULUM_LOGGER_EXPECTED_OUTPUT "${CMAKE_CURRENT_SOURCE_DIR}/test/pendulum_logger") @@ -110,8 +111,9 @@ if(BUILD_TESTING) ) file(MAKE_DIRECTORY "${CMAKE_CURRENT_BINARY_DIR}/test_pendulum__${rmw_implementation}") - ament_add_pytest_test(test_pendulum__${rmw_implementation} + add_launch_test( "${CMAKE_CURRENT_BINARY_DIR}/test_pendulum__${rmw_implementation}.py" + TARGET test_pendulum__${rmw_implementation} TIMEOUT 20 ENV RCL_ASSERT_RMW_ID_MATCHES=${rmw_implementation} @@ -125,8 +127,9 @@ if(BUILD_TESTING) PROPERTIES DEPENDS "test_pendulum__${rmw_implementation} test_pendulum__${rmw_implementation}") endif() - ament_add_pytest_test(test_pendulum_teleop__${rmw_implementation} + add_launch_test( "${CMAKE_CURRENT_BINARY_DIR}/test_pendulum_teleop__${rmw_implementation}.py" + TARGET test_pendulum_teleop__${rmw_implementation} TIMEOUT 20 ENV RCL_ASSERT_RMW_ID_MATCHES=${rmw_implementation} diff --git a/pendulum_control/package.xml b/pendulum_control/package.xml index 2326912c9..ae40384d0 100644 --- a/pendulum_control/package.xml +++ b/pendulum_control/package.xml @@ -29,6 +29,8 @@ ament_lint_common launch launch_testing + launch_testing_ament_cmake + launch_testing_ros rmw_implementation_cmake ros2run diff --git a/pendulum_control/test/test_pendulum_demo.py.in b/pendulum_control/test/test_pendulum_demo.py.in index 4194634d9..480db95ae 100644 --- a/pendulum_control/test/test_pendulum_demo.py.in +++ b/pendulum_control/test/test_pendulum_demo.py.in @@ -3,16 +3,18 @@ import os +import unittest + 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 -from launch_testing.output import get_default_filtered_prefixes +from launch.actions import OpaqueFunction + +import launch_testing +import launch_testing.asserts +import launch_testing_ros -def setup(): +def generate_test_description(ready_fn): os.environ['OSPL_VERBOSITY'] = '8' # 8 = OS_NONE # bare minimum formatting for console output matching os.environ['RCUTILS_CONSOLE_OUTPUT_FORMAT'] = '{message}' @@ -20,63 +22,55 @@ def setup(): # this ensures a correct sync of prints from processes executed within the launch file. os.environ['RCUTILS_CONSOLE_STDOUT_LINE_BUFFERED'] = '1' - -def test_executable(): - launch_test = LaunchTestService() launch_description = LaunchDescription() - rmw_implementation = '@rmw_implementation@' - pendulum_logger_executable = '@RCLCPP_DEMO_PENDULUM_LOGGER_EXECUTABLE@' - pendulum_logger_output_file = '@RCLCPP_DEMO_PENDULUM_LOGGER_EXPECTED_OUTPUT@' - pendulum_logger_name = 'pendulum_logger' + pendulum_logger_process = ExecuteProcess( + cmd=['@RCLCPP_DEMO_PENDULUM_LOGGER_EXECUTABLE@'], + name='pendulum_logger', + output='screen' + ) + launch_description.add_action(pendulum_logger_process) - action = launch_test.add_fixture_action( - launch_description, ExecuteProcess( - cmd=[pendulum_logger_executable], - name=pendulum_logger_name, - output='screen' - ), exit_allowed=True + pendulum_demo_process = ExecuteProcess( + cmd=['@RCLCPP_DEMO_PENDULUM_DEMO_EXECUTABLE@', '-i', '1000'], + name='pendulum_demo', + output='screen' ) - launch_test.add_output_test( - launch_description, action, - output_test=create_output_test_from_file(pendulum_logger_output_file), + launch_description.add_action(pendulum_demo_process) + + launch_description.add_action( + OpaqueFunction(function=lambda context: ready_fn()) ) + return launch_description, locals() - pendulum_demo_executable = '@RCLCPP_DEMO_PENDULUM_DEMO_EXECUTABLE@' - pendulum_demo_output_file = '@RCLCPP_DEMO_PENDULUM_DEMO_EXPECTED_OUTPUT@' - pendulum_demo_name = 'pendulum_demo' - filtered_prefixes = get_default_filtered_prefixes() - if rmw_implementation.startswith('rmw_connext'): - # This output can be caused by a small QoS depth leading to samples being discarded. - # Since we are optimizing for performance with a depth of 1, we can ignore it. - filtered_prefixes.append( - b'PRESWriterHistoryDriver_completeBeAsynchPub:!make_sample_reclaimable' - ) +class TestPendulumDemo(unittest.TestCase): - action = launch_test.add_fixture_action( - launch_description, ExecuteProcess( - cmd=[pendulum_demo_executable, '-i', '1000'], - name=pendulum_demo_name, - output='screen' + def test_pendulum_logger_output(self, proc_output, pendulum_logger_process): + """Test logger output against expectations.""" + proc_output.assertWaitFor( + expected_output=launch_testing.tools.expected_output_from_file( + path='@RCLCPP_DEMO_PENDULUM_LOGGER_EXPECTED_OUTPUT@' + ), process=pendulum_logger_process, timeout=10 ) - ) - launch_test.add_output_test( - launch_description, action, - output_test=create_output_test_from_file(pendulum_demo_output_file), - output_filter=create_output_lines_filter( + + def test_pendulum_demo_output(self, proc_output, pendulum_demo_process): + """Test demo output against expectations.""" + rmw_implementation = '@rmw_implementation@' + from launch_testing.tools.output import get_default_filtered_prefixes + filtered_prefixes = get_default_filtered_prefixes() + if rmw_implementation.startswith('rmw_connext'): + # This output can be caused by a small QoS depth leading to samples being discarded. + # Since we are optimizing for performance with a depth of 1, we can ignore it. + filtered_prefixes.append( + 'PRESWriterHistoryDriver_completeBeAsynchPub:!make_sample_reclaimable' + ) + output_filter = launch_testing_ros.tools.basic_output_filter( filtered_prefixes=filtered_prefixes, filtered_rmw_implementation=rmw_implementation ) - ) - - launch_service = LaunchService(debug=True) - 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) + "'. " - - -if __name__ == '__main__': - test_executable() + proc_output.assertWaitFor( + expected_output=launch_testing.tools.expected_output_from_file( + path='@RCLCPP_DEMO_PENDULUM_DEMO_EXPECTED_OUTPUT@' + ), process=pendulum_demo_process, output_filter=output_filter, timeout=15 + ) diff --git a/pendulum_control/test/test_pendulum_teleop.py.in b/pendulum_control/test/test_pendulum_teleop.py.in index 24ebc53e2..8b56912c6 100644 --- a/pendulum_control/test/test_pendulum_teleop.py.in +++ b/pendulum_control/test/test_pendulum_teleop.py.in @@ -4,16 +4,18 @@ import os import sys +import unittest + 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 -from launch_testing.output import get_default_filtered_prefixes +from launch.actions import OpaqueFunction + +import launch_testing +import launch_testing.asserts +import launch_testing_ros -def setup(): +def generate_test_description(ready_fn): os.environ['OSPL_VERBOSITY'] = '8' # 8 = OS_NONE # bare minimum formatting for console output matching os.environ['RCUTILS_CONSOLE_OUTPUT_FORMAT'] = '{message}' @@ -21,67 +23,58 @@ def setup(): # this ensures a correct sync of prints from processes executed within the launch file. os.environ['RCUTILS_CONSOLE_STDOUT_LINE_BUFFERED'] = '1' - -def test_executable(): - launch_test = LaunchTestService() launch_description = LaunchDescription() - - rmw_implementation = '@rmw_implementation@' - pendulum_demo_executable = '@RCLCPP_DEMO_PENDULUM_DEMO_EXECUTABLE@' - pendulum_demo_output_file = '@RCLCPP_DEMO_PENDULUM_DEMO_TELEOP_EXPECTED_OUTPUT@' - pendulum_demo_name = 'pendulum_demo' - - action = launch_test.add_fixture_action( - launch_description, ExecuteProcess( - cmd=[pendulum_demo_executable, '-i', '0'], - name=pendulum_demo_name, - output='screen' - ), exit_allowed=True + pendulum_demo_process = ExecuteProcess( + cmd=['@RCLCPP_DEMO_PENDULUM_DEMO_EXECUTABLE@', '-i', '0'], + name='pendulum_demo', + output='screen' ) - launch_test.add_output_test( - launch_description, action, - output_test=create_output_test_from_file(pendulum_demo_output_file), - ) - - pendulum_teleop_executable = '@RCLCPP_DEMO_PENDULUM_TELEOP_EXECUTABLE@' - pendulum_teleop_output_file = '@RCLCPP_DEMO_PENDULUM_TELEOP_EXPECTED_OUTPUT@' - pendulum_teleop_name = 'pendulum_teleop' - - filtered_prefixes = get_default_filtered_prefixes() - if rmw_implementation.startswith('rmw_connext'): - # This output can be caused by a small QoS depth leading to samples being discarded. - # Since we are optimizing for performance with a depth of 1, we can ignore it. - filtered_prefixes.append( - b'PRESWriterHistoryDriver_completeBeAsynchPub:!make_sample_reclaimable' - ) + launch_description.add_action(pendulum_demo_process) execute_with_delay_command = os.path.join( os.path.abspath(os.path.dirname(__file__)), 'execute_with_delay.py') - action = launch_test.add_test_action( - launch_description, ExecuteProcess( - cmd=[sys.executable, execute_with_delay_command, '500', pendulum_teleop_executable, '100'], - name=pendulum_teleop_name, - output='screen' - ) + pendulum_teleop_process = ExecuteProcess( + cmd=[sys.executable, execute_with_delay_command, '500', + '@RCLCPP_DEMO_PENDULUM_TELEOP_EXECUTABLE@', '100'], + name='pendulum_teleop', + output='screen' ) - launch_test.add_output_test( - launch_description, action, - output_test=create_output_test_from_file(pendulum_teleop_output_file), - output_filter=create_output_lines_filter( - filtered_prefixes=filtered_prefixes, - filtered_rmw_implementation=rmw_implementation - ) + launch_description.add_action(pendulum_teleop_process) + + launch_description.add_action( + OpaqueFunction(function=lambda context: ready_fn()) ) + return launch_description, locals() - 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) + "'. " \ - 'Maybe the client did not receive any messages?' +class TestPendulumTeleop(unittest.TestCase): + def test_pendulum_demo_output(self, proc_output, pendulum_demo_process): + """Test demo output against expectations.""" + proc_output.assertWaitFor( + expected_output=launch_testing.tools.expected_output_from_file( + path='@RCLCPP_DEMO_PENDULUM_DEMO_TELEOP_EXPECTED_OUTPUT@' + ), process=pendulum_demo_process, timeout=10 + ) -if __name__ == '__main__': - test_executable() + def test_pendulum_teleop_output(self, proc_output, pendulum_teleop_process): + """Test teleop output against expectations.""" + rmw_implementation = '@rmw_implementation@' + from launch_testing.tools.output import get_default_filtered_prefixes + filtered_prefixes = get_default_filtered_prefixes() + if rmw_implementation.startswith('rmw_connext'): + # This output can be caused by a small QoS depth leading to samples being discarded. + # Since we are optimizing for performance with a depth of 1, we can ignore it. + filtered_prefixes.append( + 'PRESWriterHistoryDriver_completeBeAsynchPub:!make_sample_reclaimable' + ) + output_filter = launch_testing_ros.tools.basic_output_filter( + filtered_prefixes=filtered_prefixes, + filtered_rmw_implementation=rmw_implementation + ) + proc_output.assertWaitFor( + expected_output=launch_testing.tools.expected_output_from_file( + path='@RCLCPP_DEMO_PENDULUM_TELEOP_EXPECTED_OUTPUT@' + ), process=pendulum_teleop_process, output_filter=output_filter, timeout=10 + )