Add a test to check if trajectory message is sent

Author: christophfroehlich

example_7/CMakeLists.txt

@@ -108,6 +108,7 @@ if(BUILD_TESTING)
   ament_add_pytest_test(example_7_urdf_xacro test/test_urdf_xacro.py)
   ament_add_pytest_test(view_example_7_launch test/test_view_robot_launch.py)
   ament_add_pytest_test(run_example_7_launch test/test_r6bot_controller_launch.py)
+  ament_add_pytest_test(test_send_trajectory_launch test/test_send_trajectory_launch.py)
 endif()
 
 ## EXPORTS

example_7/test/test_send_trajectory_launch.py

@@ -0,0 +1,95 @@
+# Copyright (c) 2024 AIT - Austrian Institute of Technology GmbH
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are met:
+#
+#    * Redistributions of source code must retain the above copyright
+#      notice, this list of conditions and the following disclaimer.
+#
+#    * Redistributions in binary form must reproduce the above copyright
+#      notice, this list of conditions and the following disclaimer in the
+#      documentation and/or other materials provided with the distribution.
+#
+#    * Neither the name of the {copyright_holder} nor the names of its
+#      contributors may be used to endorse or promote products derived from
+#      this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
+# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+# POSSIBILITY OF SUCH DAMAGE.
+#
+# Author: Christoph Froehlich
+
+import os
+import pytest
+import unittest
+
+from ament_index_python.packages import get_package_share_directory
+from launch import LaunchDescription
+from launch.actions import IncludeLaunchDescription
+from launch.launch_description_sources import PythonLaunchDescriptionSource
+from launch_testing.actions import ReadyToTest
+from launch_testing_ros import WaitForTopics
+from trajectory_msgs.msg import JointTrajectory
+
+import launch_testing.markers
+import rclpy
+
+
+# Executes the given launch file and checks if all nodes can be started
+@pytest.mark.rostest
+def generate_test_description():
+    launch_include = IncludeLaunchDescription(
+        PythonLaunchDescriptionSource(
+            os.path.join(
+                get_package_share_directory("ros2_control_demo_example_7"),
+                "launch/send_trajectory.launch.py",
+            )
+        )
+    )
+
+    return LaunchDescription([launch_include, ReadyToTest()])
+
+
+# This is our test fixture. Each method is a test case.
+# These run alongside the processes specified in generate_test_description()
+class TestFixture(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        rclpy.init()
+
+    @classmethod
+    def tearDownClass(cls):
+        rclpy.shutdown()
+
+    def setUp(self):
+        self.node = rclpy.create_node("test_node")
+
+    def tearDown(self):
+        self.node.destroy_node()
+
+    def test_check_if_msgs_published(self):
+        topic = "/r6bot_controller/joint_trajectory"
+        wait_for_topics = WaitForTopics([(topic, JointTrajectory)], timeout=20.0)
+        assert wait_for_topics.wait(), f"Topic '{topic}' not found!"
+        msgs = wait_for_topics.received_messages(topic)
+        msg = msgs[0]
+        assert len(msg.points) == 200, "Wrong number of joints in message"
+        wait_for_topics.shutdown()
+
+
+@launch_testing.post_shutdown_test()
+# These tests are run after the processes in generate_test_description() have shutdown.
+class TestDescriptionCraneShutdown(unittest.TestCase):
+
+    def test_exit_codes(self, proc_info):
+        """Check if the processes exited normally."""
+        launch_testing.asserts.assertExitCodes(proc_info)