ros2doctor: add and test node/parameter reporting

- Add and update tests: test_node.py, test_parameter.py, test_api.py, common.py

Signed-off-by: BhuvanB404<bhuvanb1408@gmail.com>
Signed-off-by: BhuvanB404 <bhuvanb1408@gmail.com>

Author: BhuvanB404

ros2doctor/ros2doctor/api/node.py

@@ -4,7 +4,7 @@
 # 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
+#     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,
@@ -13,20 +13,38 @@
 # limitations under the License.
 
 
-from collections import defaultdict
+from collections import Counter
+from collections import namedtuple
 
+from rclpy.node import HIDDEN_NODE_PREFIX
 from ros2cli.node.strategy import NodeStrategy
 from ros2doctor.api import DoctorCheck
 from ros2doctor.api import DoctorReport
 from ros2doctor.api import Report
 from ros2doctor.api import Result
 from ros2doctor.api.format import doctor_warn
-from ros2node.api import get_node_names
 
 
-def has_duplicates(values):
-    """Find out if there are any exact duplicates in a list of strings."""
-    return len(set(values)) < len(values)
+NodeName = namedtuple('NodeName', ('name', 'namespace', 'full_name'))
+
+
+def get_node_names(*, node, include_hidden_nodes=False):
+    node_names_and_namespaces = node.get_node_names_and_namespaces()
+    return [
+        NodeName(
+            name=t[0],
+            namespace=t[1],
+            full_name=t[1] + ('' if t[1].endswith('/') else '/') + t[0],
+        )
+        for t in node_names_and_namespaces
+        if (include_hidden_nodes or (t[0] and not t[0].startswith(HIDDEN_NODE_PREFIX)))
+    ]
+
+
+def find_duplicates(values):
+    """Return values that appear more than once."""
+    counts = Counter(values)
+    return [v for v, c in counts.items() if c > 1]
 
 
 class NodeCheck(DoctorCheck):
@@ -40,15 +58,10 @@ def check(self):
         with NodeStrategy(None) as node:
             node_list = get_node_names(node=node, include_hidden_nodes=True)
             node_names = [n.full_name for n in node_list]
-            if has_duplicates(node_names):
-                name_counts = defaultdict(int)
-                for name in node_names:
-                    name_counts[name] += 1
-
-                duplicates = [name for name, count in name_counts.items() if count > 1]
-                for duplicate in duplicates:
-                    doctor_warn(f'Duplicate node name: {duplicate}')
-                    result.add_warning()
+            duplicates = find_duplicates(node_names)
+            for duplicate in duplicates:
+                doctor_warn(f'Duplicate node name: {duplicate}')
+                result.add_warning()
         return result
 
 

ros2doctor/ros2doctor/api/parameter.py

@@ -4,7 +4,7 @@
 # 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
+#     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,
@@ -13,8 +13,10 @@
 # limitations under the License.
 
 
-from rcl_interfaces.srv import ListParameters
 import rclpy
+from rclpy.parameter import parameter_value_to_python
+from rclpy.parameter_client import AsyncParameterClient
+
 from ros2cli.node.direct import DirectNode
 from ros2cli.node.strategy import NodeStrategy
 from ros2doctor.api import DoctorCheck
@@ -24,33 +26,40 @@
 from ros2doctor.api.format import doctor_warn
 
 
-def call_list_parameters(node, node_name, namespace='/'):
+def call_list_parameters(node, node_name):
     """Call the list_parameters service for a specific node."""
     try:
-        # Create service name and client for the target node's list_parameters service
-        service_name = f"{namespace.rstrip('/')}/{node_name}/list_parameters"
-        if service_name.startswith('//'):
-            service_name = service_name[1:]
-
-        client = node.create_client(ListParameters, service_name)
-
-        if not client.wait_for_service(timeout_sec=1.0):
+        client = AsyncParameterClient(node, node_name)
+        ready = client.wait_for_services(timeout_sec=2.0)
+        if not ready:
             return None
+        future = client.list_parameters()
+        rclpy.spin_until_future_complete(node, future, timeout_sec=2.0)
+        if future.done() and future.result() is not None:
+            # Return the parameter names list
+            return future.result().result.names
+        return None
+    except Exception:
+        return None
 
-        request = ListParameters.Request()
-        future = client.call_async(request)
 
-        # Spin until the service call completes or times out
-        rclpy.spin_until_future_complete(node, future, timeout_sec=2.0)
+def call_get_parameters(node, node_name, parameter_names):
+    """
+    Call the get_parameters service for a specific node.
 
-        if future.done():
-            response = future.result()
-            node.destroy_client(client)
-            return response
-        else:
-            node.destroy_client(client)
+    Return the parameter values.
+    """
+    try:
+        client = AsyncParameterClient(node, node_name)
+        ready = client.wait_for_services(timeout_sec=2.0)
+        if not ready:
             return None
-
+        future = client.get_parameters(parameter_names)
+        rclpy.spin_until_future_complete(node, future, timeout_sec=2.0)
+        if future.done() and future.result() is not None:
+            # Return the parameter values list
+            return future.result().values
+        return None
     except Exception:
         return None
 
@@ -65,15 +74,18 @@ def check(self):
         result = Result()
         with NodeStrategy(None) as node:
             try:
-                node_names_and_namespaces = node.get_node_names_and_namespaces()
+                node_names_and_namespaces = \
+                    node.get_node_names_and_namespaces()
             except Exception:
                 node_names_and_namespaces = []
             with DirectNode(None) as param_node:
                 for node_name, namespace in node_names_and_namespaces:
-                    response = call_list_parameters(param_node.node, node_name, namespace)
-                    if response is None:
-                        full_name = f"{namespace.rstrip('/')}/{node_name}"
-                        doctor_warn(f'Node {full_name} has no parameter services.')
+                    full_name = f"{namespace.rstrip('/')}/{node_name}"
+                    param_names = call_list_parameters(
+                        param_node.node, full_name)
+                    if param_names is None:
+                        doctor_warn(
+                            f'Node {full_name} has no parameter services.')
                         result.add_warning()
         return result
 
@@ -88,7 +100,8 @@ def report(self):
         report = Report('PARAMETER LIST')
         with NodeStrategy(None) as node:
             try:
-                node_names_and_namespaces = node.get_node_names_and_namespaces()
+                node_names_and_namespaces = \
+                    node.get_node_names_and_namespaces()
             except Exception:
                 node_names_and_namespaces = []
             if not node_names_and_namespaces:
@@ -103,15 +116,29 @@ def report(self):
             with DirectNode(None) as param_node:
                 for node_name, namespace in sorted(node_names_and_namespaces):
                     nodes_checked += 1
-                    response = call_list_parameters(param_node.node, node_name, namespace)
-                    if response and hasattr(response, 'result') and response.result:
-                        result = response.result
-                        param_names = result.names if hasattr(result, 'names') else []
-                        if param_names:
-                            total_param_count += len(param_names)
-                            full_name = f"{namespace.rstrip('/')}/{node_name}"
-
-                            report.add_to_report('node', full_name)
+                    full_name = f"{namespace.rstrip('/')}/{node_name}"
+                    param_names = call_list_parameters(
+                        param_node.node, full_name)
+
+                    if param_names:
+                        total_param_count += len(param_names)
+                        report.add_to_report('node', full_name)
+                        param_values = call_get_parameters(
+                            param_node.node, full_name, param_names
+                        )
+                        if (
+                            param_values and
+                            len(param_values) == len(param_names)
+                        ):
+                            params_with_values = sorted(
+                                zip(param_names, param_values),
+                                key=lambda x: x[0]
+                            )
+                            for name, value_msg in params_with_values:
+                                value = parameter_value_to_python(value_msg)
+                                report.add_to_report(
+                                    'parameter', f'{name}: {value}')
+                        else:
                             for param_name in sorted(param_names):
                                 report.add_to_report('parameter', param_name)
 

ros2doctor/test/common.py

@@ -33,3 +33,32 @@ def generate_expected_service_report(services: Iterable[str], serv_counts: Itera
         expected_report.add_to_report('service count', serv_count)
         expected_report.add_to_report('client count', cli_count)
     return expected_report
+
+
+def generate_expected_node_report(node_count: int, nodes: Iterable[str]) -> Report:
+    expected_report = Report('NODE LIST')
+    if node_count == 0:
+        expected_report.add_to_report('node count', 0)
+        expected_report.add_to_report('node', 'none')
+    else:
+        expected_report.add_to_report('node count', node_count)
+        for node in nodes:
+            expected_report.add_to_report('node', node)
+    return expected_report
+
+
+def generate_expected_parameter_report(nodes_checked: int, param_count: int,
+                                       node_params: Iterable[tuple]) -> Report:
+    expected_report = Report('PARAMETER LIST')
+    if nodes_checked == 0:
+        expected_report.add_to_report('total nodes checked', 0)
+        expected_report.add_to_report('total parameter count', 0)
+        expected_report.add_to_report('parameter', 'none')
+    else:
+        for node_name, params in node_params:
+            expected_report.add_to_report('node', node_name)
+            for param in params:
+                expected_report.add_to_report('parameter', param)
+        expected_report.add_to_report('total nodes checked', nodes_checked)
+        expected_report.add_to_report('total parameter count', param_count)
+    return expected_report

ros2doctor/test/fixtures/complex_node.py

@@ -0,0 +1,69 @@
+# 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 rclpy
+from rclpy.action import ActionServer
+from rclpy.executors import ExternalShutdownException
+from rclpy.node import Node
+from rclpy.qos import qos_profile_system_default
+
+from test_msgs.action import Fibonacci
+from test_msgs.msg import Arrays
+from test_msgs.msg import Strings
+from test_msgs.srv import BasicTypes
+
+
+class ComplexNode(Node):
+
+    def __init__(self):
+        super().__init__('complex_node')
+        self.publisher = self.create_publisher(
+            Arrays, 'arrays', qos_profile_system_default
+        )
+        self.subscription = self.create_subscription(
+            Strings, 'strings', lambda msg: None, qos_profile_system_default
+        )
+        self.server = self.create_service(BasicTypes, 'basic', lambda req, res: res)
+        self.action_server = ActionServer(
+            self, Fibonacci, 'fibonacci', self.action_callback
+        )
+        self.timer = self.create_timer(1.0, self.pub_callback)
+
+    def destroy_node(self):
+        self.timer.destroy()
+        self.publisher.destroy()
+        self.subscription.destroy()
+        self.server.destroy()
+        self.action_server.destroy()
+        super().destroy_node()
+
+    def pub_callback(self):
+        self.publisher.publish(Arrays())
+
+    def action_callback(self, goal_handle):
+        goal_handle.succeed()
+        return Fibonacci.Result()
+
+
+def main(args=None):
+    try:
+        with rclpy.init(args=args):
+            node = ComplexNode()
+            rclpy.spin(node)
+    except (KeyboardInterrupt, ExternalShutdownException):
+        print('node stopped cleanly')
+
+
+if __name__ == '__main__':
+    main()

ros2doctor/test/fixtures/parameter_fixtures_node.py

@@ -0,0 +1,44 @@
+# Copyright 2025 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.
+
+"""Test fixture node with parameters for ros2doctor testing."""
+
+import rclpy
+from rclpy.executors import ExternalShutdownException
+from rclpy.parameter import PARAMETER_SEPARATOR_STRING
+
+
+def main():
+    try:
+        with rclpy.init():
+            node = rclpy.create_node('parameter_node')
+            # Declare various parameter types for testing
+            node.declare_parameter('bool_param', True)
+            node.declare_parameter('int_param', 42)
+            node.declare_parameter('double_param', 3.14)
+            node.declare_parameter('str_param', 'hello')
+            node.declare_parameter('bool_array_param', [True, False])
+            node.declare_parameter('int_array_param', [1, 2, 3])
+            node.declare_parameter('double_array_param', [1.0, 2.0])
+            node.declare_parameter('str_array_param', ['foo', 'bar'])
+            node.declare_parameter(
+                'nested' + PARAMETER_SEPARATOR_STRING + 'param', 'nested_value'
+            )
+            rclpy.spin(node)
+    except (KeyboardInterrupt, ExternalShutdownException):
+        print('parameter_node stopped cleanly')
+
+
+if __name__ == '__main__':
+    main()