|
| 1 | +#!/usr/bin/env python3 |
| 2 | +# Copyright 2025 Robert Kwan |
| 3 | +# |
| 4 | +# Licensed under the Apache License, Version 2.0 (the "License"); |
| 5 | +# you may not use this file except in compliance with the License. |
| 6 | +# You may obtain a copy of the License at |
| 7 | +# |
| 8 | +# http://www.apache.org/licenses/LICENSE-2.0 |
| 9 | +# |
| 10 | +# Unless required by applicable law or agreed to in writing, software |
| 11 | +# distributed under the License is distributed on an "AS IS" BASIS, |
| 12 | +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 13 | +# See the License for the specific language governing permissions and |
| 14 | +# limitations under the License. |
| 15 | + |
| 16 | +import pytest |
| 17 | +import unittest |
| 18 | +from launch import LaunchDescription |
| 19 | +import launch_testing |
| 20 | +from launch_testing.actions import ReadyToTest |
| 21 | +import launch_ros.actions |
| 22 | +from launch.substitutions import FileContent, PathSubstitution |
| 23 | +from launch_ros.substitutions import FindPackageShare |
| 24 | + |
| 25 | +import rclpy |
| 26 | + |
| 27 | +from controller_manager.test_utils import check_controllers_running |
| 28 | + |
| 29 | +from controller_manager.launch_utils import ( |
| 30 | + generate_controllers_spawner_launch_description_from_dict, |
| 31 | +) |
| 32 | + |
| 33 | + |
| 34 | +@pytest.mark.launch_test |
| 35 | +def generate_test_description(): |
| 36 | + """ |
| 37 | + Generate launch description for testing the dict-based spawner helper. |
| 38 | +
|
| 39 | + Uses the combined controller YAML installed with the package. |
| 40 | + """ |
| 41 | + |
| 42 | + urdf = FileContent( |
| 43 | + PathSubstitution(FindPackageShare("ros2_control_test_assets")) |
| 44 | + / "urdf" |
| 45 | + / "test_hardware_components.urdf" |
| 46 | + ) |
| 47 | + robot_description = {"robot_description": urdf} |
| 48 | + |
| 49 | + # The dictionary keys are the controller names to be spawned/started. |
| 50 | + # Values can be empty lists since config is provided via the main YAML. |
| 51 | + ctrl_dict = { |
| 52 | + "test_broadcaster": [], |
| 53 | + "controller1": [], |
| 54 | + "controller2": [], |
| 55 | + } |
| 56 | + controller_list = list(ctrl_dict.keys()) |
| 57 | + |
| 58 | + # ===== CREATE LAUNCH DESCRIPTION ===== |
| 59 | + return LaunchDescription( |
| 60 | + [ |
| 61 | + launch_ros.actions.Node( |
| 62 | + package="robot_state_publisher", |
| 63 | + executable="robot_state_publisher", |
| 64 | + namespace="", |
| 65 | + output="both", |
| 66 | + parameters=[robot_description], |
| 67 | + ), |
| 68 | + launch_ros.actions.Node( |
| 69 | + package="controller_manager", |
| 70 | + executable="ros2_control_node", |
| 71 | + namespace="", |
| 72 | + parameters=[ |
| 73 | + robot_description, |
| 74 | + PathSubstitution(FindPackageShare("controller_manager")) |
| 75 | + / "test" |
| 76 | + / "test_launch_utils" |
| 77 | + / "test_ros2_control_node_combined.yaml", |
| 78 | + ], |
| 79 | + output="both", |
| 80 | + ), |
| 81 | + generate_controllers_spawner_launch_description_from_dict( |
| 82 | + controller_info_dict=ctrl_dict, |
| 83 | + extra_spawner_args=["--inactive"], |
| 84 | + ), |
| 85 | + ReadyToTest(), |
| 86 | + ] |
| 87 | + ), {"controller_list": controller_list} |
| 88 | + |
| 89 | + |
| 90 | +# Active tests |
| 91 | +class TestControllerSpawnerList(unittest.TestCase): |
| 92 | + """Active tests that run while the launch is active.""" |
| 93 | + |
| 94 | + @classmethod |
| 95 | + def setUpClass(cls): |
| 96 | + rclpy.init() |
| 97 | + |
| 98 | + @classmethod |
| 99 | + def tearDownClass(cls): |
| 100 | + rclpy.shutdown() |
| 101 | + |
| 102 | + def setUp(self): |
| 103 | + self.node = rclpy.create_node("test_controller_spawner") |
| 104 | + |
| 105 | + def tearDown(self): |
| 106 | + self.node.destroy_node() |
| 107 | + |
| 108 | + def test_controllers_start(self, proc_info, controller_list): |
| 109 | + cnames = controller_list.copy() |
| 110 | + check_controllers_running(self.node, cnames, state="inactive") |
| 111 | + |
| 112 | + # Wait for controller_spawner to finish and verify successful exit. |
| 113 | + proc_info.assertWaitForShutdown(process="spawner", timeout=30) |
| 114 | + launch_testing.asserts.assertExitCodes(proc_info, process="spawner") |
| 115 | + |
| 116 | + # Re-check controllers after spawner has exited. |
| 117 | + check_controllers_running(self.node, cnames, state="inactive") |
| 118 | + |
| 119 | + |
| 120 | +@launch_testing.post_shutdown_test() |
| 121 | +class TestShutdown(unittest.TestCase): |
| 122 | + """Post-shutdown tests.""" |
| 123 | + |
| 124 | + def test_exit_codes(self, proc_info): |
| 125 | + """Verify all processes exited successfully.""" |
| 126 | + launch_testing.asserts.assertExitCodes(proc_info, allowable_exit_codes=[0]) |
0 commit comments