check for invalid ROS discovery configuration and print warning if needed.

Signed-off-by: Tomoya Fujita <Tomoya.Fujita@sony.com>

Author: Tomoya Fujita

ros2cli/ros2cli/helpers.py

@@ -122,6 +122,27 @@ def collect_stdin():
     return lines
 
 
+def check_discovery_configuration():
+    """
+    Check for invalid ROS discovery configuration and print warning if needed.
+
+    Warns when ROS_AUTOMATIC_DISCOVERY_RANGE=OFF is set without ROS_STATIC_PEERS,
+    which results in no discovery mechanism being available.
+    """
+    discovery_range = os.environ.get('ROS_AUTOMATIC_DISCOVERY_RANGE', '')
+    static_peers = os.environ.get('ROS_STATIC_PEERS', '')
+
+    if discovery_range == 'OFF' and not static_peers:
+        print(
+            'Warning: ROS_AUTOMATIC_DISCOVERY_RANGE=OFF with no ROS_STATIC_PEERS configured.\n'
+            'No discovery mechanism is available. Results will be empty.\n'
+            'Either:\n'
+            '  - Set ROS_STATIC_PEERS to specify peers explicitly, or\n'
+            '  - Change ROS_AUTOMATIC_DISCOVERY_RANGE to LOCALHOST or SUBNET',
+            file=sys.stderr
+        )
+
+
 def get_rmw_additional_env(rmw_implementation: str) -> Dict[str, str]:
     """Get a dictionary of additional environment variables based on rmw."""
     if rmw_implementation == 'rmw_zenoh_cpp':

ros2cli/ros2cli/node/direct.py

@@ -19,13 +19,17 @@
 import rclpy.action
 
 from rclpy.parameter import Parameter
+from ros2cli.helpers import check_discovery_configuration
 from ros2cli.node import NODE_NAME_PREFIX
 DEFAULT_TIMEOUT = 0.5
 
 
 class DirectNode:
 
     def __init__(self, args, *, node_name: Optional[str] = None):
+        # Check for invalid discovery configuration
+        check_discovery_configuration()
+
         timeout_reached = False
 
         def timer_callback():

ros2cli/ros2cli/node/strategy.py

@@ -14,6 +14,7 @@
 
 from typing import Optional
 
+from ros2cli.helpers import check_discovery_configuration
 from ros2cli.node.daemon import add_arguments as add_daemon_node_arguments
 from ros2cli.node.daemon import DaemonNode
 from ros2cli.node.daemon import is_daemon_running
@@ -25,6 +26,9 @@
 class NodeStrategy:
 
     def __init__(self, args, *, node_name: Optional[str] = None):
+        # Check for invalid discovery configuration
+        check_discovery_configuration()
+
         use_daemon = not getattr(args, 'no_daemon', False)
         if use_daemon and is_daemon_running(args):
             self._daemon_node = DaemonNode(args)

ros2cli/test/test_helpers.py

@@ -0,0 +1,76 @@
+# Copyright 2026 Sony Corporation
+#
+# 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 io
+import os
+from unittest.mock import patch
+
+from ros2cli.helpers import check_discovery_configuration
+
+
+def test_check_discovery_configuration_off_without_peers():
+    """Test warning is shown when ROS_AUTOMATIC_DISCOVERY_RANGE=OFF without ROS_STATIC_PEERS."""
+    env = {
+        'ROS_AUTOMATIC_DISCOVERY_RANGE': 'OFF',
+    }
+    # Make sure ROS_STATIC_PEERS is not set
+    env_without_peers = {k: v for k, v in os.environ.items() if k != 'ROS_STATIC_PEERS'}
+    env_without_peers.update(env)
+
+    # Capture stderr
+    stderr_capture = io.StringIO()
+    with patch.dict(os.environ, env_without_peers, clear=True):
+        with patch('sys.stderr', stderr_capture):
+            check_discovery_configuration()
+
+    output = stderr_capture.getvalue()
+    assert 'Warning: ROS_AUTOMATIC_DISCOVERY_RANGE=OFF' in output
+    assert 'No discovery mechanism is available' in output
+    assert 'ROS_STATIC_PEERS' in output
+    assert 'LOCALHOST or SUBNET' in output
+
+
+def test_check_discovery_configuration_off_with_peers():
+    """Test no warning when ROS_AUTOMATIC_DISCOVERY_RANGE=OFF with ROS_STATIC_PEERS set."""
+    env = {
+        'ROS_AUTOMATIC_DISCOVERY_RANGE': 'OFF',
+        'ROS_STATIC_PEERS': '192.168.1.10;192.168.1.11',
+    }
+
+    # Capture stderr
+    stderr_capture = io.StringIO()
+    with patch.dict(os.environ, env, clear=True):
+        with patch('sys.stderr', stderr_capture):
+            check_discovery_configuration()
+
+    output = stderr_capture.getvalue()
+    assert output == '', 'Expected no warning output'
+
+
+def test_check_discovery_configuration_not_set():
+    """Test no warning when ROS_AUTOMATIC_DISCOVERY_RANGE is not set (default behavior)."""
+    # Make sure neither env var is set
+    env_clean = {
+        k: v for k, v in os.environ.items()
+        if k not in ('ROS_AUTOMATIC_DISCOVERY_RANGE', 'ROS_STATIC_PEERS')
+    }
+
+    # Capture stderr
+    stderr_capture = io.StringIO()
+    with patch.dict(os.environ, env_clean, clear=True):
+        with patch('sys.stderr', stderr_capture):
+            check_discovery_configuration()
+
+    output = stderr_capture.getvalue()
+    assert output == '', 'Expected no warning output'