Repository: felixdivo/ros2-easy-test (language: python)

Issue:
Add an easy method to call a service with a timeout
Afterward, one should adjust `ros2-easy-test/doc/source/design_and_limits.rst` accordingly.


Discussion / hints:

Fix patch:
diff --git a/.github/workflows/python-package.yaml b/.github/workflows/python-package.yaml
index 175c3d9..8c2a287 100644
--- a/.github/workflows/python-package.yaml
+++ b/.github/workflows/python-package.yaml
@@ -39,6 +39,15 @@ jobs:
         name: Lint (with ruff)
         run: |
           ruff check .
+      -
+        name: Type check main code (with mypy)
+        run: |
+          mypy ros2_easy_test
+      -
+        name: Type check tests (with mypy)
+        continue-on-error: true  # We do not want to fail the build if the type checker fails on the tests
+        run: |
+          mypy tests
       -
         name: Testing (with pytest)
         run: |
diff --git a/.readthedocs.yaml b/.readthedocs.yaml
index fc64e0a..900a04d 100644
--- a/.readthedocs.yaml
+++ b/.readthedocs.yaml
@@ -17,7 +17,7 @@ python:
 sphinx:
   builder: html
   configuration: doc/source/conf.py
-  fail_on_warning: true
+  fail_on_warning: false
 
 formats:  # in addition to the HTML variant
   - pdf
diff --git a/README.md b/README.md
index 2e35bbe..bebd7dd 100644
--- a/README.md
+++ b/README.md
@@ -81,16 +81,16 @@ Using `ROS2TestEnvironment`, you can call:
 - `publish(topic: str, message: RosMessage) -> None`
 - `listen_for_messages(topic: str, time_span: float) -> List[RosMessage]`
 - `clear_messages(topic: str) -> None` to forget all messages that have been received so far.
+- `call_service(name: str, request: Request, timeout_availability: Optional[float], timeout_call: Optional[float]) -> Response`
 
 Note that `ROS2TestEnvironment` is a `rclpy.node.Node` and thus has all the methods of a ROS2 node. 
-As an example, you can create a service by `env.create_service(...)` and then use it with `env.call(...)`.
 In addition, nothing stops you from using any other means of interacting with ROS2 that would work otherwise.
 
 ### What you can test (recommended way of obtaining messages)
 
 Using `ROS2TestEnvironment`, you can assert:
-- `assert_message_published(topic: str, timeout: float) -> RosMessage`
-- `assert_no_message_published(topic: str, timeout: float) -> None`
+- `assert_message_published(topic: str, timeout: Optional[float]) -> RosMessage`
+- `assert_no_message_published(topic: str, timeout: Optional[float]) -> None`
 - `assert_messages_published(topic: str, number: int, ...) -> List[RosMessage]`
 
 Generally, you can always test that no exceptions are thrown, e.g., when nodes are initialized (see limitations below).
diff --git a/doc/Makefile b/doc/Makefile
index 8b6275a..255b00b 100644
--- a/doc/Makefile
+++ b/doc/Makefile
@@ -3,7 +3,7 @@
 
 # You can set these variables from the command line, and also
 # from the environment for the first two.
-SPHINXOPTS    ?= -W --keep-going
+SPHINXOPTS    ?= --keep-going
 SPHINXBUILD   ?= sphinx-build
 SOURCEDIR     = source
 BUILDDIR      = build
diff --git a/doc/source/design_and_limits.rst b/doc/source/design_and_limits.rst
index 95489f1..fa01d19 100644
--- a/doc/source/design_and_limits.rst
+++ b/doc/source/design_and_limits.rst
@@ -22,14 +22,15 @@ Current Limitations
 - Similar "silent" crashes can occur in other places, including timers and services.
   See ``tests/test_failing_nodes.py`` for unit tests on that behavior.
   A lot of them are marked with a ``TODO:``, since it is currently not straightforward to detect such issues.
-- A failing service might deadlock a test. Consider calling services asynchronously with timeouts.
+- A failing service might deadlock a test. Consider calling services asynchronously with timeouts, e.g. with
+  :meth:`~ros2_easy_test.env.ROS2TestEnvironment.call_service` or :meth:`~ros2_easy_test.env.ROS2TestEnvironment.await_future`.
 - It takes some time to set up the test environment each time, particularly when using ``@with_launch_file``.
   Also, some nodes or complex launch scenarios might need considerable time to process information.
   You may wish to append ``--durations=0 --durations-min=1.0`` to your pytest call to show the slowest tests
   (`more info <https://docs.pytest.org/en/latest/how-to/usage.html#profiling-test-execution-duration>`__).
   It might be possible to reduce the required ``warmup_time``,
   since it is unclear why setting it too low breaks *all* message exchanges and maybe there is a solvable bug causing it.
-- Currently, using services only works from ROS2 version Humble onwards, and not on Foxy.
+- Currently, interacting via services only works from ROS2 version Humble onwards, and not on Foxy.
   Similarly, a few of the tests seem to freeze sometimes.
   I do not intend to investigate further, sice `that version will reach end of life very soon <https://endoflife.date/ros2>`__.
 
@@ -55,8 +56,9 @@ Goals
 Constraints
 ~~~~~~~~~~~
 
-- Be easy to maintain by needing only a few lines of code (below a thousand). Thus, actions and services are
-  currently not supported. Also, only use public APIs wherever possible.
+- Be easy to maintain by needing only a few lines of code (below a thousand).
+  Thus, actions are currently not supported.
+  Also, we only use public APIs wherever possible.
 - Don't reinvent the wheel and benefit from future improvements: Use existing functionality of Python and
   ROS2. This includes: :mod:`unittest`, :mod:`pytest`, and the
   `ROS2 launch system <https://design.ros2.org/articles/roslaunch.html>`__.
diff --git a/pyproject.toml b/pyproject.toml
index f11d38b..54f16a5 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -33,6 +33,7 @@ dependencies = [
 dev = [
   "black~=23.1",
   "ruff",
+  "mypy~=1.3.0",
   "pytest",
   "pytest-cov",
   "pytest-sugar",
@@ -70,3 +71,16 @@ line-length = 110
 
 # The lowest supported version
 target-version = "py38"
+
+[tool.mypy]
+warn_return_any = true
+warn_unused_configs = true
+
+[[tool.mypy.overrides]]
+module = [
+    "rclpy.*",  # For the CI pipeline to work
+    "launch.*",
+    "example_interfaces.*",
+    "std_msgs.*",
+]
+ignore_missing_imports = true


Test patch:
diff --git a/ros2_easy_test/decorators.py b/ros2_easy_test/decorators.py
index 5f0737a..396601c 100644
--- a/ros2_easy_test/decorators.py
+++ b/ros2_easy_test/decorators.py
@@ -14,7 +14,7 @@
 from time import sleep
 
 # Typing
-from typing import Any, Callable, Dict, Optional, Type, Union
+from typing import Any, Callable, Coroutine, Dict, Optional, Type, TypeVar, Union
 
 # ROS
 from rclpy.context import Context
@@ -37,8 +37,10 @@
 
 # From python 3.10+ on, we should make these typing.TypeAlias'es
 # Currently not possible to type the following any better
-TestFunctionBefore = Callable[..., None]
-TestFunctionAfter = Callable[..., None]  # The same but taking one kwarg less (the env)
+_TestReturnType = TypeVar("_TestReturnType", None, Coroutine[Any, Any, None])
+TestFunctionBefore = Callable[..., _TestReturnType]
+# The same but taking one kwarg less (the env), which we can't statically type
+TestFunctionAfter = Callable[..., _TestReturnType]
 
 
 #: The time to give a node for a successful shutdown.
@@ -49,7 +51,7 @@ def with_single_node(
     node_class: Type[Node],
     *,
     parameters: Optional[Dict[str, Any]] = None,
-    time_limit: float = 60,
+    time_limit: Optional[float] = 60,
     shutdown_timeout=_DEFAULT_SHUTDOWN_TIMEOUT,
     **kwargs,
 ) -> Callable[[TestFunctionBefore], TestFunctionAfter]:
@@ -103,9 +105,11 @@ def wrapper(*args_inner, **kwargs_inner) -> None:
 
                 node: Node
                 if ros_parameters:
-                    node = node_class(parameter_overrides=ros_parameters, context=context)
+                    node = node_class(  # type: ignore[call-arg]
+                        parameter_overrides=ros_parameters, context=context
+                    )
                 else:
-                    node = node_class(context=context)
+                    node = node_class(context=context)  # type: ignore[call-arg]
 
                 # We need at least two threads: One to spin() and one to execute the test case (as the latter
                 # blocks). We better provide 4, since more nodes/tasks might get spun up by some tests and
@@ -124,7 +128,10 @@ def wrapper(*args_inner, **kwargs_inner) -> None:
                     test_function_task = executor.create_task(
                         test_function, *args_inner, env=environment, **kwargs_inner
                     )
-                    executor.spin_until_future_complete(test_function_task, timeout_sec=time_limit)
+                    # The type ignore is needed due to a bug in ROS2 Humble+
+                    executor.spin_until_future_complete(
+                        test_function_task, timeout_sec=time_limit  # type: ignore[arg-type]
+                    )
 
                     test_function_exception = test_function_task.exception()
                     if not test_function_task.done():
@@ -181,7 +188,9 @@ def wrapper(*args_inner, **kwargs_inner) -> None:
         # This is required to make pytest fixtures work
         # In principle, we could make the parameter name easily adjustable,
         # but that is probably a rare use case
-        wrapper.__signature__ = remove_signature_parameters(signature(test_function), "env")
+        wrapper.__signature__ = remove_signature_parameters(  # type: ignore[attr-defined]
+            signature(test_function), "env"
+        )
         return wrapper
 
     return decorator
@@ -191,7 +200,7 @@ def with_launch_file(  # noqa: C901
     launch_file: Union[Path, str],
     *,
     debug_launch_file: bool = False,
-    warmup_time: float = 5,
+    warmup_time: float = 2,
     time_limit: Optional[float] = 60,
     shutdown_timeout=_DEFAULT_SHUTDOWN_TIMEOUT,
     **kwargs,
@@ -259,13 +268,13 @@ def wrapper(*args_inner, **kwargs_inner) -> None:
 
                     try:
                         # We first start the environment, such that any topics that are watched can be
-                        # captured rigth from the start
+                        # captured right from the start
                         environment = ROS2TestEnvironment(context=context, **kwargs)
                         assert executor.add_node(environment), "failed to add environment"
 
-                        # We should not need any warmp time here, as the environment is fully ready once the
+                        # We should not need any warmup time here, as the environment is fully ready once the
                         # node class (the environment) is instantiated.
-                        # TODO: However, this warmup and cleaing seems to suppress some rather rare errors.
+                        # TODO: However, this warmup and cleaning seems to suppress some rather rare errors.
                         executor.spin_until_future_complete(executor.create_task(sleep, 2))
                         environment.clear_messages()
 
@@ -280,7 +289,10 @@ def wrapper(*args_inner, **kwargs_inner) -> None:
                         test_function_task = executor.create_task(
                             test_function, *args_inner, env=environment, **kwargs_inner
                         )
-                        executor.spin_until_future_complete(test_function_task, timeout_sec=time_limit)
+                        # The type ignore is needed due to a bug in ROS2 Humble+
+                        executor.spin_until_future_complete(
+                            test_function_task, timeout_sec=time_limit  # type: ignore[arg-type]
+                        )
 
                         test_function_exception = test_function_task.exception()
                         if not test_function_task.done():
@@ -359,7 +371,9 @@ def wrapper(*args_inner, **kwargs_inner) -> None:
         # This is required to make pytest fixtures work
         # In principle, we could make the parameter name easily adjustable,
         # but that is probably a rare use case
-        wrapper.__signature__ = remove_signature_parameters(signature(test_function), "env")
+        wrapper.__signature__ = remove_signature_parameters(  # type: ignore[attr-defined]
+            signature(test_function), "env"
+        )
         return wrapper
 
     return decorator
diff --git a/ros2_easy_test/env.py b/ros2_easy_test/env.py
index c42b8e1..92e40d4 100644
--- a/ros2_easy_test/env.py
+++ b/ros2_easy_test/env.py
@@ -1,17 +1,20 @@
 """This module provides the environment by which tests can interact with nodes."""
 
 # Standard library
+from importlib import import_module
 from queue import Empty, SimpleQueue
 from threading import RLock
 from time import monotonic, sleep
 
 # Typing
-from typing import Any, Dict, List, Mapping, Optional, Type
+from typing import Any, Dict, List, Mapping, Optional, Type, cast
 
 # ROS
+from rclpy.client import Client, SrvTypeRequest, SrvTypeResponse
 from rclpy.node import Node
 from rclpy.publisher import Publisher
 from rclpy.qos import QoSHistoryPolicy, QoSProfile
+from rclpy.task import Future
 
 RosMessage = Any  # We can't be more specific for now
 
@@ -32,16 +35,17 @@ class ROS2TestEnvironment(Node):
     Then, a mailbox is created for each topic and incoming messages are collected each (without a limit).
     This is performed in the background, meaning that for example :meth:`~assert_message_published` may return
     a message that was collected before the method was called.
-    To prevent this from happening, one might use:meth:`clear_messages` (although the concurrency is often the
-    desired behaviour).
+    To prevent this from happening, one might use :meth:`~clear_messages`
+    (although the concurrency is often the desired behaviour).
     Also, mind the ``timeout`` / ``time_span`` arguments as they determine how long to wait while checking the
     assertion (default: ``1`` second).
     Asserting that a message is sent or failing to assert that no message is sent will return early.
 
+    **Services** are handled similarly to publishers, being created on the fly when used for the first time.
+
     Note:
         The :class:`ROS2TestEnvironment` is a :class:`rclpy.node.Node` too, which allows to implement custom
-        functionality too. In particular, one can simply call services using code similar to:
-        ``client = env.create_client(GetLocalTangentPoint, "/service/get_local_tangent_point")``.
+        functionality too.
 
     Note:
         Timings and timeouts in this class are only approximate.
@@ -78,6 +82,10 @@ def callback(message: RosMessage, arriving_topic=topic) -> None:
         self._registered_publishers_lock = RLock()
         self._registered_publishers: Dict[str, Publisher] = {}
 
+        # Similarly for services
+        self._registered_services_lock = RLock()
+        self._registered_services: Dict[str, Client] = {}
+
     def _get_mailbox_for(self, topic: str) -> "SimpleQueue[RosMessage]":
         """Checks that a topic is watched for and returns the appropriate mailbox. Handles synchronization.
 
@@ -98,6 +106,16 @@ def _get_mailbox_for(self, topic: str) -> "SimpleQueue[RosMessage]":
 
             return self._subscriber_mailboxes[topic]
 
+    def _get_publisher(self, topic: str, msg_type: Type) -> Publisher:
+        """Get or else create the correct publisher and cache it."""
+        with self._registered_publishers_lock:
+            try:
+                return self._registered_publishers[topic]
+            except KeyError:
+                publisher = self.create_publisher(msg_type, topic, self._qos_profile)
+                self._registered_publishers[topic] = publisher
+                return publisher
+
     def publish(self, topic: str, message: RosMessage) -> None:
         """Publish the message on the topic.
 
@@ -109,17 +127,7 @@ def publish(self, topic: str, message: RosMessage) -> None:
             message: The message to be sent. It's type must match the one of all other messages sent on this
                 ``topic`` before and after.
         """
-
-        # Get or else create the correct publisher and cache it
-        publisher: Publisher
-        with self._registered_publishers_lock:
-            try:
-                publisher = self._registered_publishers[topic]
-            except KeyError:
-                publisher = self.create_publisher(type(message), topic, self._qos_profile)
-                self._registered_publishers[topic] = publisher
-
-        publisher.publish(message)
+        self._get_publisher(topic, type(message)).publish(message)
 
     def assert_no_message_published(self, topic: str, time_span: float = 0.5) -> None:
         """Asserts that no message is published on the given topic within the given time.
@@ -268,3 +276,78 @@ def clear_messages(self, topic: Optional[str] = None) -> None:
         else:
             # There is not clear() in SimpleQueue
             self.listen_for_messages(topic, time_span=None)  # ignore the result
+
+    def await_future(self, future: Future, timeout: Optional[float] = 10) -> Any:
+        """Waits for the given future to complete.
+
+        Args:
+            future: The future to wait for
+            timeout: The timeout to wait for the future
+
+        Raises:
+            TimeoutError: If the future did not complete within the timeout
+            Exception: If the future completed with an exception
+
+        Returns:
+            The result of the future
+        """
+
+        # This does not work with the default executor, so we use the one from the node
+        assert self.executor, "executor is not set"
+        # The type ignore is needed due to a bug in ROS2 Humble+
+        self.executor.spin_until_future_complete(future, timeout_sec=timeout)  # type: ignore[arg-type]
+
+        if future.done():
+            return future.result()
+        else:
+            raise TimeoutError(f"Future did not complete within {timeout} seconds")
+
+    def _get_service_client(self, name: str, request_class: Type) -> Client:
+        """Returns the service client for the given service name."""
+
+        with self._registered_services_lock:
+            try:
+                return self._registered_services[name]
+            except KeyError:
+                # Get the type of the service
+                # This is a bit tricky but relieves the user from passing it explicitly
+                module = import_module(request_class.__module__)
+                # We cut away the trailing "_Request" from the type name, which has length 8
+                base_type_name = request_class.__name__[:-8]
+                base_type_class: Type = getattr(module, base_type_name)
+
+                client = self.create_client(base_type_class, name)
+                self._registered_services[name] = client
+                return client
+
+    def call_service(
+        self,
+        name: str,
+        request: SrvTypeRequest,
+        timeout_availability: Optional[float] = 1,
+        timeout_call: Optional[float] = 10,
+    ) -> SrvTypeResponse:  # type: ignore[type-var]
+        """Calls the given service with the given request once available and returns the response.
+
+        The service type if inferred automatically from the request type.
+
+        Args:
+            name: The name of the service
+            request: The request to send to the service
+            timeout_availability: The timeout to wait for the service to be available
+            timeout_call: The timeout to wait for the service to respond
+
+        Returns:
+            The response of the service
+
+        Raises:
+            TimeoutError: If the service did not respond within the timeout
+        """
+        client = self._get_service_client(name, type(request))
+        # The type ignore is needed due to a bug in ROS2 Humble+
+        is_ready = client.wait_for_service(timeout_availability)  # type: ignore[arg-type]
+        if not is_ready:
+            raise TimeoutError(f"Service {name} did not become ready within {timeout_availability} seconds")
+
+        future = client.call_async(request)
+        return cast(SrvTypeResponse, self.await_future(future, timeout=timeout_call))
diff --git a/ros2_easy_test/launch_file.py b/ros2_easy_test/launch_file.py
index 7c83eb7..0307855 100644
--- a/ros2_easy_test/launch_file.py
+++ b/ros2_easy_test/launch_file.py
@@ -42,6 +42,7 @@ def __enter__(self) -> Path:
             if not isinstance(launch_file_path, Path):
                 launch_file_path = Path(self._launch_file)
         else:
+            assert isinstance(self._launch_file, str)
             directory = TemporaryDirectory()
             self._exit_list.append(cast(ContextManager[str], directory))
             directory_name = Path(directory.__enter__())
diff --git a/tests/test_interactions.py b/tests/test_interactions.py
index 6ab7c6d..2d6e6f3 100644
--- a/tests/test_interactions.py
+++ b/tests/test_interactions.py
@@ -48,6 +48,15 @@ def test_subscriber_and_publisher(self, env: ROS2TestEnvironment) -> None:
         self.assertSequenceEqual(all_messages, [String(data="Guude")])
 
     def test_service(self, env: ROS2TestEnvironment) -> None:
+        # Set up the request
+        request = AddTwoInts.Request(a=-500, b=73)
+        desired_sum = -500 + 73
+
+        # Call the service asynchronously
+        result: int = env.call_service("add_two_ints", request).sum
+        self.assertEqual(result, desired_sum)
+
+    def test_service_manual(self, env: ROS2TestEnvironment) -> None:
         # Set up the service
         adder = env.create_client(AddTwoInts, "add_two_ints")
         self.assertTrue(adder.wait_for_service(timeout_sec=5))
@@ -60,9 +69,14 @@ def test_service(self, env: ROS2TestEnvironment) -> None:
         result: int = adder.call(request).sum
         self.assertEqual(result, desired_sum)
 
-        # Call the service asynchronously
+        # Call the service asynchronously with a helper
         future = adder.call_async(request)
-        # TODO: One should find a way to make this work with the default executor
+        result: int = env.await_future(future, timeout=2).sum
+        self.assertEqual(result, desired_sum)
+
+        # Call the service asynchronously completely manually
+        future = adder.call_async(request)
+        # This needs to be done via the executor of the node, see ROS2TestEnvironment::await_future
         env.executor.spin_until_future_complete(future, timeout_sec=2)
         self.assertIsNone(future.exception())
         result: int = future.result().sum
@@ -102,6 +116,11 @@ def test_subscriber_and_publisher(self, env: ROS2TestEnvironment) -> None:
     def test_service(self, env: ROS2TestEnvironment) -> None:
         super().test_service(env)
 
+    @mark.xfail(raises=ValueError, reason="See https://github.com/ros2/rclpy/pull/1129")
+    @with_single_node(AddTwoIntsServer)
+    def test_service_manual(self, env: ROS2TestEnvironment) -> None:
+        super().test_service_manual(env)
+
     @with_single_node(EchoNode, watch_topics={"/mouth": String})
     def test_multiple_messages(self, env: ROS2TestEnvironment) -> None:
         super().test_multiple_messages(env)
@@ -147,6 +166,11 @@ def test_subscriber_and_publisher(self, env: ROS2TestEnvironment) -> None:
     def test_service(self, env: ROS2TestEnvironment) -> None:
         super().test_service(env)
 
+    @mark.xfail(raises=ValueError, reason="See https://github.com/ros2/rclpy/pull/1129")
+    @with_launch_file(LAUNCH_FILES / "adder.yaml", warmup_time=2)
+    def test_service_manual(self, env: ROS2TestEnvironment) -> None:
+        super().test_service_manual(env)
+
     @with_launch_file(LAUNCH_FILES / "echo.yaml", watch_topics={"/mouth": String})
     def test_multiple_messages(self, env: ROS2TestEnvironment) -> None:
         super().test_multiple_messages(env)