Adds example tests for launch descriptions

launch_ros/examples/lifecycle_pub_sub_launch.py

@@ -31,8 +31,7 @@
 import lifecycle_msgs.msg
 
 
-def main(argv=sys.argv[1:]):
-    """Main."""
+def generate_launch_description() -> launch.LaunchDescription:
     ld = launch.LaunchDescription()
 
     # Prepare the talker node.
@@ -84,6 +83,14 @@ def main(argv=sys.argv[1:]):
     ld.add_action(talker_node)
     ld.add_action(emit_event_to_request_that_talker_does_configure_transition)
 
+    return ld
+
+
+def main(argv=sys.argv[1:]):
+    """Main."""
+
+    ld = generate_launch_description()
+
     print('Starting introspection of launch description...')
     print('')
 

launch_ros/examples/pub_sub_launch.py

@@ -19,6 +19,8 @@
 sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))  # noqa
 sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..', 'launch'))  # noqa
 
+import launch
+
 from launch import LaunchDescription
 from launch import LaunchIntrospector
 from launch import LaunchService
@@ -27,17 +29,24 @@
 import launch_ros.actions
 
 
-def main(argv=sys.argv[1:]):
-    """Main."""
-    ld = LaunchDescription([
+def generate_launch_description() -> LaunchDescription:
+    return LaunchDescription([
         launch_ros.actions.Node(
-            package='demo_nodes_cpp', node_executable='talker', output='screen',
-            remappings=[('chatter', 'my_chatter')]),
+            package='demo_nodes_cpp', node_executable='talker',
+            output='screen', remappings=[('chatter', 'my_chatter')]
+        ),
         launch_ros.actions.Node(
-            package='demo_nodes_cpp', node_executable='listener', output='screen',
-            remappings=[('chatter', 'my_chatter')]),
+            package='demo_nodes_cpp', node_executable='listener',
+            output='screen', remappings=[('chatter', 'my_chatter')]
+        ),
     ])
 
+
+def main(argv=sys.argv[1:]):
+    """Main."""
+
+    ld = generate_launch_description()
+
     print('Starting introspection of launch description...')
     print('')
 

launch_testing/examples/test_pub_sub_launch.py

@@ -0,0 +1,230 @@
+# Copyright 2019 Open Source Robotics Foundation, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""
+Some example tests around ROS 2 talker/listener nodes.
+
+.. note::
+
+To maximize code reuse, this test assumes that execution and executable
+abstractions are decoupled (see https://github.com/ros2/launch/issues/114).
+"""
+
+import sys
+import pytest
+
+from launch import LaunchDescription
+from launch.actions import Execute
+from launch.events.process import ProcessStarted
+from launch.launch_description_sources import \
+    get_launch_description_from_python_launch_file
+from launch_ros import get_default_launch_description
+from launch_ros.executables import Node
+from launch_ros.executables import ComposableNode
+from launch_ros.executables import ComposableNodeProcess
+import launch_ros.events.lifecycle as lifecyle
+
+from launch_testing import TestLaunchService
+from launch_testing.actions import PyTest
+from launch_testing.actions import GTest
+from launch_testing.actions import Assert
+from launch_testing.actions import AssertSequenceOnce
+
+import launch_testing.predicates as predicates
+import launch_testing.variables as variables
+
+
+@pytest.fixture
+def ros_test_launch_service() -> TestLaunchService:
+    """
+    A test fixture providing a test-aware launch service (i.e. one
+    that listens to events fired by `launch_testing.actions.Assert`
+    and `launch_testing.actions.Test` actions and reacts accordingly).
+    """
+    ls = TestLaunchService()
+    # TODO(hidmic): consider integrating ROS specific actions and the
+    # launch service in some way to ease launch file composition without
+    # repetition).
+    ls.include_launch_description(
+        get_default_launch_description(prefix_output_with_name=False)
+    )
+    return ls
+
+
+@pytest.fixture
+def pub_sub_launch_description() -> LaunchDescription:
+    """
+    A test fixture providing the basic pub/sub launch description.
+    """
+    # TODO(hidmic): implement functionality to lookup launch files
+    # on a per package basis
+    launch_path = get_launch_path(
+        package_name='launch_ros', launch_file_path='pub_sub_launch.py'
+    )
+    return get_launch_description_from_python_launch_file(launch_path)
+
+
+def test_pub_sub_launch(ros_test_launch_service, pub_sub_launch_description):
+    """
+    Tests an existing simple pub/sub system.
+    """
+    # Launches a pytest that also happens to be a node with
+    # a 30s timeout.
+    pub_sub_launch_description.add_action(
+        # TODO(hidmic): allow launching non-installed
+        # executables within the current package.
+        PyTest(Node(
+            node_executable='pub_test.py',
+            remappings=[('chatter', 'my_chatter')]
+        ), timeout=30)
+    )
+    # Launches a gtest that also happens to be a node with
+    # a 30s timeout.
+    pub_sub_launch_description.add_action(
+        GTest(Node(
+            node_executable='sub_test',
+            remappings=[('chatter', 'my_chatter')]
+        ), timeout=30)
+    )
+
+    ros_test_launch_service.include_launch_description(pub_sub_launch_description)
+
+    # TODO(hidmic): implement launch_testing specific pytest plugin to aggregate
+    # all test result information at the launch system-level and below.
+    assert ros_test_launch_service.run() == 0
+
+
+@pytest.fixture
+def lifecycle_pub_sub_launch_description() -> LaunchDescription:
+    """
+    A test fixture providing the lifecycle pub/sub launch description.
+    """
+    # TODO(hidmic): implement functionality to lookup launch files
+    # on a per package basis
+    launch_path = get_launch_path(
+        package_name='launch_ros',
+        launch_file_path='lifecycle_pub_sub_launch.py'
+    )
+    return get_launch_description_from_python_launch_file(launch_path)
+
+
+def test_lifecycle_pub_sub_launch(ros_test_launch_service,
+                                  lifecycle_pub_sub_launch_description):
+    """
+    Tests an existing lifecycle pub/sub nodes setup for execution
+    in proper sequence.
+    """
+    # Asserts that a lifecycle.ChangeState event within a 30s window
+    # is followed by a process.ProcessStarted event within a 10s window
+    # once (40s total timeout).
+    lifecycle_pub_sub_launch_description.add_action(
+        AssertSequenceOnce([
+            variables.EventEmitted(
+                event_type=lifecycle.ChangeState
+            ),
+            variables.EventEmitted(
+                event_type=ProcessStarted
+            )
+        ], timeout=[30, 10])
+    )
+
+    ros_test_launch_service.include_launch_description(
+        lifecycle_pub_sub_launch_description
+    )
+    # TODO(hidmic): implement launch_testing specific pytest plugin to
+    # aggregate all test result information at the launch system-level
+    # and below.
+    assert ros_test_launch_service.run() == 0
+
+
+def test_pub_sub(ros_test_launch_service):
+    """
+    Tests a basic pub/sub setup for proper output matching.
+    """
+    ld = launch.LaunchDescription()
+    talker_node_action = Execute(
+        launch_ros.actions.Node(
+            package='demo_nodes_cpp',
+            node_executable='talker',
+            remappings=[('chatter', 'my_chatter')]
+        ), output='screen'
+    )
+    ld.add_action(talker_node_action)
+    listener_node_action = Execute(
+        launch_ros.actions.Node(
+            package='demo_nodes_py',
+            node_executable='listener',
+            remappings=[('chatter', 'my_chatter')]
+        ),  output='screen'
+    )
+    ld.add_action(listener_node_action)
+    # Asserts that not a single talker screen output
+    # line contains the ERROR string throughout the duration
+    # of the test (i.e. it completes when all other tests
+    # have completed). The assertion is strict (i.e. failure
+    # on a single output event means failure altogether).
+    ld.add_action(
+        Assert(not predicates.regex_match(
+            variables.Output(talker_node_action),
+            pattern='.*ERROR.*', flags=None
+        ), timeout=None, strict=True)
+    )
+    # Asserts that a the listener outputs 'I heard...'
+    # at least once in the first 10s of the test. The
+    # assertion is not strict (i.e. failure on a single
+    # output event does not mean failure altogether).
+    ld.add_action(
+        Assert(predicates.count(predicates.regex_match(
+            variables.Output(listener_node_action),
+            pattern='I heard.*', flags=None
+        )) > 0, message='No message received!', timeout=10)
+    )
+    ros_test_launch_service.include_launch_description(ld)
+
+    # TODO(hidmic): implement launch_testing specific pytest plugin to
+    # aggregate all test result information at the launch system-level
+    # and below.
+    assert ros_test_launch_service.run() == 0
+
+
+def test_intra_comm_pub_sub(ros_test_launch_service):
+    """
+    Tests a intra process communication between composable
+    pub/sub nodes via gtest.
+    """
+    # Launches a gtest that happens to be a container
+    # for composable nodes.
+    ld = LaunchDescription([
+        GTest(ComposableNodeProcess(
+            container_executable='intra_proc_comm_test_container',
+            composable_node_descriptions=[
+                ComposableNode(
+                    package_name='composition',
+                    node_plugin_name='talker_component',
+                    name='my_talker_component'
+                ),
+                ComposableNode(
+                    package_name='composition',
+                    node_plugin_name='listener_component'
+                    name='my_listener_component'
+                ),
+            ]
+        ), timeout=40),
+    ])
+    ros_test_launch_service.include_launch_description(ld)
+
+    # TODO(hidmic): implement launch_testing specific pytest plugin to
+    # aggregate all test result information at the launch system-level
+    # and below.
+    assert ros_test_launch_service.run() == 0

launch_testing/examples/test_system_launch.py

@@ -0,0 +1,90 @@
+# Copyright 2019 Open Source Robotics Foundation, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import sys
+import pytest
+
+from datetime import timedelta
+
+from launch import LaunchDescription
+from launch.actions import Execute
+from launch.actions import RegisterEventHandler
+from launch.executables import Process
+from launch.event_handlers import OnProcessExited
+from launch_testing import TestLaunchService
+from launch_testing.actions import AssertOnce
+
+import launch_testing.predicates as predicates
+import launch_testing.variables as variables
+
+
+@pytest.fixture
+def launch_service() -> TestLaunchService:
+    """
+    A test fixture providing a test-aware launch service (i.e. one
+    that listens to events fired by `launch_testing.actions.Assert`
+    and `launch_testing.actions.Test` actions and reacts accordingly).
+    """
+    return TestLaunchService()
+
+
+def test_file_compression(launch_service):
+    ld = LaunchDescription()
+
+    prelist_bags_action = Execute(Process(
+        cmd='ls *.bag', cwd='/var/log/bags', shell=True
+    ), output='screen')
+    ld.add_action(prelist_bags_action)
+
+    compression_action = Execute(Process(
+        cmd='bzip2 /var/log/bags/*.bag', shell=True
+    ), output='screen', prefix='time')
+    ld.add_action(compression_action)
+
+    @predicates.custom
+    def parse_timedelta(string_value):
+        match = re.match(
+            '(?P<minutes>\d+)m(?P<seconds>\d\.\d{3})', string_value
+        )
+        return timedelta(**{
+            name: float(value) if value else 0
+            for name, value in match.groupdict()
+        })
+
+    ld.add_action(
+        AssertOnce(
+            parse_timedelta(predicates.regex_match(
+                variables.Output(compression_action),
+                pattern='^real (\d+m\d\.\d{3}s)$', group=1
+            )) < timedelta(minutes=1, seconds=30) and
+            parse_timedelta(predicates.regex_match(
+                variables.Output(compression_action),
+                pattern='^sys (\d+m\d\.\d{3}s)$', group=1
+            )) < timedelta(seconds=30), timeout=120
+        )
+    )
+
+    postlist_bags_action = Execute(Process(
+        cmd='ls *.bag.bz2', cwd='/var/log/bags', shell=True
+    ), output='screen')
+    assert_all_bags_compressed = AssertOnce(
+        predicates.count(variables.Output(postlist_bags_action))
+        == predicates.count(variables.Output(prelist_bags_action))
+    )
+    ld.add_action(RegisterEventHandler(OnProcessExit(
+        target_action=compression_action,
+        on_exit=[postlist_bags_action, assert_all_bags_compressed]
+    )))
+
+    assert launch_service.run() == 0