diff --git a/ros2param/ros2param/verb/load.py b/ros2param/ros2param/verb/load.py index 71715d586..ac537bed0 100644 --- a/ros2param/ros2param/verb/load.py +++ b/ros2param/ros2param/verb/load.py @@ -12,23 +12,53 @@ # See the License for the specific language governing permissions and # limitations under the License. +from rclpy.parameter import parameter_dict_from_yaml_file +from ros2cli.helpers import wait_for from ros2cli.node.direct import DirectNode from ros2cli.node.strategy import add_arguments from ros2cli.node.strategy import NodeStrategy from ros2node.api import get_absolute_node_name +from ros2node.api import get_node_names from ros2node.api import NodeNameCompleter from ros2node.api import wait_for_node from ros2param.api import load_parameter_file from ros2param.verb import VerbExtension +def _parameter_file_matches_node(parameter_file, node_name, use_wildcard): + try: + parameter_dict_from_yaml_file( + parameter_file, use_wildcard, target_nodes=[node_name]) + except RuntimeError: + # The complete file is validated before this helper is used. For a + # validated file, rclpy raises RuntimeError here when the selected node + # contributes no parameters. + return False + return True + + +def _get_matching_node_names(node, parameter_file, use_wildcard, include_hidden_nodes): + discovered_nodes = get_node_names( + node=node, include_hidden_nodes=include_hidden_nodes) + local_node_name = None + if node.daemon_node is None: + local_node_name = node.direct_node.get_fully_qualified_name() + + return [ + node_name + for node_name in sorted({n.full_name for n in discovered_nodes}) + if node_name != local_node_name + and _parameter_file_matches_node(parameter_file, node_name, use_wildcard) + ] + + class LoadVerb(VerbExtension): - """Load parameter file for a node.""" + """Load a parameter file for one node or all matching nodes.""" def add_arguments(self, parser, cli_name): # noqa: D102 add_arguments(parser) arg = parser.add_argument( - 'node_name', help='Name of the ROS node') + 'node_name', nargs='?', help='Name of the ROS node (omit to load all matching nodes)') arg.completer = NodeNameCompleter( include_hidden_nodes_key='include_hidden_nodes') parser.add_argument( @@ -48,12 +78,42 @@ def add_arguments(self, parser, cli_name): # noqa: D102 '(default: waits indefinitely)') def main(self, *, args): # noqa: D102 - node_name = get_absolute_node_name(args.node_name) + use_wildcard = not args.no_use_wildcard + + if args.node_name is not None: + node_name = get_absolute_node_name(args.node_name) + with NodeStrategy(args) as node: + if not wait_for_node(node, node_name, args.include_hidden_nodes, args.timeout): + return 'Node not found' + + with DirectNode(args) as node: + load_parameter_file( + node=node, node_name=node_name, parameter_file=args.parameter_file, + use_wildcard=use_wildcard, timeout=args.service_timeout) + return + + # Validate the complete file before per-node matching. This keeps malformed + # file errors distinct from an otherwise-valid file not selecting a node. + parameter_dict_from_yaml_file(args.parameter_file, use_wildcard) + with NodeStrategy(args) as node: - if not wait_for_node(node, node_name, args.include_hidden_nodes, args.timeout): - return 'Node not found' + matching_node_names = [] + + def matching_node_available(): + nonlocal matching_node_names + matching_node_names = _get_matching_node_names( + node, + args.parameter_file, + use_wildcard, + args.include_hidden_nodes, + ) + return bool(matching_node_names) + + if not wait_for(matching_node_available, args.timeout): + return 'No matching nodes found' with DirectNode(args) as node: - load_parameter_file( - node=node, node_name=node_name, parameter_file=args.parameter_file, - use_wildcard=not args.no_use_wildcard, timeout=args.service_timeout) + for node_name in matching_node_names: + load_parameter_file( + node=node, node_name=node_name, parameter_file=args.parameter_file, + use_wildcard=use_wildcard, timeout=args.service_timeout) diff --git a/ros2param/test/test_load_all_nodes.py b/ros2param/test/test_load_all_nodes.py new file mode 100644 index 000000000..aea40bbe7 --- /dev/null +++ b/ros2param/test/test_load_all_nodes.py @@ -0,0 +1,125 @@ +# Copyright 2026 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 argparse +from types import SimpleNamespace +from unittest.mock import call +from unittest.mock import MagicMock +from unittest.mock import patch + +from ros2param.verb.load import LoadVerb + + +def _args(**kwargs): + values = { + 'node_name': None, + 'parameter_file': 'params.yaml', + 'no_use_wildcard': False, + 'include_hidden_nodes': False, + 'timeout': 1, + 'service_timeout': None, + } + values.update(kwargs) + return SimpleNamespace(**values) + + +def _run_wait_for_once(predicate, timeout): + return predicate() + + +def test_parser_accepts_parameter_file_without_node_name(): + parser = argparse.ArgumentParser() + LoadVerb().add_arguments(parser, 'ros2 param load') + + args = parser.parse_args(['params.yaml']) + + assert args.node_name is None + assert args.parameter_file == 'params.yaml' + + +def test_loads_parameter_file_for_all_matching_nodes(): + args = _args(timeout=3) + strategy = MagicMock() + strategy.daemon_node = MagicMock() + strategy_context = MagicMock() + strategy_context.__enter__.return_value = strategy + strategy_context.__exit__.return_value = False + direct = MagicMock() + direct_context = MagicMock() + direct_context.__enter__.return_value = direct + direct_context.__exit__.return_value = False + + discovered_nodes = [ + SimpleNamespace(full_name='/first'), + SimpleNamespace(full_name='/second'), + SimpleNamespace(full_name='/unmatched'), + ] + + def parse_parameter_file(parameter_file, use_wildcard, target_nodes=None): + if target_nodes == ['/unmatched']: + # Per-node matching must not depend on the text of this error. + raise RuntimeError('no parameters selected') + return {'parameter': object()} + + with patch('ros2param.verb.load.NodeStrategy', return_value=strategy_context), \ + patch('ros2param.verb.load.DirectNode', return_value=direct_context), \ + patch('ros2param.verb.load.get_node_names', return_value=discovered_nodes), \ + patch( + 'ros2param.verb.load.parameter_dict_from_yaml_file', + side_effect=parse_parameter_file + ), \ + patch( + 'ros2param.verb.load.wait_for', side_effect=_run_wait_for_once + ) as wait_for, \ + patch('ros2param.verb.load.load_parameter_file') as load_parameter_file: + result = LoadVerb().main(args=args) + + assert result is None + assert wait_for.call_args.args[1] == 3 + assert load_parameter_file.call_args_list == [ + call( + node=direct, node_name='/first', parameter_file='params.yaml', + use_wildcard=True, timeout=None), + call( + node=direct, node_name='/second', parameter_file='params.yaml', + use_wildcard=True, timeout=None), + ] + + +def test_returns_error_when_no_running_node_matches_file(): + args = _args() + strategy = MagicMock() + strategy.daemon_node = MagicMock() + strategy_context = MagicMock() + strategy_context.__enter__.return_value = strategy + strategy_context.__exit__.return_value = False + + def parse_parameter_file(parameter_file, use_wildcard, target_nodes=None): + if target_nodes is not None: + raise RuntimeError('no parameters selected') + return {'parameter': object()} + + with patch('ros2param.verb.load.NodeStrategy', return_value=strategy_context), \ + patch( + 'ros2param.verb.load.get_node_names', + return_value=[SimpleNamespace(full_name='/unmatched')] + ), \ + patch( + 'ros2param.verb.load.parameter_dict_from_yaml_file', + side_effect=parse_parameter_file + ), \ + patch('ros2param.verb.load.wait_for', side_effect=_run_wait_for_once): + result = LoadVerb().main(args=args) + + assert result == 'No matching nodes found' diff --git a/ros2param/test/test_verb_load.py b/ros2param/test/test_verb_load.py index 9570cef50..c2e4317c9 100644 --- a/ros2param/test/test_verb_load.py +++ b/ros2param/test/test_verb_load.py @@ -241,7 +241,7 @@ def test_verb_load_missing_args(self): assert param_load_command.exit_code != launch_testing.asserts.EXIT_OK assert launch_testing.tools.expect_output( expected_lines=['ros2 param load: error: the following arguments are required: ' - 'node_name, parameter_file'], + 'parameter_file'], text=param_load_command.output, strict=False ) @@ -249,8 +249,7 @@ def test_verb_load_missing_args(self): assert param_load_command.wait_for_shutdown(timeout=TEST_TIMEOUT) assert param_load_command.exit_code != launch_testing.asserts.EXIT_OK assert launch_testing.tools.expect_output( - expected_lines=['ros2 param load: error: the following arguments are required: ' - 'parameter_file'], + expected_lines=['No such file or directory'], text=param_load_command.output, strict=False ) diff --git a/ros2param/test/test_verb_load_all_nodes.py b/ros2param/test/test_verb_load_all_nodes.py new file mode 100644 index 000000000..319ccecf7 --- /dev/null +++ b/ros2param/test/test_verb_load_all_nodes.py @@ -0,0 +1,171 @@ +# Copyright 2026 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 contextlib +import os +import sys +import tempfile +import time +import unittest +import xmlrpc + +from launch import LaunchDescription +from launch.actions import ExecuteProcess +from launch.actions import RegisterEventHandler +from launch.actions import ResetEnvironment +from launch.actions import SetEnvironmentVariable +from launch.event_handlers import OnShutdown +from launch_ros.actions import Node +import launch_testing +import launch_testing.actions +import launch_testing.asserts +import launch_testing.markers +import launch_testing.tools +from launch_testing_ros.actions import EnableRmwIsolation +import launch_testing_ros.tools +import pytest +import rclpy +from rclpy.utilities import get_available_rmw_implementations + +from ros2cli.helpers import get_rmw_additional_env +from ros2cli.node.strategy import NodeStrategy + + +TEST_NODE = 'test_node' +TEST_NAMESPACE = '/foo' +TEST_TIMEOUT = 20.0 + + +@pytest.mark.rostest +@launch_testing.parametrize('rmw_implementation', get_available_rmw_implementations()) +def generate_test_description(rmw_implementation): + path_to_fixtures = os.path.join(os.path.dirname(__file__), 'fixtures') + additional_env = get_rmw_additional_env(rmw_implementation) + set_env_actions = [SetEnvironmentVariable(k, v) for k, v in additional_env.items()] + parameter_node = Node( + executable=sys.executable, + name=TEST_NODE, + namespace=TEST_NAMESPACE, + arguments=[os.path.join(path_to_fixtures, 'parameter_node.py')], + ) + + return LaunchDescription([ + ExecuteProcess( + cmd=['ros2', 'daemon', 'stop'], + name='daemon-stop', + on_exit=[ + *set_env_actions, + EnableRmwIsolation(), + RegisterEventHandler(OnShutdown(on_shutdown=[ + ExecuteProcess( + cmd=['ros2', 'daemon', 'stop'], + name='daemon-stop-isolated', + additional_env=dict(additional_env), + ), + ResetEnvironment(), + ])), + ExecuteProcess( + cmd=['ros2', 'daemon', 'start'], + name='daemon-start', + on_exit=[parameter_node, launch_testing.actions.ReadyToTest()], + ), + ], + ), + ]) + + +class TestVerbLoadAllNodes(unittest.TestCase): + + @classmethod + def setUpClass( + cls, + launch_service, + proc_info, + proc_output, + rmw_implementation, + ): + output_filter = launch_testing_ros.tools.basic_output_filter( + filtered_rmw_implementation=rmw_implementation + ) + + @contextlib.contextmanager + def launch_cli(self, arguments): + action = ExecuteProcess( + cmd=['ros2', 'param', *arguments], + name='ros2param-cli', + output='screen', + ) + with launch_testing.tools.launch_process( + launch_service, + action, + proc_info, + proc_output, + output_filter=output_filter, + ) as process: + yield process + + cls.launch_cli = launch_cli + + def setUp(self): + start_time = time.time() + with NodeStrategy(None) as node: + while (time.time() - start_time) < TEST_TIMEOUT: + try: + services = node.get_service_names_and_types_by_node( + TEST_NODE, TEST_NAMESPACE + ) + except rclpy.node.NodeNameNonExistentError: + continue + except ConnectionRefusedError: + continue + except xmlrpc.client.Fault as e: + if 'NodeNameNonExistentError' in e.faultString: + continue + raise + + if f'{TEST_NAMESPACE}/{TEST_NODE}/set_parameters' in { + name for name, _ in services + }: + return + self.fail(f'CLI daemon failed to find test node after {TEST_TIMEOUT} seconds') + + def test_load_parameter_file_without_node_name(self): + with tempfile.TemporaryDirectory() as tmpdir: + filepath = os.path.join(tmpdir, 'params.yaml') + with open(filepath, 'w') as f: + f.write( + f'{TEST_NAMESPACE}/{TEST_NODE}:\n' + ' ros__parameters:\n' + ' str_param: Loaded by all-node mode\n' + ) + + with self.launch_cli(['load', filepath, '--timeout', '3']) as process: + assert process.wait_for_shutdown(timeout=TEST_TIMEOUT) + assert process.exit_code == launch_testing.asserts.EXIT_OK + + with self.launch_cli([ + 'get', + f'{TEST_NAMESPACE}/{TEST_NODE}', + 'str_param', + '--hide-type', + '--timeout', + '3', + ]) as process: + assert process.wait_for_shutdown(timeout=TEST_TIMEOUT) + assert process.exit_code == launch_testing.asserts.EXIT_OK + assert launch_testing.tools.expect_output( + expected_lines=['Loaded by all-node mode'], + text=process.output, + strict=True, + )