Add goal file input to ros2 action send_goal

ros2action/ros2action/verb/send_goal.py

@@ -61,6 +61,9 @@ def add_arguments(self, parser, cli_name):
         group.add_argument(
             '--stdin', action='store_true',
             help='Read goal from standard input')
+        group.add_argument(
+            '--goal-file', metavar='FILE',
+            help='Read goal request values from a YAML file')
         arg.completer = ActionGoalPrototypeCompleter(action_type_key='action_type')
         parser.add_argument(
             '-f', '--feedback', action='store_true',
@@ -81,6 +84,14 @@ def main(self, *, args):
 
         if args.stdin:
             goal = collect_stdin()
+        elif args.goal_file:
+            try:
+                with open(args.goal_file, 'r', encoding='utf-8') as goal_file:
+                    goal = goal_file.read()
+            except OSError as e:
+                return f'Failed to read goal file: {e}'
+            if not goal.strip():
+                return 'Goal file is empty'
         else:
             goal = args.goal
 

ros2action/test/test_send_goal_file.py

@@ -0,0 +1,74 @@
+# 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.
+
+from argparse import ArgumentParser
+from unittest.mock import patch
+
+from ros2action.verb.send_goal import SendGoalVerb
+
+
+def _parse_goal_file(path):
+    parser = ArgumentParser()
+    verb = SendGoalVerb()
+    verb.add_arguments(parser, 'action')
+    args = parser.parse_args([
+        '/fibonacci',
+        'example_interfaces/action/Fibonacci',
+        '--goal-file', str(path),
+    ])
+    return verb, args
+
+
+def test_send_goal_reads_goal_file_when_command_runs(tmp_path):
+    goal_file = tmp_path / 'goal.yaml'
+    goal_file.write_text('order: 7\n', encoding='utf-8')
+    verb, args = _parse_goal_file(goal_file)
+
+    assert args.goal_file == str(goal_file)
+
+    with patch('ros2action.verb.send_goal.send_goal') as mock_send_goal:
+        result = verb.main(args=args)
+
+    assert result is mock_send_goal.return_value
+    mock_send_goal.assert_called_once_with(
+        '/fibonacci',
+        'example_interfaces/action/Fibonacci',
+        'order: 7\n',
+        None,
+        None,
+    )
+
+
+def test_send_goal_rejects_empty_goal_file(tmp_path):
+    goal_file = tmp_path / 'empty.yaml'
+    goal_file.write_text('', encoding='utf-8')
+    verb, args = _parse_goal_file(goal_file)
+
+    with patch('ros2action.verb.send_goal.send_goal') as mock_send_goal:
+        result = verb.main(args=args)
+
+    assert result == 'Goal file is empty'
+    mock_send_goal.assert_not_called()
+
+
+def test_send_goal_reports_goal_file_read_error(tmp_path):
+    missing_goal_file = tmp_path / 'missing.yaml'
+    verb, args = _parse_goal_file(missing_goal_file)
+
+    with patch('ros2action.verb.send_goal.send_goal') as mock_send_goal:
+        result = verb.main(args=args)
+
+    assert result.startswith('Failed to read goal file:')
+    assert str(missing_goal_file) in result
+    mock_send_goal.assert_not_called()

ros2action/test/test_send_goal_file_cli.py

@@ -0,0 +1,142 @@
+# 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 re
+import sys
+import tempfile
+import unittest
+
+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
+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
+from rclpy.utilities import get_available_rmw_implementations
+
+from ros2cli.helpers import get_rmw_additional_env
+
+
+if sys.platform.startswith('win'):
+    pytest.skip(
+        'CLI tests can block for a pathological amount of time on Windows.',
+        allow_module_level=True,
+    )
+
+
+@pytest.mark.rostest
+@launch_testing.parametrize('rmw_implementation', get_available_rmw_implementations())
+def generate_test_description(rmw_implementation):
+    action_server = os.path.join(
+        os.path.dirname(__file__), 'fixtures', 'fibonacci_action_server.py'
+    )
+    additional_env = get_rmw_additional_env(rmw_implementation)
+    set_env_actions = [SetEnvironmentVariable(k, v) for k, v in additional_env.items()]
+
+    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=[
+                        ExecuteProcess(cmd=[sys.executable, action_server]),
+                        launch_testing.actions.ReadyToTest(),
+                    ],
+                ),
+            ],
+        ),
+    ])
+
+
+class TestSendGoalFileCLI(unittest.TestCase):
+
+    @classmethod
+    def setUpClass(
+        cls,
+        launch_service,
+        proc_info,
+        proc_output,
+        rmw_implementation,
+    ):
+        @contextlib.contextmanager
+        def launch_action_command(self, arguments):
+            action = ExecuteProcess(
+                cmd=['ros2', 'action', *arguments],
+                name='ros2action-goal-file-cli',
+                output='screen',
+                additional_env={'PYTHONUNBUFFERED': '1'},
+            )
+            with launch_testing.tools.launch_process(
+                launch_service,
+                action,
+                proc_info,
+                proc_output,
+                output_filter=launch_testing_ros.tools.basic_output_filter(
+                    filtered_rmw_implementation=rmw_implementation
+                ),
+            ) as process:
+                yield process
+
+        cls.launch_action_command = launch_action_command
+
+    @launch_testing.markers.retry_on_failure(times=5, delay=1)
+    def test_goal_file_reaches_action_server(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            goal_file = os.path.join(tmpdir, 'goal.yaml')
+            with open(goal_file, 'w', encoding='utf-8') as f:
+                f.write('order: 5\n')
+
+            with self.launch_action_command([
+                'send_goal',
+                '/test/fibonacci',
+                'test_msgs/action/Fibonacci',
+                '--goal-file',
+                goal_file,
+            ]) as process:
+                assert process.wait_for_shutdown(timeout=10)
+
+        assert process.exit_code == launch_testing.asserts.EXIT_OK
+        assert launch_testing.tools.expect_output(
+            expected_lines=[
+                'Sending goal:',
+                '     order: 5',
+                re.compile('Goal accepted with ID: [a-f0-9]+'),
+                'Goal finished with status: SUCCEEDED',
+            ],
+            text=process.output,
+            strict=False,
+        )