Fix tests with get_type_description service and param present (#838)

Signed-off-by: Emerson Knapp <[email protected]>

Author: emersonknapp

ros2cli/ros2cli/daemon/__init__.py

@@ -75,7 +75,8 @@ def serve(server, *, timeout=2 * 60 * 60):
         ros_domain_id = get_ros_domain_id()
         node_args = argparse.Namespace(
             node_name_suffix=f'_daemon_{ros_domain_id}_{uuid.uuid4().hex}',
-            start_parameter_services=False)
+            start_parameter_services=False,
+            start_type_description_service=False)
         with NetworkAwareNode(node_args) as node:
             functions = [
                 node.get_name,

ros2cli/ros2cli/node/direct.py

@@ -40,6 +40,7 @@ def timer_callback():
         start_parameter_services = getattr(
             args, 'start_parameter_services', False)
         use_sim_time = getattr(args, 'use_sim_time', False)
+        start_type_description_service = getattr(args, 'start_type_description_service', True)
 
         if node_name is None:
             node_name = NODE_NAME_PREFIX + node_name_suffix
@@ -48,7 +49,8 @@ def timer_callback():
             node_name,
             start_parameter_services=start_parameter_services,
             parameter_overrides=[
-                Parameter('use_sim_time', value=use_sim_time)
+                Parameter('use_sim_time', value=use_sim_time),
+                Parameter('start_type_description_service', value=start_type_description_service),
             ], automatically_declare_parameters_from_overrides=True)
 
         timeout = getattr(args, 'spin_time', DEFAULT_TIMEOUT)

ros2param/test/test_verb_dump.py

@@ -63,6 +63,7 @@
     '    - 2\n'
     '    - 3\n'
     '    int_param: 42\n'
+    '    start_type_description_service: true\n'
     '    str_array_param:\n'
     '    - foo\n'
     '    - bar\n'

ros2param/test/test_verb_list.py

@@ -180,6 +180,7 @@ def test_verb_list(self):
                 '  foo.str_param',
                 '  int_array_param',
                 '  int_param',
+                '  start_type_description_service',
                 '  str_array_param',
                 '  str_param',
                 '  use_sim_time'],

ros2param/test/test_verb_load.py

@@ -66,6 +66,7 @@
     '    - 3\n'
     '    - 3\n'
     '    int_param: -42\n'
+    '    start_type_description_service: true\n'
     '    str_array_param:\n'
     '    - a_foo\n'
     '    - a_bar\n'

ros2service/test/test_cli.py

@@ -16,7 +16,6 @@
 import functools
 import itertools
 import os
-import re
 import sys
 import unittest
 
@@ -105,6 +104,16 @@ def generate_test_description(rmw_implementation):
 
 
 class TestROS2ServiceCLI(unittest.TestCase):
+    expected_builtin_services = {
+        'describe_parameters': 'rcl_interfaces/srv/DescribeParameters',
+        'get_parameter_types': 'rcl_interfaces/srv/GetParameterTypes',
+        'get_parameters': 'rcl_interfaces/srv/GetParameters',
+        'get_type_description': 'type_description_interfaces/srv/GetTypeDescription',
+        'list_parameters': 'rcl_interfaces/srv/ListParameters',
+        'set_parameters': 'rcl_interfaces/srv/SetParameters',
+        'set_parameters_atomically': 'rcl_interfaces/srv/SetParametersAtomically',
+    }
+    builtin_service_count = len(expected_builtin_services)
 
     @classmethod
     def setUpClass(
@@ -138,6 +147,13 @@ def launch_service_command(self, arguments):
                 yield service_command
         cls.launch_service_command = launch_service_command
 
+    @classmethod
+    def get_expected_builtin_services(cls, namespace: str, with_types: bool = False) -> list[str]:
+        return [
+            f'{namespace}/{srv_name}' + (f' [{srv_type}]' if with_types else '')
+            for srv_name, srv_type in cls.expected_builtin_services.items()
+        ]
+
     @launch_testing.markers.retry_on_failure(times=5, delay=1)
     def test_list_services(self):
         with self.launch_service_command(arguments=['list']) as service_command:
@@ -146,9 +162,7 @@ def test_list_services(self):
         assert launch_testing.tools.expect_output(
             expected_lines=itertools.chain(
                 ['/my_ns/echo'],
-                itertools.repeat(re.compile(
-                    r'/my_ns/echo_server/.*parameter.*'
-                ), 6)
+                self.get_expected_builtin_services('/my_ns/echo_server'),
             ),
             text=service_command.output,
             strict=True
@@ -164,13 +178,9 @@ def test_list_hidden(self):
         assert launch_testing.tools.expect_output(
             expected_lines=itertools.chain(
                 ['/my_ns/_echo'],
-                itertools.repeat(re.compile(
-                    r'/my_ns/_hidden_echo_server/.*parameter.*'
-                ), 6),
+                self.get_expected_builtin_services('/my_ns/_hidden_echo_server'),
                 ['/my_ns/echo'],
-                itertools.repeat(re.compile(
-                    r'/my_ns/echo_server/.*parameter.*'
-                ), 6)
+                self.get_expected_builtin_services('/my_ns/echo_server'),
             ),
             text=service_command.output,
             strict=True
@@ -184,10 +194,7 @@ def test_list_with_types(self):
         assert launch_testing.tools.expect_output(
             expected_lines=itertools.chain(
                 ['/my_ns/echo [test_msgs/srv/BasicTypes]'],
-                itertools.repeat(re.compile(
-                    r'/my_ns/echo_server/.*parameter.*'
-                    r' \[rcl_interfaces/srv/.*Parameter.*\]'
-                ), 6)
+                self.get_expected_builtin_services('/my_ns/echo_server', with_types=True)
             ),
             text=service_command.output,
             strict=True
@@ -200,7 +207,7 @@ def test_list_count(self):
         assert service_command.exit_code == launch_testing.asserts.EXIT_OK
         output_lines = service_command.output.splitlines()
         assert len(output_lines) == 1
-        assert int(output_lines[0]) == 7
+        assert int(output_lines[0]) == self.builtin_service_count + 1
 
     @launch_testing.markers.retry_on_failure(times=5, delay=1)
     def test_find(self):