Fix concurrent spinning of the test_node

controller_manager/test/test_ros2_control_node_launch.py

@@ -37,16 +37,13 @@
 from launch_testing.actions import ReadyToTest
 import launch_testing.markers
 import launch_ros.actions
-from sensor_msgs.msg import JointState
 
 import rclpy
 from controller_manager.test_utils import (
     check_controllers_running,
-    check_if_js_published,
     check_node_running,
 )
 from controller_manager.launch_utils import generate_controllers_spawner_launch_description
-import threading
 
 
 # Executes the given launch file and checks if all nodes can be started
@@ -105,31 +102,13 @@ def setUp(self):
     def tearDown(self):
         self.node.destroy_node()
 
-    def timer_callback(self):
-        js_msg = JointState()
-        js_msg.name = ["joint0"]
-        js_msg.position = [0.0]
-        self.pub.publish(js_msg)
-
-    def publish_joint_states(self):
-        self.pub = self.node.create_publisher(JointState, "/joint_states", 10)
-        self.timer = self.node.create_timer(1.0, self.timer_callback)
-        rclpy.spin(self.node)
-
     def test_node_start(self):
         check_node_running(self.node, "controller_manager")
 
     def test_controllers_start(self):
         cnames = ["ctrl_with_parameters_and_type"]
         check_controllers_running(self.node, cnames, state="active")
 
-    def test_check_if_msgs_published(self):
-        # we don't have a joint_state_broadcaster in this repo,
-        # publish joint states manually to test check_if_js_published
-        thread = threading.Thread(target=self.publish_joint_states)
-        thread.start()
-        check_if_js_published("/joint_states", ["joint0"])
-
 
 @launch_testing.post_shutdown_test()
 # These tests are run after the processes in generate_test_description() have shutdown.

controller_manager/test/test_test_utils.py

@@ -0,0 +1,76 @@
+# Copyright (c) 2025 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 unittest
+from sensor_msgs.msg import JointState
+
+import rclpy
+from controller_manager.test_utils import (
+    check_if_js_published,
+)
+import threading
+
+
+# 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 timer_callback(self):
+        js_msg = JointState()
+        js_msg.name = ["joint0"]
+        js_msg.position = [0.0]
+        self.pub.publish(js_msg)
+
+    def publish_joint_states(self):
+        self.pub = self.node.create_publisher(JointState, "/joint_states", 10)
+        self.timer = self.node.create_timer(1.0, self.timer_callback)
+        rclpy.spin(self.node)
+
+    def test_check_if_msgs_published(self):
+        # we don't have a joint_state_broadcaster in this repo,
+        # publish joint states manually to test check_if_js_published
+        thread = threading.Thread(target=self.publish_joint_states)
+        thread.start()
+        check_if_js_published("/joint_states", ["joint0"])
+
+    # Note: Other test cases are in test_ros2_control_node_launch.py