Please linters + improve test setup and teardown

Signed-off-by: Ivan Santiago Paunovic <[email protected]>

Author: ivanpauno

rclpy/rclpy/qos_overriding_options.py

@@ -26,8 +26,8 @@
 from rclpy.node import Node
 from rclpy.parameter import Parameter
 from rclpy.publisher import Publisher
-from rclpy.qos import QoSProfile
 from rclpy.qos import QoSPolicyKind
+from rclpy.qos import QoSProfile
 from rclpy.subscription import Subscription
 
 
@@ -36,29 +36,27 @@ class InvalidQosOverridesError(Exception):
 
 
 class QoSOverridingOptions:
-    """
-    Options to customize qos parameter overrides.
-    """
+    """Options to customize qos parameter overrides."""
 
     def __init__(
         self,
         *,
         policy_kinds: Iterable[QoSPolicyKind],
         callback: Optional[Callable[[QoSProfile], bool]] = None,
-        id: Optional[Text] = None
+        entity_id: Optional[Text] = None
     ):
         """
         Construct a QoSOverridingOptions object.
 
         :param policy_kinds: Qos kinds that will have a declared parameter.
         :param callback: Callback that will be used to validate the qos profile
             after the paramter overrides get applied.
-        :param id: Optional identifier, to disambiguate in the case that different qos
+        :param entity_id: Optional identifier, to disambiguate in the case that different qos
             policies for the same topic are desired.
         """
         self._policy_kinds = policy_kinds
         self._callback = callback
-        self._id = id
+        self._entity_id = entity_id
 
     @property
     def policy_kinds(self) -> Iterable[QoSPolicyKind]:
@@ -71,20 +69,20 @@ def callback(self) -> Optional[Callable[[QoSProfile], bool]]:
         return self._callback
 
     @property
-    def id(self) -> Optional[Text]:
+    def entity_id(self) -> Optional[Text]:
         """Get the optional entity id."""
-        return self._id
+        return self._entity_id
 
     @classmethod
     def with_default_policies(
         cls, *,
         callback: Optional[Callable[[QoSProfile], bool]] = None,
-        id: Optional[Text] = None
+        entity_id: Optional[Text] = None
     ) -> 'QoSOverridingOptions':
         return cls(
             policy_kinds=(QoSPolicyKind.HISTORY, QoSPolicyKind.DEPTH, QoSPolicyKind.RELIABILITY),
             callback=callback,
-            id=id,
+            entity_id=entity_id,
         )
 
 
@@ -95,8 +93,7 @@ def _declare_qos_parameteres(
     qos: QoSProfile,
     options: QoSOverridingOptions) -> QoSProfile:
     """
-    Internal.
-    Declares qos parameters for a Publisher or a Subscription.
+    Internal, declare qos parameters for a Publisher or a Subscription.
 
     :param entity_type: Either `rclpy.node.Publisher` or `rclpy.node.Subscription`.
     :param node: Node used to declare the parameters.
@@ -108,9 +105,9 @@ def _declare_qos_parameteres(
     if not issubclass(entity_type, (Publisher, Subscription)):
         raise TypeError("Argument `entity_type` should be a subclass of Publisher or Subscription")
     entity_type_str = 'publisher' if issubclass(entity_type, Publisher) else Subscription
-    id_suffix = '' if options.id is None else f'_{options.id}'
+    id_suffix = '' if options.entity_id is None else f'_{options.entity_id}'
     name = f'qos_overrides.{topic_name}.{entity_type_str}{id_suffix}.' '{}'
-    description = '{}' f' for {entity_type_str} `{topic_name}` with id `{id}`'
+    description = '{}' f' for {entity_type_str} `{topic_name}` with id `{options.entity_id}`'
     allowed_policies = _get_allowed_policies(entity_type)
     for policy in options.policy_kinds:
         if policy not in allowed_policies:
@@ -153,12 +150,6 @@ def _get_qos_policy_parameter(qos: QoSProfile, policy: QoSPolicyKind) -> Paramet
 
 
 def _override_qos_policy_with_param(qos: QoSProfile, policy: QoSPolicyKind, param: Parameter):
-    # policy_enum_map = {
-    #     QoSPolicyKind.LIVELINESS: rclpy.qos.LivelinessPolicy,
-    #     QoSPolicyKind.RELIABILITY: rclpy.qos.LivelinessPolicy,
-    #     QoSPolicyKind.LIVELINESS: rclpy.qos.LivelinessPolicy,
-    #     QoSPolicyKind.LIVELINESS: rclpy.qos.LivelinessPolicy,
-    # }
     value = param.value
     policy_name = policy.name.lower()
     if policy in (

rclpy/test/test_qos_overriding_options.py

@@ -18,15 +18,22 @@
 from rclpy.node import Node
 from rclpy.parameter import Parameter
 from rclpy.publisher import Publisher
-from rclpy.qos import QoSPolicyKind, QoSProfile
+from rclpy.qos import QoSPolicyKind
+from rclpy.qos import QoSProfile
 from rclpy.qos_overriding_options import _declare_qos_parameteres
 from rclpy.qos_overriding_options import InvalidQosOverridesError
 from rclpy.qos_overriding_options import QoSOverridingOptions
 
 
-def test_declare_qos_parameters():
+@pytest.fixture(autouse=True)
+def init_shutdown():
     rclpy.init()
-    node = Node("my_node")
+    yield
+    rclpy.shutdown()
+
+
+def test_declare_qos_parameters():
+    node = Node('my_node')
     _declare_qos_parameteres(
         Publisher, node, '/my_topic', QoSProfile(depth=10),
         QoSOverridingOptions.with_default_policies()
@@ -38,18 +45,15 @@ def test_declare_qos_parameters():
         ('/my_topic.publisher.history', 'keep_last'),
         ('/my_topic.publisher.reliability', 'reliable'),
     )
-    for actual, expected in zip (
+    for actual, expected in zip(
         sorted(qos_overrides.items(), key=lambda x: x[0]), expected_params
     ):
         assert actual[0] == expected[0]  # same param name
         assert actual[1].value == expected[1]  # same param value
 
-    rclpy.shutdown()
-
 
 def test_declare_qos_parameters_with_overrides():
-    rclpy.init()
-    node = Node("my_node", parameter_overrides=[
+    node = Node('my_node', parameter_overrides=[
         Parameter('qos_overrides./my_topic.publisher.depth', value=100),
         Parameter('qos_overrides./my_topic.publisher.reliability', value='best_effort'),
     ])
@@ -64,17 +68,14 @@ def test_declare_qos_parameters_with_overrides():
         ('/my_topic.publisher.history', 'keep_last'),
         ('/my_topic.publisher.reliability', 'best_effort'),
     )
-    for actual, expected in zip (
+    for actual, expected in zip(
         sorted(qos_overrides.items(), key=lambda x: x[0]), expected_params
     ):
         assert actual[0] == expected[0]  # same param name
         assert actual[1].value == expected[1]  # same param value
 
-    rclpy.shutdown()
-
 
 def test_declare_qos_parameters_with_happy_callback():
-    rclpy.init()
     node = Node("my_node")
     _declare_qos_parameteres(
         Publisher, node, '/my_topic', QoSProfile(depth=10),
@@ -87,27 +88,28 @@ def test_declare_qos_parameters_with_happy_callback():
         ('/my_topic.publisher.history', 'keep_last'),
         ('/my_topic.publisher.reliability', 'reliable'),
     )
-    rclpy.shutdown()
+    for actual, expected in zip(
+        sorted(qos_overrides.items(), key=lambda x: x[0]), expected_params
+    ):
+        assert actual[0] == expected[0]  # same param name
+        assert actual[1].value == expected[1]  # same param value
 
 
 def test_declare_qos_parameters_with_unhappy_callback():
-    rclpy.init()
-    node = Node("my_node")
+    node = Node('my_node')
 
     with pytest.raises(InvalidQosOverridesError):
         _declare_qos_parameteres(
             Publisher, node, '/my_topic', QoSProfile(depth=10),
             QoSOverridingOptions.with_default_policies(callback=lambda x: False)
         )
-    rclpy.shutdown()
 
 
 def test_declare_qos_parameters_with_id():
-    rclpy.init()
-    node = Node("my_node")
+    node = Node('my_node')
     _declare_qos_parameteres(
         Publisher, node, '/my_topic', QoSProfile(depth=10),
-        QoSOverridingOptions.with_default_policies(id='i_have_an_id')
+        QoSOverridingOptions.with_default_policies(entity_id='i_have_an_id')
     )
     qos_overrides = node.get_parameters_by_prefix('qos_overrides')
     assert len(qos_overrides) == 3
@@ -116,10 +118,8 @@ def test_declare_qos_parameters_with_id():
         ('/my_topic.publisher_i_have_an_id.history', 'keep_last'),
         ('/my_topic.publisher_i_have_an_id.reliability', 'reliable'),
     )
-    for actual, expected in zip (
+    for actual, expected in zip(
         sorted(qos_overrides.items(), key=lambda x: x[0]), expected_params
     ):
         assert actual[0] == expected[0]  # same param name
         assert actual[1].value == expected[1]  # same param value
-
-    rclpy.shutdown()