ros2doctor: add and test node/parameter reporting

- Add and update tests: test_node.py, test_parameter.py, test_api.py, common.py
- Add test coverage for node and parameter for ros2doctor report
Signed-off-by: BhuvanB404<bhuvanb1408@gmail.com>
Signed-off-by: BhuvanB404 <bhuvanb1408@gmail.com>

Author: BhuvanB404

ros2doctor/ros2doctor/api/parameter.py

@@ -13,8 +13,8 @@
 # limitations under the License.
 
 
-from rcl_interfaces.srv import ListParameters
 import rclpy
+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 +24,19 @@
 from ros2doctor.api.format import doctor_warn
 
 
-def call_list_parameters(node, node_name, namespace='/'):
-    """Call the list_parameters service for a specific node."""
+def call_list_parameters(node, node_name):
+    """Call the list_parameters service for a specific node using AsyncParameterClient."""
     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
-
-        request = ListParameters.Request()
-        future = client.call_async(request)
-
-        # Spin until the service call completes or times out
+        future = client.list_parameters()
         rclpy.spin_until_future_complete(node, future, timeout_sec=2.0)
-
-        if future.done():
-            response = future.result()
-            node.destroy_client(client)
-            return response
-        else:
-            node.destroy_client(client)
-            return None
-
+        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
 
@@ -70,9 +56,9 @@ def check(self):
                 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}"
+                    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
@@ -103,17 +89,14 @@ 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)
-                            for param_name in sorted(param_names):
-                                report.add_to_report('parameter', param_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)
+                        for param_name in sorted(param_names):
+                            report.add_to_report('parameter', param_name)
 
             report.add_to_report('total nodes checked', nodes_checked)
             report.add_to_report('total parameter count', total_param_count)

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,67 @@
+# Copyright 2019 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()

ros2doctor/test/test_api.py

@@ -14,6 +14,7 @@
 
 import unittest
 
+from common import generate_expected_node_report
 from common import generate_expected_service_report
 from common import generate_expected_topic_report
 
@@ -25,6 +26,10 @@
 
 import pytest
 
+from ros2doctor.api.node import NodeCheck
+from ros2doctor.api.node import NodeReport
+from ros2doctor.api.parameter import ParameterCheck
+from ros2doctor.api.parameter import ParameterReport
 from ros2doctor.api.service import ServiceReport
 from ros2doctor.api.topic import TopicCheck
 from ros2doctor.api.topic import TopicReport
@@ -74,3 +79,29 @@ def test_no_service_report(self):
         self.assertEqual(report.name, expected_report.name)
         self.assertEqual(report.items, expected_report.items)
         self.assertEqual(report, expected_report)
+
+    def test_node_check(self):
+        """Assume no duplicate nodes exist in a clean environment."""
+        node_check = NodeCheck()
+        check_result = node_check.check()
+        self.assertEqual(check_result.error, 0)
+        self.assertEqual(check_result.warning, 0)
+
+    def test_no_node_report(self):
+        """Assume no nodes are running in a clean environment."""
+        report = NodeReport().report()
+        expected_report = generate_expected_node_report(0, [])
+        self.assertEqual(report.name, expected_report.name)
+        self.assertEqual(report.name, 'NODE LIST')
+
+    def test_parameter_check(self):
+        """Test parameter check runs without errors in clean environment."""
+        param_check = ParameterCheck()
+        check_result = param_check.check()
+        self.assertEqual(check_result.error, 0)
+
+    def test_no_parameter_report(self):
+        """Assume no parameters in a clean environment."""
+        report = ParameterReport().report()
+        self.assertEqual(report.name, 'PARAMETER LIST')
+        self.assertIsNotNone(report)