Running integration tests with mock hardware

ur_robot_driver/CMakeLists.txt

@@ -230,6 +230,11 @@ if(BUILD_TESTING)
   find_package(ur_msgs REQUIRED)
   find_package(launch_testing_ament_cmake)
 
+  add_launch_test(test/test_mock_hardware.py
+    TIMEOUT
+      800
+  )
+
   if(${UR_ROBOT_DRIVER_BUILD_INTEGRATION_TESTS})
     add_launch_test(test/launch_args.py
       TIMEOUT
@@ -243,7 +248,7 @@ if(BUILD_TESTING)
       TIMEOUT
         180
     )
-    add_launch_test(test/robot_driver.py
+    add_launch_test(test/integration_test_scaled_joint_controller.py
       TIMEOUT
         800
     )
@@ -259,6 +264,18 @@ if(BUILD_TESTING)
       TIMEOUT
         500
     )
+    add_launch_test(test/integration_test_config_controller.py
+      TIMEOUT
+        800
+    )
+    add_launch_test(test/integration_test_passthrough_controller.py
+      TIMEOUT
+        800
+    )
+    add_launch_test(test/integration_test_io_controller.py
+      TIMEOUT
+        800
+    )
     add_launch_test(test/integration_test_tool_contact.py
       TIMEOUT
         800

ur_robot_driver/test/integration_test_config_controller.py

@@ -0,0 +1,89 @@
+#!/usr/bin/env python
+# Copyright 2025, Universal Robots A/S
+#
+# 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.
+
+import os
+import sys
+import time
+import unittest
+
+import launch_testing
+import pytest
+import rclpy
+from rclpy.node import Node
+
+sys.path.append(os.path.dirname(__file__))
+from test_common import (  # noqa: E402
+    ControllerManagerInterface,
+    DashboardInterface,
+    IoStatusInterface,
+    ConfigurationInterface,
+    generate_driver_test_description,
+)
+
+
+@pytest.mark.launch_test
+@launch_testing.parametrize("tf_prefix", [(""), ("my_ur_")])
+def generate_test_description(tf_prefix):
+    return generate_driver_test_description(tf_prefix=tf_prefix)
+
+
+class ConfigControllerTest(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        # Initialize the ROS context
+        rclpy.init()
+        cls.node = Node("config_controller_test")
+        time.sleep(1)
+        cls.init_robot(cls)
+
+    @classmethod
+    def tearDownClass(cls):
+        # Shutdown the ROS context
+        cls.node.destroy_node()
+        rclpy.shutdown()
+
+    def init_robot(self):
+        self._dashboard_interface = DashboardInterface(self.node)
+        self._controller_manager_interface = ControllerManagerInterface(self.node)
+        self._io_status_controller_interface = IoStatusInterface(self.node)
+        self._configuration_controller_interface = ConfigurationInterface(self.node)
+
+    def setUp(self):
+        self._dashboard_interface.start_robot()
+        time.sleep(1)
+        self.assertTrue(self._io_status_controller_interface.resend_robot_program().success)
+
+    #
+    # Test functions
+    #
+
+    def test_get_robot_software_version(self):
+        self.assertGreater(
+            self._configuration_controller_interface.get_robot_software_version().major, 1
+        )

ur_robot_driver/test/integration_test_controller_switch.py

@@ -60,21 +60,17 @@
 
 
 @pytest.mark.launch_test
-@launch_testing.parametrize(
-    "tf_prefix",
-    [""],
-    # [(""), ("my_ur_")],
-)
+@launch_testing.parametrize("tf_prefix", [(""), ("my_ur_")])
 def generate_test_description(tf_prefix):
     return generate_driver_test_description(tf_prefix=tf_prefix)
 
 
-class RobotDriverTest(unittest.TestCase):
+class ControllerSwitchTest(unittest.TestCase):
     @classmethod
     def setUpClass(cls):
         # Initialize the ROS context
         rclpy.init()
-        cls.node = Node("robot_driver_test")
+        cls.node = Node("controller_switching_test")
         time.sleep(1)
         cls.init_robot(cls)
 

ur_robot_driver/test/integration_test_force_mode.py

@@ -88,12 +88,12 @@ def generate_test_description(tf_prefix):
     return generate_driver_test_description(tf_prefix=tf_prefix)
 
 
-class RobotDriverTest(unittest.TestCase):
+class ForceModeTest(unittest.TestCase):
     @classmethod
     def setUpClass(cls):
         # Initialize the ROS context
         rclpy.init()
-        cls.node = Node("robot_driver_test")
+        cls.node = Node("force_mode_test")
         time.sleep(1)
         cls.init_robot(cls)
 

ur_robot_driver/test/integration_test_io_controller.py

@@ -0,0 +1,132 @@
+#!/usr/bin/env python
+# Copyright 2025, Universal Robots A/S
+#
+# 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.
+
+import logging
+import os
+import sys
+import time
+import unittest
+
+import launch_testing
+import pytest
+import rclpy
+from rclpy.node import Node
+from ur_msgs.msg import IOStates
+
+sys.path.append(os.path.dirname(__file__))
+from test_common import (  # noqa: E402
+    ControllerManagerInterface,
+    DashboardInterface,
+    IoStatusInterface,
+    generate_driver_test_description,
+)
+
+
+@pytest.mark.launch_test
+@launch_testing.parametrize("tf_prefix", [(""), ("my_ur_")])
+def generate_test_description(tf_prefix):
+    return generate_driver_test_description(tf_prefix=tf_prefix)
+
+
+class IOControllerTest(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        # Initialize the ROS context
+        rclpy.init()
+        cls.node = Node("io_controller_test")
+        time.sleep(1)
+        cls.init_robot(cls)
+
+    @classmethod
+    def tearDownClass(cls):
+        # Shutdown the ROS context
+        cls.node.destroy_node()
+        rclpy.shutdown()
+
+    def init_robot(self):
+        self._dashboard_interface = DashboardInterface(self.node)
+        self._controller_manager_interface = ControllerManagerInterface(self.node)
+        self._io_status_controller_interface = IoStatusInterface(self.node)
+
+    def setUp(self):
+        self._dashboard_interface.start_robot()
+        time.sleep(1)
+        self.assertTrue(self._io_status_controller_interface.resend_robot_program().success)
+
+    #
+    # Test functions
+    #
+
+    def test_set_io(self):
+        """Test to set an IO and check whether it has been set."""
+        # Create io callback to verify result
+        io_msg = None
+
+        def io_msg_cb(msg):
+            nonlocal io_msg
+            io_msg = msg
+
+        io_states_sub = self.node.create_subscription(
+            IOStates,
+            "/io_and_status_controller/io_states",
+            io_msg_cb,
+            rclpy.qos.qos_profile_system_default,
+        )
+
+        # Set pin 0 to 1.0
+        test_pin = 0
+
+        logging.info("Setting pin %d to 1.0", test_pin)
+        self._io_status_controller_interface.set_io(fun=1, pin=test_pin, state=1.0)
+
+        # Wait until the pin state has changed
+        pin_state = False
+        end_time = time.time() + 5
+        while not pin_state and time.time() < end_time:
+            rclpy.spin_once(self.node, timeout_sec=0.1)
+            if io_msg is not None:
+                pin_state = io_msg.digital_out_states[test_pin].state
+
+        self.assertEqual(pin_state, 1.0)
+
+        # Set pin 0 to 0.0
+        logging.info("Setting pin %d to 0.0", test_pin)
+        self._io_status_controller_interface.set_io(fun=1, pin=test_pin, state=0.0)
+
+        # Wait until the pin state has changed back
+        end_time = time.time() + 5
+        while pin_state and time.time() < end_time:
+            rclpy.spin_once(self.node, timeout_sec=0.1)
+            if io_msg is not None:
+                pin_state = io_msg.digital_out_states[test_pin].state
+
+        self.assertEqual(pin_state, 0.0)
+
+        # Clean up io subscription
+        self.node.destroy_subscription(io_states_sub)