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

Issue:
QoSProfile is hardcoded
Transient local topics cannot be send nor received because the `DurabilityPolicy` cannot be set to `TRANSIENT_LOCAL`


Discussion / hints:

Fix patch:
diff --git a/README.md b/README.md
index 69a126d..b726026 100644
--- a/README.md
+++ b/README.md
@@ -48,6 +48,7 @@ def test_simple_publisher(env: ROS2TestEnvironment) -> None:
     response: String = env.assert_message_published("/chatter", timeout=5)
     assert response.data == "Hello World: 0"
 ```
+It is also possible to specify custom QoSProfiles for the `ROS2TestEnvironment` to use when subscribing or publishing on the specified topics. See [`tests/demo/latching_topic_test.py`](tests/demo/latching_topic_test.py) for an example. If no profile is specified, the default `QoSProfile(history=QoSHistoryPolicy.KEEP_ALL)` is used.
 
 You can optionally provide more parameters to the test setting, i.e., additionally pass `parameters={"some.thing": 30.2}` to the decorator.
 The argument of the test function receiving the `ROS2TestEnvironment` *must* be named `env`.


Test patch:
diff --git a/ros2_easy_test/env.py b/ros2_easy_test/env.py
index 92e40d4..f85c12b 100644
--- a/ros2_easy_test/env.py
+++ b/ros2_easy_test/env.py
@@ -1,6 +1,7 @@
 """This module provides the environment by which tests can interact with nodes."""
 
 # Standard library
+from collections import defaultdict
 from importlib import import_module
 from queue import Empty, SimpleQueue
 from threading import RLock
@@ -53,9 +54,18 @@ class ROS2TestEnvironment(Node):
     Args:
         watch_topics: The topics (and their type) to watch for incoming messages.
             You may pass ``None`` to not watch any topic.
+        qos_profile: The topics and their QoS profile to subscribe to or publish on.
+            You may pass ``None`` to use the default profile.
+            All topics without an explicit profile use ``QoSProfile(history=QoSHistoryPolicy.KEEP_ALL)``.
     """
 
-    def __init__(self, *, watch_topics: Optional[Mapping[str, Type]] = None, **kwargs) -> None:
+    def __init__(
+        self,
+        *,
+        watch_topics: Optional[Mapping[str, Type]] = None,
+        qos_profiles: Optional[Mapping[str, QoSProfile]] = None,
+        **kwargs,
+    ) -> None:
         super().__init__("_testing_ROS2TestEnvironment", **kwargs)
 
         # Collects the messages that were received
@@ -63,7 +73,9 @@ def __init__(self, *, watch_topics: Optional[Mapping[str, Type]] = None, **kwarg
         self._subscriber_mailboxes_lock = RLock()
         self._subscriber_mailboxes: Dict[str, SimpleQueue[RosMessage]] = {}
 
-        self._qos_profile = QoSProfile(history=QoSHistoryPolicy.KEEP_ALL)
+        self._qos_profiles = defaultdict(
+            lambda: QoSProfile(history=QoSHistoryPolicy.KEEP_ALL), qos_profiles or {}
+        )
 
         # Set up the subscribers, which place the messages in the mailboxes
         for topic, message_type in (watch_topics or {}).items():
@@ -76,7 +88,7 @@ def callback(message: RosMessage, arriving_topic=topic) -> None:
                 self._subscriber_mailboxes[topic] = SimpleQueue()
 
             # After the mailboxes are set up, we can start receiving messages
-            self.create_subscription(message_type, topic, callback, self._qos_profile)
+            self.create_subscription(message_type, topic, callback, self._qos_profiles[topic])
 
         # Prepare to collect publishers; these are set up on the fly when used the first time in `publish()`
         self._registered_publishers_lock = RLock()
@@ -112,7 +124,7 @@ def _get_publisher(self, topic: str, msg_type: Type) -> Publisher:
             try:
                 return self._registered_publishers[topic]
             except KeyError:
-                publisher = self.create_publisher(msg_type, topic, self._qos_profile)
+                publisher = self.create_publisher(msg_type, topic, self._qos_profiles[topic])
                 self._registered_publishers[topic] = publisher
                 return publisher
 
diff --git a/tests/demo/latching_topic_test.py b/tests/demo/latching_topic_test.py
new file mode 100644
index 0000000..d89a703
--- /dev/null
+++ b/tests/demo/latching_topic_test.py
@@ -0,0 +1,22 @@
+# QoS Profile
+from rclpy.qos import DurabilityPolicy, HistoryPolicy, QoSProfile
+
+# ROS2 interfaces
+from std_msgs.msg import String
+
+# Testing
+from ros2_easy_test import ROS2TestEnvironment, with_single_node
+
+# Module under test and interfaces
+from ..example_nodes.well_behaved import LatchingNode
+
+
+@with_single_node(
+    LatchingNode,
+    watch_topics={"/hello": String},
+    qos_profiles={
+        "/hello": QoSProfile(durability=DurabilityPolicy.TRANSIENT_LOCAL, history=HistoryPolicy.KEEP_ALL)
+    },
+)
+def test_plain_method(env: ROS2TestEnvironment) -> None:
+    env.assert_message_published("/hello")
diff --git a/tests/example_nodes/well_behaved.py b/tests/example_nodes/well_behaved.py
index 6212aaf..e36a346 100644
--- a/tests/example_nodes/well_behaved.py
+++ b/tests/example_nodes/well_behaved.py
@@ -1,6 +1,6 @@
 from example_interfaces.srv import AddTwoInts
 from rclpy.node import Node
-from rclpy.qos import QoSHistoryPolicy, QoSProfile
+from rclpy.qos import DurabilityPolicy, HistoryPolicy, QoSHistoryPolicy, QoSProfile
 from std_msgs.msg import String
 
 
@@ -32,6 +32,14 @@ def __init__(self, *args, **kwargs):
         self.create_subscription(String, "/ear", mouth.publish, qos_profile)
 
 
+class LatchingNode(Node):
+    def __init__(self, *args, **kwargs):
+        super().__init__("latching_publisher", *args, **kwargs)
+
+        qos_profile = QoSProfile(durability=DurabilityPolicy.TRANSIENT_LOCAL, history=HistoryPolicy.KEEP_ALL)
+        self.create_publisher(String, "/hello", qos_profile).publish(String(data="Hello World"))
+
+
 class AddTwoIntsServer(Node):
     def __init__(self, *args, **kwargs):
         super().__init__("add_two_ints_server", *args, **kwargs)